Add a formula generator for SoC in the LogicalMeter (#137)

This formula calculates and streams the average SoC of the active
batteries.

This is temporarily part of the `LogicalMeter` and will be moved to the
`BatteryPool` within the next few releases.

Author: sahas-subramanian-frequenz

src/frequenz/sdk/timeseries/logical_meter/_exceptions.py

@@ -0,0 +1,8 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Formula Engine Exceptions."""
+
+
+class FormulaEngineError(Exception):
+    """An error occured while fetching metrics or applying the formula on them."""

src/frequenz/sdk/timeseries/logical_meter/_formula_engine.py

@@ -7,13 +7,14 @@
 
 import asyncio
 from datetime import datetime
-from typing import Dict, List, Optional, Set
+from typing import Dict, List, Optional, Set, Tuple
 
 from frequenz.channels import Receiver
 
 from .. import Sample
 from ._formula_steps import (
     Adder,
+    Averager,
     Divider,
     FormulaStep,
     MetricFetcher,
@@ -218,6 +219,20 @@ def push_metric(
         )
         self._steps.append(fetcher)
 
+    def push_average(self, metrics: List[Tuple[str, Receiver[Sample], bool]]) -> None:
+        """Push an average calculator into the engine.
+
+        Args:
+            metrics: list of arguments to pass to each `MetricFetcher`.
+        """
+        fetchers: List[MetricFetcher] = []
+        for metric in metrics:
+            fetcher = self._metric_fetchers.setdefault(
+                metric[0], MetricFetcher(*metric)
+            )
+            fetchers.append(fetcher)
+        self._steps.append(Averager(fetchers))
+
     def build(self) -> FormulaEngine:
         """Finalize and build the formula engine.
 

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

@@ -4,6 +4,7 @@
 """Generators for formulas from component graphs."""
 
 from ._battery_power_formula import BatteryPowerFormula
+from ._battery_soc_formula import BatterySoCFormula
 from ._formula_generator import (
     ComponentNotFound,
     FormulaGenerationError,
@@ -22,6 +23,7 @@
     #
     "GridPowerFormula",
     "BatteryPowerFormula",
+    "BatterySoCFormula",
     "PVPowerFormula",
     #
     # Exceptions

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

@@ -0,0 +1,102 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Formula generator from component graph for Grid Power."""
+
+import asyncio
+
+from frequenz.channels import Receiver
+
+from .....sdk import microgrid
+from ....microgrid.component import ComponentCategory, ComponentMetricId, InverterType
+from ... import Sample
+from .._formula_engine import FormulaEngine
+from ._formula_generator import (
+    ComponentNotFound,
+    FormulaGenerationError,
+    FormulaGenerator,
+)
+
+
+class _ActiveBatteryReceiver(Receiver[Sample]):
+    """Returns a Sample from a battery, only if the attached inverter is active."""
+
+    def __init__(self, inv_recv: Receiver[Sample], bat_recv: Receiver[Sample]):
+        self._inv_recv = inv_recv
+        self._bat_recv = bat_recv
+
+    async def ready(self) -> None:
+        """Wait until the next Sample is ready."""
+        await asyncio.gather(self._inv_recv.ready(), self._bat_recv.ready())
+
+    def consume(self) -> Sample:
+        """Return the next Sample.
+
+        Returns:
+            the next Sample.
+        """
+        inv = self._inv_recv.consume()
+        bat = self._bat_recv.consume()
+        if inv.value is None:
+            return inv
+        return bat
+
+
+class BatterySoCFormula(FormulaGenerator):
+    """Creates a formula engine from the component graph for calculating battery soc."""
+
+    async def generate(
+        self,
+    ) -> FormulaEngine:
+        """Make a formula for the average battery soc of a microgrid.
+
+        If there's no data coming from an inverter or a battery, the corresponding
+        battery will be excluded from the calculation.
+
+        Returns:
+            A formula engine that will calculate average battery soc values.
+
+        Raises:
+            ComponentNotFound: if there are no batteries in the component graph, or if
+                they don't have an inverter as a predecessor.
+            FormulaGenerationError: If a battery has a non-inverter predecessor
+                in the component graph.
+        """
+        builder = self._get_builder(ComponentMetricId.ACTIVE_POWER)
+        component_graph = microgrid.get().component_graph
+        inv_bat_pairs = {
+            comp: component_graph.successors(comp.component_id)
+            for comp in component_graph.components()
+            if comp.category == ComponentCategory.INVERTER
+            and comp.type == InverterType.BATTERY
+        }
+
+        if not inv_bat_pairs:
+            raise ComponentNotFound(
+                "Unable to find any battery inverters in the component graph."
+            )
+
+        soc_streams = []
+        for inv, bats in inv_bat_pairs.items():
+            bat = list(bats)[0]
+            if len(bats) != 1:
+                raise FormulaGenerationError(
+                    f"Expected exactly one battery for inverter {inv}, got {bats}"
+                )
+
+            # pylint: disable=protected-access
+            soc_recv = _ActiveBatteryReceiver(
+                await builder._get_resampled_receiver(
+                    inv.component_id, ComponentMetricId.ACTIVE_POWER
+                ),
+                await builder._get_resampled_receiver(
+                    bat.component_id, ComponentMetricId.SOC
+                ),
+            )
+            # pylint: enable=protected-access
+
+            soc_streams.append((f"{bat.component_id}", soc_recv, False))
+
+        builder.push_average(soc_streams)
+
+        return builder.build()

src/frequenz/sdk/timeseries/logical_meter/_formula_steps.py

@@ -3,12 +3,15 @@
 
 """Steps for building formula engines with."""
 
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
 from typing import List, Optional
 
 from frequenz.channels import Receiver
 
 from .. import Sample
+from ._exceptions import FormulaEngineError
 
 
 class FormulaStep(ABC):
@@ -157,6 +160,54 @@ def apply(self, _: List[Optional[float]]) -> None:
         """No-op."""
 
 
+class Averager(FormulaStep):
+    """A formula step for calculating average."""
+
+    def __init__(self, fetchers: List[MetricFetcher]) -> None:
+        """Create an `Averager` instance.
+
+        Args:
+            fetchers: MetricFetchers for the metrics that need to be averaged.
+        """
+        self._fetchers = fetchers
+
+    def __repr__(self) -> str:
+        """Return a string representation of the step.
+
+        Returns:
+            A string representation of the step.
+        """
+        return f"avg({', '.join(repr(f) for f in self._fetchers)})"
+
+    def apply(self, eval_stack: List[Optional[float]]) -> None:
+        """Calculate average of given metrics, push the average to the eval_stack.
+
+        Args:
+            eval_stack: An evaluation stack, to append the calculated average to.
+
+        Raises:
+            FormulaEngineError: when metric fetchers are unable to fetch values.
+        """
+        value_count = 0
+        total = 0.0
+        for fetcher in self._fetchers:
+            next_val = fetcher.value
+            if next_val is None:
+                raise FormulaEngineError(
+                    "Unable to fetch a value from the resampling actor."
+                )
+            if next_val.value is None:
+                continue
+            value_count += 1
+            total += next_val.value
+        if value_count == 0:
+            avg = 0.0
+        else:
+            avg = total / value_count
+
+        eval_stack.append(avg)
+
+
 class MetricFetcher(FormulaStep):
     """A formula step for fetching a value from a metric Receiver."""
 
@@ -186,6 +237,15 @@ async def fetch_next(self) -> Optional[Sample]:
         self._next_value = await self._stream.receive()
         return self._next_value
 
+    @property
+    def value(self) -> Optional[Sample]:
+        """Get the next value in the stream.
+
+        Returns:
+            Next value in the stream.
+        """
+        return self._next_value
+
     def __repr__(self) -> str:
         """Return a string representation of the step.
 

src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py

@@ -19,6 +19,7 @@
 from ._formula_engine import FormulaEngine
 from ._formula_generators import (
     BatteryPowerFormula,
+    BatterySoCFormula,
     FormulaGenerator,
     GridPowerFormula,
     PVPowerFormula,
@@ -187,3 +188,14 @@ async def pv_power(self) -> Receiver[Sample]:
             A *new* receiver that will stream PV power production values.
         """
         return await self._get_formula_stream("pv_power", PVPowerFormula)
+
+    async def _soc(self) -> Receiver[Sample]:
+        """Fetch the SoC of the active batteries in the microgrid.
+
+        NOTE: This method is part of the logical meter only temporarily, and will get
+        moved to the `BatteryPool` within the next few releases.
+
+        Returns:
+            A *new* receiver that will stream average SoC of active batteries.
+        """
+        return await self._get_formula_stream("soc", BatterySoCFormula)

src/frequenz/sdk/timeseries/logical_meter/_resampled_formula_builder.py

@@ -40,21 +40,19 @@ def __init__(
         self._metric_id = metric_id
         super().__init__()
 
-    async def _get_resampled_receiver(self, component_id: int) -> Receiver[Sample]:
+    async def _get_resampled_receiver(
+        self, component_id: int, metric_id: ComponentMetricId
+    ) -> Receiver[Sample]:
         """Get a receiver with the resampled data for the given component id.
 
-        This receiver would contain data for the `metric_id` specified when creating the
-        `ResampledFormulaBuilder` instance.
-
         Args:
             component_id: The component id for which to get a resampled data receiver.
+            metric_id: A metric ID to fetch for all components in this formula.
 
         Returns:
             A receiver to stream resampled data for the given component id.
         """
-        request = ComponentMetricRequest(
-            self._namespace, component_id, self._metric_id, None
-        )
+        request = ComponentMetricRequest(self._namespace, component_id, metric_id, None)
         await self._resampler_subscription_sender.send(request)
         return self._channel_registry.new_receiver(request.get_channel_name())
 
@@ -68,7 +66,7 @@ async def push_component_metric(
             nones_are_zeros: Whether to treat None values from the stream as 0s.  If
                 False, the returned value will be a None.
         """
-        receiver = await self._get_resampled_receiver(component_id)
+        receiver = await self._get_resampled_receiver(component_id, self._metric_id)
         self.push_metric(f"#{component_id}", receiver, nones_are_zeros)
 
     async def from_string(

tests/timeseries/mock_microgrid.py

@@ -9,7 +9,9 @@
 from typing import Iterator, Tuple
 
 from frequenz.api.microgrid import microgrid_pb2
-from frequenz.api.microgrid.common_pb2 import AC, Metric
+from frequenz.api.microgrid.battery_pb2 import Battery
+from frequenz.api.microgrid.battery_pb2 import Data as BatteryData
+from frequenz.api.microgrid.common_pb2 import AC, Metric, MetricAggregation
 from frequenz.api.microgrid.inverter_pb2 import Data as InverterData
 from frequenz.api.microgrid.inverter_pb2 import Inverter
 from frequenz.api.microgrid.inverter_pb2 import Type as InverterType
@@ -120,19 +122,36 @@ def inverter_msg(value: float) -> ComponentData:
                     ),
                 )
 
+            def battery_msg(value: float) -> ComponentData:
+                timestamp = Timestamp()
+                timestamp.GetCurrentTime()
+                return ComponentData(
+                    id=request.id,
+                    ts=timestamp,
+                    battery=Battery(data=BatteryData(soc=MetricAggregation(avg=value))),
+                )
+
             if request.id % 10 == cls.inverter_id_suffix:
                 next_msg = inverter_msg
             elif (
                 request.id % 10 == cls.meter_id_suffix
                 or request.id == cls.main_meter_id
             ):
                 next_msg = meter_msg
+            elif request.id % 10 == cls.battery_id_suffix:
+                next_msg = battery_msg
             else:
                 raise RuntimeError(
                     f"Component id {request.id} unsupported by MockMicrogrid"
                 )
             for value in range(1, 10):
-                yield next_msg(value=value + request.id)
+                # for inverters with component_id > 100, send only half the messages.
+                if request.id % 10 == cls.inverter_id_suffix:
+                    if request.id < 100 or value <= 5:
+                        yield next_msg(value=value + request.id)
+                else:
+                    yield next_msg(value=value + request.id)
+
                 time.sleep(0.1)
 
         mocker.patch.object(servicer, "GetComponentData", get_component_data)