Add wrappers for sensor data

Again, these wrappers are done with 0.17 in mind, and trying to be
forward-compatible, although sensor data in 0.15 is already more similar
to 0.17 (compared to component data).

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/client/microgrid/_sensor_proto.py

@@ -4,13 +4,25 @@
 """Loading of SensorDataSamples objects from protobuf messages."""
 
 import logging
+from collections.abc import Set
+from datetime import datetime
 
 from frequenz.api.common import components_pb2
-from frequenz.api.microgrid import microgrid_pb2
+from frequenz.api.microgrid import common_pb2, microgrid_pb2, sensor_pb2
+from frequenz.client.base import conversion
 
 from ._id import SensorId
 from ._lifetime import Lifetime
-from .sensor import Sensor
+from ._util import enum_from_proto
+from .sensor import (
+    Sensor,
+    SensorDataSamples,
+    SensorErrorCode,
+    SensorMetric,
+    SensorMetricSample,
+    SensorStateCode,
+    SensorStateSample,
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -90,3 +102,88 @@ def sensor_from_proto_with_issues(
         model_name=model_name,
         operational_lifetime=Lifetime(),
     )
+
+
+def sensor_data_samples_from_proto(
+    message: microgrid_pb2.ComponentData,
+    metrics: Set[sensor_pb2.SensorMetric.ValueType],
+) -> SensorDataSamples:
+    """Convert a protobuf component data message to a sensor data object.
+
+    Args:
+        message: The protobuf message to convert.
+        metrics: A set of metrics to filter the samples.
+
+    Returns:
+        The resulting `SensorDataSamples` object.
+    """
+    # At some point it might make sense to also log issues found in the samples, but
+    # using a naive approach like in `component_from_proto` might spam the logs too
+    # much, as we can receive several samples per second, and if a component is in
+    # a unrecognized state for long, it will mean we will emit the same log message
+    # again and again.
+    ts = conversion.to_datetime(message.ts)
+    return SensorDataSamples(
+        sensor_id=SensorId(message.id),
+        metrics=[
+            sensor_metric_sample_from_proto(ts, sample)
+            for sample in message.sensor.data.sensor_data
+            if sample.sensor_metric in metrics
+        ],
+        states=[sensor_state_sample_from_proto(ts, message.sensor)],
+    )
+
+
+def sensor_metric_sample_from_proto(
+    sampled_at: datetime, message: sensor_pb2.SensorData
+) -> SensorMetricSample:
+    """Convert a protobuf message to a `SensorMetricSample` object.
+
+    Args:
+        sampled_at: The time at which the sample was taken.
+        message: The protobuf message to convert.
+
+    Returns:
+        The resulting `SensorMetricSample` object.
+    """
+    return SensorMetricSample(
+        sampled_at=sampled_at,
+        metric=enum_from_proto(message.sensor_metric, SensorMetric),
+        value=message.value,
+    )
+
+
+def sensor_state_sample_from_proto(
+    sampled_at: datetime, message: sensor_pb2.Sensor
+) -> SensorStateSample:
+    """Convert a protobuf message to a `SensorStateSample` object.
+
+    Args:
+        sampled_at: The time at which the sample was taken.
+        message: The protobuf message to convert.
+
+    Returns:
+        The resulting `SensorStateSample` object.
+    """
+    # In v0.15 the enum has 3 values, UNSPECIFIED, OK, and ERROR. In v0.17
+    # (common v0.6), it also have 3 values with the same tags, but OK is renamed
+    # to ON, so this conversion should work fine for both versions.
+    state = enum_from_proto(message.state.component_state, SensorStateCode)
+    errors: set[SensorErrorCode | int] = set()
+    warnings: set[SensorErrorCode | int] = set()
+    for error in message.errors:
+        match error.level:
+            case common_pb2.ErrorLevel.ERROR_LEVEL_CRITICAL:
+                errors.add(enum_from_proto(error.code, SensorErrorCode))
+            case common_pb2.ErrorLevel.ERROR_LEVEL_WARN:
+                warnings.add(enum_from_proto(error.code, SensorErrorCode))
+            case _:
+                # If we don´t know the level we treat it as an error just to be safe.
+                errors.add(enum_from_proto(error.code, SensorErrorCode))
+
+    return SensorStateSample(
+        sampled_at=sampled_at,
+        states=frozenset([state]),
+        warnings=frozenset(warnings),
+        errors=frozenset(errors),
+    )

src/frequenz/client/microgrid/sensor.py

@@ -7,12 +7,37 @@
 sensors in a microgrid environment. [`Sensor`][frequenz.client.microgrid.sensor.Sensor]s
 measure various physical metrics in the surrounding environment, such as temperature,
 humidity, and solar irradiance.
+
+# Streaming Sensor Data Samples
+
+This package also provides several data structures for handling sensor readings
+and states:
+
+* [`SensorDataSamples`][frequenz.client.microgrid.sensor.SensorDataSamples]:
+    Represents a collection of sensor data samples.
+* [`SensorErrorCode`][frequenz.client.microgrid.sensor.SensorErrorCode]:
+    Defines error codes that a sensor can report.
+* [`SensorMetric`][frequenz.client.microgrid.sensor.SensorMetric]: Enumerates
+    the different metrics a sensor can measure (e.g., temperature, voltage).
+* [`SensorMetricSample`][frequenz.client.microgrid.sensor.SensorMetricSample]:
+    Represents a single sample of a sensor metric, including its value and
+    timestamp.
+* [`SensorStateCode`][frequenz.client.microgrid.sensor.SensorStateCode]:
+    Defines codes representing the operational state of a sensor.
+* [`SensorStateSample`][frequenz.client.microgrid.sensor.SensorStateSample]:
+    Represents a single sample of a sensor's state, including its state code
+    and timestamp.
 """
 
 import dataclasses
+import enum
+from dataclasses import dataclass
+from datetime import datetime
+from typing import assert_never
 
 from ._id import SensorId
 from ._lifetime import Lifetime
+from .metrics import AggregatedMetricValue, AggregationMethod
 
 
 @dataclasses.dataclass(frozen=True, kw_only=True)
@@ -48,3 +73,170 @@ def __str__(self) -> str:
         """Return a human-readable string representation of this instance."""
         name = f":{self.name}" if self.name else ""
         return f"<{type(self).__name__}:{self.id}{name}>"
+
+
+@enum.unique
+class SensorMetric(enum.Enum):
+    """The metrics that can be reported by sensors in the microgrid.
+
+    These metrics correspond to various sensor readings primarily related to
+    environmental conditions and physical measurements.
+    """
+
+    UNSPECIFIED = 0
+    """Default value (this should not be normally used and usually indicates an issue)."""
+
+    TEMPERATURE = 1
+    """Temperature, in Celsius (°C)."""
+
+    HUMIDITY = 2
+    """Humidity, in percentage (%)."""
+
+    PRESSURE = 3
+    """Pressure, in Pascal (Pa)."""
+
+    IRRADIANCE = 4
+    """Irradiance / Radiation flux, in watts per square meter (W / m²)."""
+
+    VELOCITY = 5
+    """Velocity, in meters per second (m / s)."""
+
+    ACCELERATION = 6
+    """Acceleration in meters per second per second (m / s²)."""
+
+    ANGLE = 7
+    """Angle, in degrees with respect to the (magnetic) North (°)."""
+
+    DEW_POINT = 8
+    """Dew point, in Celsius (°C).
+
+    The temperature at which the air becomes saturated with water vapor.
+    """
+
+
+@enum.unique
+class SensorStateCode(enum.Enum):
+    """The various states that a sensor can be in."""
+
+    UNSPECIFIED = 0
+    """Default value (this should not be normally used and usually indicates an issue)."""
+
+    ON = 1
+    """The sensor is up and running."""
+
+    ERROR = 2
+    """The sensor is in an error state."""
+
+
+@enum.unique
+class SensorErrorCode(enum.Enum):
+    """The various errors that can occur in sensors."""
+
+    UNSPECIFIED = 0
+    """Default value (this should not be normally used and usually indicates an issue)."""
+
+    UNKNOWN = 1
+    """An unknown or undefined error.
+
+    This is used when the error can be retrieved from the sensor but it doesn't match
+    any known error or can't be interpreted for some reason.
+    """
+
+    INTERNAL = 2
+    """An internal error within the sensor."""
+
+
+@dataclass(frozen=True, kw_only=True)
+class SensorStateSample:
+    """A sample of state, warnings, and errors for a sensor at a specific time."""
+
+    sampled_at: datetime
+    """The time at which this state was sampled."""
+
+    states: frozenset[SensorStateCode | int]
+    """The set of states of the sensor.
+
+    If the reported state is not known by the client (it could happen when using an
+    older version of the client with a newer version of the server), it will be
+    represented as an `int` and **not** the
+    [`SensorStateCode.UNSPECIFIED`][frequenz.client.microgrid.sensor.SensorStateCode.UNSPECIFIED]
+    value (this value is used only when the state is not known by the server).
+    """
+
+    warnings: frozenset[SensorErrorCode | int]
+    """The set of warnings for the sensor."""
+
+    errors: frozenset[SensorErrorCode | int]
+    """The set of errors for the sensor.
+
+    This set will only contain errors if the sensor is in an error state.
+    """
+
+
+@dataclass(frozen=True, kw_only=True)
+class SensorMetricSample:
+    """A sample of a sensor metric at a specific time.
+
+    This represents a single sample of a specific metric, the value of which is either
+    measured at a particular time.
+    """
+
+    sampled_at: datetime
+    """The moment when the metric was sampled."""
+
+    metric: SensorMetric | int
+    """The metric that was sampled."""
+
+    # In the protocol this is float | AggregatedMetricValue, but for live data we can't
+    # receive the AggregatedMetricValue, so we limit this to float for now.
+    value: float | AggregatedMetricValue | None
+    """The value of the sampled metric."""
+
+    def as_single_value(
+        self, *, aggregation_method: AggregationMethod = AggregationMethod.AVG
+    ) -> float | None:
+        """Return the value of this sample as a single value.
+
+        if [`value`][frequenz.client.microgrid.sensor.SensorMetricSample.value] is a `float`,
+        it is returned as is. If `value` is an
+        [`AggregatedMetricValue`][frequenz.client.microgrid.metrics.AggregatedMetricValue],
+        the value is aggregated using the provided `aggregation_method`.
+
+        Args:
+            aggregation_method: The method to use to aggregate the value when `value` is
+                a `AggregatedMetricValue`.
+
+        Returns:
+            The value of the sample as a single value, or `None` if the value is `None`.
+        """
+        match self.value:
+            case float() | int():
+                return self.value
+            case AggregatedMetricValue():
+                match aggregation_method:
+                    case AggregationMethod.AVG:
+                        return self.value.avg
+                    case AggregationMethod.MIN:
+                        return self.value.min
+                    case AggregationMethod.MAX:
+                        return self.value.max
+                    case unexpected:
+                        assert_never(unexpected)
+            case None:
+                return None
+            case unexpected:
+                assert_never(unexpected)
+
+
+@dataclass(frozen=True, kw_only=True)
+class SensorDataSamples:
+    """An aggregate of multiple metrics, states, and errors of a sensor."""
+
+    sensor_id: SensorId
+    """The unique identifier of the sensor."""
+
+    metrics: list[SensorMetricSample]
+    """The metrics sampled from the sensor."""
+
+    states: list[SensorStateSample]
+    """The states sampled from the sensor."""