hololinked.client.abstractions.ConsumedThingEvent

Source code in hololinked\client\abstractions.py

class ConsumedThingEvent:

    __slots__ = ['__name__', '__qualname__', '__doc__', 
                '_callbacks', '_subscribed', '_thread', '_thread_callbacks', '_logger']

    # event subscription
    # Dont add class doc otherwise __doc__ in slots will conflict with class variable

    def __init__(self, 
                resource: EventAffordance,
                **kwargs
                ) -> None:
        """
        Parameters
        ----------
        resource: EventAffordance
            dataclass object representing the event
        """
        self._resource = resource
        self._callbacks = None        
        self._thread_callbacks = False
        self._logger = kwargs.get('logger', None)


    def add_callbacks(self, *callbacks : typing.Union[typing.List[typing.Callable], typing.Callable]) -> None:
        """
        add callbacks to the event

        Parameters
        ----------
        *callbacks: typing.List[typing.Callable] | typing.Callable
            callback or list of callbacks to add
        """
        if not self._callbacks:
            self._callbacks = [] 
        self._callbacks.extend(callbacks)

    def subscribe(self, 
                callbacks: typing.Union[typing.List[typing.Callable], typing.Callable], 
                thread_callbacks: bool = False,
                deserialize: bool = True
            ) -> None:
        """
        subscribe to the event

        Parameters
        ----------
        callbacks: typing.List[typing.Callable] | typing.Callable
            callback or list of callbacks to add
        thread_callbacks: bool
            whether to run each callback in a separate thread
        deserialize: bool
            whether to deserialize the event payload
        """
        raise NotImplementedError("implement subscribe per protocol")

    def unsubscribe(self, join_thread: bool = True):
        """
        unsubscribe from the event

        Parameters
        ----------
        join_thread: bool
            whether to join the event thread after unsubscribing
        """
        raise NotImplementedError("implement unsubscribe per protocol")

    def listen(self):
        """
        listen to events and call the callbacks
        """
        raise NotImplementedError("implement listen per protocol")

Functions

__init__

__init__(resource: EventAffordance, **kwargs) -> None

Parameters:

Name	Type	Description	Default
resource	EventAffordance	dataclass object representing the event	required

Source code in hololinked\client\abstractions.py

def __init__(self, 
            resource: EventAffordance,
            **kwargs
            ) -> None:
    """
    Parameters
    ----------
    resource: EventAffordance
        dataclass object representing the event
    """
    self._resource = resource
    self._callbacks = None        
    self._thread_callbacks = False
    self._logger = kwargs.get('logger', None)

subscribe

subscribe(callbacks: typing.Union[typing.List[typing.Callable], typing.Callable], thread_callbacks: bool = False, deserialize: bool = True) -> None

subscribe to the event

Parameters:

Name	Type	Description	Default
callbacks	Union[List[Callable], Callable]	callback or list of callbacks to add	required
thread_callbacks	bool	whether to run each callback in a separate thread	False
deserialize	bool	whether to deserialize the event payload	True

Source code in hololinked\client\abstractions.py

def subscribe(self, 
            callbacks: typing.Union[typing.List[typing.Callable], typing.Callable], 
            thread_callbacks: bool = False,
            deserialize: bool = True
        ) -> None:
    """
    subscribe to the event

    Parameters
    ----------
    callbacks: typing.List[typing.Callable] | typing.Callable
        callback or list of callbacks to add
    thread_callbacks: bool
        whether to run each callback in a separate thread
    deserialize: bool
        whether to deserialize the event payload
    """
    raise NotImplementedError("implement subscribe per protocol")

unsubscribe

unsubscribe(join_thread: bool = True)

unsubscribe from the event

Parameters:

Name	Type	Description	Default
join_thread	bool	whether to join the event thread after unsubscribing	True

Source code in hololinked\client\abstractions.py

def unsubscribe(self, join_thread: bool = True):
    """
    unsubscribe from the event

    Parameters
    ----------
    join_thread: bool
        whether to join the event thread after unsubscribing
    """
    raise NotImplementedError("implement unsubscribe per protocol")

listen

listen()

listen to events and call the callbacks

Source code in hololinked\client\abstractions.py

def listen(self):
    """
    listen to events and call the callbacks
    """
    raise NotImplementedError("implement listen per protocol")

add_callbacks

add_callbacks(*callbacks: typing.Union[typing.List[typing.Callable], typing.Callable]) -> None

add callbacks to the event

Parameters:

Name	Type	Description	Default
*callbacks	Union[List[Callable], Callable]	callback or list of callbacks to add	()

Source code in hololinked\client\abstractions.py

def add_callbacks(self, *callbacks : typing.Union[typing.List[typing.Callable], typing.Callable]) -> None:
    """
    add callbacks to the event

    Parameters
    ----------
    *callbacks: typing.List[typing.Callable] | typing.Callable
        callback or list of callbacks to add
    """
    if not self._callbacks:
        self._callbacks = [] 
    self._callbacks.extend(callbacks)