mirror of
https://github.com/onyx-and-iris/obsws-python.git
synced 2024-11-22 12:50:53 +00:00
54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
from typing import Callable, Iterable, Union
|
|
|
|
from .util import as_dataclass, to_camel_case, to_snake_case
|
|
|
|
|
|
class Callback:
|
|
"""Adds support for callbacks"""
|
|
|
|
def __init__(self):
|
|
"""list of current callbacks"""
|
|
|
|
self._callbacks = list()
|
|
|
|
def get(self) -> list:
|
|
"""returns a list of registered events"""
|
|
|
|
return [to_camel_case(fn.__name__) for fn in self._callbacks]
|
|
|
|
def trigger(self, event, data):
|
|
"""trigger callback on update"""
|
|
|
|
for fn in self._callbacks:
|
|
if fn.__name__ == f"on_{to_snake_case(event)}":
|
|
fn(as_dataclass(event, data.get("eventData")))
|
|
|
|
def register(self, fns: Union[Iterable, Callable]):
|
|
"""registers callback functions"""
|
|
|
|
try:
|
|
iterator = iter(fns)
|
|
for fn in iterator:
|
|
if fn not in self._callbacks:
|
|
self._callbacks.append(fn)
|
|
except TypeError as e:
|
|
if fns not in self._callbacks:
|
|
self._callbacks.append(fns)
|
|
|
|
def deregister(self, fns: Union[Iterable, Callable]):
|
|
"""deregisters a callback from _callbacks"""
|
|
|
|
try:
|
|
iterator = iter(fns)
|
|
for fn in iterator:
|
|
if fn in self._callbacks:
|
|
self._callbacks.remove(fn)
|
|
except TypeError as e:
|
|
if fns in self._callbacks:
|
|
self._callbacks.remove(fns)
|
|
|
|
def clear(self):
|
|
"""clears the _callbacks list"""
|
|
|
|
self._callbacks.clear()
|