herostools.actor.data_archiver¶

Classes¶

`HERODataArchiver`	This EventObserver subscribes to a user-definced event of a HERO and saves its payload to a file
`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.

Module Contents¶

class herostools.actor.data_archiver.HERODataArchiver(object_selector: str, event_name: str, default_metadata: dict | None = None, max_retries: int = 5, *args, **kwargs)[source]¶

Bases: heros.EventObserver

This EventObserver subscribes to a user-definced event of a HERO and saves its payload to a file path. The event payload must have the form of an iterable with the first entry being the data and the second entry being a dictionary containing metadata.

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.
default_metadata – A dictionary containing the default metadata.
max_retries – In case storing the data failed, retry storing until max_retries.

Note

Do not use this class directly, use a child class which implements the method _store(). For example, ArrayArchiver assumes the data to be array-like and saves it as a numpy array.

metadata = None¶

max_retries = 5¶

_payload_queue¶

_stop_event¶

_worker_thread¶

_process_queue() → None[source]¶: Background task to process the queue.

_stop()[source]¶: Stop the background thread gracefully.

_teardown() → None[source]¶: Teardown method is called by boss.

feed(source_name: str, data: Iterable, retry_count: int = 0) → None[source]¶

Callback function called by the source event.

Parameters:

source_name – Name of the event source (the HERO).
data – Data to be archived in the format (payload: object, metadata: dict)
retry_count – Count how often this queue item was already encountered.

abstractmethod _store(source_name: str, payload: Any, metadata: dict) → None[source]¶

Abstract method to store the payload. Must be implemented by subclasses.

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.data_archiver.ArrayArchiver(save_template: str, split_data_array: bool = False, *args, **kwargs)[source]¶

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[source]¶

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.data_archiver.JsonArchiver(save_template: str, merge_metadata: bool = False, *args, **kwargs)[source]¶

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[source]¶

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.