Implement ReceiveComponentDataStreamRequest (#178)

Author: llucax

src/frequenz/client/microgrid/_client.py

@@ -17,6 +17,7 @@
 from frequenz.api.common.v1.metrics import bounds_pb2, metric_sample_pb2
 from frequenz.api.common.v1.microgrid.components import components_pb2
 from frequenz.api.microgrid.v1 import microgrid_pb2, microgrid_pb2_grpc
+from frequenz.channels import Receiver
 from frequenz.client.base import channel, client, conversion, retry, streaming
 from frequenz.client.common.microgrid.components import ComponentId
 from google.protobuf.empty_pb2 import Empty
@@ -30,6 +31,8 @@
 from .component._component_proto import component_from_proto
 from .component._connection import ComponentConnection
 from .component._connection_proto import component_connection_from_proto
+from .component._data_samples import ComponentDataSamples
+from .component._data_samples_proto import component_data_samples_from_proto
 from .component._types import ComponentTypes
 from .metrics._bounds import Bounds
 from .metrics._metric import Metric
@@ -84,8 +87,11 @@ def __init__(
             connect=connect,
             channel_defaults=channel_defaults,
         )
-        self._broadcasters: dict[
-            ComponentId, streaming.GrpcStreamBroadcaster[Any, Any]
+        self._component_data_broadcasters: dict[
+            str,
+            streaming.GrpcStreamBroadcaster[
+                microgrid_pb2.ReceiveComponentDataStreamResponse, ComponentDataSamples
+            ],
         ] = {}
         self._sensor_data_broadcasters: dict[
             str,
@@ -118,15 +124,15 @@ async def __aexit__(
                 *(
                     broadcaster.stop()
                     for broadcaster in itertools.chain(
-                        self._broadcasters.values(),
+                        self._component_data_broadcasters.values(),
                         self._sensor_data_broadcasters.values(),
                     )
                 ),
                 return_exceptions=True,
             )
             if isinstance(exc, BaseException)
         )
-        self._broadcasters.clear()
+        self._component_data_broadcasters.clear()
         self._sensor_data_broadcasters.clear()
 
         result = None
@@ -528,6 +534,70 @@ async def add_component_bounds(  # noqa: DOC502 (Raises ApiClientError indirectl
 
         return None
 
+    # noqa: DOC502 (Raises ApiClientError indirectly)
+    def receive_component_data_samples_stream(
+        self,
+        component: ComponentId | Component,
+        metrics: Iterable[Metric | int],
+        *,
+        buffer_size: int = 50,
+    ) -> Receiver[ComponentDataSamples]:
+        """Stream data samples from a component.
+
+        At least one metric must be specified. If no metric is specified, then the
+        stream will raise an error.
+
+        Warning:
+            Components may not support all metrics. If a component does not support
+            a given metric, then the returned data stream will not contain that metric.
+
+            There is no way to tell if a metric is not being received because the
+            component does not support it or because there is a transient issue when
+            retrieving the metric from the component.
+
+            The supported metrics by a component can even change with time, for example,
+            if a component is updated with new firmware.
+
+        Args:
+            component: The component to stream data from.
+            metrics: List of metrics to return. Only the specified metrics will be
+                returned.
+            buffer_size: The maximum number of messages to buffer in the returned
+                receiver. After this limit is reached, the oldest messages will be
+                dropped.
+
+        Returns:
+            The data stream from the component.
+        """
+        component_id = _get_component_id(component)
+        metrics_set = frozenset([_get_metric_value(m) for m in metrics])
+        key = f"{component_id}-{hash(metrics_set)}"
+        broadcaster = self._component_data_broadcasters.get(key)
+        if broadcaster is None:
+            client_id = hex(id(self))[2:]
+            stream_name = f"microgrid-client-{client_id}-component-data-{key}"
+            # Alias to avoid too long lines linter errors
+            # pylint: disable-next=invalid-name
+            Request = microgrid_pb2.ReceiveComponentDataStreamRequest
+            broadcaster = streaming.GrpcStreamBroadcaster(
+                stream_name,
+                lambda: aiter(
+                    self.stub.ReceiveComponentDataStream(
+                        Request(
+                            component_id=_get_component_id(component),
+                            filter=Request.ComponentDataStreamFilter(
+                                metrics=metrics_set
+                            ),
+                        ),
+                        timeout=DEFAULT_GRPC_CALL_TIMEOUT,
+                    )
+                ),
+                lambda msg: component_data_samples_from_proto(msg.data),
+                retry_strategy=self._retry_strategy,
+            )
+            self._component_data_broadcasters[key] = broadcaster
+        return broadcaster.new_receiver(maxsize=buffer_size)
+
 
 class Validity(enum.Enum):
     """The duration for which a given list of bounds will stay in effect."""

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,
@@ -50,6 +51,7 @@
     UnspecifiedComponent,
 )
 from ._relay import Relay
+from ._state_sample import ComponentErrorCode, ComponentStateCode, ComponentStateSample
 from ._status import ComponentStatus
 from ._types import (
     ComponentTypes,
@@ -69,6 +71,10 @@
     "Component",
     "ComponentCategory",
     "ComponentConnection",
+    "ComponentDataSamples",
+    "ComponentErrorCode",
+    "ComponentStateCode",
+    "ComponentStateSample",
     "ComponentStatus",
     "ComponentTypes",
     "Converter",

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)),
+    )