Add a Sensor class

This class will be used to list and retrieve sensors information from
the microgrid. The class is based on version 0.17 of the microgrid API
instead of the current 0.15 version, so users can already use the new
interface and make migrating to 0.17 more straight forward.

It is also implemented having in mind that sensor types will be removed
in future releases (and it is actually not being set properly in the
0.15 service), so sensor types are being ignored.

To be compatible with 0.17, sensors also have a "fake" lifetime that is
not really retrieved from the microgrid (it is filled with a "eternal"
lifetime, so sensors are present always).

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

Author: llucax

src/frequenz/client/microgrid/_sensor_proto.py

@@ -0,0 +1,92 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of SensorDataSamples objects from protobuf messages."""
+
+import logging
+
+from frequenz.api.common import components_pb2
+from frequenz.api.microgrid import microgrid_pb2
+
+from ._id import SensorId
+from ._lifetime import Lifetime
+from .sensor import Sensor
+
+_logger = logging.getLogger(__name__)
+
+
+def sensor_from_proto(message: microgrid_pb2.Component) -> Sensor:
+    """Convert a protobuf message to a `Sensor` instance.
+
+    Args:
+        message: The protobuf message.
+
+    Returns:
+        The resulting sensor instance.
+    """
+    major_issues: list[str] = []
+    minor_issues: list[str] = []
+
+    sensor = sensor_from_proto_with_issues(
+        message, major_issues=major_issues, minor_issues=minor_issues
+    )
+
+    if major_issues:
+        _logger.warning(
+            "Found issues in sensor: %s | Protobuf message:\n%s",
+            ", ".join(major_issues),
+            message,
+        )
+    if minor_issues:
+        _logger.debug(
+            "Found minor issues in sensor: %s | Protobuf message:\n%s",
+            ", ".join(minor_issues),
+            message,
+        )
+
+    return sensor
+
+
+def sensor_from_proto_with_issues(
+    message: microgrid_pb2.Component,
+    *,
+    major_issues: list[str],
+    minor_issues: list[str],
+) -> Sensor:
+    """Convert a protobuf message to a sensor instance and collect issues.
+
+    Args:
+        message: The protobuf message.
+        major_issues: A list to append major issues to.
+        minor_issues: A list to append minor issues to.
+
+    Returns:
+        The resulting sensor instance.
+    """
+    sensor_id = SensorId(message.id)
+
+    name = message.name or None
+    if name is None:
+        minor_issues.append("name is empty")
+
+    manufacturer = message.manufacturer or None
+    if manufacturer is None:
+        minor_issues.append("manufacturer is empty")
+
+    model_name = message.model_name or None
+    if model_name is None:
+        minor_issues.append("model_name is empty")
+
+    if (
+        message.category
+        is not components_pb2.ComponentCategory.COMPONENT_CATEGORY_SENSOR
+    ):
+        major_issues.append(f"unexpected category for sensor ({message.category})")
+
+    return Sensor(
+        id=sensor_id,
+        name=name,
+        manufacturer=manufacturer,
+        model_name=model_name,
+        operational_lifetime=Lifetime(),
+    )

src/frequenz/client/microgrid/sensor.py

@@ -0,0 +1,50 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Microgrid sensors.
+
+This package provides classes and utilities for working with different types of
+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.
+"""
+
+import dataclasses
+
+from ._id import SensorId
+from ._lifetime import Lifetime
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True)
+class Sensor:
+    """Measures environmental metrics in the microgrid."""
+
+    id: SensorId
+    """This sensor's ID."""
+
+    name: str | None = None
+    """The name of this sensor."""
+
+    manufacturer: str | None = None
+    """The manufacturer of this sensor."""
+
+    model_name: str | None = None
+    """The model name of this sensor."""
+
+    operational_lifetime: Lifetime = dataclasses.field(default_factory=Lifetime)
+    """The operational lifetime of this sensor."""
+
+    @property
+    def identity(self) -> SensorId:
+        """The identity of this sensor.
+
+        This uses the sensor ID to identify a sensor without considering the
+        other attributes, so even if a sensor state changed, the identity
+        remains the same.
+        """
+        return self.id
+
+    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}>"

tests/test_sensor.py

