Implement GridReactivePowerFormula (frequenz-floss#1086)

Author: ela-kotulska-frequenz

RELEASE_NOTES.md

@@ -18,6 +18,8 @@ and adapt your imports if you are using these types.
   - `force_polling`: Whether to force file polling to check for changes. Default is `True`.
   - `polling_interval`: The interval to check for changes. Only relevant if polling is enabled. Default is 1 second.
 
+- Add a new method `microgrid.grid().reactive_power` to stream reactive power at the grid connection point.
+
 ## Bug Fixes
 
 - Many long running async tasks including metric streamers in the BatteryPool now have automatic recovery in case of exceptions.

pyproject.toml

@@ -31,7 +31,7 @@ dependencies = [
   # (plugins.mkdocstrings.handlers.python.import)
   "frequenz-client-microgrid >= 0.5.1, < 0.6.0",
   "frequenz-channels >= 1.2.0, < 2.0.0",
-  "frequenz-quantities == 1.0.0rc1",
+  "frequenz-quantities == 1.0.0rc2",
   "networkx >= 2.8, < 4",
   "numpy >= 1.26.4, < 2",
   "typing_extensions >= 4.6.1, < 5",

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/formula_engine/_formula_generators/__init__.py

@@ -17,6 +17,7 @@
 from ._grid_current_formula import GridCurrentFormula
 from ._grid_power_3_phase_formula import GridPower3PhaseFormula
 from ._grid_power_formula import GridPowerFormula
+from ._grid_reactive_power_formula import GridReactivePowerFormula
 from ._producer_power_formula import ProducerPowerFormula
 from ._pv_power_formula import PVPowerFormula
 
@@ -33,6 +34,7 @@
     "ConsumerPowerFormula",
     "GridPower3PhaseFormula",
     "GridPowerFormula",
+    "GridReactivePowerFormula",
     "BatteryPowerFormula",
     "EVChargerPowerFormula",
     "PVPowerFormula",

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

@@ -18,7 +18,7 @@
     FormulaGenerator,
     FormulaGeneratorConfig,
 )
-from ._simple_power_formula import SimplePowerFormula
+from ._simple_formula import SimplePowerFormula
 
 _logger = logging.getLogger(__name__)
 

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

@@ -3,20 +3,18 @@
 
 """Formula generator from component graph for Grid Power."""
 
-from frequenz.client.microgrid import Component, ComponentCategory, ComponentMetricId
+
+from frequenz.client.microgrid import Component, ComponentMetricId
 from frequenz.quantities import Power
 
 from .._formula_engine import FormulaEngine
 from ._fallback_formula_metric_fetcher import FallbackFormulaMetricFetcher
-from ._formula_generator import (
-    ComponentNotFound,
-    FormulaGenerator,
-    FormulaGeneratorConfig,
-)
-from ._simple_power_formula import SimplePowerFormula
+from ._formula_generator import FormulaGeneratorConfig
+from ._grid_power_formula_base import GridPowerFormulaBase
+from ._simple_formula import SimplePowerFormula
 
 
-class GridPowerFormula(FormulaGenerator[Power]):
+class GridPowerFormula(GridPowerFormulaBase[Power]):
     """Creates a formula engine from the component graph for calculating grid power."""
 
     def generate(  # noqa: DOC502
@@ -32,70 +30,18 @@ def generate(  # noqa: DOC502
             ComponentNotFound: when the component graph doesn't have a `GRID` component.
         """
         builder = self._get_builder(
-            "grid-power", ComponentMetricId.ACTIVE_POWER, Power.from_watts
+            "grid-power",
+            ComponentMetricId.ACTIVE_POWER,
+            Power.from_watts,
         )
-        grid_successors = self._get_grid_component_successors()
-
-        components = {
-            c
-            for c in grid_successors
-            if c.category
-            in {
-                ComponentCategory.INVERTER,
-                ComponentCategory.EV_CHARGER,
-                ComponentCategory.METER,
-            }
-        }
-
-        if not components:
-            raise ComponentNotFound("No grid successors found")
-
-        # generate a formula that just adds values from all components that are
-        # directly connected to the grid.  If the requested formula type is
-        # `PASSIVE_SIGN_CONVENTION`, there is nothing more to do.  If the requested
-        # formula type is `PRODUCTION`, the formula output is negated, then clipped to
-        # 0.  If the requested formula type is `CONSUMPTION`, the formula output is
-        # already positive, so it is just clipped to 0.
-        #
-        # So the formulas would look like:
-        #  - `PASSIVE_SIGN_CONVENTION`: `(grid-successor-1 + grid-successor-2 + ...)`
-        #  - `PRODUCTION`: `max(0, -(grid-successor-1 + grid-successor-2 + ...))`
-        #  - `CONSUMPTION`: `max(0, (grid-successor-1 + grid-successor-2 + ...))`
-        if self._config.allow_fallback:
-            fallbacks = self._get_fallback_formulas(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, comp in enumerate(components):
-                if idx > 0:
-                    builder.push_oper("+")
-
-                builder.push_component_metric(
-                    comp.component_id,
-                    nones_are_zeros=(comp.category != ComponentCategory.METER),
-                )
-
-        return builder.build()
+        return self._generate(builder)
 
     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 producer power.
+        The primary component is the one that will be used to calculate the grid power.
         If it is not available, the fallback formula will be used instead.
         Fallback formulas calculate the grid power using the fallback components.
         Fallback formulas are wrapped in `FallbackFormulaMetricFetcher`.

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

@@ -0,0 +1,106 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Base formula generator from component graph for Grid Power."""
+
+from abc import ABC, abstractmethod
+
+from frequenz.client.microgrid import Component, ComponentCategory
+
+from ..._base_types import QuantityT
+from .._formula_engine import FormulaEngine
+from .._resampled_formula_builder import ResampledFormulaBuilder
+from ._fallback_formula_metric_fetcher import FallbackFormulaMetricFetcher
+from ._formula_generator import ComponentNotFound, FormulaGenerator
+
+
+class GridPowerFormulaBase(FormulaGenerator[QuantityT], ABC):
+    """Base class for grid power formula generators."""
+
+    def _generate(
+        self, builder: ResampledFormulaBuilder[QuantityT]
+    ) -> FormulaEngine[QuantityT]:
+        """Generate a formula for calculating grid power from the component graph.
+
+        Args:
+            builder: The builder to use to create the formula.
+
+        Returns:
+            A formula engine that will calculate grid power values.
+
+        Raises:
+            ComponentNotFound: when the component graph doesn't have a `GRID` component.
+        """
+        grid_successors = self._get_grid_component_successors()
+
+        components = {
+            c
+            for c in grid_successors
+            if c.category
+            in {
+                ComponentCategory.INVERTER,
+                ComponentCategory.EV_CHARGER,
+                ComponentCategory.METER,
+            }
+        }
+
+        if not components:
+            raise ComponentNotFound("No grid successors found")
+
+        # generate a formula that just adds values from all components that are
+        # directly connected to the grid.  If the requested formula type is
+        # `PASSIVE_SIGN_CONVENTION`, there is nothing more to do.  If the requested
+        # formula type is `PRODUCTION`, the formula output is negated, then clipped to
+        # 0.  If the requested formula type is `CONSUMPTION`, the formula output is
+        # already positive, so it is just clipped to 0.
+        #
+        # So the formulas would look like:
+        #  - `PASSIVE_SIGN_CONVENTION`: `(grid-successor-1 + grid-successor-2 + ...)`
+        #  - `PRODUCTION`: `max(0, -(grid-successor-1 + grid-successor-2 + ...))`
+        #  - `CONSUMPTION`: `max(0, (grid-successor-1 + grid-successor-2 + ...))`
+        if self._config.allow_fallback:
+            fallbacks = self._get_fallback_formulas(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, comp in enumerate(components):
+                if idx > 0:
+                    builder.push_oper("+")
+
+                builder.push_component_metric(
+                    comp.component_id,
+                    nones_are_zeros=(comp.category != ComponentCategory.METER),
+                )
+
+        return builder.build()
+
+    @abstractmethod
+    def _get_fallback_formulas(
+        self, components: set[Component]
+    ) -> dict[Component, FallbackFormulaMetricFetcher[QuantityT] | None]:
+        """Find primary and fallback components and create fallback formulas.
+
+        The primary component is the one that will be used to calculate the producer power.
+        If it is not available, the fallback formula will be used instead.
+        Fallback formulas calculate the grid power using the fallback components.
+        Fallback formulas are wrapped in `FallbackFormulaMetricFetcher`.
+
+        Args:
+            components: The producer components.
+
+        Returns:
+            A dictionary mapping primary components to their FallbackFormulaMetricFetcher.
+        """

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

@@ -0,0 +1,81 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph for Grid Reactive Power."""
+
+
+from frequenz.client.microgrid import Component, ComponentMetricId
+from frequenz.quantities import ReactivePower
+
+from .._formula_engine import FormulaEngine
+from ._fallback_formula_metric_fetcher import FallbackFormulaMetricFetcher
+from ._formula_generator import FormulaGeneratorConfig
+from ._grid_power_formula_base import GridPowerFormulaBase
+from ._simple_formula import SimpleReactivePowerFormula
+
+
+class GridReactivePowerFormula(GridPowerFormulaBase[ReactivePower]):
+    """Creates a formula engine from the component graph for calculating grid reactive power."""
+
+    def generate(  # noqa: DOC502
+        # * ComponentNotFound is raised indirectly by _get_grid_component_successors
+        self,
+    ) -> FormulaEngine[ReactivePower]:
+        """Generate a formula for calculating grid reactive power from the component graph.
+
+        Returns:
+            A formula engine that will calculate grid reactive power values.
+
+        Raises:
+            ComponentNotFound: when the component graph doesn't have a `GRID` component.
+        """
+        builder = self._get_builder(
+            "grid_reactive_power_formula",
+            ComponentMetricId.REACTIVE_POWER,
+            ReactivePower.from_volt_amperes_reactive,
+        )
+        return self._generate(builder)
+
+    def _get_fallback_formulas(
+        self, components: set[Component]
+    ) -> dict[Component, FallbackFormulaMetricFetcher[ReactivePower] | None]:
+        """Find primary and fallback components and create fallback formulas.
+
+        The primary component is the one that will be used to calculate the grid reactive power.
+        If it is not available, the fallback formula will be used instead.
+        Fallback formulas calculate the grid power using the fallback components.
+        Fallback formulas are wrapped in `FallbackFormulaMetricFetcher`.
+
+        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[ReactivePower] | 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 = SimpleReactivePowerFormula(
+                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