onyx_online/voicemeeter-api-python
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/voicemeeter-api-python.git synced 2024-09-27 13:17:00 +01:00
Code Issues Packages Projects Releases Wiki Activity

extend subject class to support callbacks

Browse Source
This commit is contained in:
onyx-and-iris 2023-06-23 17:31:49 +01:00
parent c23a6aff6d
commit b2f57a9e60
1 changed files with 45 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
voicemeeterlib/subject.py
View File

@ -4,10 +4,8 @@ logger = logging.getLogger(__name__)
class Subject: class Subject:
"""Adds support for observers"""
def __init__(self): def __init__(self):
"""list of current observers""" """Adds support for observers and callbacks"""
self._observers = list() self._observers = list()
self.logger = logger.getChild(self.__class__.__name__) self.logger = logger.getChild(self.__class__.__name__)
@ -18,34 +16,57 @@ class Subject:
return self._observers return self._observers
def notify(self, modifier): def notify(self, event):
"""run callbacks on update""" """run callbacks on update"""
[o.on_update(modifier) for o in self._observers] for o in self._observers:
if hasattr(o, "on_update"):
o.on_update(event)
else:
if o.__name__ == f"on_{event}":
o()
def add(self, observer): def add(self, observer):
"""adds an observer to _observers""" """adds an observer to observers"""
if observer not in self._observers:
self._observers.append(observer)
self.logger.info(f"{type(observer).__name__} added to event observers")
else:
self.logger.error(
f"Failed to add {type(observer).__name__} to event observers"
)
def remove(self, observer):
"""removes an observer from _observers"""
try: try:
self._observers.remove(observer) iterator = iter(observer)
self.logger.info(f"{type(observer).__name__} removed from event observers") for o in iterator:
if o not in self._observers:
self._observers.append(o)
self.logger.info(f"{o} added to event observers")
else:
self.logger.error(f"Failed to add {o} to event observers")
except TypeError:
if observer not in self._observers:
self._observers.append(observer)
self.logger.info(f"{observer} added to event observers")
else:
self.logger.error(f"Failed to add {observer} to event observers")
register = add
def remove(self, observer):
"""removes an observer from observers"""
try:
iterator = iter(observer)
for o in iterator:
try:
self._observers.remove(o)
self.logger.info(f"{o} removed from event observers")
except ValueError: except ValueError:
self.logger.error( self.logger.error(f"Failed to remove {o} from event observers")
f"Failed to remove {type(observer).__name__} from event observers" except TypeError:
) try:
self._observers.remove(observer)
self.logger.info(f"{observer} removed from event observers")
except ValueError:
self.logger.error(f"Failed to remove {observer} from event observers")
deregister = remove
def clear(self): def clear(self):
"""clears the _observers list""" """clears the observers list"""
self._observers.clear() self._observers.clear()