Change battery status when component stopped sending data

Set timer for battery and inverter messages. If no message
was received for some time, then 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).

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

Author: ela-kotulska-frequenz

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
@@ -185,8 +193,12 @@ 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
@@ -232,6 +244,8 @@ 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,
         )
@@ -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:
             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")
 

tests/actor/test_battery_status.py

@@ -190,21 +190,30 @@ class Message(Generic[T]):
 class FakeSelect:
     """Helper class to mock Select object used in BatteryStatusTracker"""
 
-    def __init__(
+    # This is just number of Select instance attributes.
+    def __init__(  # pylint: disable=too-many-arguments
         self,
         battery: Optional[BatteryData] = None,
         inverter: Optional[InverterData] = None,
         request_result: Optional[SetPowerResult] = None,
+        battery_timer_flag: bool = False,
+        inverter_timer_flag: bool = False,
     ) -> None:
         """Create FakeSelect instance
 
         Args:
             battery: Expected battery message. Defaults to None.
             inverter: Expected inverter message. Defaults to None.
+            battery_timer_flag: If true the battery data timer will be set to indicate
+                that no messages have been received for a while.
+            inverter_timer_flag: If true the inverter data timer will be set to indicate
+                that no messages have been received for a while.
             request_result: Expected SetPowerResult message. Defaults to None.
         """
         self.battery = None if battery is None else Message(battery)
         self.inverter = None if inverter is None else Message(inverter)
+        self.inverter_timer = inverter_timer_flag
+        self.battery_timer = battery_timer_flag
         self.request_result = (
             None if request_result is None else Message(request_result)
         )
@@ -536,13 +545,6 @@ async def test_sync_blocking_interrupted_with_with_max_data(
                 assert tracker._update_status(select) is None  # type: ignore[arg-type]
                 time.shift(timeout)
 
-            # Battery message should be to old after 5 seconds.
-            select = FakeSelect(
-                request_result=SetPowerResult(succeed={1}, failed={106})
-            )
-            status = tracker._update_status(select)  # type: ignore[arg-type]
-            assert status is Status.NOT_WORKING
-
             await tracker.stop()
 
     @time_machine.travel("2022-01-01 00:00 UTC", tick=False)
@@ -597,6 +599,69 @@ async def test_sync_blocking_interrupted_with_invalid_message(
 
         await tracker.stop()
 
+    @time_machine.travel("2022-01-01 00:00 UTC", tick=False)
+    async def test_timers(
+        self, mock_microgrid: MockMicrogridClient, mocker: MockerFixture
+    ) -> None:
+        """Test if messages changes battery status/
+
+        Tests uses FakeSelect to test status in sync way.
+        Otherwise we would have lots of async calls and waiting.
+
+        Args:
+            mock_microgrid: mock_microgrid fixture
+            mocker: pytest mocker instance
+        """
+        status_channel = Broadcast[Status]("battery_status")
+        request_result_channel = Broadcast[SetPowerResult]("request_result")
+
+        tracker = BatteryStatusTracker(
+            BATTERY_ID,
+            max_data_age_sec=5,
+            max_blocking_duration_sec=30,
+            status_sender=status_channel.new_sender(),
+            request_result_receiver=request_result_channel.new_receiver(),
+        )
+
+        battery_timer_spy = mocker.spy(tracker._battery.data_recv_timer, "reset")
+        inverter_timer_spy = mocker.spy(tracker._inverter.data_recv_timer, "reset")
+
+        assert tracker.battery_id == BATTERY_ID
+        assert tracker._last_status == Status.NOT_WORKING
+
+        select = FakeSelect(inverter=inverter_data(component_id=INVERTER_ID))
+        assert tracker._update_status(select) is None  # type: ignore[arg-type]
+
+        select = FakeSelect(battery=battery_data(component_id=BATTERY_ID))
+        assert tracker._update_status(select) is Status.WORKING  # type: ignore[arg-type]
+
+        assert battery_timer_spy.call_count == 1
+
+        select = FakeSelect(battery_timer_flag=True)
+        assert tracker._update_status(select) is Status.NOT_WORKING  # type: ignore[arg-type]
+
+        assert battery_timer_spy.call_count == 1
+
+        select = FakeSelect(battery=battery_data(component_id=BATTERY_ID))
+        assert tracker._update_status(select) is Status.WORKING  # type: ignore[arg-type]
+
+        assert battery_timer_spy.call_count == 2
+
+        select = FakeSelect(inverter_timer_flag=True)
+        assert tracker._update_status(select) is Status.NOT_WORKING  # type: ignore[arg-type]
+
+        select = FakeSelect(battery_timer_flag=True)
+        assert tracker._update_status(select) is None  # type: ignore[arg-type]
+
+        select = FakeSelect(battery=battery_data(component_id=BATTERY_ID))
+        assert tracker._update_status(select) is None  # type: ignore[arg-type]
+
+        select = FakeSelect(inverter=inverter_data(component_id=INVERTER_ID))
+        assert tracker._update_status(select) is Status.WORKING  # type: ignore[arg-type]
+
+        assert inverter_timer_spy.call_count == 2
+        await tracker.stop()
+
     @time_machine.travel("2022-01-01 00:00 UTC", tick=False)
     async def test_async_battery_status(
         self, mock_microgrid: MockMicrogridClient