Replace FormulaEngine with the new Formula

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/sdk/timeseries/battery_pool/_battery_pool.py

@@ -17,14 +17,10 @@
 
 from ... import timeseries
 from ..._internal._channels import MappingReceiverFetcher, ReceiverFetcher
-from ...microgrid import _power_distributing, _power_managing
+from ...microgrid import _power_distributing, _power_managing, connection_manager
 from ...timeseries import Sample
 from .._base_types import SystemBounds
-from ..formula_engine import FormulaEngine
-from ..formula_engine._formula_generators import (
-    BatteryPowerFormula,
-    FormulaGeneratorConfig,
-)
+from ..formulas._formula import Formula
 from ._battery_pool_reference_store import BatteryPoolReferenceStore
 from ._methods import SendOnUpdate
 from ._metric_calculator import (
@@ -195,7 +191,7 @@ def component_ids(self) -> abc.Set[ComponentId]:
         return self._pool_ref_store._batteries
 
     @property
-    def power(self) -> FormulaEngine[Power]:
+    def power(self) -> Formula[Power]:
         """Fetch the total power of the batteries in the pool.
 
         This formula produces values that are in the Passive Sign Convention (PSC).
@@ -210,15 +206,12 @@ def power(self) -> FormulaEngine[Power]:
             A FormulaEngine that will calculate and stream the total power of all
                 batteries in the pool.
         """
-        engine = self._pool_ref_store._formula_pool.from_power_formula_generator(
+        engine = self._pool_ref_store._formula_pool.from_power_formula(
             "battery_pool_power",
-            BatteryPowerFormula,
-            FormulaGeneratorConfig(
-                component_ids=self._pool_ref_store._batteries,
-                allow_fallback=True,
+            connection_manager.get().component_graph.battery_formula(
+                self._pool_ref_store._batteries
             ),
         )
-        assert isinstance(engine, FormulaEngine)
         return engine
 
     @property

src/frequenz/sdk/timeseries/battery_pool/_battery_pool_reference_store.py

@@ -21,7 +21,7 @@
 from ...microgrid._power_distributing import Result
 from ...microgrid._power_distributing._component_status import ComponentPoolStatus
 from ...microgrid._power_managing._base_classes import Proposal, ReportRequest
-from ..formula_engine._formula_engine_pool import FormulaEnginePool
+from ..formulas._formula_pool import FormulaPool
 from ._methods import MetricAggregator
 
 
@@ -117,7 +117,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self._power_dist_results_fetcher: ReceiverFetcher[Result] = (
             power_distribution_results_fetcher
         )
-        self._formula_pool: FormulaEnginePool = FormulaEnginePool(
+        self._formula_pool: FormulaPool = FormulaPool(
             self._namespace,
             self._channel_registry,
             resampler_subscription_sender,

src/frequenz/sdk/timeseries/consumer.py

@@ -9,10 +9,10 @@
 from frequenz.quantities import Power
 
 from .._internal._channels import ChannelRegistry
+from ..microgrid import connection_manager
 from ..microgrid._data_sourcing import ComponentMetricRequest
-from .formula_engine import FormulaEngine
-from .formula_engine._formula_engine_pool import FormulaEnginePool
-from .formula_engine._formula_generators import ConsumerPowerFormula
+from .formulas._formula import Formula
+from .formulas._formula_pool import FormulaPool
 
 
 class Consumer:
@@ -52,7 +52,7 @@ class Consumer:
         ```
     """
 
-    _formula_pool: FormulaEnginePool
+    _formula_pool: FormulaPool
     """The formula engine pool to generate consumer metrics."""
 
     def __init__(
@@ -67,14 +67,14 @@ def __init__(
             resampler_subscription_sender: The sender to use for resampler subscriptions.
         """
         namespace = f"consumer-{uuid.uuid4()}"
-        self._formula_pool = FormulaEnginePool(
+        self._formula_pool = FormulaPool(
             namespace,
             channel_registry,
             resampler_subscription_sender,
         )
 
     @property
-    def power(self) -> FormulaEngine[Power]:
+    def power(self) -> Formula[Power]:
         """Fetch the consumer power for the microgrid.
 
         This formula produces values that are in the Passive Sign Convention (PSC).
@@ -88,12 +88,10 @@ def power(self) -> FormulaEngine[Power]:
         Returns:
             A FormulaEngine that will calculate and stream consumer power.
         """
-        engine = self._formula_pool.from_power_formula_generator(
+        return self._formula_pool.from_power_formula(
             "consumer_power",
-            ConsumerPowerFormula,
+            connection_manager.get().component_graph.consumer_formula(),
         )
-        assert isinstance(engine, FormulaEngine)
-        return engine
 
     async def stop(self) -> None:
         """Stop all formula engines."""

src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py

@@ -12,15 +12,11 @@
 from frequenz.quantities import Current, Power
 
 from ..._internal._channels import MappingReceiverFetcher, ReceiverFetcher
-from ...microgrid import _power_distributing, _power_managing
+from ...microgrid import _power_distributing, _power_managing, connection_manager
 from ...timeseries import Bounds
 from .._base_types import SystemBounds
-from ..formula_engine import FormulaEngine, FormulaEngine3Phase
-from ..formula_engine._formula_generators import (
-    EVChargerCurrentFormula,
-    EVChargerPowerFormula,
-    FormulaGeneratorConfig,
-)
+from ..formulas._formula import Formula
+from ..formulas._formula_3_phase import Formula3Phase
 from ._ev_charger_pool_reference_store import EVChargerPoolReferenceStore
 from ._result_types import EVChargerPoolReport
 
@@ -118,7 +114,7 @@ def component_ids(self) -> abc.Set[ComponentId]:
         return self._pool_ref_store.component_ids
 
     @property
-    def current_per_phase(self) -> FormulaEngine3Phase[Current]:
+    def current_per_phase(self) -> Formula3Phase[Current]:
         """Fetch the total current for the EV Chargers in the pool.
 
         This formula produces values that are in the Passive Sign Convention (PSC).
@@ -133,20 +129,15 @@ def current_per_phase(self) -> FormulaEngine3Phase[Current]:
             A FormulaEngine that will calculate and stream the total current of all EV
                 Chargers.
         """
-        engine = (
-            self._pool_ref_store.formula_pool.from_3_phase_current_formula_generator(
-                "ev_charger_total_current",
-                EVChargerCurrentFormula,
-                FormulaGeneratorConfig(
-                    component_ids=self._pool_ref_store.component_ids
-                ),
-            )
+        return self._pool_ref_store.formula_pool.from_current_3_phase_formula(
+            "ev_charger_total_current",
+            connection_manager.get().component_graph.ev_charger_formula(
+                self._pool_ref_store.component_ids
+            ),
         )
-        assert isinstance(engine, FormulaEngine3Phase)
-        return engine
 
     @property
-    def power(self) -> FormulaEngine[Power]:
+    def power(self) -> Formula[Power]:
         """Fetch the total power for the EV Chargers in the pool.
 
         This formula produces values that are in the Passive Sign Convention (PSC).
@@ -161,15 +152,12 @@ def power(self) -> FormulaEngine[Power]:
             A FormulaEngine that will calculate and stream the total power of all EV
                 Chargers.
         """
-        engine = self._pool_ref_store.formula_pool.from_power_formula_generator(
+        return self._pool_ref_store.formula_pool.from_power_formula(
             "ev_charger_power",
-            EVChargerPowerFormula,
-            FormulaGeneratorConfig(
-                component_ids=self._pool_ref_store.component_ids,
+            connection_manager.get().component_graph.ev_charger_formula(
+                self._pool_ref_store.component_ids
             ),
         )
-        assert isinstance(engine, FormulaEngine)
-        return engine
 
     @property
     def power_status(self) -> ReceiverFetcher[EVChargerPoolReport]:

src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool_reference_store.py

@@ -17,7 +17,7 @@
 from ...microgrid._power_distributing import ComponentPoolStatus, Result
 from ...microgrid._power_managing._base_classes import Proposal, ReportRequest
 from .._base_types import SystemBounds
-from ..formula_engine._formula_engine_pool import FormulaEnginePool
+from ..formulas._formula_pool import FormulaPool
 from ._system_bounds_tracker import EVCSystemBoundsTracker
 
 
@@ -82,7 +82,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self.power_bounds_subs: dict[str, asyncio.Task[None]] = {}
 
         self.namespace: str = f"ev-charger-pool-{uuid.uuid4()}"
-        self.formula_pool = FormulaEnginePool(
+        self.formula_pool = FormulaPool(
             self.namespace,
             self.channel_registry,
             self.resampler_subscription_sender,

src/frequenz/sdk/timeseries/grid.py

@@ -14,21 +14,15 @@
 
 from frequenz.channels import Sender
 from frequenz.client.microgrid.component import GridConnectionPoint
-from frequenz.client.microgrid.metrics import Metric
 from frequenz.quantities import Current, Power, ReactivePower
 
 from .._internal._channels import ChannelRegistry
 from ..microgrid import connection_manager
 from ..microgrid._data_sourcing import ComponentMetricRequest
 from ._fuse import Fuse
-from .formula_engine import FormulaEngine, FormulaEngine3Phase
-from .formula_engine._formula_engine_pool import FormulaEnginePool
-from .formula_engine._formula_generators import (
-    GridCurrentFormula,
-    GridPower3PhaseFormula,
-    GridPowerFormula,
-    GridReactivePowerFormula,
-)
+from .formulas._formula import Formula
+from .formulas._formula_3_phase import Formula3Phase
+from .formulas._formula_pool import FormulaPool
 
 _logger = logging.getLogger(__name__)
 
@@ -72,11 +66,11 @@ class Grid:
     lacks information about the fuse.
     """
 
-    _formula_pool: FormulaEnginePool
+    _formula_pool: FormulaPool
     """The formula engine pool to generate grid metrics."""
 
     @property
-    def power(self) -> FormulaEngine[Power]:
+    def power(self) -> Formula[Power]:
         """Fetch the grid power for the microgrid.
 
         This formula produces values that are in the Passive Sign Convention (PSC).
@@ -90,15 +84,13 @@ def power(self) -> FormulaEngine[Power]:
         Returns:
             A FormulaEngine that will calculate and stream grid power.
         """
-        engine = self._formula_pool.from_power_formula_generator(
+        return self._formula_pool.from_power_formula(
             "grid_power",
-            GridPowerFormula,
+            connection_manager.get().component_graph.grid_formula(),
         )
-        assert isinstance(engine, FormulaEngine)
-        return engine
 
     @property
-    def reactive_power(self) -> FormulaEngine[ReactivePower]:
+    def reactive_power(self) -> Formula[ReactivePower]:
         """Fetch the grid reactive power for the microgrid.
 
         This formula produces values that are in the Passive Sign Convention (PSC).
@@ -112,15 +104,13 @@ def reactive_power(self) -> FormulaEngine[ReactivePower]:
         Returns:
             A FormulaEngine that will calculate and stream grid reactive power.
         """
-        engine = self._formula_pool.from_reactive_power_formula_generator(
-            f"grid-{Metric.AC_REACTIVE_POWER.value}",
-            GridReactivePowerFormula,
+        return self._formula_pool.from_reactive_power_formula(
+            "grid_reactive_power",
+            connection_manager.get().component_graph.grid_formula(),
         )
-        assert isinstance(engine, FormulaEngine)
-        return engine
 
     @property
-    def _power_per_phase(self) -> FormulaEngine3Phase[Power]:
+    def _power_per_phase(self) -> Formula3Phase[Power]:
         """Fetch the per-phase grid power for the microgrid.
 
         This formula produces values that are in the Passive Sign Convention (PSC).
@@ -131,14 +121,13 @@ def _power_per_phase(self) -> FormulaEngine3Phase[Power]:
         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
+        return self._formula_pool.from_power_3_phase_formula(
+            "grid_power_3_phase",
+            connection_manager.get().component_graph.grid_formula(),
         )
-        assert isinstance(engine, FormulaEngine3Phase)
-        return engine
 
     @property
-    def current_per_phase(self) -> FormulaEngine3Phase[Current]:
+    def current_per_phase(self) -> Formula3Phase[Current]:
         """Fetch the per-phase grid current for the microgrid.
 
         This formula produces values that are in the Passive Sign Convention (PSC).
@@ -152,12 +141,10 @@ def current_per_phase(self) -> FormulaEngine3Phase[Current]:
         Returns:
             A FormulaEngine that will calculate and stream grid current.
         """
-        engine = self._formula_pool.from_3_phase_current_formula_generator(
+        return self._formula_pool.from_current_3_phase_formula(
             "grid_current",
-            GridCurrentFormula,
+            connection_manager.get().component_graph.grid_formula(),
         )
-        assert isinstance(engine, FormulaEngine3Phase)
-        return engine
 
     async def stop(self) -> None:
         """Stop all formula engines."""
@@ -215,7 +202,7 @@ def initialize(
             )
 
     namespace = f"grid-{uuid.uuid4()}"
-    formula_pool = FormulaEnginePool(
+    formula_pool = FormulaPool(
         namespace,
         channel_registry,
         resampler_subscription_sender,