Add FormulaGenerator class and grid_power/battery_power formulas

shsms · shsms · commit 0b1c5aebac10 · 2022-12-16T14:49:58.000+01:00
Signed-off-by: Sahas Subramanian &lt;sahas.subramanian@proton.me&gt;
diff --git a/src/frequenz/sdk/timeseries/logical_meter/_formula_generators/__init__.py b/src/frequenz/sdk/timeseries/logical_meter/_formula_generators/__init__.py
@@ -0,0 +1,14 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Generators for formulas from component graphs."""
+
+from ._battery_power_formula import BatteryPowerFormula
+from ._formula_generator import FormulaGenerator
+from ._grid_power_formula import GridPowerFormula
+
+__all__ = [
+    "GridPowerFormula",
+    "BatteryPowerFormula",
+    "FormulaGenerator",
+]
diff --git a/src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_battery_power_formula.py b/src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_battery_power_formula.py
@@ -0,0 +1,52 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph for Grid Power."""
+
+from .....sdk import microgrid
+from ....microgrid.component import ComponentCategory, ComponentMetricId, InverterType
+from .._formula_engine import FormulaEngine
+from ._formula_generator import FormulaGenerator
+
+
+class BatteryPowerFormula(FormulaGenerator):
+    """Creates a formula engine from the component graph for calculating grid power."""
+
+    async def generate(
+        self,
+    ) -> FormulaEngine:
+        """Make a formula for the cumulative AC battery power of a microgrid.
+
+        The calculation is performed by adding the Active Powers of all the inverters
+        that are attached to batteries.
+
+        If there's no data coming from an inverter, that inverter's power will be
+        treated as 0.
+
+        Returns:
+            A formula engine that will calculate cumulative battery power values.
+
+        Raises:
+            RuntimeError: if there are no batteries in the component graph, or if they
+                don't have an inverter as a predecessor.
+        """
+        builder = self._get_builder(ComponentMetricId.ACTIVE_POWER)
+        component_graph = microgrid.get().component_graph
+        battery_inverters = list(
+            comp
+            for comp in component_graph.components()
+            if comp.category == ComponentCategory.INVERTER
+            and comp.type == InverterType.BATTERY
+        )
+
+        if not battery_inverters:
+            raise RuntimeError(
+                "Unable to find any battery inverters in the component graph."
+            )
+
+        for idx, comp in enumerate(battery_inverters):
+            if idx > 0:
+                builder.push_oper("+")
+            await builder.push_component_metric(comp.component_id, nones_are_zeros=True)
+
+        return builder.build()
diff --git a/src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_formula_generator.py b/src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_formula_generator.py
@@ -0,0 +1,51 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Base class for formula generators that use the component graphs."""
+
+from abc import ABC, abstractmethod
+
+from frequenz.channels import Sender
+
+from ....actor import ChannelRegistry, ComponentMetricRequest
+from ....microgrid.component import ComponentMetricId
+from .._formula_engine import FormulaEngine
+from .._resampled_formula_builder import ResampledFormulaBuilder
+
+
+class FormulaGenerator(ABC):
+    """A class for generating formulas from the component graph."""
+
+    def __init__(
+        self,
+        namespace: str,
+        channel_registry: ChannelRegistry,
+        resampler_subscription_sender: Sender[ComponentMetricRequest],
+    ) -> None:
+        """Create a `FormulaGenerator` instance.
+
+        Args:
+            namespace: A namespace to use with the data-pipeline.
+            channel_registry: A channel registry instance shared with the resampling
+                actor.
+            resampler_subscription_sender: A sender for sending metric requests to the
+                resampling actor.
+        """
+        self._channel_registry = channel_registry
+        self._resampler_subscription_sender = resampler_subscription_sender
+        self._namespace = namespace
+
+    def _get_builder(
+        self, component_metric_id: ComponentMetricId
+    ) -> ResampledFormulaBuilder:
+        builder = ResampledFormulaBuilder(
+            self._namespace,
+            self._channel_registry,
+            self._resampler_subscription_sender,
+            component_metric_id,
+        )
+        return builder
+
+    @abstractmethod
+    async def generate(self) -> FormulaEngine:
+        """Generate a formula engine, based on the component graph."""
diff --git a/src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_grid_power_formula.py b/src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_grid_power_formula.py
@@ -0,0 +1,67 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph for Grid Power."""
+
+from .....sdk import microgrid
+from ....microgrid.component import ComponentCategory, ComponentMetricId
+from .._formula_engine import FormulaEngine
+from ._formula_generator import FormulaGenerator
+
+
+class GridPowerFormula(FormulaGenerator):
+    """Creates a formula engine from the component graph for calculating grid power."""
+
+    async def generate(
+        self,
+    ) -> FormulaEngine:
+        """Generate a formula for calculating grid power from the component graph.
+
+        Returns:
+            A formula engine that will calculate grid power values.
+
+        Raises:
+            RuntimeError: when the component graph doesn't have a `GRID` component.
+        """
+        builder = self._get_builder(ComponentMetricId.ACTIVE_POWER)
+        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 RuntimeError(
+                "Unable to find a GRID component from the component graph."
+            )
+
+        grid_successors = component_graph.successors(grid_component.component_id)
+
+        # generate a formula that just adds values from all commponents that are
+        # directly connected to the grid.
+        for idx, comp in enumerate(grid_successors):
+            if idx > 0:
+                builder.push_oper("+")
+
+            # Ensure the device has an `ACTIVE_POWER` metric.  When inverters
+            # 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 == ComponentCategory.INVERTER:
+                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()
diff --git a/src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py b/src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py
@@ -3,17 +3,21 @@
 
 """A logical meter for calculating high level metrics for a microgrid."""
 
