herostools.actor

Submodules

• herostools.actor.data_archiver
• herostools.actor.prometheus
• herostools.actor.statemachine

Classes

ArrayArchiver	This HERODataArchiver assumes the data to be numpy-like arrays and saves them as npy files.
JsonArchiver	This HERODataArchiver assumes the data to be a dictionary and saves it as a json file.
HERODatasourceStateMachine	A class that can observe and handle the data emitted by one or more datasource HEROs.

Package Contents

class herostools.actor.ArrayArchiver(save_template: str, split_data_array: bool = False, *args, **kwargs)

Bases: HERODataArchiver

This HERODataArchiver assumes the data to be numpy-like arrays and saves them as npy files.

Parameters:

• object_selector – Zenoh object selector for the devices to subscribe to. In the simplest case this is the name of the target HERO

• event_name – Name of the event.

• save_template – The template from which the file name is generated. Jinja2 is used to generate a filename from the template using the given meta data given as a dictionary. Meta data can be supplied either by default_metadata or obtained from the payload. For an example see the json example below.

• split_data_array – If True and the payload is an array, the observer will split the array into individual frames and save them as separate files. The key _split_index can be used in save_template to specify the subframe index in the filename.

• default_metadata – A dictionary containing the default metadata to be used when generating the filename.

• max_retries – In case storing the data failed, retry storing until max_retries.

Example

The class can be started with BOSS using a json string as in the following example:

{
  "_id": "my-camera-capturer",
  "classname": "herostools.actor.ArrayArchiver",
  "arguments": {
    "object_selector": "my-camera",
    "event_name": "acquisition_data",
    "default_metadata": {
      "file_path": "/mnt/mystorage/images"
    },
    "save_template": "{{ file_path }}/testimg-{{ '%04d' % ( frame / 2 ) |round(0, 'floor') }}-{{ frame % 2 }}.npy"
  }
}

The templates generates file paths like the following:

/mnt/mystorage/images/testimg-0000-0.npy
/mnt/mystorage/images/testimg-0000-1.npy
/mnt/mystorage/images/testimg-0001-0.npy

assuming that frame is a running iterator provided by the payload metadata (i.e. a key in the metadata dictionary).

name_template

split_data_array = False

_store(source_name: str, payload: numpy.typing.NDArray[Any], metadata: dict) → None

Save data to a numpy array. The filename is generated from the jinja template using the metadata.

Parameters:

• source_name – Name of the event source (the HERO).

• payload – The actual data as a numpy array.

• metadata – The received metadata combined with the default metadata.

class herostools.actor.JsonArchiver(save_template: str, merge_metadata: bool = False, *args, **kwargs)

Bases: HERODataArchiver

This HERODataArchiver assumes the data to be a dictionary and saves it as a json file.

Parameters:

• object_selector – Zenoh object selector for the devices to subscribe to. In the simplest case this is the name of the target HERO

• event_name – Name of the event.

• save_template –

  The template from which the file name is generated. Jinja2 is used to generate a filename from the template using the given meta data given as a dictionary. Meta data can be supplied either by default_metadata or obtained from the payload. For an example see the json example below.

• merge_metadata – If True, merge the accompanying metadata to the data itself under the key metadata..

• default_metadata – A dictionary containing the default metadata to be used when generating the filename.

name_template

merge_metadata = False

_store(source_name: str, payload: dict, metadata: dict) → None

Save data to a json file. The filename is generated from the jinja template using the metadata.

Parameters:

• source_name – Name of the event source (the HERO).

• payload – The actual data.

• metadata – The received metadata combined with the default metadata.

class herostools.actor.HERODatasourceStateMachine(loop, *args, http_port: int = 9090, bind_address: str = 'localhost', metrics_endpoint='/metrics', object_selector: str = '*', labels: dict = {}, **kwargs)

Bases: heros.DatasourceObserver

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 observable_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)

cache

_http_port = 9090

_bind_address = 'localhost'

_metrics_endpoint = '/metrics'

_global_labels

async _http_handle_metrics(request)

async _start_webserver()

_metric_name(key)

create a clean name for a metric without special characters matching [a-z0-9_].

_unit_name(unit)

cleans units to not contain special characters

_update(source_name, data)

update the values in the cache for source_name

_convert_to_metrics(dsrs: heros.datasource.types.DatasourceReturnSet, prefix=None)

get_cache()

clear()