Metrics and SenML¶

@dataclass
class Metric:
    """A single metric definition used to describe sensor data.

    Metrics are declared in a plugin's ``Meta.out_metrics`` or
    ``Meta.in_metrics``. They define the shape of data the plugin
    produces or consumes.

    Attributes:
        name: Human-readable metric name (e.g. ``"temperature"``).
        unit: Unit of measure (e.g. ``"°C"``). Use an empty string when
            the metric is dimensionless.
        type: `MetricType` — ``TEMPORAL`` for instantaneous readings,
            ``INCREMENTAL`` for cumulative counters.
        data_type: `MetricDataType` — ``NUMBER``, ``STRING``, or ``BOOLEAN``.
        id: Unique identifier, equal to ``name``. Set automatically
            via ``__post_init__``.

    """

    id: str = field(init=False, repr=False)
    name: str
    unit: str
    type: MetricType = field(default=MetricType.TEMPORAL)
    data_type: MetricDataType = field(default=MetricDataType.NUMBER)

    def __post_init__(self) -> None:
        self.id = self.name  # short_hash(self.name)

    # --- Concrete Methods ---

    @staticmethod
    def custom_dict(kv_pairs: Any) -> Dict:
        """Drop private fields when a metric is rendered as a dict.

        Args:
            kv_pairs: The field name/value pairs, as passed by `asdict`.

        Returns:
            The pairs whose name does not start with an underscore.

        """
        return {k: v for k, v in kv_pairs if not k.startswith("_")}

    def as_dict(self) -> Dict:
        """Render the metric as a plain dict, without its private fields."""
        return asdict(self, dict_factory=self.custom_dict)

class MetricType(IntEnum):
    """Wire representation of how a metric's values accumulate."""

    TEMPORAL = 0
    INCREMENTAL = auto()

class MetricDataType(IntEnum):
    """Wire representation of a metric's value type."""

    STRING = 0
    NUMBER = auto()
    BOOLEAN = auto()

@dataclass
class SenMLRecord:
    """A single SenML record representing one measurement or command.

    Follows the SenML (Sensor Measurement Lists) JSON format defined in
    RFC 8428. Only the fields relevant to the ToloMEO protocol are used.

    Attributes:
        bn: Base name — identifies the source device or service.
        n: Name — identifies the specific measurement or command.
        t: Timestamp as a Unix epoch float (optional).
        u: Unit of measure (optional, e.g. ``"°C"``).
        v: Numeric value (optional).
        vs: String value — used for JSON-encoded command payloads (optional).

    """

    bn: str
    n: str
    t: float | None = None
    u: str | None = None
    v: float | None = None
    vs: str | None = None

    # --- Concrete Methods ---

    @staticmethod
    def custom_dict(kv_pairs: Any) -> Dict:
        """Drop unset fields when a record is rendered as a dict.

        Args:
            kv_pairs: The field name/value pairs, as passed by `asdict`.

        Returns:
            The pairs whose value is not None. SenML omits absent fields
            rather than sending them as nulls.

        """
        return {k: v for k, v in kv_pairs if v is not None}

    def as_dict(self) -> Dict:
        """Render the record as a plain dict, omitting its unset fields."""
        return asdict(self, dict_factory=self.custom_dict)

class SenMLRecordBuilder:
    """Builds a single `SenMLRecord` from raw data using a configurable factory.

    Args:
        base_name: The ``bn`` field value for all records built by this instance.
        factory_class: The `SenMLRecordFactory` subclass used to construct records.

    """

    def __init__(self, base_name: str, factory_class: Type[SenMLRecordFactory]) -> None:
        self._base_name = base_name
        self._factory_class = factory_class

    # --- Concrete Methods ---

    def build(self, data: Any) -> SenMLRecord:
        """Build one `SenMLRecord` from ``data`` using the configured factory.

        Args:
            data: Raw input passed to ``factory_class.create``.

        Returns:
            A fully constructed `SenMLRecord`.

        """
        return self._factory_class.create(self._base_name, data)

class MetricNotFoundError(ValueError):
    """Base exception raised when a metric is not found."""

    ...

class DuplicateMetricError(ValueError):
    """Raised when attempting to add a metric that already exists.

    Args:
        id: The id of the metric.
        name: The name of the metric.

    """

    def __init__(self, id: str, name: str):
        super().__init__(f"Duplicate metric ID: {id} (name: {name})")