herostools.actor.archiver¶

Submodules¶

Exceptions¶

ArtifactReadError

Raised when an artifact cannot be read from its configured storage location.

Classes¶

`ArrayArchiver`	HERODataArchiver that saves numpy arrays as `.npy` files.
`ArtifactStorage`	Abstract base for artifact key-value stores.
`InMemoryArtifactStorage`	Dict-backed in-memory ArtifactStorage — useful for testing and dry runs.
`S3ArtifactStorage`	ArtifactStorage backed by any S3-compatible object store.
`HERODataArchiver`	Base EventObserver that subscribes to a HERO event and archives its payload.
`JsonArchiver`	HERODataArchiver that saves dict payloads as `.json` files.
`PostgresArchiver`	HEROS event adapter for `PostgresRecordStore`.
`PostgresDatabase`	Pure SQL layer for the archiver schema — stores and retrieves clean JSON.
`PostgresRecordChangeListener`	Yield batches of changed record identifiers from PostgreSQL notifications.
`PostgresRecordStore`	HEROS-independent PostgreSQL record store with artifact handling.
`ZarrArchiver`	HERODataArchiver that stores numpy array payloads in a zarr store.

Package Contents¶

class herostools.actor.archiver.ArrayArchiver(save_template: str, split_data_array: bool = False, *args, **kwargs)[source]¶

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

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.

exception herostools.actor.archiver.ArtifactReadError(artifact_id: str, bucket: str | None)[source]¶

Bases: RuntimeError

Raised when an artifact cannot be read from its configured storage location.

Parameters:

artifact_id – Identifier of the artifact that could not be read.
bucket – Effective bucket used for the read.

artifact_id¶

bucket¶

class herostools.actor.archiver.ArtifactStorage(max_artifacts_per_batch: int = DEFAULT_MAX_ARTIFACTS_PER_BATCH)[source]¶

Bases: abc.ABC

Abstract base for artifact key-value stores.

Implementations must support put/get/remove by string artifact_id and optionally list all stored IDs. All mutating methods accept an optional bucket override; when omitted they use the backend’s default bucket.

_max_artifacts_per_batch = 100¶

property default_bucket: str | None¶: Default bucket name for this storage backend, or None if not applicable.

abstractmethod put(artifact_id: str, data: numpy.ndarray, bucket: str | None = None) → str | None[source]¶

Store a numpy array under the given artifact_id.

Parameters:

artifact_id – Unique key (UUID string).
data – Array to store.
bucket – Bucket override; uses backend default when None.

Returns:

The effective bucket name used for storage, or None for backends without a bucket concept (e.g. in-memory).

abstractmethod get(artifact_id: str, bucket: str | None = None) → numpy.ndarray[source]¶

Retrieve a numpy array by artifact_id.

Parameters:

artifact_id – Key to look up.
bucket – Bucket override; uses backend default when None.

Returns:

The stored numpy array.

abstractmethod remove(artifact_id: str, bucket: str | None = None) → None[source]¶

Delete the artifact with the given artifact_id.

Parameters:

artifact_id – Key to delete.
bucket – Bucket override; uses backend default when None.

abstractmethod list(bucket: str | None = None) → list[str][source]¶

Return all artifact_ids currently stored.

Parameters:: bucket – Bucket override; uses backend default when None.
Returns:: List of artifact_id strings.

get_many(artifact_ids: list[str], bucket: str | None = None) → list[numpy.ndarray][source]¶

Retrieve multiple arrays by artifact_id, in the same order as the input.

Parameters:

artifact_ids – Keys to look up.
bucket – Bucket override forwarded to each get() call.

Returns:

List of numpy arrays in the same order as artifact_ids.

get_artifact(artifact_id: str, bucket: str | None = None) → numpy.ndarray[source]¶

Retrieve one decoded artifact by its ID and optional bucket.

Parameters:

artifact_id – Identifier of the artifact to retrieve.
bucket – Bucket override; uses default_bucket when None.

Returns:

The decoded numpy array.

Raises:

ArtifactReadError – If the artifact cannot be read.

get_artifacts(locations: collections.abc.Sequence[tuple[str, str | None]]) → collections.abc.Mapping[str, numpy.ndarray][source]¶

Retrieve a bounded set of artifacts addressed by ID and bucket.

The returned mapping is keyed only by artifact ID. An ID may therefore occur at one effective bucket only within a request. Identical locations are read once and represented once in the result.