@@ -0,0 +1,152 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the Sensor and sensor data classes."""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from frequenz.client.microgrid import Lifetime, SensorId
+from frequenz.client.microgrid.sensor import Sensor
+
+
+@pytest.fixture
+def now() -> datetime:
+    """Get the current time."""
+    return datetime.now(timezone.utc)
+
+
+def test_sensor_creation_defaults() -> None:
+    """Test Sensor defaults are as expected."""
+    sensor = Sensor(id=SensorId(1))
+
+    assert sensor.id == SensorId(1)
+    assert sensor.name is None
+    assert sensor.manufacturer is None
+    assert sensor.model_name is None
+    assert sensor.operational_lifetime == Lifetime()
+
+
+def test_sensor_creation_full(now: datetime) -> None:
+    """Test Sensor creation with all fields."""
+    start = now
+    end = start + timedelta(days=1)
+    sensor = Sensor(
+        id=SensorId(1),
+        name="test-sensor",
+        manufacturer="Test Manufacturer",
+        model_name="Test Model",
+        operational_lifetime=Lifetime(
+            start=start,
+            end=end,
+        ),
+    )
+
+    assert sensor.id == SensorId(1)
+    assert sensor.name == "test-sensor"
+    assert sensor.manufacturer == "Test Manufacturer"
+    assert sensor.model_name == "Test Model"
+    assert sensor.operational_lifetime.start == start
+    assert sensor.operational_lifetime.end == end
+
+
+@pytest.mark.parametrize(
+    "name,expected_str",
+    [(None, "<Sensor:SID1>"), ("test-sensor", "<Sensor:SID1:test-sensor>")],
+    ids=["no-name", "with-name"],
+)
+def test_sensor_str(name: str | None, expected_str: str) -> None:
+    """Test string representation of a sensor."""
+    sensor = Sensor(
+        id=SensorId(1),
+        name=name,
+        manufacturer="Test Manufacturer",
+        model_name="Test Model",
+        operational_lifetime=Lifetime(
+            start=datetime.now(timezone.utc),
+            end=datetime.now(timezone.utc) + timedelta(days=1),
+        ),
+    )
+    assert str(sensor) == expected_str
+
+
+_SENSOR = Sensor(
+    id=SensorId(1),
+    name="test",
+    manufacturer="Test Mfg",
+    model_name="Model A",
+)
+
+_DIFFERENT_NAME = Sensor(
+    id=_SENSOR.id,
+    name="different",
+    manufacturer=_SENSOR.manufacturer,
+    model_name=_SENSOR.model_name,
+)
+
+_DIFFERENT_ID = Sensor(
+    id=SensorId(2),
+    name=_SENSOR.name,
+    manufacturer=_SENSOR.manufacturer,
+    model_name=_SENSOR.model_name,
+)
+
+_DIFFERENT_BOTH_ID = Sensor(
+    id=SensorId(2),
+    name=_SENSOR.name,
+    manufacturer=_SENSOR.manufacturer,
+    model_name=_SENSOR.model_name,
+)
+
+
+@pytest.mark.parametrize(
+    "comp,expected",
+    [
+        pytest.param(_SENSOR, True, id="self"),
+        pytest.param(_DIFFERENT_NAME, False, id="other-name"),
+        pytest.param(_DIFFERENT_ID, False, id="other-id"),
+        pytest.param(_DIFFERENT_BOTH_ID, False, id="other-both-ids"),
+    ],
+    ids=lambda o: str(o.id) if isinstance(o, Sensor) else str(o),
+)
+def test_sensor_equality(comp: Sensor, expected: bool) -> None:
+    """Test sensor equality."""
+    assert (_SENSOR == comp) is expected
+    assert (comp == _SENSOR) is expected
+    assert (_SENSOR != comp) is not expected
+    assert (comp != _SENSOR) is not expected
+
+
+@pytest.mark.parametrize(
+    "comp,expected",
+    [
+        pytest.param(_SENSOR, True, id="self"),
+        pytest.param(_DIFFERENT_NAME, True, id="other-name"),
+        pytest.param(_DIFFERENT_ID, False, id="other-id"),
+        pytest.param(_DIFFERENT_BOTH_ID, False, id="other-both-ids"),
+    ],
+)
+def test_sensor_identity(comp: Sensor, expected: bool) -> None:
+    """Test sensor identity."""
+    assert (_SENSOR.identity == comp.identity) is expected
+    assert comp.identity == comp.id
+
+
+_ALL_SENSORS_PARAMS = [
+    pytest.param(_SENSOR, id="comp"),
+    pytest.param(_DIFFERENT_NAME, id="name"),
+    pytest.param(_DIFFERENT_ID, id="id"),
+    pytest.param(_DIFFERENT_BOTH_ID, id="both_ids"),
+]
+
+
+@pytest.mark.parametrize("comp1", _ALL_SENSORS_PARAMS)
+@pytest.mark.parametrize("comp2", _ALL_SENSORS_PARAMS)
+def test_sensor_hash(comp1: Sensor, comp2: Sensor) -> None:
+    """Test that the Sensor hash is consistent."""
+    # We can only say the hash are the same if the sensors are equal, if they
+    # are not, they could still have the same hash (and they will if they have
+    # only different non-hashable attributes)
+    if comp1 == comp2:
+        assert hash(comp1) == hash(comp2)