+from __future__ import annotations
+
 import asyncio
 import logging
 import uuid
-from typing import Dict, List
+from typing import Dict, List, Type
 
 from frequenz.channels import Broadcast, Receiver, Sender
 
 from ...actor import ChannelRegistry, ComponentMetricRequest
+from ...microgrid import ComponentGraph
 from ...microgrid.component import ComponentMetricId
 from .. import Sample
 from ._formula_engine import FormulaEngine
+from ._formula_generators import BatteryPowerFormula, FormulaGenerator, GridPowerFormula
 from ._resampled_formula_builder import ResampledFormulaBuilder
 
 logger = logging.Logger(__name__)
@@ -35,6 +39,7 @@ def __init__(
         self,
         channel_registry: ChannelRegistry,
         resampler_subscription_sender: Sender[ComponentMetricRequest],
+        component_graph: ComponentGraph,
     ) -> None:
         """Create a `LogicalMeter instance`.
 
@@ -43,14 +48,15 @@ def __init__(
                 actor.
             resampler_subscription_sender: A sender for sending metric requests to the
                 resampling actor.
+            component_graph: The component graph representing the microgrid.
         """
         self._channel_registry = channel_registry
         self._resampler_subscription_sender = resampler_subscription_sender
 
         # Use a randomly generated uuid to create a unique namespace name for the local
         # meter to use when communicating with the resampling actor.
         self._namespace = f"logical-meter-{uuid.uuid4()}"
-
+        self._component_graph = component_graph
         self._output_channels: Dict[str, Broadcast[Sample]] = {}
         self._tasks: List[asyncio.Task[None]] = []
 
@@ -118,3 +124,49 @@ async def start_formula(
             )
         )
         return out_chan.new_receiver()
+
+    async def _get_formula_stream(
+        self,
+        channel_key: str,
+        generator: Type[FormulaGenerator],
+    ) -> Receiver[Sample]:
+        if channel_key in self._output_channels:
+            return self._output_channels[channel_key].new_receiver()
+
+        formula_engine = await generator(
+            self._namespace, self._channel_registry, self._resampler_subscription_sender
+        ).generate()
+        out_chan = Broadcast[Sample](channel_key)
+        self._output_channels[channel_key] = out_chan
+        self._tasks.append(
+            asyncio.create_task(
+                self._run_formula(formula_engine, out_chan.new_sender())
+            )
+        )
+        return out_chan.new_receiver()
+
+    async def grid_power(self) -> Receiver[Sample]:
+        """Fetch the grid power for the microgrid.
+
+        If a formula engine to calculate grid power is not already running, it
+        will be started.  Else, we'll just get a new receiver to the already
+        existing data stream.
+
+        Returns:
+            A *new* receiver that will stream grid_power values.
+
+        """
+        return await self._get_formula_stream("grid_power", GridPowerFormula)
+
+    async def battery_power(self) -> Receiver[Sample]:
+        """Fetch the cumulative battery power in the microgrid.
+
+        If a formula engine to calculate cumulative battery power is not
+        already running, it will be started.  Else, we'll just get a new
+        receiver to the already existing data stream.
+
+        Returns:
+            A *new* receiver that will stream battery_power values.
+
+        """
+        return await self._get_formula_stream("battery_power", BatteryPowerFormula)
