diff --git a/roborock/devices/traits/b01/q10/status.py b/roborock/devices/traits/b01/q10/status.py index 7f44a526..329ff104 100644 --- a/roborock/devices/traits/b01/q10/status.py +++ b/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 @@ -16,11 +20,22 @@ class StatusTrait(Q10Status): This is a thin wrapper around Q10Status that provides the Trait interface. The current values reflect the most recently received data from the device. - New values can be requited through the `Q10PropertiesApi`'s `refresh` method. + New values can be requested 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) diff --git a/tests/devices/traits/b01/q10/test_status.py b/tests/devices/traits/b01/q10/test_status.py index b2c56ee9..06f5b196 100644 --- a/tests/devices/traits/b01/q10/test_status.py +++ b/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]