heros.events ============ .. py:module:: heros.events Classes ------- .. autoapisummary:: heros.events.Callback heros.events.CallbackStorage heros.events.EventHandler heros.events.LocalEventHandler heros.events.RemoteEventHandler heros.events.EventDescriptor heros.events.LocalEventDescriptor heros.events.RemoteEventDescriptor Functions --------- .. autoapisummary:: heros.events.remote_hero_method_to_str heros.events.reliable heros.events.event Module Contents --------------- .. py:function:: remote_hero_method_to_str(func: types.MethodType) -> str .. py:class:: Callback(func: collections.abc.Callable, origin: str | None = None) Represent a callback function. .. py:attribute:: func .. py:attribute:: origin :value: None .. py:method:: __eq__(other) -> bool Check for equality with other `Callback`. :param other: other callback instance. :returns: equality result. :rtype: bool .. py:method:: __call__(*args, **kwargs) Call the callback function. .. py:method:: __hash__() Generate a hash value for this callback using the name in case of remote hero methods or the callable itself for builtins or local callables. :returns: calculated hash. :rtype: int .. py:method:: to_dict() -> dict Generate a dictionary representation of this callback. :returns: dictionary with keys: name, origin, is_remote_hero and func :rtype: dict .. py:class:: CallbackStorage Store all callbacks. .. py:attribute:: _callbacks .. py:method:: __iter__() Generate an iteration for this iterable. .. py:method:: __contains__(func: collections.abc.Callable) -> bool Implements the `in` operation for this class. .. py:method:: append(func: collections.abc.Callable, origin: str | None = None) -> str Append a given callable to the storage. :param func: `callable` to append. :param origin: `str` (default: `None`) indicating the origin of the callback. :returns: name of the callback. :rtype: str .. py:method:: remove(func: collections.abc.Callable) -> bool Remove a callable from storage. :param func: `callable` to remove. :returns: truth value indicating if the callable was a callback. :rtype: bool .. py:method:: is_callback(func: collections.abc.Callable) -> bool Check if given `callable` is a callback. :param func: `callable` to check. :returns: `callable` is a callback :rtype: bool .. py:method:: get_callbacks() -> list Get a list of all callbacks dictionaries. :returns: dictionary representation of all callbacks :rtype: list .. py:class:: EventHandler Base class for event handlers. .. py:attribute:: __name__ :type: str :value: '' .. py:method:: connect(callback: collections.abc.Callable) :abstractmethod: Connect a callback function. .. py:method:: disconnect(callback: collections.abc.Callable) :abstractmethod: Disconnect a callback function. .. py:method:: is_callback(func: collections.abc.Callable) -> bool :abstractmethod: Check if `func` is a callback .. py:method:: get_callbacks() :abstractmethod: Return all callbacks .. py:class:: LocalEventHandler(instance: heros.heros.HERO, func, reliability: zenoh.Reliability, congestion_control: zenoh.CongestionControl) Bases: :py:obj:`EventHandler` Handles event connections for a specific instance. .. py:attribute:: instance .. py:attribute:: func .. py:attribute:: callbacks .. py:method:: __call__(*args, **kwargs) Call the original function and trigger callbacks. .. py:method:: connect(callback: collections.abc.Callable, origin: str | None = None) -> str Connect a callback function to be triggered when the method is called. :param callback: `callable` to connect. :param origin: (optional) `str` indicting origin of this callback. :returns: name of the callback. :rtype: str .. py:method:: disconnect(callback: collections.abc.Callable) -> bool Disconnect a callback function. :param callback: `callable` to disconnect. :returns: truth value indicating if the callable was a callback. :rtype: bool .. py:method:: is_callback(func: collections.abc.Callable) -> bool Check if given callable is already a registered callback. :param callback: `callable` to check. :returns: truth value indicating if the callable is a callback. :rtype: bool .. py:method:: get_callbacks() -> list Return a list of registered callback functions. :returns: dictionary representation of all callbacks :rtype: list .. py:class:: RemoteEventHandler(instance, func: collections.abc.Callable | None = None) Bases: :py:obj:`EventHandler` Handles remote events for a specific instance. .. py:attribute:: instance .. py:attribute:: callbacks .. py:method:: __call__(payload) Call the original function and trigger callbacks. .. py:method:: connect(callback: collections.abc.Callable) -> str Connect a callback function to be triggered when the method is called. :param callback: `callable` to connect. :param origin: (optional) `str` indicting origin of this callback. :returns: name of the callback. :rtype: str .. py:method:: disconnect(callback: collections.abc.Callable) -> bool Disconnect a callback function. :param callback: `callable` to disconnect. :returns: truth value indicating if the callable was a callback. :rtype: bool .. py:method:: is_callback(func: collections.abc.Callable) -> bool Check if given callable is already a registered callback. :param callback: `callable` to check. :returns: truth value indicating if the callable is a callback. :rtype: bool .. py:method:: get_callbacks() -> list Return a list of registered callback functions. :returns: dictionary representation of all callbacks :rtype: list .. py:class:: EventDescriptor(func: collections.abc.Callable | None = None, handler_config: dict | None = None) A descriptor to handle instance-specific event connections. .. py:attribute:: func :value: None .. py:attribute:: handler_config .. py:attribute:: _instances .. py:method:: _get_event_handler_cls() :staticmethod: :abstractmethod: .. py:method:: __get__(instance, owner) Ensure the method and event-handling functions are bound to the instance. :param self: the EventDescriptor instance :param instance: the owning `LocalHERO`/`RemoteHERO`. :param owner: .. py:class:: LocalEventDescriptor(func: collections.abc.Callable | None = None, handler_config: dict | None = None) Bases: :py:obj:`EventDescriptor` Descriptor of `@event` decorated methods of a `LocalHERO`. .. py:method:: _get_event_handler_cls() :staticmethod: .. py:class:: RemoteEventDescriptor(func: collections.abc.Callable | None = None, handler_config: dict | None = None) Bases: :py:obj:`EventDescriptor` Descriptor of remote representations of events in a `RemoteHERO`. .. py:method:: _get_event_handler_cls() :staticmethod: .. py:function:: reliable(func: collections.abc.Callable | LocalEventDescriptor) Decorator to make an event reliable. This decorator can be used in addition to the @event decorator to make an event use the zenoh `RELIABLE reliability `_ and `BLOCK congestion `_. This guarantees that the message is received at all subscribers. Using the default can lead to lost messages under high load or changing network topologies. Use this if you observe missing data from e.g. cameras. .. note:: Can be used on arbitrary events with the :external:ref:`BOSS decorator injector mechanism `. The following example makes the emitted images from a camera reliable. .. code-block:: json { "_id": "my_camera", "classname": "herosdevices.hardware.ids.PeakCompatibleCamera", "arguments": { "cam_id": "1409f4e7c3b1" } "extra_decorators": [ ["acquisition_data", "heros.events.reliable"] ] } .. py:function:: event(func: collections.abc.Callable) Decorator for events. .. note:: Only use on methods bound to objects.