Parameters:

locations – Artifact ID and optional bucket pairs.

Returns:

Mapping of artifact IDs to decoded numpy arrays.

Raises:

ValueError – If the request exceeds the batch limit or one artifact ID is requested from multiple effective buckets.
ArtifactReadError – If an artifact cannot be read.

ensure_bucket(bucket: str | None) → None[source]¶

Ensure a bucket is available for writes when the backend requires it.

Parameters:: bucket – Bucket to ensure, or None for backends without buckets.

put_many(artifact_ids: list[str], arrays: list[numpy.ndarray], bucket: str | None = None) → None[source]¶

Store multiple arrays under their respective artifact_ids.

Parameters:

artifact_ids – Unique keys.
arrays – Arrays to store, paired with artifact_ids by position.
bucket – Bucket override forwarded to each put() call.

remove_many(artifact_ids: list[str], bucket: str | None = None) → None[source]¶

Delete multiple artifacts by artifact_id.

Parameters:

artifact_ids – Keys to delete.
bucket – Bucket override forwarded to each remove() call.

class herostools.actor.archiver.InMemoryArtifactStorage(max_artifacts_per_batch: int = DEFAULT_MAX_ARTIFACTS_PER_BATCH)[source]¶

Bases: ArtifactStorage

Dict-backed in-memory ArtifactStorage — useful for testing and dry runs.

The bucket parameter is accepted on all methods for interface compatibility but is ignored; all artifacts share a single in-memory dict regardless of bucket.

_store: dict[str, numpy.ndarray]¶

put(artifact_id: str, data: numpy.ndarray, bucket: str | None = None) → str | None[source]¶

Store array in memory.

Parameters:

artifact_id – Unique key.
data – Array to store.
bucket – Ignored.

Returns:

None (no bucket concept for in-memory storage).

get(artifact_id: str, bucket: str | None = None) → numpy.ndarray[source]¶

Retrieve array from memory.

Parameters:

artifact_id – Key to look up.
bucket – Ignored.

Returns:

The stored numpy array.

remove(artifact_id: str, bucket: str | None = None) → None[source]¶

Remove array from memory.

Parameters:

artifact_id – Key to delete.
bucket – Ignored.

list(bucket: str | None = None) → list[str][source]¶

Return all stored keys.

Parameters:: bucket – Ignored.
Returns:: List of artifact_id strings.

class herostools.actor.archiver.S3ArtifactStorage(endpoint_url: str | None = 'http://localhost:9000', bucket: str = 'artifacts', access_key: str = 'heros', secret_key: str = 'heros', max_workers: int = 10, max_artifacts_per_batch: int = DEFAULT_MAX_ARTIFACTS_PER_BATCH, create_buckets: bool = True)[source]¶

Bases: ArtifactStorage

ArtifactStorage backed by any S3-compatible object store.

Numpy arrays are serialised via np.save/np.load into BytesIO and stored as S3 objects keyed by artifact_id (UUID string). The bucket is created on first use if it does not already exist.

