Add fallback formula feature to ConsumerPowerFormula

Signed-off-by: Elzbieta Kotulska <[email protected]>

Author: ela-kotulska-frequenz

RELEASE_NOTES.md

@@ -14,6 +14,7 @@
   - PVPowerFormula
   - ProducerPowerFormula
   - BatteryPowerFormula
+  - ConsumerPowerFormula
 
 ## Bug Fixes
 

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

@@ -11,11 +11,14 @@
 from ..._quantities import Power
 from .._formula_engine import FormulaEngine
 from .._resampled_formula_builder import ResampledFormulaBuilder
+from ._fallback_formula_metric_fetcher import FallbackFormulaMetricFetcher
 from ._formula_generator import (
     NON_EXISTING_COMPONENT_ID,
     ComponentNotFound,
     FormulaGenerator,
+    FormulaGeneratorConfig,
 )
+from ._simple_power_formula import SimplePowerFormula
 
 _logger = logging.getLogger(__name__)
 
@@ -27,6 +30,25 @@ class ConsumerPowerFormula(FormulaGenerator[Power]):
     are not part of a battery, CHP, PV or EV charger chain.
     """
 
+    def _are_grid_meters(self, grid_successors: set[Component]) -> bool:
+        """Check if the grid successors are grid meters.
+
+        Args:
+            grid_successors: The successors of the grid component.
+
+        Returns:
+            True if the provided components are grid meters, False otherwise.
+        """
+        component_graph = connection_manager.get().component_graph
+        return all(
+            successor.category == ComponentCategory.METER
+            and not component_graph.is_battery_chain(successor)
+            and not component_graph.is_chp_chain(successor)
+            and not component_graph.is_pv_chain(successor)
+            and not component_graph.is_ev_charger_chain(successor)
+            for successor in grid_successors
+        )
+
     def generate(self) -> FormulaEngine[Power]:
         """Generate formula for calculating consumer power from the component graph.
 
@@ -48,15 +70,7 @@ def generate(self) -> FormulaEngine[Power]:
         if not grid_successors:
             raise ComponentNotFound("No components found in the component graph.")
 
-        component_graph = connection_manager.get().component_graph
-        if all(
-            successor.category == ComponentCategory.METER
-            and not component_graph.is_battery_chain(successor)
-            and not component_graph.is_chp_chain(successor)
-            and not component_graph.is_pv_chain(successor)
-            and not component_graph.is_ev_charger_chain(successor)
-            for successor in grid_successors
-        ):
+        if self._are_grid_meters(grid_successors):
             return self._gen_with_grid_meter(builder, grid_successors)
 
         return self._gen_without_grid_meter(builder, self._get_grid_component())
@@ -112,13 +126,30 @@ def non_consumer_component(component: Component) -> bool:
                 grid_meter.component_id, nones_are_zeros=False
             )
 
-        # push all non consumer components and subtract them from the grid meters
-        for component in non_consumer_components:
-            builder.push_oper("-")
-            builder.push_component_metric(
-                component.component_id,
-                nones_are_zeros=component.category != ComponentCategory.METER,
-            )
+        if self._config.allow_fallback:
+            fallbacks = self._get_fallback_formulas(non_consumer_components)
+
+            for idx, (primary_component, fallback_formula) in enumerate(
+                fallbacks.items()
+            ):
+                builder.push_oper("-")
+
+                # should only be the case if the component is not a meter
+                builder.push_component_metric(
+                    primary_component.component_id,
+                    nones_are_zeros=(
+                        primary_component.category != ComponentCategory.METER
+                    ),
+                    fallback=fallback_formula,
+                )
+        else:
+            # push all non consumer components and subtract them from the grid meters
+            for component in non_consumer_components:
+                builder.push_oper("-")
+                builder.push_component_metric(
+                    component.component_id,
+                    nones_are_zeros=component.category != ComponentCategory.METER,
+                )
 
         return builder.build()
 
@@ -175,13 +206,76 @@ def consumer_component(component: Component) -> bool:
             )
             return builder.build()
 
-        for idx, component in enumerate(consumer_components):
-            if idx > 0:
-                builder.push_oper("+")
+        if self._config.allow_fallback:
+            fallbacks = self._get_fallback_formulas(consumer_components)
+
+            for idx, (primary_component, fallback_formula) in enumerate(
+                fallbacks.items()
+            ):
+                if idx > 0:
+                    builder.push_oper("+")
+
+                # should only be the case if the component is not a meter
+                builder.push_component_metric(
+                    primary_component.component_id,
+                    nones_are_zeros=(
+                        primary_component.category != ComponentCategory.METER
+                    ),
+                    fallback=fallback_formula,
+                )
+        else:
+            for idx, component in enumerate(consumer_components):
+                if idx > 0:
+                    builder.push_oper("+")
+
+                builder.push_component_metric(
+                    component.component_id,
+                    nones_are_zeros=component.category != ComponentCategory.METER,
+                )
 
