Add BatteryPool implementation for aggregating battery-inverter data (#205)

BatteryPool implementation for aggregating battery-inverter metrics into
higher level metrics

Author: ela-kotulska-frequenz

benchmarks/power_distribution/power_distributor.py

@@ -10,10 +10,11 @@
 from dataclasses import dataclass
 from typing import Any, Coroutine, Dict, List, Set  # pylint: disable=unused-import
 
-from frequenz.channels import Bidirectional
+from frequenz.channels import Bidirectional, Broadcast
 
 from frequenz.sdk import microgrid
 from frequenz.sdk.actor.power_distributing import (
+    BatteryStatus,
     Error,
     Ignored,
     OutOfBound,
@@ -123,7 +124,11 @@ async def run_test(  # pylint: disable=too-many-locals
         user_id: channel.service_handle for user_id, channel in channels.items()
     }
 
-    distributor = PowerDistributingActor(service_channels)
+    battery_status_channel = Broadcast[BatteryStatus]("battery-status")
+
+    distributor = PowerDistributingActor(
+        service_channels, battery_status_sender=battery_status_channel.new_sender()
+    )
 
     tasks: List[Coroutine[Any, Any, List[Result]]] = []
     for user_id, channel in channels.items():

examples/battery_pool.py

@@ -0,0 +1,92 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Script with an example how to use BatteryPool."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import timedelta
+from typing import Any, Dict
+
+from frequenz.channels import Broadcast, Receiver
+from frequenz.channels.util import MergeNamed
+
+from frequenz.sdk import microgrid
+from frequenz.sdk.actor import (
+    ChannelRegistry,
+    ComponentMetricRequest,
+    DataSourcingActor,
+)
+from frequenz.sdk.actor.power_distributing import PowerDistributingActor
+from frequenz.sdk.actor.power_distributing._battery_pool_status import BatteryStatus
+from frequenz.sdk.microgrid.component import ComponentCategory
+from frequenz.sdk.timeseries.battery_pool import BatteryPool
+
+HOST = "microgrid.sandbox.api.frequenz.io"  # it should be the host name.
+PORT = 61060
+
+
+def create_battery_pool() -> BatteryPool:
+    """Create battery pool.
+
+    It needs many instance to be created before.
+
+    Returns:
+        BatteryPool instance ready to use.
+    """
+    channel_registry = ChannelRegistry(name="data-registry")
+
+    # Create a channels for sending/receiving subscription requests
+    data_source_request_channel = Broadcast[ComponentMetricRequest](
+        "data-source", resend_latest=True
+    )
+
+    # Instantiate a data sourcing actor
+    _ = DataSourcingActor(
+        request_receiver=data_source_request_channel.new_receiver(
+            "data_sourcing_receiver"
+        ),
+        registry=channel_registry,
+    )
+
+    battery_status_channel = Broadcast[BatteryStatus]("batteries-status")
+    _ = PowerDistributingActor(
+        users_channels={},
+        battery_status_sender=battery_status_channel.new_sender(),
+    )
+
+    batteries = microgrid.get().component_graph.components(
+        component_category={ComponentCategory.BATTERY}
+    )
+
+    return BatteryPool(
+        batteries_id=set(battery.component_id for battery in batteries),
+        batteries_status_receiver=battery_status_channel.new_receiver(
+            name="battery_pool", maxsize=1
+        ),
+        min_update_interval=timedelta(seconds=0.2),
+    )
+
+
+async def main() -> None:
+    """Create the battery pool, activate all formulas and listen for any update."""
+    logging.basicConfig(
+        level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s:%(message)s"
+    )
+    await microgrid.initialize(host=HOST, port=PORT)
+
+    battery_pool = create_battery_pool()
+    receivers: Dict[str, Receiver[Any]] = {
+        "soc": await battery_pool.soc(maxsize=1),
+        "capacity": await battery_pool.capacity(maxsize=1),
+        "power_bounds": await battery_pool.power_bounds(maxsize=1),
+    }
+
+    merged_channel = MergeNamed[Any](**receivers)
+    async for metric_name, metric in merged_channel:
+        print(f"Received new {metric_name}: {metric}")
+
+
+asyncio.run(main())

examples/battery_status.py

@@ -10,8 +10,13 @@
 import asyncio
 import logging
 
+from frequenz.channels import Broadcast
+
 from frequenz.sdk import microgrid
-from frequenz.sdk.actor.power_distributing._battery_pool_status import BatteryPoolStatus
+from frequenz.sdk.actor.power_distributing._battery_pool_status import (
+    BatteryPoolStatus,
+    BatteryStatus,
+)
 from frequenz.sdk.microgrid.component import ComponentCategory
 
 _logger = logging.getLogger(__name__)
@@ -31,13 +36,19 @@ async def main() -> None:
             component_category={ComponentCategory.BATTERY}
         )
     }
