Change battery status when component stopped sending data (#207)

Set timer for battery and inverter messages. If no message was received
for some time, check timestamp and update battery status.
Previously the timestamp was checked after received request result.
Because of that BatteryStatus could not work alone (without instance
that sends request result).

Author: ela-kotulska-frequenz

examples/battery_status.py

@@ -0,0 +1,44 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Example how to run BatteryPoolStatus as separate instance.
+
+This is not needed for user but simplifies testing and debugging and understanding
+this feature.
+"""
+
+import asyncio
+import logging
+
+from frequenz.sdk import microgrid
+from frequenz.sdk.actor.power_distributing._battery_pool_status import BatteryPoolStatus
+from frequenz.sdk.microgrid.component import ComponentCategory
+
+_logger = logging.getLogger(__name__)
+HOST = "microgrid.sandbox.api.frequenz.io"  # it should be the host name.
+PORT = 61060
+
+
+async def main() -> None:
+    """Start BatteryPoolStatus to see how it works."""
+    logging.basicConfig(
+        level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s:%(message)s"
+    )
+    await microgrid.initialize(HOST, PORT)
+    batteries = {
+        bat.component_id
+        for bat in microgrid.get().component_graph.components(
+            component_category={ComponentCategory.BATTERY}
+        )
+    }
+
+    batteries_status = BatteryPoolStatus(
+        battery_ids=batteries,
+        max_data_age_sec=5,
+        max_blocking_duration_sec=30,
+    )
+
+    await batteries_status.join()
+
+
+asyncio.run(main())

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

@@ -94,8 +94,8 @@ def __init__(
         self._current_status = BatteryStatus(working=set(), uncertain=set())
 
         # Channel for sending results of requests to the batteries
-        request_result_channel = Broadcast[SetPowerResult]("battery_request_status")
-        self._request_result_sender = request_result_channel.new_sender()
+        set_power_result_channel = Broadcast[SetPowerResult]("battery_request_status")
+        self._set_power_result_sender = set_power_result_channel.new_sender()
 
         self._batteries: Dict[str, BatteryStatusTracker] = {}
 
@@ -112,7 +112,7 @@ def __init__(
                 max_data_age_sec=max_data_age_sec,
                 max_blocking_duration_sec=max_blocking_duration_sec,
                 status_sender=channel.sender,
-                request_result_receiver=request_result_channel.new_receiver(
+                set_power_result_receiver=set_power_result_channel.new_receiver(
                     f"battery_{battery_id}_request_status"
                 ),
             )
@@ -123,6 +123,17 @@ def __init__(
 
         self._task = asyncio.create_task(self._run())
 
+    async def join(self) -> None:
+        """Await for the battery pool, and return when the task completes.
+
+        It will not terminate the program while BatteryPool is working.
+        BatteryPool can be stopped with the `stop` method.
+        This method is not needed in source code, because BatteryPool is owned
+        by the internal code.
+        It is needed only when user needs to run his own instance of the BatteryPool.
+        """
+        await self._task
+
     async def stop(self) -> None:
         """Stop tracking batteries status."""
         await cancel_and_await(self._task)
@@ -178,7 +189,7 @@ async def update_status(
             succeed_batteries: Batteries that succeed request
             failed_batteries: Batteries that failed request
         """
-        await self._request_result_sender.send(
+        await self._set_power_result_sender.send(
             SetPowerResult(succeed_batteries, failed_batteries)
         )
 

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

@@ -16,7 +16,7 @@
 from frequenz.api.microgrid.common_pb2 import ErrorLevel
 from frequenz.api.microgrid.inverter_pb2 import ComponentState as InverterComponentState
 from frequenz.channels import Receiver, Sender
-from frequenz.channels.util import Select
+from frequenz.channels.util import Select, Timer
 
 from frequenz.sdk._internal.asyncio import cancel_and_await
 from frequenz.sdk.microgrid import get as get_microgrid
@@ -62,8 +62,16 @@ class SetPowerResult:
 @dataclass
 class _ComponentStreamStatus:
     component_id: int
+    """Component id."""
+
+    data_recv_timer: Timer
+    """Timer that is set when no component data has been received for some time."""
+
     last_msg_timestamp: datetime = datetime.now(tz=timezone.utc)
+    """Timestamp of the last message from the component."""
+
     last_msg_correct: bool = False
+    """Flag whether last message was correct or not."""
 
 
 @dataclass
@@ -156,7 +164,7 @@ def __init__(  # pylint: disable=too-many-arguments
         max_data_age_sec: float,
         max_blocking_duration_sec: float,
         status_sender: Sender[Status],
-        request_result_receiver: Receiver[SetPowerResult],
+        set_power_result_receiver: Receiver[SetPowerResult],
     ) -> None:
         """Create class instance.
 
@@ -169,7 +177,7 @@ def __init__(  # pylint: disable=too-many-arguments
             max_blocking_duration_sec: This value tell what should be the maximum
                 timeout used for blocking failing component.
             status_sender: Channel to send status updates.
-            request_result_receiver: Channel to receive results of the requests to the
+            set_power_result_receiver: Channel to receive results of the requests to the
                 components.
 
         Raises:
@@ -185,14 +193,18 @@ def __init__(  # pylint: disable=too-many-arguments
         if inverter_id is None:
             raise RuntimeError(f"Can't find inverter adjacent to battery: {battery_id}")
 
-        self._battery = _ComponentStreamStatus(battery_id)
-        self._inverter = _ComponentStreamStatus(inverter_id)
+        self._battery = _ComponentStreamStatus(
+            battery_id, data_recv_timer=Timer(max_data_age_sec)
+        )
+        self._inverter = _ComponentStreamStatus(
+            inverter_id, data_recv_timer=Timer(max_data_age_sec)
+        )
 
         # Select needs receivers that can be get in async way only.
         self._select = None
 
         self._task = asyncio.create_task(
-            self._run(status_sender, request_result_receiver)
+            self._run(status_sender, set_power_result_receiver)
         )
 
     @property
@@ -214,15 +226,15 @@ async def stop(self) -> None:
     async def _run(
         self,
         status_sender: Sender[Status],
-        request_result_receiver: Receiver[SetPowerResult],
+        set_power_result_receiver: Receiver[SetPowerResult],
     ) -> None:
-        """Process data from the components and request_result_receiver.
+        """Process data from the components and set_power_result_receiver.
 
         New status is send only when it change.
 
         Args:
             status_sender: Channel to send status updates.
-            request_result_receiver: Channel to receive results of the requests to the
+            set_power_result_receiver: Channel to receive results of the requests to the
                 components.
         """
         api_client = get_microgrid().api_client
@@ -232,8 +244,10 @@ async def _run(
 
         self._select = Select(
             battery=battery_receiver,
+            battery_timer=self._battery.data_recv_timer,
+            inverter_timer=self._inverter.data_recv_timer,
             inverter=inverter_receiver,
-            request_result=request_result_receiver,
+            set_power_result=set_power_result_receiver,
         )
 
         while True:
@@ -255,6 +269,7 @@ def _update_status(self, select: Select) -> Optional[Status]:
                 and self._no_critical_error(msg.inner)
             )
             self._battery.last_msg_timestamp = msg.inner.timestamp
+            self._battery.data_recv_timer.reset()
 
         elif msg := select.inverter:
             self._inverter.last_msg_correct = (
@@ -263,37 +278,42 @@ def _update_status(self, select: Select) -> Optional[Status]:
                 and self._no_critical_error(msg.inner)
             )
             self._inverter.last_msg_timestamp = msg.inner.timestamp
+            self._inverter.data_recv_timer.reset()
 
-        elif msg := select.request_result:
+        elif msg := select.set_power_result:
             result: SetPowerResult = msg.inner
             if self.battery_id in result.succeed:
                 self._blocking_status.unblock()
 
-            elif self.battery_id in result.failed:
-                # check if component stopped sending data
-                if self._battery.last_msg_correct and self._is_timestamp_outdated(
-                    self._battery.last_msg_timestamp
-                ):
-                    self._battery.last_msg_correct = False
-
-                if self._inverter.last_msg_correct and self._is_timestamp_outdated(
-                    self._inverter.last_msg_timestamp
-                ):
-                    self._inverter.last_msg_correct = False
-
-                component_correct = (
-                    self._battery.last_msg_correct and self._inverter.last_msg_correct
+            elif (
+                self.battery_id in result.failed
+                and self._last_status != Status.NOT_WORKING
+            ):
+                duration = self._blocking_status.block()
+
+                if duration > 0:
+                    _logger.warning(
+                        "battery %d failed last response. block it for %f sec",
+                        self.battery_id,
+                        duration,
+                    )
+        elif msg := select.battery_timer:
+            if self._battery.last_msg_correct:
+                self._battery.last_msg_correct = False
+                _logger.warning(
+                    "Battery %d stopped sending data, last timestamp: %s",
+                    self._battery.component_id,
+                    self._battery.last_msg_timestamp,
                 )
-                if component_correct:
-                    # if component is sending data, but request fails, then block it
-                    # for some time
-                    duration = self._blocking_status.block()
-                    if duration > 0:
-                        _logger.warning(
-                            "battery %d failed last response. block it for %f sec",
-                            self.battery_id,
-                            duration,
-                        )
+        elif msg := select.inverter_timer:
+            if self._inverter.last_msg_correct:
+                self._inverter.last_msg_correct = False
+                _logger.warning(
+                    "Inverter %d stopped sending data, last timestamp: %s",
+                    self._inverter.component_id,
+                    self._inverter.last_msg_timestamp,
+                )
+
         else:
             _logger.error("Unknown message returned from select")