Source code for herostools.actor.archiver.json_archiver

# coding:utf-8

import json
from pathlib import Path

from jinja2 import Template

from .base import HERODataArchiver




class JsonArchiver(HERODataArchiver):
    """HERODataArchiver that saves dict payloads as ``.json`` files.

    Args:
        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.
    """

    def __init__(
        self,
        save_template: str,
        merge_metadata: bool = False,
        *args,
        **kwargs,
    ):
        self.name_template = Template(save_template)
        self.merge_metadata = merge_metadata
        HERODataArchiver.__init__(self, *args, **kwargs)



    def _store(self, source_name: str, payload: dict, metadata: dict) -> None:
        """Save the payload as a ``.json`` file.

        Args:
            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``.
        """
        metadata = self.metadata | metadata
        if self.merge_metadata:
            payload["metadata"] = metadata
        full_path = Path(self.name_template.render(metadata))
        full_path.parent.mkdir(parents=True, exist_ok=True)
        full_path.write_text(json.dumps(payload, indent=4))