+    battery_status_channel = Broadcast[BatteryStatus]("battery-status-channel")
 
     batteries_status = BatteryPoolStatus(
+        battery_status_sender=battery_status_channel.new_sender(),
         battery_ids=batteries,
         max_data_age_sec=5,
         max_blocking_duration_sec=30,
     )
 
+    battery_status_receiver = battery_status_channel.new_receiver()
+    async for status in battery_status_receiver:
+        print(f"Received new battery status {status}")
+
     await batteries_status.join()
 
 

examples/power_distribution.py

@@ -27,6 +27,7 @@
     ResamplerConfig,
 )
 from frequenz.sdk.actor.power_distributing import (
+    BatteryStatus,
     PowerDistributingActor,
     Request,
     Result,
@@ -194,11 +195,14 @@ async def run() -> None:
         )
     }
 
+    battery_status_channel = Broadcast[BatteryStatus]("battery-status")
+
     power_distributor = PowerDistributingActor(
         users_channels={
             key: channel.service_handle
             for key, channel in power_distributor_channels.items()
         },
+        battery_status_sender=battery_status_channel.new_sender(),
     )
 
     # Channel to communicate between actors.

src/frequenz/sdk/_internal/_constants.py

@@ -0,0 +1,21 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Module with constants shared between instances of the sdk.
+
+To be replaced by ConfigManager.
+"""
+
+RECEIVER_MAX_SIZE = 50
+"""Default buffer size of the receiver."""
+
+WAIT_FOR_COMPONENT_DATA_SEC: float = 2
+"""Delay the start of the application to wait for the data."""
+
+MAX_BATTERY_DATA_AGE_SEC: float = 2
+"""Max time difference for the battery or inverter data to be considered as reliable.
+
+If battery or inverter stopped sending data, then this is the maximum time when its
+last message should be considered as valid. After that time, component data
+should not be used.
+"""

src/frequenz/sdk/actor/power_distributing/__init__.py

@@ -10,6 +10,7 @@
 PowerDistributingActor and send requests for charging or discharging power.
 """
 
+from ._battery_pool_status import BatteryStatus
 from .power_distributing import PowerDistributingActor
 from .request import Request
 from .result import Error, Ignored, OutOfBound, PartialFailure, Result, Success
@@ -23,4 +24,5 @@
     "Ignored",
     "OutOfBound",
     "PartialFailure",
+    "BatteryStatus",
 ]

src/frequenz/sdk/actor/power_distributing/_battery_pool_status.py

@@ -9,7 +9,7 @@
 from dataclasses import dataclass
 from typing import Dict, Set
 
-from frequenz.channels import Broadcast, Receiver
+from frequenz.channels import Broadcast, Receiver, Sender
 from frequenz.channels.util import MergeNamed
 
 from ..._internal.asyncio import cancel_and_await
