heros

Submodules

• heros.capabilities
• heros.datasource
• heros.event
• heros.helper
• heros.heros
• heros.inspect
• heros.serdes
• heros.zenoh

Attributes

local_only
force_remote
default_session_manager
log

Classes

Capability
MethodCapability
AttributeCapability	An attribute capability describes a single variable of the remote object.
EventCapability	An event capability describes the ability of a remote object to notify upon a certain event.
RemoteEventDescriptor	Descriptor of remote representations of events in a RemoteHERO.
HEROPeer	A HEROPeer provides the minimal interface to establish the HERO communication on top of the zenoh backend.
HERO	A HEROPeer provides the minimal interface to establish the HERO communication on top of the zenoh backend.
RemoteHERO	Creates a local stub object from a remote HERO such that it seems like the remote object is a local object.
LocalHERO	Base class for objects exposed through HEROS.
HEROObserver	A HEROObserver keeps track of the HEROs in a given realm by monitoring its zenoh liveliness tokens.
LocalDatasourceHERO	A datasource is a HERO that can provide information on a standardized interface.
PolledLocalDatasourceHERO	This local HERO periodically triggers the event "new_data" to poll and publish data.
DatasourceObserver	A class that can observe and handle the data emitted by one or more datasource HEROs.

Functions

is_hero_event(→ bool)	Check if a callable is a event.
is_local_only(→ bool)	Check if a callable is a local only.
is_force_remote(→ bool)	Check if a callable is a force remote.
is_hero_method(→ bool)	Check if a callable is a method of a (remote) hero.
mark_hero_method(→ callable)	Mark a callable as a method of a (remote) hero.
object_name_from_keyexpr(key_expr, ns_objects, realm)
event(func)	Decorator for events.

Package Contents

class heros.Capability

name: str

flavor: ClassVar[str] = 'undefined'

to_dict()

static from_dict(d: dict)

class heros.MethodCapability

Bases: Capability

flavor: ClassVar[str] = 'method'

parameters: list[Parameter] = []

return_type: str = 'None'

static from_method(m: Callable, name: str | None = None) → MethodCapability

to_signature() → inspect.Signature

to_dict() → dict

static from_dict(d: dict) → MethodCapability

Generate a method capabilities object from a defining dictionary.

Args: definition of the capability according to the standard

call_dict(*args, **kwargs) → dict

This returns a dict that assigns the given parameter to the parameters of ourself. It takes care that positional and keyword arguments are handled correctly

Parameters:

• *args – positional arguments

• **kwargs – keyword arguments

Returns:

dict with parameter assignments

Return type:

dict

class heros.AttributeCapability

Bases: Capability

An attribute capability describes a single variable of the remote object. It is exposed under the name of the capability.

Parameters:

• name – name of the capability

• type – data type. E.g. “str”, “int”, “float”, “list”, …

• access – Read and/or write access. “r” for read, “w” for write, and “rw” for both

flavor: ClassVar[str] = 'attribute'

type: str

access: str = 'rw'

to_dict() → dict

static from_dict(d: dict) → AttributeCapability

class heros.EventCapability

Bases: Capability

An event capability describes the ability of a remote object to notify upon a certain event.

flavor: ClassVar[str] = 'event'

static from_dict(d: dict) → EventCapability

class heros.RemoteEventDescriptor(func: callable = None)

Bases: EventDescriptor

Descriptor of remote representations of events in a RemoteHERO.

static _get_event_handler_cls()

heros.is_hero_event(func: callable) → bool

Check if a callable is a event.

Parameters:

func – callable to check

Returns:

The value of the marker. False if the marker is not present.

heros.is_local_only(func: callable) → bool

Check if a callable is a local only.

Parameters:

func – callable to check

Returns:

The value of the marker. False if the marker is not present.

heros.is_force_remote(func: callable) → bool

Check if a callable is a force remote.

Parameters:

func – callable to check

Returns:

The value of the marker. False if the marker is not present.

