herostools.actor
================

.. py:module:: herostools.actor


Submodules
----------

.. toctree::
   :maxdepth: 1

   /autoapi/herostools/actor/archiver/index
   /autoapi/herostools/actor/prometheus/index
   /autoapi/herostools/actor/statemachine/index


Classes
-------

.. autoapisummary::

   herostools.actor.ArrayArchiver
   herostools.actor.JsonArchiver
   herostools.actor.PostgresArchiver
   herostools.actor.ZarrArchiver
   herostools.actor.HERODatasourceStateMachine


Package Contents
----------------

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

   Bases: :py:obj:`herostools.actor.archiver.base.HERODataArchiver`


   HERODataArchiver that saves numpy arrays as ``.npy`` files.

   :param object_selector: Zenoh object selector for the devices to subscribe to.
   :param event_name: Name of the event.
   :param save_template: Jinja2 template used to generate the output file path from
                         the payload metadata.
   :param 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.
   :param default_metadata: Default metadata available to the filename template.
   :param max_retries: Number of times to retry a failed store before dropping the item.

   .. rubric:: 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


   .. py:attribute:: name_template


   .. py:attribute:: split_data_array
      :value: False



   .. py:method:: _store(source_name: str, payload: numpy.typing.NDArray[Any], metadata: dict) -> None

      Save the payload as a numpy ``.npy`` file.

      :param source_name: Name of the event source (the HERO).
      :param payload: Data to save as a numpy array.
      :param metadata: Incoming metadata merged with ``default_metadata``.



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

   Bases: :py:obj:`herostools.actor.archiver.base.HERODataArchiver`


   HERODataArchiver that saves dict payloads as ``.json`` files.

   :param object_selector: Zenoh object selector for the devices to subscribe to.
   :param event_name: Name of the event.
   :param save_template: Jinja2 template used to generate the output file path from
                         the payload metadata.
   :param merge_metadata: When ``True``, the metadata dict is embedded in the saved
                          JSON under the key ``metadata``.
   :param default_metadata: Default metadata available to the filename template.
   :param max_retries: Number of times to retry a failed store before dropping the item.


   .. py:attribute:: name_template


   .. py:attribute:: merge_metadata
      :value: False



   .. py:method:: _store(source_name: str, payload: dict, metadata: dict) -> None

      Save the payload as a ``.json`` file.

      :param source_name: Name of the event source (the HERO).
      :param payload: Data to save as a JSON-serialisable dict.
      :param metadata: Incoming metadata merged with ``default_metadata``.



.. py:class:: 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: :py:obj:`herostools.actor.archiver.base.HERODataArchiver`, :py:obj:`PostgresRecordStore`


   HEROS event adapter for :class:`PostgresRecordStore`.

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


   .. py:attribute:: _change_listener


   .. py:attribute:: _listener_thread


   .. py:method:: record_changed(identifiers: list[str]) -> list[str]

      Publish changed record identifiers to HEROS subscribers.

      :param identifiers: Deduplicated identifiers from PostgreSQL notifications.

      :returns: The published identifiers.



   .. py:method:: _store(source_name: str, payload: Any, metadata: dict) -> None

      Store a queued HEROS payload through the record store.

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



   .. py:method:: _listen_loop() -> None

      Publish PostgreSQL change notifications through the HEROS event.



   .. py:method:: _process_queue() -> None

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



   .. py:method:: _teardown() -> None

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



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

   Bases: :py:obj:`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.

   :param store_template: Jinja2 template rendered to the zarr store root path.
   :param array_path_template: Jinja2 template rendered to the array key within the store.
   :param object_selector: Zenoh object selector for the devices to subscribe to.
   :param event_name: Name of the event.
   :param default_metadata: Default metadata available to both templates.
   :param max_retries: Number of times to retry a failed store before dropping the item.


   .. py:attribute:: store_template


   .. py:attribute:: array_path_template


   .. py:method:: _store(source_name: str, payload: numpy.typing.NDArray[Any], metadata: dict) -> None

      Write the payload array into the zarr store.

      :param source_name: Name of the event source (the HERO).
      :param payload: Numpy array to store.
      :param metadata: Incoming metadata merged with ``default_metadata``.



.. py:class:: HERODatasourceStateMachine(loop, *args, http_port: int = 9090, bind_address: str = 'localhost', metrics_endpoint='/metrics', object_selector: str = '*', labels: dict | None = None, **kwargs)

   Bases: :py:obj:`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.

   :param object_selector: selector to specify which objects to observe. This becomes part of a zenoh selector and thus
   :param can be anything that makes sense in the selector. Defaults to * to observe all HEROs in the realm.:


   .. py:attribute:: cache


   .. py:attribute:: _http_port
      :value: 9090



   .. py:attribute:: _bind_address
      :value: 'localhost'



   .. py:attribute:: _metrics_endpoint
      :value: '/metrics'



   .. py:attribute:: _global_labels


   .. py:method:: _http_handle_metrics(request)
      :async:



   .. py:method:: _start_webserver()
      :async:



   .. py:method:: _metric_name(key)

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



   .. py:method:: _unit_name(unit)

      cleans units to not contain special characters



   .. py:method:: _update(source_name, data)

      update the values in the cache for source_name



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


   .. py:method:: get_cache()


   .. py:method:: clear()


