diff --git a/roborock/devices/device.py b/roborock/devices/device.py index f6226f07..b7132b61 100644 --- a/roborock/devices/device.py +++ b/roborock/devices/device.py @@ -202,6 +202,8 @@ async def connect(self) -> None: await self.v1_properties.start() elif self.b01_q10_properties is not None: await self.b01_q10_properties.start() + elif self.b01_q7_properties is not None: + await self.b01_q7_properties.start() except RoborockException: # Expected: start() can fail transiently. Unsubscribe before propagating # so the retry by connect_loop() gets a clean channel. @@ -230,6 +232,8 @@ async def close(self) -> None: self.v1_properties.close() if self.b01_q10_properties is not None: await self.b01_q10_properties.close() + if self.b01_q7_properties is not None: + await self.b01_q7_properties.close() if self._unsub: self._unsub() self._unsub = None diff --git a/roborock/devices/rpc/b01_q7_channel.py b/roborock/devices/rpc/b01_q7_channel.py index e2bc7b12..d01257b3 100644 --- a/roborock/devices/rpc/b01_q7_channel.py +++ b/roborock/devices/rpc/b01_q7_channel.py @@ -53,6 +53,10 @@ async def send_map_command( """Send a map command and get decoded bytes.""" ... + async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Callable[[], None]: + """Subscribe to unsolicited map pushes, invoking callback with decoded SCMap bytes.""" + ... + def _matches_map_response(response_message: RoborockMessage, *, version: bytes | None) -> bytes | None: """Return raw map payload bytes for matching MAP_RESPONSE messages.""" @@ -208,6 +212,25 @@ async def send_map_command( return decode_map_payload(raw_payload, map_key=self._map_key) + async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Callable[[], None]: + """Subscribe to unsolicited ``MAP_RESPONSE`` pushes. + + The device streams full SCMap frames on its own during cleaning; the + callback receives the decoded (inflated) SCMap bytes for each frame. + """ + + def on_message(message: RoborockMessage) -> None: + if (raw_payload := _matches_map_response(message, version=B01_VERSION)) is None: + return + try: + decoded = decode_map_payload(raw_payload, map_key=self._map_key) + except RoborockException as ex: + _LOGGER.debug("Failed to decode pushed B01 map payload: %s", ex) + return + callback(decoded) + + return await self._mqtt_channel.subscribe(on_message) + def create_b01_q7_channel( device: HomeDataDevice, diff --git a/roborock/devices/traits/b01/q7/__init__.py b/roborock/devices/traits/b01/q7/__init__.py index 35a29144..c19735b8 100644 --- a/roborock/devices/traits/b01/q7/__init__.py +++ b/roborock/devices/traits/b01/q7/__init__.py @@ -3,6 +3,7 @@ Potentially other devices may fall into this category in the future. """ +from collections.abc import Callable from typing import Any from roborock import B01Props @@ -71,6 +72,19 @@ def __init__( self._map_rpc_channel, self.map, ) + self._unsub_map_pushes: Callable[[], None] | None = None + + async def start(self) -> None: + """Start listening for unsolicited map pushes from the device.""" + if self._unsub_map_pushes is not None: + return + self._unsub_map_pushes = await self._map_rpc_channel.subscribe_map_pushes(self.map_content.update_from_push) + + async def close(self) -> None: + """Stop listening for unsolicited map pushes.""" + if self._unsub_map_pushes is not None: + self._unsub_map_pushes() + self._unsub_map_pushes = None async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None: """Query the device for the values of the given Q7 properties.""" diff --git a/roborock/devices/traits/b01/q7/map_content.py b/roborock/devices/traits/b01/q7/map_content.py index 0becf91a..d316ad4f 100644 --- a/roborock/devices/traits/b01/q7/map_content.py +++ b/roborock/devices/traits/b01/q7/map_content.py @@ -9,6 +9,7 @@ """ import asyncio +import logging from dataclasses import dataclass from vacuum_map_parser_base.map_data import MapData @@ -16,12 +17,14 @@ from roborock.data import RoborockBase from roborock.devices.rpc.b01_q7_channel import Q7MapRpcChannel from roborock.devices.traits import Trait +from roborock.devices.traits.common import TraitUpdateListener from roborock.exceptions import RoborockException from roborock.map.b01_map_parser import B01MapParser, B01MapParserConfig from roborock.roborock_typing import RoborockB01Q7Methods from .map import MapTrait +_LOGGER = logging.getLogger(__name__) _TRUNCATE_LENGTH = 20 @@ -49,7 +52,7 @@ def __repr__(self) -> str: return f"MapContent(image_content={img!r}, map_data={self.map_data!r})" -class MapContentTrait(MapContent, Trait): +class MapContentTrait(MapContent, Trait, TraitUpdateListener): """Trait for fetching parsed map content for Q7 devices.""" def __init__( @@ -59,7 +62,8 @@ def __init__( *, map_parser_config: B01MapParserConfig | None = None, ) -> None: - super().__init__() + MapContent.__init__(self) + TraitUpdateListener.__init__(self, logger=_LOGGER) self._map_rpc_channel = map_rpc_channel self._map_trait = map_trait self._map_parser = B01MapParser(map_parser_config) @@ -82,6 +86,23 @@ async def refresh(self) -> None: {"map_id": map_id}, ) + self._parse_and_store(raw_payload) + + def update_from_push(self, raw_payload: bytes) -> None: + """Store an unsolicited SCMap frame pushed by the device during cleaning. + + Pushed frames carry the live robot pose and cleaning path, so the + rendered image stays current without polling. + """ + try: + self._parse_and_store(raw_payload) + except RoborockException as ex: + _LOGGER.debug("Failed to parse pushed B01 map frame: %s", ex) + return + self._notify_update() + + def _parse_and_store(self, raw_payload: bytes) -> None: + """Parse decoded SCMap bytes and update the cached fields.""" try: parsed_data = self._map_parser.parse(raw_payload) except RoborockException: diff --git a/tests/devices/rpc/test_b01_q7_channel.py b/tests/devices/rpc/test_b01_q7_channel.py index 21c6d31a..087291d9 100644 --- a/tests/devices/rpc/test_b01_q7_channel.py +++ b/tests/devices/rpc/test_b01_q7_channel.py @@ -293,3 +293,60 @@ async def test_send_command_general_exception( with pytest.raises(RuntimeError, match="Generic publish crash"): await channel.send_command("prop.get", {"property": ["status"]}) + + +async def test_subscribe_map_pushes_delivers_decoded_frames( + device: HomeDataDevice, + product: HomeDataProduct, + fake_channel: FakeChannel, + message_builder: B01MessageBuilder, +) -> None: + """Unsolicited MAP_RESPONSE frames are decoded and delivered to the callback.""" + channel = create_b01_q7_channel(device, product, fake_channel) # type: ignore[arg-type] + + received: list[bytes] = [] + unsub = await channel.subscribe_map_pushes(received.append) + + with patch( + "roborock.devices.rpc.b01_q7_channel.decode_map_payload", + return_value=b"inflated-payload", + ): + fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload")) + + assert received == [b"inflated-payload"] + + # Non-map messages are filtered out. + fake_channel.notify_subscribers(message_builder.build({"status": 1})) + assert received == [b"inflated-payload"] + + # After unsubscribing, further frames are not delivered. + unsub() + with patch( + "roborock.devices.rpc.b01_q7_channel.decode_map_payload", + return_value=b"inflated-payload", + ): + fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload")) + assert received == [b"inflated-payload"] + + +async def test_subscribe_map_pushes_skips_undecodable_frames( + device: HomeDataDevice, + product: HomeDataProduct, + fake_channel: FakeChannel, + message_builder: B01MessageBuilder, +) -> None: + """Frames that fail map decoding are skipped without breaking the subscription.""" + channel = create_b01_q7_channel(device, product, fake_channel) # type: ignore[arg-type] + + received: list[bytes] = [] + await channel.subscribe_map_pushes(received.append) + + fake_channel.notify_subscribers(message_builder.build_map_response(b"!!! not base64 !!!")) + assert received == [] + + with patch( + "roborock.devices.rpc.b01_q7_channel.decode_map_payload", + return_value=b"inflated-payload", + ): + fake_channel.notify_subscribers(message_builder.build_map_response(b"raw-map-payload")) + assert received == [b"inflated-payload"] diff --git a/tests/devices/traits/b01/q7/conftest.py b/tests/devices/traits/b01/q7/conftest.py index efc9d166..03f36db6 100644 --- a/tests/devices/traits/b01/q7/conftest.py +++ b/tests/devices/traits/b01/q7/conftest.py @@ -1,3 +1,4 @@ +from collections.abc import Callable from typing import Any import pytest @@ -14,6 +15,8 @@ def __init__(self) -> None: self.published_commands: list[tuple[Any, Any]] = [] self.response_queue: list[Any] = [] self.side_effect: Exception | None = None + self.map_push_callback: Callable[[bytes], None] | None = None + self.map_push_subscribe_count = 0 async def send_command(self, command: Any, params: Any = None) -> Any: if self.side_effect: @@ -29,6 +32,15 @@ async def send_map_command(self, command: Any, params: Any = None) -> bytes: return self.response_queue.pop(0) return b"" + async def subscribe_map_pushes(self, callback: Callable[[bytes], None]) -> Callable[[], None]: + self.map_push_subscribe_count += 1 + self.map_push_callback = callback + + def unsub() -> None: + self.map_push_callback = None + + return unsub + @pytest.fixture(name="fake_channel") def fake_channel_fixture() -> FakeQ7Channel: diff --git a/tests/devices/traits/b01/q7/test_map_content.py b/tests/devices/traits/b01/q7/test_map_content.py index b8753d77..ff4f5c1d 100644 --- a/tests/devices/traits/b01/q7/test_map_content.py +++ b/tests/devices/traits/b01/q7/test_map_content.py @@ -87,3 +87,76 @@ async def test_q7_map_content_refresh_errors_without_map_list( with pytest.raises(RoborockException, match="Unable to determine current map ID"): await q7_api.map_content.refresh() + + +async def test_q7_map_content_updates_from_push( + q7_api: Q7PropertiesApi, + fake_channel: FakeQ7Channel, +): + """Unsolicited map pushes update the cached map and notify listeners.""" + await q7_api.start() + assert fake_channel.map_push_callback is not None + + updates: list[bool] = [] + q7_api.map_content.add_update_listener(lambda: updates.append(True)) + + dummy_map_data = MapData() + parsed_map_data = ParsedMapData( + image_content=b"pngbytes", + map_data=dummy_map_data, + ) + with patch( + "roborock.devices.traits.b01.q7.map_content.B01MapParser.parse", + return_value=parsed_map_data, + ): + fake_channel.map_push_callback(b"pushed-payload") + + assert q7_api.map_content.image_content == b"pngbytes" + assert q7_api.map_content.raw_api_response == b"pushed-payload" + assert updates == [True] + + await q7_api.close() + assert fake_channel.map_push_callback is None + + +async def test_q7_map_content_push_parse_failure_keeps_previous_map( + q7_api: Q7PropertiesApi, + fake_channel: FakeQ7Channel, +): + """A malformed pushed frame is dropped without clearing cached content.""" + await q7_api.start() + assert fake_channel.map_push_callback is not None + + dummy_map_data = MapData() + parsed_map_data = ParsedMapData( + image_content=b"pngbytes", + map_data=dummy_map_data, + ) + with patch( + "roborock.devices.traits.b01.q7.map_content.B01MapParser.parse", + return_value=parsed_map_data, + ): + fake_channel.map_push_callback(b"good-payload") + + updates: list[bool] = [] + q7_api.map_content.add_update_listener(lambda: updates.append(True)) + + fake_channel.map_push_callback(b"not a map") + + assert q7_api.map_content.image_content == b"pngbytes" + assert q7_api.map_content.raw_api_response == b"good-payload" + assert updates == [] + await q7_api.close() + + +async def test_q7_start_is_idempotent( + q7_api: Q7PropertiesApi, + fake_channel: FakeQ7Channel, +): + """Repeated start() calls must not leak additional subscriptions.""" + await q7_api.start() + await q7_api.start() + + assert fake_channel.map_push_subscribe_count == 1 + await q7_api.close() + assert fake_channel.map_push_callback is None