Add option to get batteries status via PowerDistributingActor

Add batteries_status_sender to the PowerDistributingActor.
This sender will send batteries status when it change.

Signed-off-by: ela-kotulska-frequenz <[email protected]>

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

tests/actor/test_battery_pool_status.py

@@ -6,15 +6,20 @@
 from typing import Set
 
 import pytest
+from frequenz.channels import Broadcast
 from pytest_mock import MockerFixture
 
-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
 
 from ..utils.mock_microgrid import MockMicrogridClient
 from .test_battery_status import battery_data, component_graph, inverter_data
 
 
+# pylint: disable=protected-access
 class TestBatteryPoolStatus:
     """Tests for BatteryPoolStatus"""
 
@@ -47,8 +52,11 @@ async def test_batteries_status(self, mock_microgrid: MockMicrogridClient) -> No
                 component_category={ComponentCategory.BATTERY}
             )
         }
+        battery_status_channel = Broadcast[BatteryStatus]("battery_status")
+        battery_status_recv = battery_status_channel.new_receiver(maxsize=1)
         batteries_status = BatteryPoolStatus(
             battery_ids=batteries,
+            battery_status_sender=battery_status_channel.new_sender(),
             max_data_age_sec=5,
             max_blocking_duration_sec=30,
         )
@@ -69,6 +77,8 @@ async def test_batteries_status(self, mock_microgrid: MockMicrogridClient) -> No
         )
         await asyncio.sleep(0.1)
         assert batteries_status.get_working_batteries(batteries) == expected_working
+        msg = await asyncio.wait_for(battery_status_recv.receive(), timeout=0.2)
+        assert msg == batteries_status._current_status
 
         assert await mock_microgrid.send(
             inverter_data(component_id=batteries_list[1] - 1)
@@ -83,6 +93,8 @@ async def test_batteries_status(self, mock_microgrid: MockMicrogridClient) -> No
         expected_working = set(batteries_list)
         await asyncio.sleep(0.1)
         assert batteries_status.get_working_batteries(batteries) == expected_working
+        msg = await asyncio.wait_for(battery_status_recv.receive(), timeout=0.2)
+        assert msg == batteries_status._current_status
 
         await batteries_status.update_status(
             succeed_batteries={106}, failed_batteries={206, 306}