Fetch and stream 3-phase power (#847)

The 3-phase active power is temporary needed to calculate the power
factor of the microgrid. The power factor is not available through [the
microgrid API in the current version used in the
SDK](https://github.com/frequenz-floss/frequenz-api-common/blob/v0.4.0/proto/frequenz/api/common/v1/metrics/electrical.proto#L139-L199).

So the 3-phase active power is temporary exposed through the microgrid
API until the migration to the latest version is
completed. Then the 3-phase power factor can be fetched and streamed
through the microgrid API.

Temporary workaround for #841.

Author: daniel-zullo-frequenz

RELEASE_NOTES.md

@@ -14,4 +14,4 @@
 
 ## Bug Fixes
 
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+- Fix grid current formula generator to add the operator `+` to the engine only when the component category is handled.

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

@@ -28,6 +28,9 @@
 
 _MeterDataMethods: dict[ComponentMetricId, Callable[[MeterData], float]] = {
     ComponentMetricId.ACTIVE_POWER: lambda msg: msg.active_power,
+    ComponentMetricId.ACTIVE_POWER_PHASE_1: lambda msg: msg.active_power_per_phase[0],
+    ComponentMetricId.ACTIVE_POWER_PHASE_2: lambda msg: msg.active_power_per_phase[1],
+    ComponentMetricId.ACTIVE_POWER_PHASE_3: lambda msg: msg.active_power_per_phase[2],
     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],
@@ -59,6 +62,9 @@
 
 _InverterDataMethods: dict[ComponentMetricId, Callable[[InverterData], float]] = {
     ComponentMetricId.ACTIVE_POWER: lambda msg: msg.active_power,
+    ComponentMetricId.ACTIVE_POWER_PHASE_1: lambda msg: msg.active_power_per_phase[0],
+    ComponentMetricId.ACTIVE_POWER_PHASE_2: lambda msg: msg.active_power_per_phase[1],
+    ComponentMetricId.ACTIVE_POWER_PHASE_3: lambda msg: msg.active_power_per_phase[2],
     ComponentMetricId.ACTIVE_POWER_INCLUSION_LOWER_BOUND: lambda msg: (
         msg.active_power_inclusion_lower_bound
     ),
@@ -82,6 +88,9 @@
 
 _EVChargerDataMethods: dict[ComponentMetricId, Callable[[EVChargerData], float]] = {
     ComponentMetricId.ACTIVE_POWER: lambda msg: msg.active_power,
+    ComponentMetricId.ACTIVE_POWER_PHASE_1: lambda msg: msg.active_power_per_phase[0],
+    ComponentMetricId.ACTIVE_POWER_PHASE_2: lambda msg: msg.active_power_per_phase[1],
+    ComponentMetricId.ACTIVE_POWER_PHASE_3: lambda msg: msg.active_power_per_phase[2],
     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],

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

@@ -195,6 +195,13 @@ class ComponentMetricId(Enum):
     ACTIVE_POWER = "active_power"
     """Active power."""
 
+    ACTIVE_POWER_PHASE_1 = "active_power_phase_1"
+    """Active power in phase 1."""
+    ACTIVE_POWER_PHASE_2 = "active_power_phase_2"
+    """Active power in phase 2."""
+    ACTIVE_POWER_PHASE_3 = "active_power_phase_3"
+    """Active power in phase 3."""
+
     CURRENT_PHASE_1 = "current_phase_1"
     """Current in phase 1."""
     CURRENT_PHASE_2 = "current_phase_2"

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

@@ -68,6 +68,9 @@ class MeterData(ComponentData):
             -ve current means supply into the grid.
     """
 
+    active_power_per_phase: tuple[float, float, float]
+    """The AC active power for phase/line 1,2 and 3 respectively."""
+
     current_per_phase: tuple[float, float, float]
     """AC current in Amperes (A) for phase/line 1,2 and 3 respectively.
             +ve current means consumption, away from the grid.
@@ -96,6 +99,11 @@ def from_proto(cls, raw: microgrid_pb.ComponentData) -> MeterData:
             component_id=raw.id,
             timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
             active_power=raw.meter.data.ac.power_active.value,
+            active_power_per_phase=(
+                raw.meter.data.ac.phase_1.power_active.value,
+                raw.meter.data.ac.phase_2.power_active.value,
+                raw.meter.data.ac.phase_3.power_active.value,
+            ),
             current_per_phase=(
                 raw.meter.data.ac.phase_1.current.value,
                 raw.meter.data.ac.phase_2.current.value,
@@ -231,6 +239,9 @@ class InverterData(ComponentData):
             -ve current means supply into the grid.
     """
 
+    active_power_per_phase: tuple[float, float, float]
+    """The AC active power for phase/line 1, 2 and 3 respectively."""
+
     current_per_phase: tuple[float, float, float]
     """AC current in Amperes (A) for phase/line 1, 2 and 3 respectively.
             +ve current means consumption, away from the grid.
@@ -312,6 +323,11 @@ def from_proto(cls, raw: microgrid_pb.ComponentData) -> InverterData:
             component_id=raw.id,
             timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
             active_power=raw.inverter.data.ac.power_active.value,
+            active_power_per_phase=(
+                raw.inverter.data.ac.phase_1.power_active.value,
+                raw.inverter.data.ac.phase_2.power_active.value,
+                raw.inverter.data.ac.phase_3.power_active.value,
+            ),
             current_per_phase=(
                 raw.inverter.data.ac.phase_1.current.value,
                 raw.inverter.data.ac.phase_2.current.value,
@@ -345,6 +361,9 @@ class EVChargerData(ComponentData):
         -ve current means supply into the grid.
     """
 
+    active_power_per_phase: tuple[float, float, float]
+    """The AC active power for phase/line 1,2 and 3 respectively."""
+
     current_per_phase: tuple[float, float, float]
     """AC current in Amperes (A) for phase/line 1,2 and 3 respectively.
         +ve current means consumption, away from the grid.
@@ -424,6 +443,11 @@ def from_proto(cls, raw: microgrid_pb.ComponentData) -> EVChargerData:
             component_id=raw.id,
             timestamp=raw.ts.ToDatetime(tzinfo=timezone.utc),
             active_power=raw_power.value,
+            active_power_per_phase=(
+                raw.ev_charger.data.ac.phase_1.power_active.value,
+                raw.ev_charger.data.ac.phase_2.power_active.value,
+                raw.ev_charger.data.ac.phase_3.power_active.value,
+            ),
             current_per_phase=(
                 raw.ev_charger.data.ac.phase_1.current.value,
                 raw.ev_charger.data.ac.phase_2.current.value,

src/frequenz/sdk/timeseries/formula_engine/_formula_engine_pool.py

@@ -51,6 +51,7 @@ def __init__(
         )
         self._string_engines: dict[str, FormulaEngine[Quantity]] = {}
         self._power_engines: dict[str, FormulaEngine[Power]] = {}
+        self._power_3_phase_engines: dict[str, FormulaEngine3Phase[Power]] = {}
         self._current_engines: dict[str, FormulaEngine3Phase[Current]] = {}
 
     def from_string(
@@ -123,6 +124,40 @@ def from_power_formula_generator(
         self._power_engines[channel_key] = engine
         return engine
 
+    def from_power_3_phase_formula_generator(
+        self,
+        channel_key: str,
+        generator: type[FormulaGenerator[Power]],
+        config: FormulaGeneratorConfig = FormulaGeneratorConfig(),
+    ) -> FormulaEngine3Phase[Power]:
+        """Get a formula engine that streams 3-phase power values.
+
+        Args:
+            channel_key: The string to uniquely identify the formula.
+            generator: The formula generator.
+            config: The config to initialize the formula generator with.
+
+        Returns:
+            A formula engine that streams [3-phase][frequenz.sdk.timeseries.Sample3Phase]
+            power values.
+        """
+        from ._formula_engine import (  # pylint: disable=import-outside-toplevel
+            FormulaEngine3Phase,
+        )
+
+        if channel_key in self._power_3_phase_engines:
+            return self._power_3_phase_engines[channel_key]
+
+        engine = generator(
+            self._namespace,
+            self._channel_registry,
+            self._resampler_subscription_sender,
+            config,
+        ).generate()
+        assert isinstance(engine, FormulaEngine3Phase)
+        self._power_3_phase_engines[channel_key] = engine
+        return engine
+
     def from_3_phase_current_formula_generator(
         self,
         channel_key: str,
@@ -163,5 +198,7 @@ async def stop(self) -> None:
             await string_engine._stop()  # pylint: disable=protected-access
         for power_engine in self._power_engines.values():
             await power_engine._stop()  # pylint: disable=protected-access
+        for power_3_phase_engine in self._power_3_phase_engines.values():
+            await power_3_phase_engine._stop()  # pylint: disable=protected-access
         for current_engine in self._current_engines.values():
             await current_engine._stop()  # pylint: disable=protected-access

src/frequenz/sdk/timeseries/formula_engine/_formula_generators/__init__.py

@@ -15,6 +15,7 @@
     FormulaGeneratorConfig,
 )
 from ._grid_current_formula import GridCurrentFormula
+from ._grid_power_3_phase_formula import GridPower3PhaseFormula
 from ._grid_power_formula import GridPowerFormula
 from ._producer_power_formula import ProducerPowerFormula
 from ._pv_power_formula import PVPowerFormula
@@ -30,6 +31,7 @@
     #
     "CHPPowerFormula",
     "ConsumerPowerFormula",
+    "GridPower3PhaseFormula",
     "GridPowerFormula",
     "BatteryPowerFormula",
     "EVChargerPowerFormula",

src/frequenz/sdk/timeseries/formula_engine/_formula_generators/_grid_current_formula.py

@@ -52,9 +52,6 @@ def _gen_phase_formula(
         # generate a formula that just adds values from all components that are
         # directly connected to the grid.
         for idx, comp in enumerate(grid_successors):
-            if idx > 0:
-                builder.push_oper("+")
-
             # When inverters or ev chargers produce `None` samples, those
             # inverters are excluded from the calculation by treating their
             # `None` values as `0`s.
@@ -71,6 +68,9 @@ def _gen_phase_formula(
             else:
                 continue
 
+            if idx > 0:
+                builder.push_oper("+")
+
             builder.push_component_metric(
                 comp.component_id, nones_are_zeros=nones_are_zeros
             )

src/frequenz/sdk/timeseries/formula_engine/_formula_generators/_grid_power_3_phase_formula.py

@@ -0,0 +1,90 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph for 3-phase Grid Power."""
+
+from ....microgrid.component import Component, ComponentCategory, ComponentMetricId
+from ..._quantities import Power
+from .._formula_engine import FormulaEngine, FormulaEngine3Phase
+from ._formula_generator import FormulaGenerator
+
+
+class GridPower3PhaseFormula(FormulaGenerator[Power]):
+    """Create a formula engine for calculating the grid 3-phase power."""
+
+    def generate(  # noqa: DOC502
+        # ComponentNotFound is raised indirectly by _get_grid_component_successors
+        self,
+    ) -> FormulaEngine3Phase[Power]:
+        """Generate a formula for calculating grid 3-phase power.
+
+        Raises:
+            ComponentNotFound: when the component graph doesn't have a `GRID` component.
+
+        Returns:
+            A formula engine that will calculate grid 3-phase power values.
+        """
+        grid_successors = self._get_grid_component_successors()
+
+        return FormulaEngine3Phase(
+            "grid-power-3-phase",
+            Power.from_watts,
+            (
+                self._gen_phase_formula(
+                    grid_successors, ComponentMetricId.ACTIVE_POWER_PHASE_1
+                ),
+                self._gen_phase_formula(
+                    grid_successors, ComponentMetricId.ACTIVE_POWER_PHASE_2
+                ),
+                self._gen_phase_formula(
+                    grid_successors, ComponentMetricId.ACTIVE_POWER_PHASE_3
+                ),
+            ),
+        )
+
+    def _gen_phase_formula(
+        self,
+        grid_successors: set[Component],
+        metric_id: ComponentMetricId,
+    ) -> FormulaEngine[Power]:
+        """Generate a formula for calculating grid 3-phase power from the component graph.
+
+        Generate a formula that adds values from all components that are directly
+        connected to the grid.
+
+        Args:
+            grid_successors: The set of components that are directly connected to the grid.
+            metric_id: The metric to use for the formula.
+
+        Returns:
+            A formula engine that will calculate grid 3-phase power values.
+        """
+        formula_builder = self._get_builder(
+            "grid-power-3-phase", metric_id, Power.from_watts
+        )
+
+        for idx, comp in enumerate(grid_successors):
+            # When inverters or EV chargers produce `None` samples, they are
+            # excluded from the calculation by treating their `None` values
+            # as `0`s.
+            #
+            # This is not possible for Meters, so when they produce `None`
+            # values, those values get propagated as the output.
+            if comp.category in (
+                ComponentCategory.INVERTER,
+                ComponentCategory.EV_CHARGER,
+            ):
+                nones_are_zeros = True
+            elif comp.category == ComponentCategory.METER:
+                nones_are_zeros = False
+            else:
+                continue
+
+            if idx > 0:
+                formula_builder.push_oper("+")
+
+            formula_builder.push_component_metric(
+                comp.component_id, nones_are_zeros=nones_are_zeros
+            )
+
+        return formula_builder.build()

src/frequenz/sdk/timeseries/grid.py

@@ -21,7 +21,11 @@
 from ._quantities import Current, Power
 from .formula_engine import FormulaEngine, FormulaEngine3Phase
 from .formula_engine._formula_engine_pool import FormulaEnginePool
-from .formula_engine._formula_generators import GridCurrentFormula, GridPowerFormula
+from .formula_engine._formula_generators import (
+    GridCurrentFormula,
+    GridPower3PhaseFormula,
+    GridPowerFormula,
+)
 
 if TYPE_CHECKING:
     # Break circular import
@@ -95,6 +99,24 @@ def power(self) -> FormulaEngine[Power]:
         assert isinstance(engine, FormulaEngine)
         return engine
 
+    @property
+    def _power_3_phase(self) -> FormulaEngine3Phase[Power]:
+        """Fetch the grid 3-phase power for the microgrid.
+
+        This formula produces values that are in the Passive Sign Convention (PSC).
+
+        A receiver from the formula engine can be created using the
+        `new_receiver`method.
+
+        Returns:
+            A FormulaEngine that will calculate and stream grid 3-phase power.
+        """
+        engine = self._formula_pool.from_power_3_phase_formula_generator(
+            "grid_power_3_phase", GridPower3PhaseFormula
+        )
+        assert isinstance(engine, FormulaEngine3Phase)
+        return engine
+
     @property
     def current(self) -> FormulaEngine3Phase[Current]:
         """Fetch the grid current for the microgrid.

tests/microgrid/test_component_data.py

@@ -53,14 +53,17 @@ def test_inverter_data() -> None:
                     phase_1=electrical_pb2.AC.ACPhase(
                         current=metrics_pb2.Metric(value=12.3),
                         voltage=metrics_pb2.Metric(value=229.8),
+                        power_active=metrics_pb2.Metric(value=33.1),
                     ),
                     phase_2=electrical_pb2.AC.ACPhase(
                         current=metrics_pb2.Metric(value=23.4),
                         voltage=metrics_pb2.Metric(value=230.0),
+                        power_active=metrics_pb2.Metric(value=33.3),
                     ),
                     phase_3=electrical_pb2.AC.ACPhase(
                         current=metrics_pb2.Metric(value=34.5),
                         voltage=metrics_pb2.Metric(value=230.2),
+                        power_active=metrics_pb2.Metric(value=33.8),
                     ),
                 ),
             ),
@@ -78,6 +81,7 @@ def test_inverter_data() -> None:
     ]
     assert inv_data.frequency == pytest.approx(50.1)
     assert inv_data.active_power == pytest.approx(100.2)
+    assert inv_data.active_power_per_phase == pytest.approx((33.1, 33.3, 33.8))
     assert inv_data.current_per_phase == pytest.approx((12.3, 23.4, 34.5))
     assert inv_data.voltage_per_phase == pytest.approx((229.8, 230.0, 230.2))
     assert inv_data.active_power_inclusion_lower_bound == pytest.approx(-51_000.0)