From 6c667d50b8ef3c3c6ce5d39a73443d7db67af5e8 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sat, 25 Jul 2026 17:22:04 +0400 Subject: [PATCH] fix: request Q10 maps through dpMultiMap --- roborock/cli.py | 28 ++- roborock/devices/traits/b01/q10/__init__.py | 106 +++++++++- roborock/devices/traits/b01/q10/map.py | 14 ++ roborock/map/b01_q10_map_parser.py | 4 +- tests/devices/traits/b01/q10/test_map.py | 223 +++++++++++++++++++- tests/devices/traits/b01/q10/test_status.py | 11 +- tests/protocols/test_b01_q10_protocol.py | 20 ++ 7 files changed, 373 insertions(+), 33 deletions(-) diff --git a/roborock/cli.py b/roborock/cli.py index 9def26af..c61b0d99 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -594,26 +594,28 @@ async def maps(ctx, device_id: str): await _display_v1_trait(context, device_id, lambda v1: v1.maps) -# The Q10 pushes its map ~9s after a dpRequestDps; firmware throttles pushes to -# ~once per 60-70s, so a single request is answered quickly but rapid re-requests -# may not be. This bounds how long a one-shot CLI command waits for that push. +# The Q10 publishes its map asynchronously after a dpMultiMap list/get request. +# Firmware throttles pushes to ~once per 60-70s, so rapid re-requests may not be +# answered immediately. This bounds how long a one-shot CLI command waits. _Q10_MAP_PUSH_TIMEOUT = 30.0 async def _await_q10_map_push( properties: Q10PropertiesApi, predicate: Callable[[], bool], + add_source_listener: Callable[[Callable[[], None]], Callable[[], None]], *, timeout: float = _Q10_MAP_PUSH_TIMEOUT, allow_cached_on_timeout: bool = False, ) -> bool: """Nudge a Q10 to push its map/trace and wait for a fresh update. - The Q10 map API is entirely push-driven: there is no synchronous get-map - request. A ``dpRequestDps`` causes the device to publish a ``MAP_RESPONSE``, - which the device's subscribe loop feeds into the map trait. Here we register - an update listener, send the request, and wait for a newly pushed update to - satisfy ``predicate``. Returns whether it did within ``timeout``. + The Q10 map response remains asynchronous: ``refresh`` starts a + ``dpMultiMap`` list/get exchange, after which the device publishes a + ``MAP_RESPONSE`` that its subscribe loop feeds into the map trait. Here we + register a packet-specific listener, send the request, and wait for a newly + pushed update to satisfy ``predicate``. Returns whether it did within + ``timeout``. """ loop = asyncio.get_running_loop() updated: asyncio.Future[None] = loop.create_future() @@ -622,7 +624,7 @@ def on_update() -> None: if predicate() and not updated.done(): updated.set_result(None) - unsub = properties.map.add_update_listener(on_update) + unsub = add_source_listener(on_update) try: await properties.refresh() await asyncio.wait_for(updated, timeout=timeout) @@ -648,6 +650,7 @@ async def map_image(ctx, device_id: str, output_file: str): await _await_q10_map_push( properties, lambda: properties.map.image_content is not None, + properties.map._add_map_packet_listener, allow_cached_on_timeout=True, ) image_content = properties.map.image_content @@ -706,7 +709,11 @@ async def q10_position(ctx, device_id: str, include_path: bool): click.echo("Feature not supported by device") return properties = device.b01_q10_properties - got_trace = await _await_q10_map_push(properties, lambda: bool(properties.map.path)) + got_trace = await _await_q10_map_push( + properties, + lambda: bool(properties.map.path), + properties.map._add_trace_packet_listener, + ) if not got_trace: click.echo("No live trace available (the robot only reports position while cleaning).") return @@ -871,6 +878,7 @@ async def rooms(ctx, device_id: str): await _await_q10_map_push( properties, lambda: properties.map.image_content is not None, + properties.map._add_map_packet_listener, allow_cached_on_timeout=True, ) click.echo(dump_json({room.id: room.name for room in properties.map.rooms})) diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index 3c8c73ff..a6205cd1 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -2,10 +2,12 @@ import asyncio import logging +from typing import Any from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP from roborock.devices.rpc.b01_q10_channel import B01Q10Channel from roborock.devices.traits import Trait +from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message @@ -38,6 +40,25 @@ ] _LOGGER = logging.getLogger(__name__) +_MAP_LIST_REQUEST_TIMEOUT = 30.0 + + +def _map_id_from_list_response(response: Any) -> int | str | None: + """Return the first usable map ID from a ``dpMultiMap`` list response.""" + if not isinstance(response, dict) or response.get("op") != "list": + return None + data = response.get("data") + if not isinstance(data, list): + return None + for map_info in data: + if not isinstance(map_info, dict): + continue + map_id = map_info.get("id") + if isinstance(map_id, int) and not isinstance(map_id, bool): + return map_id + if isinstance(map_id, str) and map_id: + return map_id + return None class Q10PropertiesApi(Trait): @@ -115,6 +136,9 @@ def __init__(self, channel: B01Q10Channel) -> None: self._map_dps, ] self._subscribe_task: asyncio.Task[None] | None = None + self._map_request_lock = asyncio.Lock() + self._map_list_request_token: object | None = None + self._map_list_requested_at: float | None = None async def start(self) -> None: """Start any necessary subscriptions for the trait.""" @@ -132,22 +156,55 @@ async def close(self) -> None: async def refresh(self) -> None: """Refresh all traits.""" - # Sending the REQUEST_DPS will cause the device to send all DPS values - # to the device. Updates will be received by the subscribe loop below. + # Status and map retrieval use separate Q10 requests. A bare REQUEST_DPS + # reliably refreshes status but does not reliably make every firmware + # publish its map. await self.command.send(B01_Q10_DP.REQUEST_DPS, params={}) + await self.request_map() + + async def request_map(self) -> None: + """Request the current saved map through the Q10 multi-map protocol. + + The list response arrives asynchronously on the subscribe stream. + ``_handle_message`` extracts its first map ID and follows up with the + matching ``get`` command; the resulting protocol-301 map packet is + routed to :attr:`map`. + """ + async with self._map_request_lock: + now = asyncio.get_running_loop().time() + if ( + self._map_list_request_token is not None + and self._map_list_requested_at is not None + and now - self._map_list_requested_at < _MAP_LIST_REQUEST_TIMEOUT + ): + return + token = object() + self._map_list_request_token = token + self._map_list_requested_at = now + sent = False + try: + await self.command.send( + B01_Q10_DP.COMMON, + {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}}, + ) + sent = True + finally: + if not sent and self._map_list_request_token is token: + self._map_list_request_token = None + self._map_list_requested_at = None async def _subscribe_loop(self) -> None: """Persistent loop dispatching decoded messages to the read-model traits.""" async for message in self._channel.subscribe_stream(): - self._handle_message(message) + await self._handle_message(message) - def _handle_message(self, message: Q10Message) -> None: + async def _handle_message(self, message: Q10Message) -> None: """Route a single decoded message to the trait responsible for it. - Map and trace packets arrive as protocol-301 ``MAP_RESPONSE`` pushes (the - Q10 is entirely push-driven: there is no synchronous get-map request, a - ``dpRequestDps`` just nudges the device to publish its current map). DPS - updates feed the read-model traits. More traits can be dispatched here below. + Map and trace packets arrive as protocol-301 ``MAP_RESPONSE`` pushes. + A ``dpMultiMap`` list response completes the asynchronous request flow + started by :meth:`request_map`; other DPS updates feed the read-model + traits. """ if isinstance(message, Q10MapPacket): self.map.update_from_map_packet(message) @@ -159,6 +216,39 @@ def _handle_message(self, message: Q10Message) -> None: # only updates the fields that it is responsible for. for trait in self._updatable_traits: trait.update_from_dps(message.dps) + await self._request_map_from_list_response(message.dps) + + async def _request_map_from_list_response(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: + """Request map content after receiving our pending map-list response.""" + response = decoded_dps.get(B01_Q10_DP.MULTI_MAP) + if self._map_list_request_token is None or not isinstance(response, dict) or response.get("op") != "list": + return + + token = self._map_list_request_token + if (map_id := _map_id_from_list_response(response)) is None: + if self._map_list_request_token is token: + self._map_list_request_token = None + self._map_list_requested_at = None + _LOGGER.debug("Q10 map list response did not contain a usable map ID") + return + + try: + await self.command.send( + B01_Q10_DP.COMMON, + { + str(B01_Q10_DP.MULTI_MAP.code): { + "op": "get", + "id": map_id, + } + }, + ) + except RoborockException as ex: + # A failed follow-up must not kill the persistent subscribe loop. + _LOGGER.debug("Failed to request Q10 map content: %s", ex) + finally: + if self._map_list_request_token is token: + self._map_list_request_token = None + self._map_list_requested_at = None def create(channel: B01Q10Channel) -> Q10PropertiesApi: diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index ee51352a..c2a4e9b6 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -14,9 +14,11 @@ """ import logging +from collections.abc import Callable from dataclasses import dataclass, field from typing import Any +from roborock.callbacks import CallbackList from roborock.data import RoborockBase from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener @@ -89,6 +91,8 @@ def __init__( self._map_packet: Q10MapPacket | None = None self._trace_packet: Q10TracePacket | None = None self._image_content: bytes | None = None + self._map_packet_callbacks: CallbackList[None] = CallbackList(_LOGGER) + self._trace_packet_callbacks: CallbackList[None] = CallbackList(_LOGGER) self._map_dps.add_update_listener(self._map_dps_updated) @property @@ -121,12 +125,22 @@ def update_from_map_packet(self, packet: Q10MapPacket) -> None: self._map_packet = packet self._render() self._notify_update() + self._map_packet_callbacks(None) def update_from_trace_packet(self, packet: Q10TracePacket) -> None: """Store a trace-protocol update and render the latest sources.""" self._trace_packet = packet self._render() self._notify_update() + self._trace_packet_callbacks(None) + + def _add_map_packet_listener(self, callback: Callable[[], None]) -> Callable[[], None]: + """Register an internal callback for decoded map packets.""" + return self._map_packet_callbacks.add_callback(lambda _: callback()) + + def _add_trace_packet_listener(self, callback: Callable[[], None]) -> Callable[[], None]: + """Register an internal callback for decoded trace packets.""" + return self._trace_packet_callbacks.add_callback(lambda _: callback()) def _map_dps_updated(self) -> None: """Render after the low-level DPS source changes.""" diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index 062f85f3..f7fd68f8 100644 --- a/roborock/map/b01_q10_map_parser.py +++ b/roborock/map/b01_q10_map_parser.py @@ -1,7 +1,7 @@ """Parser for Roborock Q10 (B01/ss07) map packets. -Q10 devices deliver map data as a protocol-301 ``MAP_RESPONSE`` message (pushed a -few seconds after a ``dpRequestDps`` request). Unlike the Q7 ``SCMap`` protobuf +Q10 devices deliver map data as a protocol-301 ``MAP_RESPONSE`` message after a +``dpMultiMap`` list/get request. Unlike the Q7 ``SCMap`` protobuf format, the Q10 uses a custom, unencrypted binary packet: - ``01 01`` marker, then a ``u32be`` map id (bytes 2-5) and two consecutive diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 61120a7e..aa9e5194 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -1,10 +1,10 @@ """Tests for the Q10 B01 map content trait. -The Q10 map API is push-driven: the device publishes ``MAP_RESPONSE`` messages -which the protocol layer decodes into typed map/trace packets; the trait updates -its cached state from them via ``update_from_map_packet`` / -``update_from_trace_packet`` (there is no synchronous get-map request). These -tests cover that state management; the pixel/geometry work it drives is tested in +The Q10 map API uses an asynchronous ``dpMultiMap`` list/get exchange. The +device then publishes ``MAP_RESPONSE`` messages which the protocol layer +decodes into typed map/trace packets; the trait updates its cached state from +them via ``update_from_map_packet`` / ``update_from_trace_packet``. These tests +cover that state management; the pixel/geometry work it drives is tested in ``tests/map/test_b01_q10_render.py``. """ @@ -13,7 +13,7 @@ from collections.abc import AsyncGenerator from pathlib import Path from typing import cast -from unittest.mock import Mock, patch +from unittest.mock import AsyncMock, Mock, patch import pytest @@ -29,7 +29,7 @@ parse_trace_packet, ) from roborock.map.b01_q10_render import Q10MapOverlays -from roborock.protocols.b01_q10_protocol import Q10Message +from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message from .conftest import FakeB01Q10Channel @@ -111,7 +111,10 @@ async def test_await_q10_map_push_waits_for_fresh_update() -> None: properties.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1, 2)])) got_trace = await _await_q10_map_push( - cast(Q10PropertiesApi, properties), lambda: bool(properties.map.path), timeout=0.01 + cast(Q10PropertiesApi, properties), + lambda: bool(properties.map.path), + properties.map._add_trace_packet_listener, + timeout=0.01, ) assert got_trace is False @@ -122,7 +125,10 @@ async def test_await_q10_map_push_returns_true_after_update() -> None: properties = _FakeQ10PropertiesWithTrace() got_trace = await _await_q10_map_push( - cast(Q10PropertiesApi, properties), lambda: bool(properties.map.path), timeout=0.01 + cast(Q10PropertiesApi, properties), + lambda: bool(properties.map.path), + properties.map._add_trace_packet_listener, + timeout=0.01, ) assert got_trace is True @@ -136,6 +142,7 @@ async def test_await_q10_map_push_can_fall_back_to_cached_map_on_timeout() -> No got_map = await _await_q10_map_push( cast(Q10PropertiesApi, properties), lambda: properties.map.image_content is not None, + properties.map._add_map_packet_listener, timeout=0.01, allow_cached_on_timeout=True, ) @@ -144,6 +151,30 @@ async def test_await_q10_map_push_can_fall_back_to_cached_map_on_timeout() -> No assert properties.refresh_count == 1 +async def test_await_q10_map_push_ignores_overlay_only_render() -> None: + """A DPS recomposition cannot masquerade as a fresh map packet.""" + map_dps = MapDpsTrait() + properties = _FakeQ10Properties() + properties.map = MapContentTrait(map_dps) + properties.map.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) + + async def refresh_overlay_only() -> None: + properties.refresh_count += 1 + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: _zone_blob()}) + + properties.refresh = refresh_overlay_only # type: ignore[method-assign] + + got_map = await _await_q10_map_push( + cast(Q10PropertiesApi, properties), + lambda: properties.map.image_content is not None, + properties.map._add_map_packet_listener, + timeout=0.01, + ) + + assert got_map is False + assert properties.refresh_count == 1 + + # --- Integration through the Q10PropertiesApi subscribe loop ----------------- @@ -204,6 +235,180 @@ async def test_subscribe_loop_routes_trace_push( assert q10_api.map.robot_position is not None +async def test_map_list_response_requests_first_map( + q10_api: Q10PropertiesApi, + mock_channel: FakeB01Q10Channel, + message_queue: asyncio.Queue[Q10Message], +) -> None: + """A requested map list is followed by a get for its first map ID.""" + await q10_api.request_map() + assert mock_channel.published_commands == [ + ( + B01_Q10_DP.COMMON, + {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}}, + ) + ] + + message_queue.put_nowait( + Q10DpsUpdate( + dps={ + B01_Q10_DP.MULTI_MAP: { + "data": [ + {"id": "12345", "name": "Current", "timestamp": 1}, + {"id": "67890", "name": "Other", "timestamp": 2}, + ], + "op": "list", + "result": 1, + } + } + ) + ) + + await _wait_for(lambda: len(mock_channel.published_commands) == 2) + assert mock_channel.published_commands[1] == ( + B01_Q10_DP.COMMON, + { + str(B01_Q10_DP.MULTI_MAP.code): { + "op": "get", + "id": "12345", + } + }, + ) + + +async def test_unsolicited_map_list_does_not_request_map( + q10_api: Q10PropertiesApi, + mock_channel: FakeB01Q10Channel, + message_queue: asyncio.Queue[Q10Message], +) -> None: + """A list response not initiated by this API cannot trigger a map get.""" + message_queue.put_nowait( + Q10DpsUpdate( + dps={ + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": "12345"}], + "op": "list", + "result": 1, + } + } + ) + ) + + await asyncio.sleep(0.01) + assert mock_channel.published_commands == [] + + +async def test_concurrent_map_requests_are_coalesced(q10_api: Q10PropertiesApi) -> None: + """Only one list request remains pending when callers refresh together.""" + started = asyncio.Event() + release = asyncio.Event() + + async def send_list(*_args, **_kwargs) -> None: + started.set() + await release.wait() + + with patch.object(q10_api.command, "send", side_effect=send_list) as send: + first = asyncio.create_task(q10_api.request_map()) + await started.wait() + second = asyncio.create_task(q10_api.request_map()) + await asyncio.sleep(0) + release.set() + await asyncio.gather(first, second) + + send.assert_awaited_once() + + +async def test_cancelled_map_request_can_be_retried(q10_api: Q10PropertiesApi) -> None: + """Cancellation clears only its own pending list request.""" + started = asyncio.Event() + blocked = asyncio.Event() + + async def send_list(*_args, **_kwargs) -> None: + started.set() + await blocked.wait() + + with patch.object(q10_api.command, "send", side_effect=send_list): + request = asyncio.create_task(q10_api.request_map()) + await started.wait() + request.cancel() + with pytest.raises(asyncio.CancelledError): + await request + + with patch.object(q10_api.command, "send", new=AsyncMock()) as retry: + await q10_api.request_map() + + retry.assert_awaited_once() + + +async def test_map_request_stays_coalesced_through_get( + q10_api: Q10PropertiesApi, + message_queue: asyncio.Queue[Q10Message], +) -> None: + """The pending flow includes the follow-up get publish.""" + get_started = asyncio.Event() + release_get = asyncio.Event() + get_finished = asyncio.Event() + + async def send_command(_dp, params) -> None: + operation = params[str(B01_Q10_DP.MULTI_MAP.code)]["op"] + if operation == "get": + get_started.set() + await release_get.wait() + get_finished.set() + + response = Q10DpsUpdate( + dps={ + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": "12345"}], + "op": "list", + "result": 1, + } + } + ) + with patch.object(q10_api.command, "send", side_effect=send_command) as send: + await q10_api.request_map() + message_queue.put_nowait(response) + await get_started.wait() + + await q10_api.request_map() + assert send.await_count == 2 + + release_get.set() + await get_finished.wait() + await q10_api.request_map() + assert send.await_count == 3 + + +async def test_failed_map_get_does_not_stop_subscription( + q10_api: Q10PropertiesApi, + message_queue: asyncio.Queue[Q10Message], +) -> None: + """An expected command failure cannot stop later map-protocol updates.""" + + async def send_command(_dp, params) -> None: + if params[str(B01_Q10_DP.MULTI_MAP.code)]["op"] == "get": + raise RoborockException("device rejected map request") + + with patch.object(q10_api.command, "send", side_effect=send_command): + await q10_api.request_map() + message_queue.put_nowait( + Q10DpsUpdate( + dps={ + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": "12345"}], + "op": "list", + "result": 1, + } + } + ) + ) + message_queue.put_nowait(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes())) + + await _wait_for(lambda: bool(q10_api.map.path)) + + assert q10_api.map.robot_position is not None + + # --- Source composition + rendering ------------------------------------------ diff --git a/tests/devices/traits/b01/q10/test_status.py b/tests/devices/traits/b01/q10/test_status.py index edd4bcb1..8f301483 100644 --- a/tests/devices/traits/b01/q10/test_status.py +++ b/tests/devices/traits/b01/q10/test_status.py @@ -128,10 +128,13 @@ async def test_status_trait_refresh( # Send a refresh command await q10_api.refresh() - assert len(mock_channel.published_commands) == 1 - command, params = mock_channel.published_commands[0] - assert command == B01_Q10_DP.REQUEST_DPS - assert params == {} + assert mock_channel.published_commands == [ + (B01_Q10_DP.REQUEST_DPS, {}), + ( + B01_Q10_DP.COMMON, + {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}}, + ), + ] # Push the response message into the queue message_queue.put_nowait(message) diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index f5d916d1..5fddc41a 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -41,6 +41,26 @@ def test_decode_message_dps_update() -> None: assert decoded == Q10DpsUpdate(dps={B01_Q10_DP.BATTERY: 100}) +def test_decode_message_common_multi_map_list() -> None: + """A nested dpCommon map-list response is flattened into dpMultiMap.""" + message = _message( + b'{"dps":{"101":{"61":{"data":[{"id":"12345","name":"Map","timestamp":1}],"op":"list","result":1}}}}', + RoborockMessageProtocol.RPC_RESPONSE, + ) + + decoded = decode_message(message) + + assert decoded == Q10DpsUpdate( + dps={ + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": "12345", "name": "Map", "timestamp": 1}], + "op": "list", + "result": 1, + } + } + ) + + def test_decode_message_map_packet() -> None: """A MAP_RESPONSE 01 01 payload decodes into a Q10MapPacket.""" message = _message(MAP_FIXTURE.read_bytes(), RoborockMessageProtocol.MAP_RESPONSE)