herostools.actor

Submodules

• herostools.actor.archiver
• herostools.actor.prometheus
• herostools.actor.statemachine

Classes

ArrayArchiver	HERODataArchiver that saves numpy arrays as .npy files.
JsonArchiver	HERODataArchiver that saves dict payloads as .json files.
PostgresArchiver	HEROS event adapter for PostgresRecordStore.
ZarrArchiver	HERODataArchiver that stores numpy array payloads in a zarr store.
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: herostools.actor.archiver.base.HERODataArchiver

HERODataArchiver that saves numpy arrays as .npy files.

Parameters:

• object_selector – Zenoh object selector for the devices to subscribe to.

• event_name – Name of the event.

• save_template – Jinja2 template used to generate the output file path from the payload metadata.

• split_data_array – When True, each row of a 2-D array is saved as a separate file. Use {{ _split_index }} in save_template to embed the row index in the filename.

• default_metadata – Default metadata available to the filename template.

• max_retries – Number of times to retry a failed store before dropping the item.

Example

BOSS json configuration:

{
  "_id": "my-camera-capturer",
  "classname": "herostools.archiver.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"
  }
}

Generates paths such as:

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

name_template

split_data_array = False

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

Save the payload as a numpy .npy file.

Parameters:

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

• payload – Data to save as a numpy array.

• metadata – Incoming metadata merged with default_metadata.

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

Bases: herostools.actor.archiver.base.HERODataArchiver

HERODataArchiver that saves dict payloads as .json files.

Parameters:

• object_selector – Zenoh object selector for the devices to subscribe to.

• event_name – Name of the event.

• save_template – Jinja2 template used to generate the output file path from the payload metadata.

• merge_metadata – When True, the metadata dict is embedded in the saved JSON under the key metadata.

• default_metadata – Default metadata available to the filename template.

• max_retries – Number of times to retry a failed store before dropping the item.

name_template

merge_metadata = False

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

Save the payload as a .json file.

Parameters:

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

• payload – Data to save as a JSON-serialisable dict.

• metadata – Incoming metadata merged with default_metadata.

class herostools.actor.PostgresArchiver(db_url: str, identifier_key: str = 'identifier', allow_purge: bool = False, artifact_storage_kwargs: dict | None = _USE_DEFAULT_S3, array_key_template: str = '{{ source_name }}', use_single_bucket: bool = False, retention_days: int = 0, artifact_storage: herostools.actor.archiver.artifact_storage.ArtifactStorage | None = None, *args, **kwargs)

Bases: herostools.actor.archiver.base.HERODataArchiver, PostgresRecordStore

HEROS event adapter for PostgresRecordStore.

Subscribes to incoming data events, queues and retries writes, and publishes record_changed notifications. Use PostgresRecordStore for local current-state queries without HEROS or notification connections.

_change_listener

_listener_thread

record_changed(identifiers: list[str]) → list[str]

Publish changed record identifiers to HEROS subscribers.

Parameters:

identifiers – Deduplicated identifiers from PostgreSQL notifications.

Returns:

The published identifiers.

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

Store a queued HEROS payload through the record store.

Parameters:

• source_name – Name of the event source.

• payload – The data to store.

• metadata – Metadata merged into the payload.

_listen_loop() → None

Publish PostgreSQL change notifications through the HEROS event.

_process_queue() → None

Drain the queue, then close this thread’s DB connection on exit.

_teardown() → None

Stop worker and listener threads; worker closes its own DB connection.

class herostools.actor.ZarrArchiver(store_template: str, array_path_template: str, *args, **kwargs)

Bases: herostools.actor.archiver.base.HERODataArchiver

HERODataArchiver that stores numpy array payloads in a zarr store.

The store root and the array key within it are both derived from Jinja2 templates rendered against the merged metadata. Opening the store in append mode (‘a’) means successive calls accumulate arrays in the same store directory.

Parameters:

• store_template – Jinja2 template rendered to the zarr store root path.

• array_path_template – Jinja2 template rendered to the array key within the store.

• object_selector – Zenoh object selector for the devices to subscribe to.

• event_name – Name of the event.

• default_metadata – Default metadata available to both templates.

• max_retries – Number of times to retry a failed store before dropping the item.

store_template

array_path_template

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

Write the payload array into the zarr store.

Parameters:

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

• payload – Numpy array to store.

• metadata – Incoming metadata merged with default_metadata.

class herostools.actor.HERODatasourceStateMachine(loop, *args, http_port: int = 9090, bind_address: str = 'localhost', metrics_endpoint='/metrics', object_selector: str = '*', labels: dict | None = None, **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()