heros.event
===========

.. py:module:: heros.event


Classes
-------

.. autoapisummary::

   heros.event.Callback
   heros.event.CallbackStorage
   heros.event.EventHandler
   heros.event.LocalEventHandler
   heros.event.RemoteEventHandler
   heros.event.EventDescriptor
   heros.event.LocalEventDescriptor
   heros.event.RemoteEventDescriptor


Functions
---------

.. autoapisummary::

   heros.event.remote_hero_method_to_str
   heros.event.reliable
   heros.event.event


Module Contents
---------------

.. py:function:: remote_hero_method_to_str(func: collections.abc.Callable) -> str

.. py:class:: Callback(func: collections.abc.Callable, origin: str = 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) -> 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: 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: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, 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:attribute:: publisher


   .. 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) -> None

      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 <https://zenoh-python.readthedocs.io/en/latest/api_reference.html#zenoh.Reliability>`_ and
   `BLOCK congestion <https://zenoh-python.readthedocs.io/en/1.8.0/api_reference.html#zenoh.CongestionControl>`_.
   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 <json_extra_decorators>`. 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.event.reliable"]
               ]
           }



.. py:function:: event(func: collections.abc.Callable)

   Decorator for events.

   .. note:: Only use on methods bound to objects.


