heros.event

Classes

Callback	Represent a callback function.
CallbackStorage	Store all callbacks.
EventHandler	Base class for event handlers.
LocalEventHandler	Handles event connections for a specific instance.
RemoteEventHandler	Handles remote events for a specific instance.
EventDescriptor	A descriptor to handle instance-specific event connections.
LocalEventDescriptor	Descriptor of @event decorated methods of a LocalHERO.
RemoteEventDescriptor	Descriptor of remote representations of events in a RemoteHERO.

Functions

remote_hero_method_to_str(→ str)
reliable(func)	Decorator to make an event reliable.
event(func)	Decorator for events.

Module Contents

heros.event.remote_hero_method_to_str(func: collections.abc.Callable) → str

class heros.event.Callback(func: collections.abc.Callable, origin: str = None)

Represent a callback function.

func

origin = None

__eq__(other) → bool

Check for equality with other Callback.

Parameters:

other – other callback instance.

Returns:

equality result.

Return type:

bool

__call__(*args, **kwargs)

Call the callback function.

__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.

Return type:

int

to_dict() → dict

Generate a dictionary representation of this callback.

Returns:

dictionary with keys: name, origin, is_remote_hero and func

Return type:

dict

class heros.event.CallbackStorage

Store all callbacks.

_callbacks

__iter__()

Generate an iteration for this iterable.

__contains__(func: collections.abc.Callable) → bool

Implements the in operation for this class.

append(func: collections.abc.Callable, origin: str = None) → str

Append a given callable to the storage.

Parameters:

• func – callable to append.

• origin – str (default: None) indicating the origin of the callback.

Returns:

name of the callback.

Return type:

str

remove(func: callable) → bool

Remove a callable from storage.

Parameters:

func – callable to remove.

Returns:

truth value indicating if the callable was a callback.

Return type:

bool

is_callback(func: collections.abc.Callable) → bool

Check if given callable is a callback.

Parameters:

func – callable to check.

Returns:

callable is a callback

Return type:

bool

get_callbacks() → list

Get a list of all callbacks dictionaries.

Returns:

dictionary representation of all callbacks

Return type:

list

class heros.event.EventHandler

Base class for event handlers.

abstractmethod connect(callback: collections.abc.Callable)

Connect a callback function.

abstractmethod disconnect(callback: collections.abc.Callable)

Disconnect a callback function.

abstractmethod is_callback(func: collections.abc.Callable) → bool

Check if func is a callback

abstractmethod get_callbacks()

Return all callbacks

class heros.event.LocalEventHandler(instance, func, reliability: zenoh.Reliability, congestion_control: zenoh.CongestionControl)

Bases: EventHandler

Handles event connections for a specific instance.

instance

func

callbacks

publisher

__call__(*args, **kwargs)

Call the original function and trigger callbacks.

connect(callback: collections.abc.Callable, origin: str | None = None) → str

Connect a callback function to be triggered when the method is called.

Parameters:

• callback – callable to connect.

• origin – (optional) str indicting origin of this callback.

Returns:

name of the callback.

Return type:

str

disconnect(callback: collections.abc.Callable) → bool

Disconnect a callback function.

Parameters:

callback – callable to disconnect.

Returns:

truth value indicating if the callable was a callback.

Return type:

bool

is_callback(func: collections.abc.Callable) → bool

Check if given callable is already a registered callback.

Parameters:

callback – callable to check.

Returns:

truth value indicating if the callable is a callback.

Return type:

bool

get_callbacks() → list

Return a list of registered callback functions.

Returns:

dictionary representation of all callbacks

Return type:

list

class heros.event.RemoteEventHandler(instance, func: collections.abc.Callable | None = None)

Bases: EventHandler

Handles remote events for a specific instance.

instance

callbacks

__call__(payload)

Call the original function and trigger callbacks.

connect(callback: collections.abc.Callable) → str

Connect a callback function to be triggered when the method is called.

Parameters:

• callback – callable to connect.

• origin – (optional) str indicting origin of this callback.

Returns:

name of the callback.

Return type:

str

disconnect(callback: collections.abc.Callable) → None

Disconnect a callback function.

Parameters:

callback – callable to disconnect.

Returns:

truth value indicating if the callable was a callback.

Return type:

bool

is_callback(func: collections.abc.Callable) → bool

Check if given callable is already a registered callback.

Parameters:

callback – callable to check.

Returns:

truth value indicating if the callable is a callback.

Return type:

bool

get_callbacks() → list

Return a list of registered callback functions.

Returns:

dictionary representation of all callbacks

Return type:

list

class heros.event.EventDescriptor(func: collections.abc.Callable | None = None, handler_config: dict | None = None)

A descriptor to handle instance-specific event connections.

func = None

handler_config

_instances

static _get_event_handler_cls()

Abstractmethod:

__get__(instance, owner)

Ensure the method and event-handling functions are bound to the instance.

Parameters:

• self – the EventDescriptor instance

• instance – the owning LocalHERO/RemoteHERO.

• owner

class heros.event.LocalEventDescriptor(func: collections.abc.Callable | None = None, handler_config: dict | None = None)

Bases: EventDescriptor

Descriptor of @event decorated methods of a LocalHERO.

static _get_event_handler_cls()

class heros.event.RemoteEventDescriptor(func: collections.abc.Callable | None = None, handler_config: dict | None = None)

Bases: EventDescriptor

Descriptor of remote representations of events in a RemoteHERO.

static _get_event_handler_cls()

heros.event.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 BOSS decorator injector mechanism. The following example makes the emitted images from a camera reliable.

{
    "_id": "my_camera",
    "classname": "herosdevices.hardware.ids.PeakCompatibleCamera",
    "arguments": {
        "cam_id": "1409f4e7c3b1"
    }
    "extra_decorators": [
          ["acquisition_data", "heros.event.reliable"]
    ]
}

heros.event.event(func: collections.abc.Callable)

Decorator for events.

Note

Only use on methods bound to objects.