Add power formula to BatteryPool (#338)

And remove `battery_power` from the `LogicalMeter`.

Author: sahas-subramanian-frequenz

RELEASE_NOTES.md

@@ -6,6 +6,12 @@
 
 ## Upgrading
 
+* Battery power is no longer available through the `LogicalMeter`, but through the `BatteryPool` (#338)
+
+  ``` python
+  battery_power_receiver = microgrid.battery_pool().power.new_receiver()
+  ```
+
 + Formulas composition has changed (#327) -
   - receivers from formulas are no longer composable.
   - formula composition is now done by composing FormulaEngine instances.
@@ -18,7 +24,7 @@
 
   self._inverter_power = (
       microgrid.logical_meter().pv_power
-      + microgrid.logical_meter().battery_power
+      + microgrid.battery_pool().power
   ).build("inverter_power")
 
   inverter_power_receiver = self._inverter_power.new_receiver()

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -10,12 +10,18 @@
 
 from __future__ import annotations
 
+import logging
 import typing
 from collections import abc
 from dataclasses import dataclass
 
 from frequenz.channels import Bidirectional, Broadcast, Sender
 
+from . import connection_manager
+from .component import ComponentCategory
+
+logger = logging.getLogger(__name__)
+
 # A number of imports had to be done inside functions where they are used, to break
 # import cycles.
 #
@@ -101,7 +107,6 @@ def logical_meter(self) -> LogicalMeter:
         Returns:
             A logical meter instance.
         """
-        from ..microgrid import connection_manager
         from ..timeseries.logical_meter import LogicalMeter
 
         if self._logical_meter is None:
@@ -171,6 +176,8 @@ def battery_pool(
 
         if key not in self._battery_pools:
             self._battery_pools[key] = BatteryPool(
+                channel_registry=self._channel_registry,
+                resampler_subscription_sender=self._resampling_request_sender(),
                 batteries_status_receiver=self._battery_status_channel.new_receiver(
                     maxsize=1
                 ),
@@ -196,6 +203,16 @@ def _start_power_distributing_actor(self) -> None:
         if self._power_distributing_actor:
             return
 
+        component_graph = connection_manager.get().component_graph
+        if not component_graph.components(
+            component_category={ComponentCategory.BATTERY}
+        ):
+            logger.warning(
+                "No batteries found in the component graph. "
+                "The power distributing actor will not be started."
+            )
+            return
+
         from ..actor.power_distributing import PowerDistributingActor
 
         # The PowerDistributingActor is started with only a single default user channel.

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_power_formula.py

@@ -6,9 +6,13 @@
 import logging
 
 from ....microgrid import connection_manager
-from ....microgrid.component import ComponentCategory, ComponentMetricId, InverterType
+from ....microgrid.component import ComponentMetricId
 from ..._formula_engine import FormulaEngine
-from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator
+from ._formula_generator import (
+    NON_EXISTING_COMPONENT_ID,
+    ComponentNotFound,
+    FormulaGenerator,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -37,28 +41,32 @@ def generate(
                 in the component graph.
         """
         builder = self._get_builder("battery-power", ComponentMetricId.ACTIVE_POWER)
-        component_graph = connection_manager.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:
+        component_ids = self._config.component_ids
+        if not component_ids:
             logger.warning(
-                "Unable to find any battery inverters in the component graph. "
+                "No Battery component IDs specified. "
                 "Subscribing to the resampling actor with a non-existing "
                 "component id, so that `0` values are sent from the formula."
             )
-            # If there are no battery inverters, we have to send 0 values as the same
-            # frequency as the other streams.  So we subscribe with a non-existing
+            # If there are no Batteries, 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.push_component_metric(
                 NON_EXISTING_COMPONENT_ID, nones_are_zeros=True
             )
             return builder.build()
 
+        component_graph = connection_manager.get().component_graph
+
+        battery_inverters = list(
+            next(iter(component_graph.predecessors(bat_id))) for bat_id in component_ids
+        )
+
+        if len(component_ids) != len(battery_inverters):
+            raise ComponentNotFound(
+                "Can't find inverters for all batteries from the component graph."
+            )
+
         for idx, comp in enumerate(battery_inverters):
             if idx > 0:
                 builder.push_oper("+")

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

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import logging
+from collections import abc
 
 from ....microgrid.component import ComponentMetricId
 from .._formula_engine import FormulaEngine, FormulaEngine3Phase
@@ -67,7 +68,7 @@ def generate(self) -> FormulaEngine3Phase:
 
     def _gen_phase_formula(
         self,
-        component_ids: set[int],
+        component_ids: abc.Set[int],
         metric_id: ComponentMetricId,
     ) -> FormulaEngine:
         builder = self._get_builder("ev-current", metric_id)

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

@@ -7,6 +7,7 @@
 
 import sys
 from abc import ABC, abstractmethod
+from collections import abc
 from dataclasses import dataclass
 
 from frequenz.channels import Sender
@@ -32,7 +33,7 @@ class ComponentNotFound(FormulaGenerationError):
 class FormulaGeneratorConfig:
     """Config for formula generators."""
 
-    component_ids: set[int] | None = None
+    component_ids: abc.Set[int] | None = None
 
 
 class FormulaGenerator(ABC):

src/frequenz/sdk/timeseries/battery_pool/battery_pool.py

@@ -6,17 +6,24 @@
 from __future__ import annotations
 
 import asyncio
+import uuid
 from collections.abc import Set
 from datetime import timedelta
 from typing import Any
 
-from frequenz.channels import Receiver
+from frequenz.channels import Receiver, Sender
 
 from ..._internal._constants import RECEIVER_MAX_SIZE
 from ..._internal.asyncio import cancel_and_await
+from ...actor import ChannelRegistry, ComponentMetricRequest
 from ...actor.power_distributing._battery_pool_status import BatteryStatus
 from ...microgrid import connection_manager
 from ...microgrid.component import ComponentCategory
+from .._formula_engine import FormulaEngine, FormulaEnginePool
+from .._formula_engine._formula_generators import (
+    BatteryPowerFormula,
+    FormulaGeneratorConfig,
+)
 from ._methods import AggregateMethod, SendOnUpdate
 from ._metric_calculator import CapacityCalculator, PowerBoundsCalculator, SoCCalculator
 from ._result_types import CapacityMetrics, PowerMetrics, SoCMetrics
@@ -29,15 +36,21 @@ class BatteryPool:
     fetching high level metrics for this subset.
     """
 
-    def __init__(
+    def __init__(  # pylint: disable=too-many-arguments
         self,
+        channel_registry: ChannelRegistry,
+        resampler_subscription_sender: Sender[ComponentMetricRequest],
         batteries_status_receiver: Receiver[BatteryStatus],
         min_update_interval: timedelta,
         batteries_id: Set[int] | None = None,
     ) -> None:
         """Create the class instance.
 
         Args:
+            channel_registry: A channel registry instance shared with the resampling
+                actor.
+            resampler_subscription_sender: A sender for sending metric requests to the
+                resampling actor.
             batteries_status_receiver: Receiver to receive status of the batteries.
                 Receivers should has maxsize = 1 to fetch only the latest status.
                 Battery status channel should has resend_latest = True.
@@ -64,13 +77,21 @@ def __init__(
 
         self._working_batteries: set[int] = set()
 
-        self._update_battery_status_task = asyncio.create_task(
-            self._update_battery_status(batteries_status_receiver)
-        )
+        if self._batteries:
+            self._update_battery_status_task = asyncio.create_task(
+                self._update_battery_status(batteries_status_receiver)
+            )
 
         self._min_update_interval = min_update_interval
         self._active_methods: dict[str, AggregateMethod[Any]] = {}
 
+        self._namespace: str = f"battery-pool-{self._batteries}-{uuid.uuid4()}"
+        self._formula_pool: FormulaEnginePool = FormulaEnginePool(
+            self._namespace,
+            channel_registry,
+            resampler_subscription_sender,
+        )
+
     @property
     def battery_ids(self) -> Set[int]:
         """Return ids of the batteries in the pool.
@@ -80,6 +101,26 @@ def battery_ids(self) -> Set[int]:
         """
         return self._batteries
 
+    @property
+    def power(self) -> FormulaEngine:
+        """Fetch the total power of the batteries in the pool.
+
+        If a formula engine to calculate this metric is not already running, it will be
+        started.
+
+        A receiver from the formula engine can be obtained by calling the `new_receiver`
+        method.
+
+        Returns:
+            A FormulaEngine that will calculate and stream the total power of all
+                batteries in the pool.
+        """
+        return self._formula_pool.from_generator(
+            "battery_pool_power",
+            BatteryPowerFormula,
+            FormulaGeneratorConfig(component_ids=self._batteries),
+        )  # type: ignore[return-value]
+
     async def soc(
         self, maxsize: int | None = RECEIVER_MAX_SIZE
     ) -> Receiver[SoCMetrics | None]:

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

@@ -15,7 +15,6 @@
 from ...microgrid.component import ComponentMetricId
 from .._formula_engine import FormulaEngine, FormulaEngine3Phase, FormulaEnginePool
 from .._formula_engine._formula_generators import (
-    BatteryPowerFormula,
     GridCurrentFormula,
     GridPowerFormula,
     PVPowerFormula,
@@ -176,24 +175,6 @@ def grid_current(self) -> FormulaEngine3Phase:
             GridCurrentFormula,
         )  # type: ignore[return-value]
 
-    @property
-    def battery_power(self) -> FormulaEngine:
-        """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.
-
-        A receiver from the formula engine can be created using the `new_receiver`
-        method.
-
-        Returns:
-            A FormulaEngine that will calculate and stream battery power.
-        """
-        return self._formula_pool.from_generator(
-            "battery_power",
-            BatteryPowerFormula,
-        )  # type: ignore[return-value]
-
     @property
     def pv_power(self) -> FormulaEngine:
         """Fetch the PV power production in the microgrid.

tests/timeseries/_formula_engine/test_formula_composition.py

@@ -28,18 +28,17 @@ async def test_formula_composition(  # pylint: disable=too-many-locals
         mockgrid.add_solar_inverters(2)
         await mockgrid.start(mocker)
         logical_meter = microgrid.logical_meter()
-
+        battery_pool = microgrid.battery_pool()
         main_meter_recv = get_resampled_stream(
+            logical_meter._namespace,  # pylint: disable=protected-access
             4,
             ComponentMetricId.ACTIVE_POWER,
         )
         grid_power_recv = logical_meter.grid_power.new_receiver()
-        battery_power_recv = logical_meter.battery_power.new_receiver()
+        battery_power_recv = battery_pool.power.new_receiver()
         pv_power_recv = logical_meter.pv_power.new_receiver()
 
-        engine = (logical_meter.pv_power + logical_meter.battery_power).build(
-            "inv_power"
-        )
+        engine = (logical_meter.pv_power + battery_pool.power).build("inv_power")
         inv_calc_recv = engine.new_receiver()
 
         count = 0
@@ -69,13 +68,12 @@ async def test_formula_composition_missing_pv(self, mocker: MockerFixture) -> No
         mockgrid = MockMicrogrid(grid_side_meter=False)
         mockgrid.add_batteries(3)
         await mockgrid.start(mocker)
+        battery_pool = microgrid.battery_pool()
         logical_meter = microgrid.logical_meter()
 
-        battery_power_recv = logical_meter.battery_power.new_receiver()
+        battery_power_recv = battery_pool.power.new_receiver()
         pv_power_recv = logical_meter.pv_power.new_receiver()
-        engine = (logical_meter.pv_power + logical_meter.battery_power).build(
-            "inv_power"
-        )
+        engine = (logical_meter.pv_power + battery_pool.power).build("inv_power")
         inv_calc_recv = engine.new_receiver()
 
         count = 0
@@ -98,13 +96,12 @@ async def test_formula_composition_missing_bat(self, mocker: MockerFixture) -> N
         mockgrid = MockMicrogrid(grid_side_meter=False)
         mockgrid.add_solar_inverters(2)
         await mockgrid.start(mocker)
+        battery_pool = microgrid.battery_pool()
         logical_meter = microgrid.logical_meter()
 
-        battery_power_recv = logical_meter.battery_power.new_receiver()
+        battery_power_recv = battery_pool.power.new_receiver()
         pv_power_recv = logical_meter.pv_power.new_receiver()
-        engine = (logical_meter.pv_power + logical_meter.battery_power).build(
-            "inv_power"
-        )
+        engine = (logical_meter.pv_power + battery_pool.power).build("inv_power")
         inv_calc_recv = engine.new_receiver()
 
         count = 0

tests/timeseries/_formula_engine/utils.py

@@ -18,6 +18,7 @@
 
 
 def get_resampled_stream(  # pylint: disable=too-many-arguments
+    namespace: str,
     comp_id: int,
     metric_id: ComponentMetricId,
 ) -> Receiver[Sample]:
@@ -27,7 +28,7 @@ def get_resampled_stream(  # pylint: disable=too-many-arguments
 
     # pylint: disable=protected-access
     builder = ResampledFormulaBuilder(
-        _data_pipeline._get().logical_meter()._namespace,
+        namespace,
         "",
         _data_pipeline._get()._channel_registry,
         _data_pipeline._get()._resampling_request_sender(),

tests/timeseries/test_ev_charger_pool.py

@@ -93,6 +93,7 @@ async def test_ev_power(  # pylint: disable=too-many-locals
         ev_pool = microgrid.ev_charger_pool()
 
         main_meter_recv = get_resampled_stream(
+            logical_meter._namespace,  # pylint: disable=protected-access
             mockgrid.main_meter_id,
             ComponentMetricId.ACTIVE_POWER,
         )