-            builder.push_component_metric(
-                component.component_id,
-                nones_are_zeros=component.category != ComponentCategory.METER,
+        return builder.build()
+
+    def _get_fallback_formulas(
+        self, components: set[Component]
+    ) -> dict[Component, FallbackFormulaMetricFetcher[Power] | None]:
+        """Find primary and fallback components and create fallback formulas.
+
+        The primary component is the one that will be used to calculate the consumer power.
+        However, if it is not available, the fallback formula will be used instead.
+        Fallback formulas calculate the consumer power using the fallback components.
+        Fallback formulas are wrapped in `FallbackFormulaMetricFetcher` to allow
+        for lazy initialization.
+
+        Args:
+            components: The producer components.
+
+        Returns:
+            A dictionary mapping primary components to their FallbackFormulaMetricFetcher.
+        """
+        fallbacks = self._get_metric_fallback_components(components)
+
+        fallback_formulas: dict[
+            Component, FallbackFormulaMetricFetcher[Power] | None
+        ] = {}
+
+        for primary_component, fallback_components in fallbacks.items():
+            if len(fallback_components) == 0:
+                fallback_formulas[primary_component] = None
+                continue
+
+            fallback_ids = [c.component_id for c in fallback_components]
+            generator = SimplePowerFormula(
+                f"{self._namespace}_fallback_{fallback_ids}",
+                self._channel_registry,
+                self._resampler_subscription_sender,
+                FormulaGeneratorConfig(
+                    component_ids=set(fallback_ids),
+                    allow_fallback=False,
+                ),
             )
 
-        return builder.build()
+            fallback_formulas[primary_component] = FallbackFormulaMetricFetcher(
+                generator
+            )
+
+        return fallback_formulas

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

@@ -0,0 +1,67 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph."""
+
+from frequenz.client.microgrid import ComponentCategory, ComponentMetricId
+
+from ....microgrid import connection_manager
+from ..._quantities import Power
+from .._formula_engine import FormulaEngine
+from ._formula_generator import FormulaGenerator
+
+
+class SimplePowerFormula(FormulaGenerator[Power]):
+    """Formula generator from component graph for calculating sum of Power.
+
+    Raises:
+        RuntimeError: If no components are defined in the config or if any
+            component is not found in the component graph.
+    """
+
+    def generate(  # noqa: DOC502
+        # * ComponentNotFound is raised indirectly by _get_grid_component()
+        # * RuntimeError is raised indirectly by connection_manager.get()
+        self,
+    ) -> FormulaEngine[Power]:
+        """Generate formula for calculating producer power from the component graph.
+
+        Returns:
+            A formula engine that will calculate the producer power.
+
+        Raises:
+            ComponentNotFound: If the component graph does not contain a producer power
+                component.
+            RuntimeError: If the grid component has a single successor that is not a
+                meter.
+        """
+        builder = self._get_builder(
+            "simple_power_formula", ComponentMetricId.ACTIVE_POWER, Power.from_watts
+        )
+
+        component_graph = connection_manager.get().component_graph
+        if self._config.component_ids is None:
+            raise RuntimeError("Power formula without component ids is not supported.")
+
+        components = component_graph.components(
+            component_ids=set(self._config.component_ids)
+        )
+
+        not_found_components = self._config.component_ids - {
+            c.component_id for c in components
+        }
+        if not_found_components:
+            raise RuntimeError(
+                f"Unable to find {not_found_components} components in the component graph. ",
+            )
+
+        for idx, component in enumerate(components):
+            if idx > 0:
+                builder.push_oper("+")
+
+            builder.push_component_metric(
+                component.component_id,
+                nones_are_zeros=component.category != ComponentCategory.METER,
+            )
+
+        return builder.build()

tests/timeseries/test_consumer.py

@@ -66,3 +66,134 @@ async def test_consumer_power_no_grid_meter_no_consumer_meter(
             assert (await consumer_power_receiver.receive()).value == Power.from_watts(
                 0.0
             )
+
+    async def test_consumer_power_fallback_formula_with_grid_meter(
+        self, mocker: MockerFixture
+    ) -> None:
+        """Test the consumer power formula with a grid meter."""
+        mockgrid = MockMicrogrid(grid_meter=True, mocker=mocker)
+        mockgrid.add_batteries(1)
+        mockgrid.add_solar_inverters(1)
+        mockgrid.add_solar_inverters(1, no_meter=True)
+
+        # formula is grid_meter - battery - pv1 - pv2
+
+        async with mockgrid, AsyncExitStack() as stack:
+            consumer = microgrid.consumer()
+            stack.push_async_callback(consumer.stop)
+            consumer_power_formula = consumer.power
+            print(consumer_power_formula)
+            consumer_power_receiver = consumer_power_formula.new_receiver()
+
+            # Note: ConsumerPowerFormula has a "nones-are-zero" rule, that says:
+            # * if the meter value is None, it should be treated as None.
+            # * for other components None is treated as 0.
+
+            # fmt: off
+            expected_input_output: list[
+                tuple[list[float | None], list[float | None], list[float | None], Power | None]
+            ] = [
+                # ([grid_meter, bat_meter, pv1_meter], [bat_inv], [pv1_inv, pv2_inv], expected_power) # noqa: E501
+                ([100, 100, -50], [100], [-200, -300], Power.from_watts(350)),
+                ([500, -200, -100], [100], [-200, -100], Power.from_watts(900)),
+                # Case 2: The meter is unavailable (None).
+                # Subscribe to the fallback inverter, but return None as the result,
+                # according to the "nones-are-zero" rule
+                ([500, None, -100], [100], [-200, -100], None),
+                ([500, None, -100], [100], [-200, -100], Power.from_watts(600)),
+                # Case 3: Second meter is unavailable (None).
+                ([500, None, None], [100], [-200, -100], None),
+                ([500, None, None], [100], [-200, -100], Power.from_watts(700)),
+                # Case 3: pv2_inv is unavailable (None).
+                # It has no fallback, so return 0 as its value according to
+                # the "nones-are-zero" rule.
+                ([500, None, None], [100], [-200, None], Power.from_watts(600)),
+                # Case 4: Grid meter is unavailable (None).
+                # It has no fallback, so return None according to the "nones-are-zero" rule.
+                ([None, 100, -50], [100], [-200, -300], None),
+                ([None, 200, -50], [100], [-200, -300], None),
+                ([100, 100, -50], [100], [-200, -300], Power.from_watts(350)),
+                # Case 5: Only grid meter is working
+                ([100, None, None], [None], [None, None], Power.from_watts(100)),
+                ([-500, None, None], [None], [None, None], Power.from_watts(-500)),
+                # Case 6: Nothing is working
+                ([None, None, None], [None], [None, None], None),
+            ]
+            # fmt: on
+
+            for idx, (
+                meter_power,
+                bat_inv_power,
+                pv_inv_power,
+                expected_power,
+            ) in enumerate(expected_input_output):
+                await mockgrid.mock_resampler.send_meter_power(meter_power)
+                await mockgrid.mock_resampler.send_bat_inverter_power(bat_inv_power)
+                await mockgrid.mock_resampler.send_pv_inverter_power(pv_inv_power)
+                mockgrid.mock_resampler.next_ts()
+
+                result = await consumer_power_receiver.receive()
+                assert result.value == expected_power, (
+                    f"Test case {idx} failed:"
+                    + f" meter_power: {meter_power}"
+                    + f" bat_inverter_power {bat_inv_power}"
+                    + f" pv_inverter_power {pv_inv_power}"
+                    + f" expected_power: {expected_power}"
+                    + f" actual_power: {result.value}"
+                )
+
+    async def test_consumer_power_fallback_formula_without_grid_meter(
+        self, mocker: MockerFixture
+    ) -> None:
+        """Test the consumer power formula with a grid meter."""
+        mockgrid = MockMicrogrid(grid_meter=False, mocker=mocker)
+        mockgrid.add_consumer_meters(2)
+        mockgrid.add_batteries(1)
+        mockgrid.add_solar_inverters(1, no_meter=True)
+
+        # formula is sum of consumer meters
+
+        async with mockgrid, AsyncExitStack() as stack:
+            consumer = microgrid.consumer()
+            stack.push_async_callback(consumer.stop)
+            consumer_power_receiver = consumer.power.new_receiver()
+
+            # Note: ConsumerPowerFormula has a "nones-are-zero" rule, that says:
+            # * if the meter value is None, it should be treated as None.
+            # * for other components None is treated as 0.
+
+            # fmt: off
+            expected_input_output: list[
+                tuple[list[float | None], list[float | None], list[float | None], Power | None]
+            ] = [
+                # ([consumer_meter1, consumer_meter2, bat_meter], [bat_inv], [pv_inv], expected_power) # noqa: E501
+                ([100, 100, -50], [100], [-200,], Power.from_watts(200)),
+                ([500, 100, -50], [100], [-200,], Power.from_watts(600)),
+                # One of the meters is invalid - should return None according to none-are-zero rule
+                ([None, 100, -50], [100], [-200,], None),
+                ([None, None, -50], [100], [-200,], None),
+                ([500, None, -50], [100], [-200,], None),
+                ([2000, 1000, None], [None], [None], Power.from_watts(3000)),
+            ]
+            # fmt: on
+
+            for idx, (
+                meter_power,
+                bat_inv_power,
+                pv_inv_power,
+                expected_power,
+            ) in enumerate(expected_input_output):
+                await mockgrid.mock_resampler.send_meter_power(meter_power)
+                await mockgrid.mock_resampler.send_bat_inverter_power(bat_inv_power)
+                await mockgrid.mock_resampler.send_pv_inverter_power(pv_inv_power)
+                mockgrid.mock_resampler.next_ts()
+
+                result = await consumer_power_receiver.receive()
+                assert result.value == expected_power, (
+                    f"Test case {idx} failed:"
+                    + f" meter_power: {meter_power}"
+                    + f" bat_inverter_power {bat_inv_power}"
+                    + f" pv_inverter_power {pv_inv_power}"
+                    + f" expected_power: {expected_power}"
+                    + f" actual_power: {result.value}"
+                )