Adapt formula engine

Author: cwasicki

src/frequenz/client/reporting/__init__.py

@@ -6,7 +6,7 @@
 This package provides a low-level interface for interacting with the reporting API.
 """
 
-
+from ._base_types import *
 from ._client import ReportingApiClient
 
 __all__ = ["ReportingApiClient"]

src/frequenz/client/reporting/formula_engine/_formula_engine.py

@@ -16,8 +16,8 @@
 
 from frequenz.channels import Broadcast, Receiver
 
-from ..._internal._asyncio import cancel_and_await
-from .. import Sample, Sample3Phase
+#from ..._internal._asyncio import cancel_and_await
+from .._base_types import Sample, Sample3Phase
 from .._quantities import Quantity, QuantityT
 from ._formula_evaluator import FormulaEvaluator
 from ._formula_formatter import format_formula
@@ -316,6 +316,7 @@ async def _run(self) -> None:
                 _logger.warning(
                     "Formula application failed: %s. Error: %s", self._name, err
                 )
+                raise
             else:
                 await sender.send(msg)
 
@@ -429,7 +430,8 @@ async def _stop(self) -> None:
         """Stop a running formula engine."""
         if self._task is None:
             return
-        await cancel_and_await(self._task)
+        # FIXME
+        #await cancel_and_await(self._task)
 
     def __add__(
         self,

src/frequenz/client/reporting/formula_engine/_formula_generators/__init__.py

@@ -3,21 +3,21 @@
 
 """Generators for formulas from component graphs."""
 
-from ._battery_power_formula import BatteryPowerFormula
-from ._chp_power_formula import CHPPowerFormula
-from ._consumer_power_formula import ConsumerPowerFormula
-from ._ev_charger_current_formula import EVChargerCurrentFormula
-from ._ev_charger_power_formula import EVChargerPowerFormula
+#from ._battery_power_formula import BatteryPowerFormula
+#from ._chp_power_formula import CHPPowerFormula
+#from ._consumer_power_formula import ConsumerPowerFormula
+#from ._ev_charger_current_formula import EVChargerCurrentFormula
+#from ._ev_charger_power_formula import EVChargerPowerFormula
 from ._formula_generator import (
     ComponentNotFound,
     FormulaGenerationError,
     FormulaGenerator,
     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 ._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
 
 __all__ = [
@@ -29,19 +29,19 @@
     #
     # Power Formula generators
     #
-    "CHPPowerFormula",
-    "ConsumerPowerFormula",
-    "GridPower3PhaseFormula",
-    "GridPowerFormula",
-    "BatteryPowerFormula",
-    "EVChargerPowerFormula",
+#    "CHPPowerFormula",
+#    "ConsumerPowerFormula",
+#    "GridPower3PhaseFormula",
+#    "GridPowerFormula",
+#    "BatteryPowerFormula",
+#    "EVChargerPowerFormula",
     "PVPowerFormula",
-    "ProducerPowerFormula",
+#    "ProducerPowerFormula",
     #
     # Current formula generators
     #
-    "GridCurrentFormula",
-    "EVChargerCurrentFormula",
+#    "GridCurrentFormula",
+#    "EVChargerCurrentFormula",
     #
     # Exceptions
     #

src/frequenz/client/reporting/formula_engine/_formula_generators/_formula_generator.py

@@ -13,11 +13,11 @@
 from typing import Generic
 
 from frequenz.channels import Sender
-from frequenz.client.microgrid import Component, ComponentCategory, ComponentMetricId
+from frequenz.client.reporting.component_graph import Component, ComponentCategory, ComponentMetricId
 
-from ...._internal._channels import ChannelRegistry
-from ....microgrid import connection_manager
-from ....microgrid._data_sourcing import ComponentMetricRequest
+#from ...._internal._channels import ChannelRegistry
+#from ....microgrid import connection_manager
+#from ....microgrid._data_sourcing import ComponentMetricRequest
 from ..._quantities import QuantityT
 from .._formula_engine import FormulaEngine, FormulaEngine3Phase
 from .._resampled_formula_builder import ResampledFormulaBuilder
@@ -57,10 +57,12 @@ class FormulaGenerator(ABC, Generic[QuantityT]):
 
     def __init__(
         self,
-        namespace: str,
-        channel_registry: ChannelRegistry,
-        resampler_subscription_sender: Sender[ComponentMetricRequest],
+        get_receiver: Callable[[int, ComponentMetricId], Receiver[Sample[Quantity]]],
+        #namespace: str,
+        #channel_registry: ChannelRegistry,
+        #resampler_subscription_sender: Sender[ComponentMetricRequest],
         config: FormulaGeneratorConfig,
+        component_graph: ComponentGraph,
     ) -> None:
         """Create a `FormulaGenerator` instance.
 
