Add formula generators for 3-phase grid and ev charger current

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

Author: shsms

src/frequenz/sdk/timeseries/logical_meter/_formula_generators/__init__.py

@@ -5,11 +5,13 @@
 
 from ._battery_power_formula import BatteryPowerFormula
 from ._battery_soc_formula import BatterySoCFormula
+from ._ev_charger_current_formula import EVChargerCurrentFormula
 from ._formula_generator import (
     ComponentNotFound,
     FormulaGenerationError,
     FormulaGenerator,
 )
+from ._grid_current_formula import GridCurrentFormula
 from ._grid_power_formula import GridPowerFormula
 from ._pv_power_formula import PVPowerFormula
 
@@ -19,13 +21,18 @@
     #
     "FormulaGenerator",
     #
-    # Formula generators
+    # Power Formula generators
     #
     "GridPowerFormula",
     "BatteryPowerFormula",
     "BatterySoCFormula",
     "PVPowerFormula",
     #
+    # Current formula generators
+    #
+    "GridCurrentFormula",
+    "EVChargerCurrentFormula",
+    #
     # Exceptions
     #
     "ComponentNotFound",

src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_ev_charger_current_formula.py

@@ -0,0 +1,90 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph for 3-phase Grid Current."""
+
+import logging
+from typing import List
+
+from .....sdk import microgrid
+from ....microgrid.component import Component, ComponentCategory, ComponentMetricId
+from .._formula_engine import FormulaEngine, FormulaEngine3Phase
+from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator
+
+logger = logging.getLogger(__name__)
+
+
+class EVChargerCurrentFormula(FormulaGenerator):
+    """Create a formula engine from the component graph for calculating grid current."""
+
+    async def generate(self) -> FormulaEngine3Phase:
+        """Generate a formula for calculating total ev current from the component graph.
+
+        Returns:
+            A formula engine that calculates total 3-phase ev charger current values.
+
+        Raises:
+            ComponentNotFound: when the component graph doesn't have a `GRID` component.
+        """
+        component_graph = microgrid.get().component_graph
+        ev_chargers = [
+            comp
+            for comp in component_graph.components()
+            if comp.category == ComponentCategory.EV_CHARGER
+        ]
+
+        if not ev_chargers:
+            logger.warning(
+                "Unable to find any EV Chargers in the component graph. "
+                "Subscribing to the resampling actor with a non-existing "
+                "component id, so that `0` values are sent from the formula."
+            )
+            # If there are no EV Chargers, we have to send 0 values as the same
+            # 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 = self._get_builder("ev-current", ComponentMetricId.ACTIVE_POWER)
+            await builder.push_component_metric(
+                NON_EXISTING_COMPONENT_ID, nones_are_zeros=True
+            )
+            engine = builder.build()
+            return FormulaEngine3Phase(
+                "ev-current",
+                (engine.new_receiver(), engine.new_receiver(), engine.new_receiver()),
+            )
+
+        return FormulaEngine3Phase(
+            "ev-current",
+            (
+                (
+                    await self._gen_phase_formula(
+                        ev_chargers, ComponentMetricId.CURRENT_PHASE_1
+                    )
+                ).new_receiver(),
+                (
+                    await self._gen_phase_formula(
+                        ev_chargers, ComponentMetricId.CURRENT_PHASE_2
+                    )
+                ).new_receiver(),
+                (
+                    await self._gen_phase_formula(
+                        ev_chargers, ComponentMetricId.CURRENT_PHASE_3
+                    )
+                ).new_receiver(),
+            ),
+        )
+
+    async def _gen_phase_formula(
+        self,
+        ev_chargers: List[Component],
+        metric_id: ComponentMetricId,
+    ) -> FormulaEngine:
+        builder = self._get_builder("ev-current", metric_id)
+
+        # generate a formula that just adds values from all ev-chargers.
+        for idx, comp in enumerate(ev_chargers):
+            if idx > 0:
+                builder.push_oper("+")
+
+            await builder.push_component_metric(comp.component_id, nones_are_zeros=True)
+
+        return builder.build()

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

@@ -5,12 +5,13 @@
 
 import sys
 from abc import ABC, abstractmethod
+from typing import Generic
 
 from frequenz.channels import Sender
 
 from ....actor import ChannelRegistry, ComponentMetricRequest
 from ....microgrid.component import ComponentMetricId
-from .._formula_engine import FormulaEngine
+from .._formula_engine import _GenericEngine
 from .._resampled_formula_builder import ResampledFormulaBuilder
 
 
@@ -25,7 +26,7 @@ class ComponentNotFound(FormulaGenerationError):
 NON_EXISTING_COMPONENT_ID = sys.maxsize
 
 
-class FormulaGenerator(ABC):
+class FormulaGenerator(ABC, Generic[_GenericEngine]):
     """A class for generating formulas from the component graph."""
 
     def __init__(
@@ -60,5 +61,5 @@ def _get_builder(
         return builder
 
     @abstractmethod
-    async def generate(self) -> FormulaEngine:
+    async def generate(self) -> _GenericEngine:
         """Generate a formula engine, based on the component graph."""

src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_grid_current_formula.py

@@ -0,0 +1,97 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph for 3-phase Grid Current."""
+
+from typing import Set
+
+from .....sdk import microgrid
+from ....microgrid.component import Component, ComponentCategory, ComponentMetricId
+from .._formula_engine import FormulaEngine, FormulaEngine3Phase
+from ._formula_generator import ComponentNotFound, FormulaGenerator
+
+
+class GridCurrentFormula(FormulaGenerator):
+    """Create a formula engine from the component graph for calculating grid current."""
+
+    async def generate(self) -> FormulaEngine3Phase:
+        """Generate a formula for calculating grid current from the component graph.
+
+        Returns:
+            A formula engine that will calculate 3-phase grid current values.
+
+        Raises:
+            ComponentNotFound: when the component graph doesn't have a `GRID` component.
+        """
+        component_graph = microgrid.get().component_graph
+        grid_component = next(
+            (
+                comp
+                for comp in component_graph.components()
+                if comp.category == ComponentCategory.GRID
+            ),
+            None,
+        )
+
+        if grid_component is None:
+            raise ComponentNotFound(
+                "Unable to find a GRID component from the component graph."
+            )
+
+        grid_successors = component_graph.successors(grid_component.component_id)
+
+        return FormulaEngine3Phase(
+            "grid-current",
+            (
+                (
+                    await self._gen_phase_formula(
+                        grid_successors, ComponentMetricId.CURRENT_PHASE_1
+                    )
+                ).new_receiver(),
+                (
+                    await self._gen_phase_formula(
+                        grid_successors, ComponentMetricId.CURRENT_PHASE_2
+                    )
+                ).new_receiver(),
+                (
+                    await self._gen_phase_formula(
+                        grid_successors, ComponentMetricId.CURRENT_PHASE_3
+                    )
+                ).new_receiver(),
+            ),
+        )
+
+    async def _gen_phase_formula(
+        self,
+        grid_successors: Set[Component],
+        metric_id: ComponentMetricId,
+    ) -> FormulaEngine:
+        builder = self._get_builder("grid-current", metric_id)
+
+        # generate a formula that just adds values from all components that are
+        # directly connected to the grid.
+        for idx, comp in enumerate(grid_successors):
+            if idx > 0:
+                builder.push_oper("+")
+
+            # When inverters or ev chargers produce `None` samples, those
+            # inverters are excluded from the calculation by treating their
+            # `None` values as `0`s.
+            #
+            # This is not possible for Meters, so when they produce `None`
+            # values, those values get propagated as the output.
+            if comp.category in (
+                ComponentCategory.INVERTER,
+                ComponentCategory.EV_CHARGER,
+            ):
+                nones_are_zeros = True
+            elif comp.category == ComponentCategory.METER:
+                nones_are_zeros = False
+            else:
+                continue
+
+            await builder.push_component_metric(
+                comp.component_id, nones_are_zeros=nones_are_zeros
+            )
+
+        return builder.build()