atomiq.heros

Attributes

heros
LocalHERO

Classes

AtomiqHERO

Base class for objects exposed through HEROS.

Functions

get_heros()	To import heros, we have to play tricks here. Since the artiq worker calls our code
stringify_component(component)	Get a stringified version of the component.

Module Contents

atomiq.heros.get_heros()

To import heros, we have to play tricks here. Since the artiq worker calls our code in the artiq.master.worker_impl.examine function and afterwards removes the imported modules from sys.modules. The PyO3 used for the python zenoh bindings does not like this since then every subsequent import heros will reinitialize the PyO3 rust extension, leading to the error

PyO3 modules compiled for CPython 3.8 or older may only be initialized once per interpreter process

To work around this, we save a reference to the heros module in the sys module and use this reference if it exists. We use the sys module since it should always exist.

atomiq.heros.heros

atomiq.heros.LocalHERO

atomiq.heros.stringify_component(component)

Get a stringified version of the component. If just a component is supplied, we return the name, if it is a tuple or list, we assume the first key to be the name and the second one to be the type.

class atomiq.heros.AtomiqHERO(experiment, realm='heros')

Bases: LocalHERO

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

name: str = ''

rid: int = -1

chunksize = -1

components: tuple[str, Ellipsis] = ()

blocks: tuple[str, Ellipsis] = ()

default_arguments: dict[str, dict] | None = None

experiment

property step_counter

property identifier

property steps_total

Return the total number of steps. Since this is only available after the prepare stage of the experiment, the method returns -1 if the prepare was not yet run.

terminate()

_set_dataset(*args, **kwargs)

Exposes the ARTIQ experiment set_dataset function.

Note, that using this function is discuraged as using native HEROS data archiving and distribution is more powerful. This function exists to provide compatibility for existing setups to migrate to atomiq more smoothly.

_mutate_dataset(*args, **kwargs)

Exposes the ARTIQ experiment mutate_dataset function.

Note, that using this function is discuraged as using native HEROS data archiving and distribution is more powerful. This function exists to provide compatibility for existing setups to migrate to atomiq more smoothly.

_append_to_dataset(*args, **kwargs)

Exposes the ARTIQ experiment append_to_dataset function.

Note, that using this function is discuraged as using native HEROS data archiving and distribution is more powerful. This function exists to provide compatibility for existing setups to migrate to atomiq more smoothly.

_get_dataset(*args, **kwargs)

Exposes the ARTIQ experiment get_dataset function.

Note, that using this function is discuraged as using native HEROS data archiving and distribution is more powerful. This function exists to provide compatibility for existing setups to migrate to atomiq more smoothly.

_get_dataset_metadata(*args, **kwargs)

Exposes the ARTIQ experiment get_dataset_metadata function.

Note, that using this function is discuraged as using native HEROS data archiving and distribution is more powerful. This function exists to provide compatibility for existing setups to migrate to atomiq more smoothly.

emit_data(data)

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, remote_hero_method, origin=None)

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 (types.FunctionType) -- the event callable, i.e. a method that is decorated with @event in the LocalHERO.

• remote_hero_method (collections.abc.Callable) -- callable to connect as a callback.

• origin (str | None) -- optional str indicating the semantic origin of the connection.

Returns:

name of the callback.

Return type:

str

_disconnect_local_hero_callback(event, remote_hero_method)

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

Parameters:

• event (types.FunctionType) -- the event callable, i.e. a method that is decorated with @event in the LocalHERO.

• remote_hero_method (collections.abc.Callable) -- 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)

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

Parameters:

event (types.FunctionType) -- the event callable, i.e. a method that is decorated with @event in the LocalHERO.

Returns:

dictionary representations of the callbacks.

Return type:

list

_name

_endpoint_base_path

_endpoints

_query_endpoint(endpoint, *args, **kwargs)

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 (str) -- 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.

Return type:

list | None

_subscribe_endpoint(endpoint, callback, *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 (str) -- 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 (collections.abc.Callable) -- 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.

_ep_discover = '_discover'

_ep_capabilities = '_capabilities'

_ep_health = '_health'

_ns_objects = '@object'

_default_encoding

_realm = 'heros'

_session_manager

_session

_subscriptions = []

_queryables = []

_query_selector(*args, **kwargs)

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, callback, *args, **kwargs)

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

Parameters:

• selector (str) -- zenoh selector for the subscription. See the zenoh documentation for valid descriptors.

• callback (collections.abc.Callable) -- 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, callback)

Parameters:

• selector (str)

• callback (collections.abc.Callable)

_get_object_info(object_name, timeout=2.0)

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

Parameters:

• object_name (str) -- 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 (float) -- timeout for the discover operation in seconds (default: 2)

Returns:

{remote_object_descriptor}}

Return type:

dict of the form {name

_discover(timeout=2.0)

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 (float) -- timeout for the discover operation in seconds (default: 2)

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)

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

Parameters:

bytes (bytearray | HEROPeer._deserialize.bytes) -- bytearray to deserialize.

__enter__()

__exit__(exc_type, exc_val, exc_tb)

__del__()