Fetch and stream 3-phase voltage (#815)

Fetch and stream the phase-to-neutral voltage from a source component.
Each phase of the phase-to-neutral voltage is fetched from the source
component individually from the resampling actor and then streamed as a
3-phase sample.

It was agreed to not use the formula engine to get voltage metrics as
voltage does not need formula composition.

Also the phase-to-phase 3-phase voltage is not implemented as in the
future the microgrid API will expose that data. The user can still
perform a simple calculation to get phase-to-phase voltage given
phase-to-neutral voltage.

Fixes #783

Author: daniel-zullo-frequenz

RELEASE_NOTES.md

@@ -46,13 +46,13 @@
 
 - The `ComponentGraph.components()` parameters `component_id` and `component_category` were renamed to `component_ids` and `component_categories`, respectively.
 
-- The `GridFrequency.component` propery was renamed to `GridFrequency.source`
+- The `GridFrequency.component` property was renamed to `GridFrequency.source`
 
 - The `microgrid.frequency()` method no longer supports passing the `component` parameter. Instead the best component is automatically selected.
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+- A new method `microgrid.voltage()` was added to allow easy access to the phase-to-neutral 3-phase voltage of the microgrid.
 
 ## Bug Fixes
 

src/frequenz/sdk/actor/_data_sourcing/microgrid_api_source.py

@@ -74,6 +74,9 @@
     ComponentMetricId.CURRENT_PHASE_1: lambda msg: msg.current_per_phase[0],
     ComponentMetricId.CURRENT_PHASE_2: lambda msg: msg.current_per_phase[1],
     ComponentMetricId.CURRENT_PHASE_3: lambda msg: msg.current_per_phase[2],
+    ComponentMetricId.VOLTAGE_PHASE_1: lambda msg: msg.voltage_per_phase[0],
+    ComponentMetricId.VOLTAGE_PHASE_2: lambda msg: msg.voltage_per_phase[1],
+    ComponentMetricId.VOLTAGE_PHASE_3: lambda msg: msg.voltage_per_phase[2],
     ComponentMetricId.FREQUENCY: lambda msg: msg.frequency,
 }
 

src/frequenz/sdk/microgrid/__init__.py

@@ -130,6 +130,7 @@
     frequency,
     grid,
     logical_meter,
+    voltage,
 )
 
 
@@ -155,4 +156,5 @@ async def initialize(host: str, port: int, resampler_config: ResamplerConfig) ->
     "frequency",
     "logical_meter",
     "metadata",
+    "voltage",
 ]

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -21,6 +21,7 @@
 from ..actor._actor import Actor
 from ..timeseries._base_types import PoolType
 from ..timeseries._grid_frequency import GridFrequency
+from ..timeseries._voltage_streamer import VoltageStreamer
 from ..timeseries.grid import Grid
 from ..timeseries.grid import get as get_grid
 from ..timeseries.grid import initialize as initialize_grid
