feat(q10): add status update listener API

Author: lboue

roborock/devices/traits/b01/q10/status.py

@@ -1,7 +1,11 @@
 """Status trait for Q10 B01 devices."""
 
 import logging
+from collections.abc import Callable
+from typing import Any
 
+from roborock.callbacks import CallbackList
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.data.b01_q10.b01_q10_containers import Q10Status
 
 from .common import DpsDataConverter
@@ -19,8 +23,23 @@ class StatusTrait(Q10Status):
     New values can be requited through the `Q10PropertiesApi`'s `refresh` method.
     """
 
-    def update_from_dps(self, decoded_dps: dict) -> None:
+    def __init__(self) -> None:
+        """Initialize the status trait."""
+        super().__init__()
+        self._update_callbacks: CallbackList[dict[B01_Q10_DP, Any]] = CallbackList(
+            logger=_LOGGER
+        )
+
+    def add_update_listener(
+        self, callback: Callable[[dict[B01_Q10_DP, Any]], None]
+    ) -> Callable[[], None]:
+        """Register a callback for decoded DPS updates.
+
+        Returns a callable to remove the listener.
+        """
+        return self._update_callbacks.add_callback(callback)
+
+    def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
         """Update the trait from raw DPS data."""
         _CONVERTER.update_from_dps(self, decoded_dps)
-        # In the future we can register listeners and notify them here on update
-        # if `update_from_dps` performed any updates.
+        self._update_callbacks(decoded_dps)

tests/devices/traits/b01/q10/test_status.py

@@ -10,6 +10,7 @@
 import pytest
 
 from roborock.data.b01_q10.b01_q10_code_mappings import (
+    B01_Q10_DP,
     YXDeviceCleanTask,
     YXDeviceState,
     YXFanLevel,
@@ -139,3 +140,22 @@ async def test_status_trait_refresh(
     assert q10_api.status.battery == 100
     assert q10_api.status.status == YXDeviceState.CHARGING_STATE
     assert q10_api.status.fan_level == YXFanLevel.NORMAL
+
+
+def test_status_trait_update_listener(q10_api: Q10PropertiesApi) -> None:
+    """Test that status listeners receive updates and can unsubscribe."""
+    updates: list[dict[B01_Q10_DP, Any]] = []
+
+    unsubscribe = q10_api.status.add_update_listener(updates.append)
+
+    first_update = {B01_Q10_DP.BATTERY: 88}
+    q10_api.status.update_from_dps(first_update)
+
+    assert updates == [first_update]
+
+    unsubscribe()
+
+    second_update = {B01_Q10_DP.BATTERY: 87}
+    q10_api.status.update_from_dps(second_update)
+
+    assert updates == [first_update]