@@ -72,13 +72,16 @@ class BatteryPoolStatus:
     def __init__(
         self,
         battery_ids: Set[int],
+        battery_status_sender: Sender[BatteryStatus],
         max_data_age_sec: float,
         max_blocking_duration_sec: float,
     ) -> None:
         """Create BatteryPoolStatus instance.
 
         Args:
             battery_ids: set of batteries ids that should be stored in pool.
+            battery_status_sender: The sender used for sending the status of the
+                batteries in the pool.
             max_data_age_sec: If component stopped sending data, then
                 this is the maximum time when its last message should be considered as
                 valid. After that time, component won't be used until it starts sending
@@ -121,7 +124,7 @@ def __init__(
             **receivers,
         )
 
-        self._task = asyncio.create_task(self._run())
+        self._task = asyncio.create_task(self._run(battery_status_sender))
 
     async def join(self) -> None:
         """Await for the battery pool, and return when the task completes.
@@ -146,23 +149,30 @@ async def stop(self) -> None:
         )
         await self._battery_status_channel.stop()
 
-    async def _run(self) -> None:
-        """Start tracking batteries status."""
+    async def _run(self, battery_status_sender: Sender[BatteryStatus]) -> None:
+        """Start tracking batteries status.
+
+        Args:
+            battery_status_sender: The sender used for sending the status of the
+                batteries in the pool.
+        """
         while True:
             try:
-                await self._update_status(self._battery_status_channel)
+                await self._update_status(battery_status_sender)
             except Exception as err:  # pylint: disable=broad-except
                 _logger.error(
                     "BatteryPoolStatus failed with error: %s. Restarting.", err
                 )
 
-    async def _update_status(self, status_channel: MergeNamed[Status]) -> None:
+    async def _update_status(
+        self, battery_status_sender: Sender[BatteryStatus]
+    ) -> None:
         """Wait for any battery to change status and update status.
 
         Args:
-            status_channel: Receivers packed in Select object.
+            battery_status_sender: Sender to send the current status of the batteries.
         """
-        async for channel_name, status in status_channel:
+        async for channel_name, status in self._battery_status_channel:
             battery_id = self._batteries[channel_name].battery_id
             if status == Status.WORKING:
                 self._current_status.working.add(battery_id)
@@ -174,7 +184,7 @@ async def _update_status(self, status_channel: MergeNamed[Status]) -> None:
                 self._current_status.working.discard(battery_id)
                 self._current_status.uncertain.discard(battery_id)
 
-            # In the future here we should send status to the subscribed actors
+            await battery_status_sender.send(self._current_status)
 
     async def update_status(
         self, succeed_batteries: Set[int], failed_batteries: Set[int]

src/frequenz/sdk/actor/power_distributing/power_distributing.py

@@ -29,7 +29,7 @@
 )
 
 import grpc
-from frequenz.channels import Bidirectional, Peekable, Receiver
+from frequenz.channels import Bidirectional, Peekable, Receiver, Sender
 from google.protobuf.empty_pb2 import Empty  # pylint: disable=no-name-in-module
 
 from ... import microgrid
@@ -44,7 +44,7 @@
     InverterData,
 )
 from ...power import DistributionAlgorithm, DistributionResult, InvBatPair
-from ._battery_pool_status import BatteryPoolStatus
+from ._battery_pool_status import BatteryPoolStatus, BatteryStatus
 from .request import Request
 from .result import Error, Ignored, OutOfBound, PartialFailure, Result, Success
 
@@ -143,13 +143,16 @@ class PowerDistributingActor:
     def __init__(
         self,
         users_channels: Dict[str, Bidirectional.Handle[Result, Request]],
+        battery_status_sender: Sender[BatteryStatus],
         wait_for_data_sec: float = 2,
     ) -> None:
         """Create class instance.
 
         Args:
             users_channels: BidirectionalHandle for each user. Key should be
                 user id and value should be BidirectionalHandle.
+            battery_status_sender: Channel for sending information which batteries are
+                working.
             wait_for_data_sec: How long actor should wait before processing first
                 request. It is a time needed to collect first components data.
         """
@@ -185,6 +188,7 @@ def __init__(
 
         self._all_battery_status = BatteryPoolStatus(
             battery_ids=set(self._bat_inv_map.keys()),
+            battery_status_sender=battery_status_sender,
             max_blocking_duration_sec=30.0,
             max_data_age_sec=10.0,
         )