@@ -72,17 +74,19 @@ def __init__(
                 resampling actor.
             config: configs for the formula generator.
         """
-        self._channel_registry: ChannelRegistry = channel_registry
-        self._resampler_subscription_sender: Sender[ComponentMetricRequest] = (
-            resampler_subscription_sender
-        )
-        self._namespace: str = namespace
+        #self._channel_registry: ChannelRegistry = channel_registry
+        #self._resampler_subscription_sender: Sender[ComponentMetricRequest] = (
+        #    resampler_subscription_sender
+        #)
+        #self._namespace: str = namespace
+        self._get_receiver = get_receiver
         self._config: FormulaGeneratorConfig = config
+        self._component_graph: ComponentGraph = component_graph
 
     @property
     def namespace(self) -> str:
         """Get the namespace for the formula generator."""
-        return self._namespace
+        return "bla" #self._namespace
 
     def _get_builder(
         self,
@@ -91,10 +95,11 @@ def _get_builder(
         create_method: Callable[[float], QuantityT],
     ) -> ResampledFormulaBuilder[QuantityT]:
         builder = ResampledFormulaBuilder(
-            self._namespace,
+            #self._namespace,
             name,
-            self._channel_registry,
-            self._resampler_subscription_sender,
+            #self._channel_registry,
+            #self._resampler_subscription_sender,
+            self._get_receiver,
             component_metric_id,
             create_method,
         )
@@ -110,7 +115,8 @@ def _get_grid_component(self) -> Component:
         Raises:
             ComponentNotFound: If the grid component is not found in the component graph.
         """
-        component_graph = connection_manager.get().component_graph
+        #component_graph = connection_manager.get().component_graph
+        component_graph = self._component_graph
         grid_component = next(
             iter(
                 component_graph.components(
@@ -134,7 +140,8 @@ def _get_grid_component_successors(self) -> set[Component]:
             ComponentNotFound: If no successor components are found in the component graph.
         """
         grid_component = self._get_grid_component()
-        component_graph = connection_manager.get().component_graph
+        #component_graph = connection_manager.get().component_graph
+        component_graph = self._component_graph
         grid_successors = component_graph.successors(grid_component.component_id)
 
         if not grid_successors:
@@ -178,7 +185,8 @@ def _get_metric_fallback_components(
                 * The keys are primary components.
                 * The values are sets of fallback components.
         """
-        graph = connection_manager.get().component_graph
+        #graph = connection_manager.get().component_graph
+        graph = self._component_graph
         fallbacks: dict[Component, set[Component]] = {}
 
         for component in components:
@@ -210,7 +218,8 @@ def _get_meter_fallback_components(self, meter: Component) -> set[Component]:
         """
         assert meter.category == ComponentCategory.METER
 
-        graph = connection_manager.get().component_graph
+        #graph = connection_manager.get().component_graph
+        graph = self._component_graph
         successors = graph.successors(meter.component_id)
 
         # All fallbacks has to be of the same type and category.
@@ -241,7 +250,8 @@ def _is_primary_fallback_pair(
         Returns:
             bool: True if the provided components are a primary-fallback pair, False otherwise.
         """
-        graph = connection_manager.get().component_graph
+        #graph = connection_manager.get().component_graph
+        graph = self._component_graph
 
         # reassign to decrease the length of the line and make code readable
         fallback = fallback_candidate

src/frequenz/client/reporting/formula_engine/_formula_generators/_pv_power_formula.py

@@ -5,9 +5,9 @@
 
 import logging
 
-from frequenz.client.microgrid import Component, ComponentCategory, ComponentMetricId
+from frequenz.client.reporting.component_graph import Component, ComponentCategory, ComponentMetricId
 
-from ....microgrid import connection_manager
+#from ....microgrid import connection_manager
 from ..._quantities import Power
 from .._formula_engine import FormulaEngine
 from ._fallback_formula_metric_fetcher import FallbackFormulaMetricFetcher
@@ -27,6 +27,7 @@ def generate(  # noqa: DOC502
         # * ComponentNotFound is raised indirectly by _get_pv_power_components
         # * RuntimeError is also raised indirectly by _get_pv_power_components
         self,
+        #component_graph: "ComponentGraph",
     ) -> FormulaEngine[Power]:
         """Make a formula for the PV power production of a microgrid.
 
@@ -42,7 +43,8 @@ def generate(  # noqa: DOC502
             "pv-power", ComponentMetricId.ACTIVE_POWER, Power.from_watts
         )
 
-        component_graph = connection_manager.get().component_graph
+        #component_graph = connection_manager.get().component_graph
+        component_graph = self._component_graph
         component_ids = self._config.component_ids
         if component_ids:
             pv_components = component_graph.components(set(component_ids))
@@ -125,13 +127,9 @@ def _get_fallback_formulas(
             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,
-                ),
+                self._get_receiver,
+                self._config,
+                self._component_graph,
             )
 
             fallback_formulas[primary_component] = FallbackFormulaMetricFetcher(

src/frequenz/client/reporting/formula_engine/_resampled_formula_builder.py

@@ -8,10 +8,10 @@
 from collections.abc import Callable
 
 from frequenz.channels import Receiver, Sender
-from frequenz.client.microgrid import ComponentMetricId
+from frequenz.client.reporting.component_graph import ComponentMetricId
 
-from ..._internal._channels import ChannelRegistry
-from ...microgrid._data_sourcing import ComponentMetricRequest
+#from ..._internal._channels import ChannelRegistry
+#from ...microgrid._data_sourcing import ComponentMetricRequest
 from .. import Sample
 from .._quantities import Quantity, QuantityT
 from ._formula_engine import FormulaBuilder, FormulaEngine
@@ -24,10 +24,11 @@ class ResampledFormulaBuilder(FormulaBuilder[QuantityT]):
 
     def __init__(  # pylint: disable=too-many-arguments
         self,
-        namespace: str,
+        #namespace: str,
         formula_name: str,
-        channel_registry: ChannelRegistry,
-        resampler_subscription_sender: Sender[ComponentMetricRequest],
+        #channel_registry: ChannelRegistry,
+        #resampler_subscription_sender: Sender[ComponentMetricRequest],
+        get_receiver: Callable[[int, ComponentMetricId], Receiver[Sample[Quantity]]],
         metric_id: ComponentMetricId,
         create_method: Callable[[float], QuantityT],
     ) -> None:
@@ -46,44 +47,18 @@ def __init__(  # pylint: disable=too-many-arguments
                 formula is for generating power values, this would be
                 `Power.from_watts`, for example.
         """
-        self._channel_registry: ChannelRegistry = channel_registry
-        self._resampler_subscription_sender: Sender[ComponentMetricRequest] = (
-            resampler_subscription_sender
-        )
-        self._namespace: str = namespace
+        #self._channel_registry: ChannelRegistry = channel_registry
+        #self._resampler_subscription_sender: Sender[ComponentMetricRequest] = (
+        #    resampler_subscription_sender
+        #)
+        #self._namespace: str = namespace
+        self._get_resampled_receiver: Callable[
+            [int, ComponentMetricId], Receiver[Sample[Quantity]]
+        ] = get_receiver
         self._metric_id: ComponentMetricId = metric_id
-        self._resampler_requests: list[ComponentMetricRequest] = []
+        self._resampler_requests: list = [] # list[ComponentMetricRequest] = []
         super().__init__(formula_name, create_method)  # type: ignore[arg-type]
 
-    def _get_resampled_receiver(
-        self, component_id: int, metric_id: ComponentMetricId
-    ) -> Receiver[Sample[QuantityT]]:
-        """Get a receiver with the resampled data for the given component id.
-
-        Args:
-            component_id: The component id for which to get a resampled data receiver.
-            metric_id: A metric ID to fetch for all components in this formula.
-
-        Returns:
-            A receiver to stream resampled data for the given component id.
-        """
-        request = ComponentMetricRequest(self._namespace, component_id, metric_id, None)
-        self._resampler_requests.append(request)
-        resampled_channel = self._channel_registry.get_or_create(
-            Sample[Quantity], request.get_channel_name()
-        )
-        resampled_receiver = resampled_channel.new_receiver().map(
-            lambda sample: Sample(
-                sample.timestamp,
-                (
-                    self._create_method(sample.value.base_value)
-                    if sample.value is not None
-                    else None
-                ),
-            )
-        )
-        return resampled_receiver
-
     async def subscribe(self) -> None:
         """Subscribe to all resampled component metric streams."""
         for request in self._resampler_requests: