Create fallback formulas for the PvPool.power

Change PVPowerFormula to get primary and fallback of the given
components. For the fallback comonents sub formulas are created
This makes sure that both main and fallback formulas have the
same behaviour and less complexity.

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

Author: ela-kotulska-frequenz

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

@@ -49,6 +49,8 @@ class FormulaGeneratorConfig:
     component_ids: abc.Set[int] | None = None
     """The component IDs to use for generating the formula."""
 
+    allow_fallback: bool = True
+
 
 class FormulaGenerator(ABC, Generic[QuantityT]):
     """A class for generating formulas from the component graph."""

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

@@ -5,12 +5,17 @@
 
 import logging
 
-from frequenz.client.microgrid import ComponentCategory, ComponentMetricId
+from frequenz.client.microgrid import Component, ComponentCategory, ComponentMetricId
 
 from ....microgrid import connection_manager
 from ..._quantities import Power
 from .._formula_engine import FormulaEngine
-from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator
+from ._fallback_formula_metric_fetcher import FallbackFormulaMetricFetcher
+from ._formula_generator import (
+    NON_EXISTING_COMPONENT_ID,
+    FormulaGenerator,
+    FormulaGeneratorConfig,
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -58,18 +63,79 @@ def generate(  # noqa: DOC502
             # frequency as the other streams.  So we subscribe with a non-existing
             # component id, just to get a `None` message at the resampling interval.
             builder.push_component_metric(
-                NON_EXISTING_COMPONENT_ID, nones_are_zeros=True
+                NON_EXISTING_COMPONENT_ID,
+                nones_are_zeros=True,
             )
             return builder.build()
 
-        for idx, component in enumerate(pv_components):
-            if idx > 0:
-                builder.push_oper("+")
+        if self._config.allow_fallback:
+            fallbacks = self._get_fallback_formulas(pv_components)
 
-            # should only be the case if the component is not a meter
-            builder.push_component_metric(
-                component.component_id,
-                nones_are_zeros=component.category != ComponentCategory.METER,
-            )
+            for idx, (primary_component, fallback_formula) in enumerate(
+                fallbacks.items()
+            ):
+                if idx > 0:
+                    builder.push_oper("+")
+
+                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(pv_components):
+                if idx > 0:
+                    builder.push_oper("+")
+
+                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 PV power.
+        If it is not available, the fallback formula will be used instead.
+        Fallback formulas calculate the PV power using the fallback components.
+        Fallback formulas are wrapped in `FallbackFormulaMetricFetcher`.
+
+        Args:
+            components: The PV components.
+
+        Returns:
+            A dictionary mapping primary components to their corresponding
+                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 = PVPowerFormula(
+                f"{self._namespace}_fallback_{fallback_ids}",
+                self._channel_registry,
+                self._resampler_subscription_sender,
+                FormulaGeneratorConfig(
+                    component_ids=set(fallback_ids),
+                    allow_fallback=False,
+                ),
+            )
+
+            fallback_formulas[primary_component] = FallbackFormulaMetricFetcher(
+                generator
+            )
+
+        return fallback_formulas