heros.is_hero_method(func: callable) → bool

Check if a callable is a method of a (remote) hero.

Parameters:

func – callable to check

Returns:

The value of the marker. False if the marker is not present.

heros.local_only

heros.force_remote

heros.mark_hero_method(func: callable) → callable

Mark a callable as a method of a (remote) hero.

Parameters:

func – callable to mark

Returns:

The marked callable.

heros.default_session_manager

heros.object_name_from_keyexpr(key_expr, ns_objects, realm, endpoint='.*')

heros.log

class heros.HEROPeer(realm: str = 'heros', session_manager=None)

A HEROPeer provides the minimal interface to establish the HERO communication on top of the zenoh backend. To this end, it provides methods to send cbor-serialized messages via the zenoh network. It establishes the @object namespace and communicates in a defined realm. Methods to discover objects in the realm and to retrieve their object information are provided.

Parameters:

• realm – Name of the realm that this HEROPeer belongs to. (default: heros)

• session – optional zenoh session to use. If none is provided, a new zenoh session will be started

_ep_discover = '_discover'

_ep_capabilities = '_capabilities'

_ep_health = '_health'

_ns_objects = '@object'

_default_encoding

_realm = 'heros'

_session_manager

_session = None

_subscriptions = []

_queryables = []

_query_selector(*args, **kwargs) → list

Send a query to an endpoint and deserialize the results. This is a low-level function.

Parameters:

• selector – The zenoh selector.

• target – zenoh target for the query

• timeout – timeout for the zenoh get command

Returns:

list of deserialized results

Return type:

list

_subscribe_selector(selector: str, callback: callable, *args, **kwargs)

Subscribe to a zenoh selector and a attach a callback. The callback receives the deserialized payload of the messages published.

Parameters:

• selector – zenoh selector for the subscription. See the zenoh documentation for valid descriptors.

• callback – method to be called for messages that match the selector. The method needs to accept one argument which is the deserialized payload of the message.

_declare_queryable(selector: str, callback: callable)

_get_object_info(object_name: str, timeout: float = 10.0) → dict

Retrieve the object information for a HERO in the current realm and with the given name.

Parameters:

• object_name – name of the HERO to get the object info for. This name is inserted into a zenoh key expression and can thus contain the corresponding wildcards.

• timeout – timeout for the discover operation in seconds (default: 10)

Returns:

{remote_object_descriptor}}

Return type:

dict of the form {name

_discover(timeout: float = 10.0) → dict

Send query to discovery endpoint of all HEROs in the current realm. All alive objects will respond and send their remote object descriptor.

Parameters:

timeout – timeout for the discover operation in seconds (default: 10)

Returns:

{remote_object_descriptor}}

Return type:

