Add ComponentDataSamples

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

Author: llucax

src/frequenz/client/microgrid/component/__init__.py

@@ -18,6 +18,7 @@
 from ._connection import ComponentConnection
 from ._converter import Converter
 from ._crypto_miner import CryptoMiner
+from ._data_samples import ComponentDataSamples
 from ._electrolyzer import Electrolyzer
 from ._ev_charger import (
     AcEvCharger,
@@ -70,6 +71,7 @@
     "Component",
     "ComponentCategory",
     "ComponentConnection",
+    "ComponentDataSamples",
     "ComponentErrorCode",
     "ComponentStateCode",
     "ComponentStateSample",

src/frequenz/client/microgrid/component/_data_samples.py

@@ -0,0 +1,25 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Definition of a component data aggregate."""
+
+from dataclasses import dataclass
+
+from frequenz.client.common.microgrid.components import ComponentId
+
+from ..metrics._sample import MetricSample
+from ._state_sample import ComponentStateSample
+
+
+@dataclass(frozen=True, kw_only=True)
+class ComponentDataSamples:
+    """An aggregate of multiple metrics, states, and errors of a component."""
+
+    component_id: ComponentId
+    """The unique identifier of the component."""
+
+    metric_samples: list[MetricSample]
+    """The metrics sampled from the component."""
+
+    states: list[ComponentStateSample]
+    """The states sampled from the component."""

src/frequenz/client/microgrid/component/_data_samples_proto.py

@@ -0,0 +1,88 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Loading of ComponentDataSamples objects from protobuf messages."""
+
+
+import logging
+from functools import partial
+
+from frequenz.api.common.v1.microgrid.components import components_pb2
+from frequenz.client.common.microgrid.components import ComponentId
+
+from ..metrics._sample_proto import metric_sample_from_proto_with_issues
+from ._data_samples import ComponentDataSamples
+from ._state_sample_proto import component_state_sample_from_proto
+
+_logger = logging.getLogger(__name__)
+
+
+def component_data_samples_from_proto(
+    message: components_pb2.ComponentData,
+) -> ComponentDataSamples:
+    """Convert a protobuf component data message to a component data object.
+
+    Args:
+        message: The protobuf message to convert.
+
+    Returns:
+        The resulting `ComponentDataSamples` object.
+    """
+    major_issues: list[str] = []
+    minor_issues: list[str] = []
+
+    samples = component_data_samples_from_proto_with_issues(
+        message, major_issues=major_issues, minor_issues=minor_issues
+    )
+
+    # This approach to logging issues might be too noisy. Samples are received
+    # very often, and sometimes can remain unchanged for a long time, leading to
+    # repeated log messages. We might need to adjust the logging strategy
+    # in the future.
+    if major_issues:
+        _logger.warning(
+            "Found issues in component data samples: %s | Protobuf message:\n%s",
+            ", ".join(major_issues),
+            message,
+        )
+
+    if minor_issues:
+        _logger.debug(
+            "Found minor issues in component data samples: %s | Protobuf message:\n%s",
+            ", ".join(minor_issues),
+            message,
+        )
+
+    return samples
+
+
+def component_data_samples_from_proto_with_issues(
+    message: components_pb2.ComponentData,
+    *,
+    major_issues: list[str],
+    minor_issues: list[str],
+) -> ComponentDataSamples:
+    """Convert a protobuf component data message to a component data object collecting issues.
+
+    Args:
+        message: The protobuf message to convert.
+        major_issues: A list to append major issues to.
+        minor_issues: A list to append minor issues to.
+
+    Returns:
+        The resulting `ComponentDataSamples` object.
+    """
+    return ComponentDataSamples(
+        component_id=ComponentId(message.component_id),
+        metric_samples=list(
+            map(
+                partial(
+                    metric_sample_from_proto_with_issues,
+                    major_issues=major_issues,
+                    minor_issues=minor_issues,
+                ),
+                message.metric_samples,
+            )
+        ),
+        states=list(map(component_state_sample_from_proto, message.states)),
+    )