Defaults point to the RustFS instance defined in docker-compose.yml (endpoint http://localhost:9000, credentials rustfs/rustfsdev).

To use AWS S3, pass endpoint_url=None and supply real IAM credentials:

S3ArtifactStorage(
    endpoint_url=None,
    bucket="my-bucket",
    access_key="MY_ACCESS_KEY",
    secret_key="MY_SECRET_KEY",
)

Parameters:

endpoint_url – S3 API URL for self-hosted backends, e.g. "http://localhost:9000" for local RustFS. Pass None to use the default AWS S3 endpoint.
bucket – Default bucket name.
access_key – Access key id (RUSTFS_ACCESS_KEY for RustFS, AWS access key id for AWS S3).
secret_key – Secret access key (RUSTFS_SECRET_KEY for RustFS, AWS secret access key for AWS S3).
create_buckets – Create missing buckets on construction and before writes. Set False for read-only clients.

_bucket = 'artifacts'¶

_max_workers = 10¶

_create_buckets = True¶

_known_buckets: set[str]¶

_bucket_lock¶

_client¶

property default_bucket: str¶: Default bucket name configured at construction time.

ensure_bucket(bucket: str | None) → None[source]¶

Ensure an S3 bucket exists, tolerating concurrent creation.

Parameters:: bucket – Bucket to ensure. None uses default_bucket.
Raises:: ClientError – If the bucket cannot be inspected or created.

put(artifact_id: str, data: numpy.ndarray, bucket: str | None = None) → str[source]¶

Serialise array to npy bytes and upload as an S3 object.

Parameters:

artifact_id – S3 object key.
data – Array to store.
bucket – Bucket override; uses default_bucket when None.

Returns:

The effective bucket name used for storage.

get(artifact_id: str, bucket: str | None = None) → numpy.ndarray[source]¶

Download and deserialise an S3 object to a numpy array.

Parameters:

artifact_id – S3 object key.
bucket – Bucket override; uses default_bucket when None.

Returns:

The stored numpy array.

remove(artifact_id: str, bucket: str | None = None) → None[source]¶

Delete an S3 object.

Parameters:

artifact_id – S3 object key to delete.
bucket – Bucket override; uses default_bucket when None.

list(bucket: str | None = None) → list[str][source]¶

Return all object keys in the bucket via paginated listing.

Parameters:: bucket – Bucket override; uses default_bucket when None.
Returns:: List of artifact_id strings.

get_many(artifact_ids: list[str], bucket: str | None = None) → list[numpy.ndarray][source]¶

Download and deserialise multiple S3 objects in parallel.

Parameters:

artifact_ids – S3 object keys to fetch.
bucket – Bucket override forwarded to each get() call.

Returns:

List of numpy arrays in the same order as artifact_ids.

put_many(artifact_ids: list[str], arrays: list[numpy.ndarray], bucket: str | None = None) → None[source]¶

Serialise and upload multiple arrays to S3 in parallel.

Parameters:

artifact_ids – S3 object keys.
arrays – Arrays to store, paired with artifact_ids by position.
bucket – Bucket override forwarded to each put() call.

remove_many(artifact_ids: list[str], bucket: str | None = None) → None[source]¶

Delete multiple S3 objects using the native batch delete API.

Sends at most 1000 keys per request as required by the S3 API.

Parameters:

artifact_ids – S3 object keys to delete.
bucket – Bucket override; uses default_bucket when None.

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

Bases: heros.EventObserver

Base EventObserver that subscribes to a HERO event and archives its payload.

The event payload may be a (data, metadata) tuple or a plain dict. Subclasses implement _store().

Parameters:

object_selector – Zenoh object selector for the devices to subscribe to.
event_name – Name of the event.
default_metadata – Default metadata merged with every incoming payload’s metadata.
max_retries – Number of times to retry a failed store before dropping the item.

metadata¶

max_retries = 5¶

_payload_queue¶

_stop_event¶

_worker_thread¶

_process_queue() → None[source]¶: Background worker that consumes the payload queue.

is_queue_drained() → bool[source]¶

Return True when all queued items have been processed.

Returns:: True if every item put on the queue has had task_done() called.

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

_teardown() → None[source]¶: Called by boss on shutdown.

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

Callback registered with the source event.

data may be a (payload, metadata) tuple or a plain dict. When a plain dict is received it is used as both payload and metadata so downstream _store implementations can still locate identifier keys regardless of which field they are in.

Parameters:

source_name – Name of the event source (the HERO).
data – Either a (payload, metadata) tuple or a plain dict payload.
retry_count – Number of times this item has already been retried.

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

Store the payload. Must be implemented by subclasses.

Parameters:

source_name – Name of the event source (the HERO).
payload – The actual data.
metadata – Incoming metadata merged with default_metadata.

class herostools.actor.archiver.JsonArchiver(save_template: str, merge_metadata: bool = False, *args, **kwargs)[source]¶

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

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

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

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[source]¶: Publish PostgreSQL change notifications through the HEROS event.

_process_queue() → None[source]¶: Drain the queue, then close this thread’s DB connection on exit.

_teardown() → None[source]¶: Stop worker and listener threads; worker closes its own DB connection.

class herostools.actor.archiver.PostgresDatabase(db_url: str, identifier_key: str = 'identifier', allow_purge: bool = False, retention_days: int = 0)[source]¶

Pure SQL layer for the archiver schema — stores and retrieves clean JSON.

Can be used independently of HEROS to insert, query, and delete records. Payloads must be clean JSON dicts (no numpy arrays). Artifact references stored in data_json are returned as plain dicts by get() without resolution; use PostgresArchiver for the full artifact pipeline.

Each identifier maps to exactly one row in records. Successive payloads for the same identifier are shallow-merged (||) at write time; later events win on key collision.

Parameters:

db_url – psycopg connection string.
identifier_key – Key used to look up the record identifier in the payload.
allow_purge – Enable the purge() method (disabled by default).
retention_days – Default retention window in days used by _find_expired(). 0 means keep forever.

_db_url¶

_local¶

_identifier_key = 'identifier'¶

_allow_purge = False¶

_retention_days = 0¶

_get_conn() → psycopg.Connection[source]¶

Return a per-thread connection, opening one if necessary.

psycopg connections are not thread-safe; each thread owns its own connection.

close() → None[source]¶: Close the calling thread’s database connection.

_store(payload: dict) → str | None[source]¶

Upsert a clean JSON payload into records, merging into any existing row.

The identifier is extracted from the payload using _identifier_key. Returns None and logs an error when the identifier key is missing.

Parameters:: payload – Clean JSON dict to store. Must not contain numpy arrays.
Returns:: The identifier string, or None if the identifier key is absent.

get_ids(tag: str, values: list[Any]) → list[str][source]¶

Return identifiers matching the given tag/key in either source.

Searches identifier_groups (post-hoc tags) and records.data_json (payload fields). Results are deduplicated and sorted.

Parameters:

tag – Tag key (identifier_groups) or top-level data_json key.
values – Accepted values (cast to str for comparison).

Returns:

Sorted, deduplicated list of matching identifiers.

get(ids: list[str]) → dict[str, dict][source]¶

Fetch the merged record for each identifier, including post-hoc tags.

Tags stored via tag_ids() are merged into the returned dict alongside data_json fields. Tags overwrite data_json values on key collision. Artifact references remain as plain dicts in the result.

Parameters:: ids – Identifiers to fetch.
Returns:: Dict mapping identifier to the merged record dict. Identifiers with no record are absent from the result.

tag_ids(ids: list[str], tag: str, value: str) → None[source]¶

Attach a tag to a list of identifiers, overwriting any existing value for that tag.

Inserts (identifier, tag, value) into identifier_groups for each identifier. If the (identifier, tag) pair already exists the value is updated in place.

Parameters:

ids – Identifiers to tag. Each must already exist in records.
tag – Tag key.
value – Tag value to assign.

Raises:

psycopg.errors.ForeignKeyViolation – If any identifier is not present in records.

purge(ids: list[str]) → None[source]¶

Delete DB rows for the given identifiers from records and identifier_groups.

Requires allow_purge=True at instantiation time. Does not touch artifact storage; use PostgresArchiver.purge() for that.

Parameters:: ids – Identifiers to delete.
Raises:: PermissionError – If allow_purge was not set to True.

_find_expired(days: int) → list[str][source]¶

Return identifiers older than days days, excluding saved ones.

Records tagged with _save set to a truthy value ('1', 'true', 'True') are excluded from the result.

Parameters:: days – Retention window in days.
Returns:: List of expired identifier strings.

class herostools.actor.archiver.PostgresRecordChangeListener(db_url: str)[source]¶

Yield batches of changed record identifiers from PostgreSQL notifications.

Parameters:: db_url – Psycopg connection string.

_db_url¶

iter_changes(stop_event: threading.Event) → collections.abc.Iterator[list[str]][source]¶

Yield deduplicated notification batches until stopped.

Parameters:: stop_event – Event that stops the notification loop.
Yields:: Lists of changed record identifiers.

class herostools.actor.archiver.PostgresRecordStore(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)[source]¶

Bases: PostgresDatabase

HEROS-independent PostgreSQL record store with artifact handling.

Owns the full artifact pipeline: numpy array detection, artifact storage writes and reads, per-bucket resolution, and artifact deletion on purge. PostgresDatabase handles only SQL; this class adds the artifact layer. It does not subscribe to HEROS events or PostgreSQL notifications.

Each identifier maps to one row in records. Successive payloads are shallow-merged at write time (||); later events win on key collision. Numpy arrays are replaced by artifact_ref dicts in data_json and stored in the configured artifact storage. Each ref embeds the bucket it was written to, enabling transparent multi-bucket resolution and deletion.

Parameters:

db_url – psycopg connection string.
identifier_key – Key used to look up the record identifier.
allow_purge – Enable the purge() method.
artifact_storage_kwargs – Passed as S3ArtifactStorage(**kwargs). Pass None to use InMemoryArtifactStorage (no persistence, useful for dev/test). Omit to use local RustFS defaults.
array_key_template – Jinja2 template rendered against per-event metadata to form the dict key for bare numpy-array payloads. source_name is always available in the template context alongside all metadata fields. Defaults to "{{ source_name }}" which preserves existing behaviour.
use_single_bucket – Store new artifacts in the configured static bucket instead of UTC daily buckets. Defaults to False.
retention_days – Default retention window in days used by cleanup(). 0 (default) means keep forever.
artifact_storage – Explicit storage instance. Mutually exclusive with artifact_storage_kwargs.

_artifact_storage¶

_array_key_template¶

_use_single_bucket = False¶

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

Store one payload after extracting any numpy-array artifacts.

A bare numpy array payload is wrapped as {template_key: array} first. metadata is shallow-merged into the payload (payload wins on key collision). Arrays are replaced by artifact_ref dicts with the bucket embedded; artifacts are uploaded to storage before the DB write.

Parameters:

source_name – Name of the event source.
payload – The actual data (dict or numpy array).
metadata – Incoming metadata merged with default_metadata.

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

Store one payload through the synchronous record-store API.

Parameters:

source_name – Name of the event source.
payload – The data to store.
metadata – Metadata merged into the payload.

get(ids: list[str], resolve_artifacts: bool = True, size_limit_bytes: int | None = None) → dict[str, dict][source]¶

Fetch merged records and optionally resolve artifact_refs to numpy arrays.

Tags stored via tag_ids() are merged into the returned dict alongside data_json fields. Tags overwrite data_json values on key collision (post-hoc annotations take priority).

Artifact_refs are resolved per-bucket: each ref’s stored bucket field determines which bucket the array is fetched from. Refs without a bucket field (legacy records) fall back to the storage default_bucket.

Parameters:

ids – Identifiers to fetch.
resolve_artifacts – Replace artifact_ref dicts with numpy arrays.
size_limit_bytes – When set, only artifacts whose stored size_bytes is at or below this threshold are fetched. Larger artifacts remain as artifact_ref dicts. Refs without size_bytes are always resolved. None resolves all regardless of size.

Returns:

Dict mapping identifier to the merged record dict. Identifiers with no record are absent from the result.

get_artifact(artifact_id: str, bucket: str | None = None) → numpy.ndarray[source]¶

Retrieve one decoded artifact from the configured artifact storage.

Parameters:

artifact_id – Identifier of the artifact to retrieve.
bucket – Bucket override; uses the storage default when None.

Returns:

The decoded numpy array.

get_artifacts(locations: collections.abc.Sequence[tuple[str, str | None]]) → collections.abc.Mapping[str, numpy.ndarray][source]¶

Retrieve a bounded set of artifacts from the configured storage.

Parameters:: locations – Artifact ID and optional bucket pairs.
Returns:: Mapping of artifact IDs to decoded numpy arrays.

purge(ids: list[str], remove_artifacts: bool = True) → None[source]¶

Delete all data for the given identifiers, including stored artifacts.

Artifacts are removed per-bucket using each ref’s stored bucket field. Refs without a bucket field fall back to default_bucket. DB rows are deleted after artifact removal.

Requires allow_purge=True at instantiation time.

Parameters:

ids – Identifiers to purge.
remove_artifacts – Also remove referenced artifacts from storage.

Raises:

PermissionError – If allow_purge was not set to True.

cleanup(retention_days: int | None = None, remove_artifacts: bool = True, dry_run: bool = True) → list[str][source]¶

Delete records older than the retention window, skipping saved ones.

Requires allow_purge=True. Records tagged with _save set to a truthy value ('1', 'true', 'True') are never deleted. Returns the list of expired identifiers (deleted or would-be-deleted).

Parameters:

retention_days – Purge records whose updated_at is older than this many days. None uses the instance default set at construction. 0 is a no-op (keep forever).
remove_artifacts – Also remove referenced artifacts from storage.
dry_run – When True (default), only return the expired identifiers without deleting anything.

Returns:

List of expired identifiers.

Raises:

PermissionError – If allow_purge was not set to True.

class herostools.actor.archiver.ZarrArchiver(store_template: str, array_path_template: str, *args, **kwargs)[source]¶

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

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.