Add support for reactive formulas in formula_engine_pool

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

Author: ela-kotulska-frequenz

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

@@ -9,7 +9,7 @@
 
 from frequenz.channels import Sender
 from frequenz.client.microgrid import ComponentMetricId
-from frequenz.quantities import Current, Power, Quantity
+from frequenz.quantities import Current, Power, Quantity, ReactivePower
 
 from ..._internal._channels import ChannelRegistry
 from ...microgrid._data_sourcing import ComponentMetricRequest
@@ -54,6 +54,7 @@ def __init__(
         self._power_engines: dict[str, FormulaEngine[Power]] = {}
         self._power_3_phase_engines: dict[str, FormulaEngine3Phase[Power]] = {}
         self._current_engines: dict[str, FormulaEngine3Phase[Current]] = {}
+        self._reactive_power_engines: dict[str, FormulaEngine[ReactivePower]] = {}
 
     def from_string(
         self,
@@ -91,6 +92,40 @@ def from_string(
 
         return formula_engine
 
+    def from_reactive_power_formula_generator(
+        self,
+        channel_key: str,
+        generator: type[FormulaGenerator[ReactivePower]],
+        config: FormulaGeneratorConfig = FormulaGeneratorConfig(),
+    ) -> FormulaEngine[ReactivePower]:
+        """Get a receiver for a formula from a generator.
+
+        Args:
+            channel_key: A string to uniquely identify the formula.
+            generator: A formula generator.
+            config: config to initialize the formula generator with.
+
+        Returns:
+            A FormulaReceiver or a FormulaReceiver3Phase instance based on what the
+                FormulaGenerator returns.
+        """
+        from ._formula_engine import (  # pylint: disable=import-outside-toplevel
+            FormulaEngine,
+        )
+
+        if channel_key in self._reactive_power_engines:
+            return self._reactive_power_engines[channel_key]
+
+        engine = generator(
+            self._namespace,
+            self._channel_registry,
+            self._resampler_subscription_sender,
+            config,
+        ).generate()
+        assert isinstance(engine, FormulaEngine)
+        self._reactive_power_engines[channel_key] = engine
+        return engine
+
     def from_power_formula_generator(
         self,
         channel_key: str,
@@ -203,3 +238,5 @@ async def stop(self) -> None:
             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
+        for reactive_power_engine in self._reactive_power_engines.values():
+            await reactive_power_engine._stop()  # pylint: disable=protected-access

src/frequenz/sdk/timeseries/grid.py

@@ -14,7 +14,7 @@
 
 from frequenz.channels import Sender
 from frequenz.client.microgrid._component import ComponentCategory, ComponentMetricId
-from frequenz.quantities import Current, Power
+from frequenz.quantities import Current, Power, ReactivePower
 
 from .._internal._channels import ChannelRegistry
 from ..microgrid import connection_manager
@@ -97,7 +97,7 @@ def power(self) -> FormulaEngine[Power]:
         return engine
 
     @property
-    def reactive_power(self) -> FormulaEngine[Power]:
+    def reactive_power(self) -> FormulaEngine[ReactivePower]:
         """Fetch the grid reactive power for the microgrid.
 
         This formula produces values that are in the Passive Sign Convention (PSC).
@@ -111,7 +111,7 @@ def reactive_power(self) -> FormulaEngine[Power]:
         Returns:
             A FormulaEngine that will calculate and stream grid reactive power.
         """
-        engine = self._formula_pool.from_power_formula_generator(
+        engine = self._formula_pool.from_reactive_power_formula_generator(
             f"grid-{ComponentMetricId.REACTIVE_POWER.value}",
             GridReactivePowerFormula,
         )