Add a formula generator for SoC in the LogicalMeter

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

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

Author: shsms

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/_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)

tests/timeseries/test_logical_meter.py

@@ -204,3 +204,46 @@ async def test_battery_and_pv_power(  # pylint: disable=too-many-locals
         assert battery_results == battery_inv_sums
         assert len(pv_results) == 10
         assert pv_results == pv_inv_sums
+
+    async def test_soc(self, mocker: MockerFixture) -> None:
+        """Test the soc calculation."""
+        mockgrid = await MockMicrogrid.new(mocker)
+        mockgrid.add_solar_inverters(2)
+        mockgrid._id_increment = 8  # pylint: disable=protected-access
+        mockgrid.add_batteries(3)
+        request_sender, channel_registry = await mockgrid.start()
+        logical_meter = LogicalMeter(
+            channel_registry,
+            request_sender,
+            microgrid.get().component_graph,
+        )
+
+        soc_recv = await logical_meter._soc()  # pylint: disable=protected-access
+
+        bat_receivers = [
+            await self._get_resampled_stream(
+                logical_meter,
+                channel_registry,
+                request_sender,
+                bat_id,
+                ComponentMetricId.SOC,
+            )
+            for bat_id in mockgrid.battery_ids
+        ]
+
+        for ctr in range(10):
+            bat_vals = []
+            for recv in bat_receivers:
+                val = await recv.receive()
+                assert val is not None and val.value is not None
+                bat_vals.append(val.value)
+
+            assert len(bat_vals) == 3
+            # After 7 values, the inverter with component_id > 100 stops sending
+            # data. And the values from the last battery goes out of the calculation.
+            # So we drop it from out control value as well.
+            if ctr >= 7:
+                bat_vals = bat_vals[:2]
+            assert (await soc_recv.receive()).value == sum(bat_vals) / len(bat_vals)
+
+        await mockgrid.cleanup()