@@ -119,6 +120,7 @@ def __init__(
         self._ev_charger_pools: dict[frozenset[int], EVChargerPool] = {}
         self._battery_pools: dict[frozenset[int], BatteryPoolReferenceStore] = {}
         self._frequency_instance: GridFrequency | None = None
+        self._voltage_instance: VoltageStreamer | None = None
 
     def frequency(self) -> GridFrequency:
         """Fetch the grid frequency for the microgrid.
@@ -134,6 +136,20 @@ def frequency(self) -> GridFrequency:
 
         return self._frequency_instance
 
+    def voltage(self) -> VoltageStreamer:
+        """Fetch the 3-phase voltage for the microgrid.
+
+        Returns:
+            The VoltageStreamer instance.
+        """
+        if not self._voltage_instance:
+            self._voltage_instance = VoltageStreamer(
+                self._resampling_request_sender(),
+                self._channel_registry,
+            )
+
+        return self._voltage_instance
+
     def logical_meter(self) -> LogicalMeter:
         """Return the logical meter instance.
 
@@ -409,6 +425,15 @@ def frequency() -> GridFrequency:
     return _get().frequency()
 
 
+def voltage() -> VoltageStreamer:
+    """Return the 3-phase voltage for the microgrid.
+
+    Returns:
+        The 3-phase voltage.
+    """
+    return _get().voltage()
+
+
 def logical_meter() -> LogicalMeter:
     """Return the logical meter instance.
 

src/frequenz/sdk/microgrid/component/_component_data.py

@@ -237,6 +237,11 @@ class InverterData(ComponentData):
             -ve current means supply into the grid.
     """
 
+    voltage_per_phase: tuple[float, float, float]
+    """The AC voltage in Volts (V) between the line and the neutral wire for
+       phase/line 1, 2 and 3 respectively.
+    """
+
     # pylint: disable=line-too-long
     active_power_inclusion_lower_bound: float
     """Lower inclusion bound for inverter power in watts.
@@ -312,6 +317,11 @@ def from_proto(cls, raw: microgrid_pb.ComponentData) -> InverterData:
                 raw.inverter.data.ac.phase_2.current.value,
                 raw.inverter.data.ac.phase_3.current.value,
             ),
+            voltage_per_phase=(
+                raw.inverter.data.ac.phase_1.voltage.value,
+                raw.inverter.data.ac.phase_2.voltage.value,
+                raw.inverter.data.ac.phase_3.voltage.value,
+            ),
             active_power_inclusion_lower_bound=raw_power.system_inclusion_bounds.lower,
             active_power_exclusion_lower_bound=raw_power.system_exclusion_bounds.lower,
             active_power_inclusion_upper_bound=raw_power.system_inclusion_bounds.upper,

src/frequenz/sdk/microgrid/component_graph.py

@@ -34,6 +34,8 @@
 
 _logger = logging.getLogger(__name__)
 
+# pylint: disable=too-many-lines
+
 
 class InvalidGraphError(Exception):
     """Exception type that will be thrown if graph data is not valid."""
@@ -289,8 +291,40 @@ def dfs(
             the condition function.
         """
 
+    @abstractmethod
+    def find_first_descendant_component(
+        self,
+        *,
+        root_category: ComponentCategory,
+        descendant_categories: Iterable[ComponentCategory],
+    ) -> Component:
+        """Find the first descendant component given root and descendant categories.
+
+        This method searches for the root component within the provided root
+        category. If multiple components share the same root category, the
+        first found one is considered as the root component.
+
+        Subsequently, it looks for the first descendant component from the root
+        component, considering only the immediate descendants.
 
-class _MicrogridComponentGraph(ComponentGraph):
+        The priority of the component to search for is determined by the order
+        of the descendant categories, with the first category having the
+        highest priority.
+
+        Args:
+            root_category: The category of the root component to search for.
+            descendant_categories: The descendant categories to search for the
+                first descendant component in.
+
+        Returns:
+            The first descendant component found in the component graph,
+            considering the specified root and descendant categories.
+        """
+
+
+class _MicrogridComponentGraph(
+    ComponentGraph
+):  # pylint: disable=too-many-public-methods
     """ComponentGraph implementation designed to work with the microgrid API.
 
     For internal-only use of the `microgrid` package.
@@ -748,6 +782,67 @@ def dfs(
 
         return component
 
+    def find_first_descendant_component(
+        self,
+        *,
+        root_category: ComponentCategory,
+        descendant_categories: Iterable[ComponentCategory],
+    ) -> Component:
+        """Find the first descendant component given root and descendant categories.
+
+        This method searches for the root component within the provided root
+        category. If multiple components share the same root category, the
+        first found one is considered as the root component.
+
+        Subsequently, it looks for the first descendant component from the root
+        component, considering only the immediate descendants.
+
+        The priority of the component to search for is determined by the order
+        of the descendant categories, with the first category having the
+        highest priority.
+
+        Args:
+            root_category: The category of the root component to search for.
+            descendant_categories: The descendant categories to search for the
+                first descendant component in.
+
+        Raises:
+            ValueError: when the root component is not found in the component
+                graph or when no component is found in the given categories.
+
+        Returns:
+            The first descendant component found in the component graph,
+            considering the specified root and descendant categories.
+        """
+        root_component = next(
+            (comp for comp in self.components(component_categories={root_category})),
+            None,
+        )
+
+        if root_component is None:
+            raise ValueError(f"Root component not found for {root_category.name}")
+
+        # Sort by component ID to ensure consistent results.
+        successors = sorted(
+            self.successors(root_component.component_id),
+            key=lambda comp: comp.component_id,
+        )
+
+        def find_component(component_category: ComponentCategory) -> Component | None:
+            return next(
+                (comp for comp in successors if comp.category == component_category),
+                None,
+            )
+
+        # Find the first component that matches the given descendant categories
+        # in the order of the categories list.
+        component = next(filter(None, map(find_component, descendant_categories)), None)
+
+        if component is None:
+            raise ValueError("Component not found in any of the descendant categories.")
+
+        return component
+
     def _validate_graph(self) -> None:
         """Check that the underlying graph data is valid.
 

src/frequenz/sdk/timeseries/_grid_frequency.py

@@ -58,9 +58,20 @@ def __init__(
             channel_registry: The channel registry to use for the grid frequency.
             source: The source component to use to receive the grid frequency.
         """
+        if not source:
+            component_graph = connection_manager.get().component_graph
+            source = component_graph.find_first_descendant_component(
+                root_category=ComponentCategory.GRID,
+                descendant_categories=(
+                    ComponentCategory.METER,
+                    ComponentCategory.INVERTER,
+                    ComponentCategory.EV_CHARGER,
+                ),
+            )
+
         self._request_sender = data_sourcing_request_sender
         self._channel_registry = channel_registry
-        self._source_component = source or GridFrequency.find_frequency_source()
+        self._source_component = source
         self._component_metric_request = create_request(
             self._source_component.component_id
         )
@@ -103,71 +114,3 @@ async def _send_request(self) -> None:
         """Send the request for grid frequency."""
         await self._request_sender.send(self._component_metric_request)
         _logger.debug("Sent request for grid frequency: %s", self._source_component)
-
-    @staticmethod
-    def find_frequency_source() -> Component:
-        """Find the source component that will be used for grid frequency.
-
-        Will use the first meter it can find to gather the frequency.
-        If no meter is available, the first inverter will be used and finally the first EV charger.
-
-        Returns:
-            The component that will be used for grid frequency.
-
-        Raises:
-            ValueError: when the component graph doesn't have a `GRID` component.
-        """
-        component_graph = connection_manager.get().component_graph
-        grid_component = next(
-            (
-                comp
-                for comp in component_graph.components()
-                if comp.category == ComponentCategory.GRID
-            ),
-            None,
-        )
-
-        if grid_component is None:
-            raise ValueError(
-                "Unable to find a GRID component from the component graph."
-            )
-
-        # Sort by component id to ensure consistent results
-        grid_successors = sorted(
-            component_graph.successors(grid_component.component_id),
-            key=lambda comp: comp.component_id,
-        )
-
-        def find_component(component_category: ComponentCategory) -> Component | None:
-            return next(
-                (
-                    comp
-                    for comp in grid_successors
-                    if comp.category == component_category
-                ),
-                None,
-            )
-
-        # Find the first component that is either a meter, inverter or EV charger
-        # with category priority in that order.
-        component = next(
-            filter(
-                None,
-                map(
-                    find_component,
-                    [
-                        ComponentCategory.METER,
-                        ComponentCategory.INVERTER,
-                        ComponentCategory.EV_CHARGER,
-                    ],
-                ),
-            ),
-            None,
-        )
-
-        if component is None:
-            raise ValueError(
-                "Unable to find a METER, INVERTER or EV_CHARGER component from the component graph."
-            )
-
-        return component