dict of the form {name

_serialize(obj)

Serialize the given object using the serializer used for this HEROPeer. Currently only CBOR is supported.

Parameters:

obj – The object to serialized. Currently only built-in types and numpy arrays are supported.

_deserialize(bytes: bytearray)

Deserialize the given byte string using the deserializer used for this HEROPeer. Currently only CBOR is supported.

Parameters:

bytes – bytearray to deserialize.

_destroy_hero()

__enter__()

__exit__(exc_type, exc_val, exc_tb)

__del__()

class heros.HERO(name: str, realm: str = 'heros', session_manager=None)

Bases: HEROPeer

A HEROPeer provides the minimal interface to establish the HERO communication on top of the zenoh backend. To this end, it provides methods to send cbor-serialized messages via the zenoh network. It establishes the @object namespace and communicates in a defined realm. Methods to discover objects in the realm and to retrieve their object information are provided.

Parameters:

• realm – Name of the realm that this HEROPeer belongs to. (default: heros)

• session – optional zenoh session to use. If none is provided, a new zenoh session will be started

_name

_endpoint_base_path = '@object/heros/Uninferable'

_endpoints

_query_endpoint(endpoint: str, *args, **kwargs) → list

Send a query to an endpoint. This is a wrapper for _query_selector that enforces to talk to the endpoint of the remote object.

Parameters:

endpoint – endpoint within this HERO. If the given endpoint does not start with the endpoint base path, the endpoint base path is prepended to generate a zenoh selector.

_subscribe_endpoint(endpoint: str, callback: callable, *args, **kwargs)

Subscribe to an endpoint of this HERO. This is a wrapper for _subscribe_selector that enforces to talk to the endpoint of the remote object.

Parameters:

• endpoint – endpoint within this HERO. If the given endpoint does not start with the endpoint base path, the endpoint base path is prepended to generate a zenoh selector.

• callback – method to be called for messages that match the selector. The method needs to accept one argument which is the deserialized payload of the message.

class heros.RemoteHERO(name: str, realm: str = 'heros', *args, **kwargs)

Bases: HERO

Creates a local stub object from a remote HERO such that it seems like the remote object is a local object. The remote HERO is identified by its name and has to be available at the given realm.

Attribute and method capabilities of the remote object are directly mapped to attributes and methods of the stub object, respectively. The signature of the methods is adapted accordingly. The remote attributes do not exist locally but are directly changed and read on the remote end. Event capabilities of the remote object are mapped to RemoteEvent objects that are members of this class. By connecting one or more callbacks to this event, the RemoteHERO can react on events triggered at the remote site.

To be able to attach attributes to this class, every instance of a RemoteHERO is created from a dynamically generated child class of RemoteHERO with the name RemoteHERO_<realm>_<HERO name>.

Note

To discover which objects are available in a certain realm, see :class:HEROObserver.

Parameters:

• name – name/identifier of the remote object

• realm – realm (think namespace) at which the object is registered. default is “heros”

_hero_tags

_hero_implements

_remote_capabilities

_liveliness_subscription

_liveliness_changed(sample)

_get_capabilities()

Obtain capabilities from remote object.

Returns:

List of capabilities of the remote device

Return type:

list[Capability]

_setattr_remote_capabilities()

Attach functions to the instance that reflect the name and signature of the capabilities of the remote object.

__eq__(other)

_destroy_hero()

__hash__()

class heros.LocalHERO(name: str, *args, realm: str = 'heros', implements: list[str] | None = None, tags: list[str] | None = None, **kwargs)

Bases: HERO

Base class for objects exposed through HEROS. Any object that should be able to be accessed remotely must be based off this class.

Parameters:

• name – name/identifier under which the object is available. Make sure this name is unique in the realm.

• realm – realm the HERO should exist in. default is “heros”

• implements – list of interfaces that are implemented by the hero

• tags – list of tags to identify and classify the hero

liveliness_token

_capabilities()

Analyze ourself (i.e. the current object) and automatically generate the capabilities of the HERO from this. For every method that doesn’t start with _ a method capability is announced. Every defined class attribute becomes an attribute capability. Every method that is defined in the class with the @event decorator becomes an event.

While scanning for the capabilities, this method directly creates the necessary callbacks and defines the zenoh queryables for the capabilities.

_destroy_hero()

_connect_local_hero_callback(event: callable, remote_hero_method: callable, origin: str = None) → str

Connect a method of RemoteHERO as a callback to an event of the LocalHERO. This leads to a new, direct P2P connection between the RemoteHERO and the LocalHERO to call the method.

Parameters:

• event – the event callable, i.e. a method that is decorated with @event in the LocalHERO.

• remote_hero_method – callable to connect as a callback.

• origin – optional str indicating the semantic origin of the connection.

Returns:

name of the callback.

Return type:

str

_disconnect_local_hero_callback(event: callable, remote_hero_method: callable) → bool

Disconnect a method of RemoteHERO from an event of the LocalHERO.

Parameters:

• event – the event callable, i.e. a method that is decorated with @event in the LocalHERO.

• remote_hero_method – callable to connect as a callback.

Returns:

truth value if the remote method was indeed a callback.

Return type:

bool

_get_local_hero_callbacks(event: callable) → list

Get a list of dictionary representations of the callbacks of an event of the LocalHERO.

Parameters:

event – the event callable, i.e. a method that is decorated with @event in the LocalHERO.

Returns:

dictionary representations of the callbacks.

Return type:

list

class heros.HEROObserver(*args, **kwargs)

Bases: HEROPeer

A HEROObserver keeps track of the HEROs in a given realm by monitoring its zenoh liveliness tokens. The member attribute known_objects always holds a list of all HEROs known to the observer.

Parameters:

• realm – Name of the realm that this HEROPeer belongs to. (default: heros)

• session – optional zenoh session to use. If none is provided, a new zenoh session will be started

known_objects

_object_added_callbacks = []

_object_removed_callbacks = []

_handle_status_change(sample)

Handle the status change of liveliness tokens.

register_object_added_callback(func: callable) → None

Register a callback that should be called when a new HERO joins the realm.

Parameters:

func – function to call when a new HERO joins the realm

register_object_removed_callback(func: callable) → None

Register a callback that should be called when a new HERO leaves the realm.

Parameters:

func – function to call when a new HERO leaves the realm

get_object(object_name: str) → RemoteHERO

Get the RemoteHERO object for the HERO with the given name.

Parameters:

object_name – name of the HERO

class heros.LocalDatasourceHERO(*args, observables: dict | None = None, **kwargs)

Bases: heros.LocalHERO

A datasource is a HERO that can provide information on a standardized interface. This interface is the event new_data. Instances in the zenoh network interested in the data provided by data sources can simply subscribe to the key expression @objects/realm/*/new_data or use the DatasourceObserver.

To make your class a LocalDatasourceHERO make it inherit this class. This class is meant for datasources that create asynchronous events on their own. When processing such an event call new_data to publish the data from this datasource.

Parameters:

• name – name/identifier under which the object is available. Make sure this name is unique in the realm.

• realm – realm the HERO should exist in. default is “heros”

observable_processor

_process_data(data)

new_data(data)

class heros.PolledLocalDatasourceHERO(*args, loop, interval: float = 5, **kwargs)

Bases: LocalDatasourceHERO

This local HERO periodically triggers the event “new_data” to poll and publish data. This class is meant for datasources that do not generate events on their own an thus should be polled on a periodical basis.

To make your class a PolledLocalDatasourceHERO make it inherit this class an implement the method _new_data. The method will get called periodically and the return value will be published as an event.

Note

The periodic calling is realized via asyncio and will thus only work if the asyncio mainloop is started.

Parameters:

• name – name/identifier under which the object is available. Make sure this name is unique in the realm.

• realm – realm the HERO should exist in. default is “heros”

• interval – time interval in seconds between consecutive calls of new_data event

datasource_interval = 5

_loop

async _trigger_datasource()

new_data()

abstractmethod _new_data()

class heros.DatasourceObserver(object_selector: str = '*', *args, **kwargs)

Bases: heros.HEROPeer

A class that can observe and handle the data emitted by one or more datasource HEROs. In particular, this class provides an efficient way to listen to the data emitted by all datasource HEROs in the realm. By not instantiating the HEROs themselves but just subscribing to the topics for the datasource, this reduces the pressure on the backing zenoh network. If, however, only the data of a few HEROs should be observed, it might make more sense to just instantiate the according RemoteHEROs and connect a callback to their new_data signal.

Parameters:

• object_selector – selector to specify which objects to observe. This becomes part of a zenoh selector and thus

• realm. (can be anything that makes sense in the selector. Defaults to * to observe all HEROs in the)

_object_selector = '*'

_new_data_callbacks = []

_subscription

_handle_new_data(key_expr: str, data)

register_new_data_callback(func: callable)

Register a callback that should be called when a new HERO joins the realm.

Parameters:

func – function to call when a new HERO joins the realm

heros.event(func: callable)

Decorator for events.

Note

Only use on methods bound to objects.