From b8bdf3a363b56c725a4b62db1461b9d2b032422c Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Sat, 29 Aug 2026 21:06:39 +0100 Subject: [PATCH 1/5] fix: request Q10 maps without starting cleaning --- roborock/cli.py | 27 +++------- roborock/devices/traits/b01/q10/map.py | 29 ++++------ tests/devices/traits/b01/q10/test_map.py | 67 ++++-------------------- 3 files changed, 27 insertions(+), 96 deletions(-) diff --git a/roborock/cli.py b/roborock/cli.py index 1969c5b7..a83caa15 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -594,9 +594,9 @@ async def maps(ctx, device_id: str): await _display_v1_trait(context, device_id, lambda v1: v1.maps) -# 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. +# The Q10 publishes its current map asynchronously after a REQUEST_DPS. 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 @@ -609,10 +609,9 @@ async def _await_q10_map_push( ) -> bool: """Request Q10 map content and wait for usable map-trait state. - A Q10 needs a saved-map ID before it can request content. The map list and - content have independent refresh schedules, so the list is requested only - when no ID is stored. The content then arrives as a later ``MAP_RESPONSE`` - and is published through the standard trait update interface. + The read-only ``REQUEST_DPS`` request returns immediately; current map + content arrives as a later ``MAP_RESPONSE`` and is published through the + standard trait update interface. """ loop = asyncio.get_running_loop() updated: asyncio.Future[None] = loop.create_future() @@ -624,20 +623,6 @@ def on_update() -> None: unsub = properties.map.add_update_listener(on_update) try: async with asyncio.timeout(timeout): - if properties.maps.current_map_id is None: - map_list_updated: asyncio.Future[None] = loop.create_future() - - def on_map_list_update() -> None: - if properties.maps.current_map_id is not None and not map_list_updated.done(): - map_list_updated.set_result(None) - - unsub_maps = properties.maps.add_update_listener(on_map_list_update) - try: - await properties.maps.refresh() - if properties.maps.current_map_id is None: - await map_list_updated - finally: - unsub_maps() await properties.map.refresh() await updated return True diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index 2d51e441..5890e688 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -7,10 +7,10 @@ * restricted zones, virtual walls and dock state arrive as ordinary DPS values. ``MapDpsTrait`` owns the low-level map-specific DPS read model. -``MapContentTrait`` uses a stored ID from ``MapsTrait`` only when it requests -content. It combines the latest map and trace packets with the map DPS state -through the pure functions in :mod:`roborock.map.b01_q10_render`. Map-list -updates do not refresh content. +``MapContentTrait`` requests a current-map push through ``REQUEST_DPS`` and +combines the latest map and trace packets with the map DPS state through the +pure functions in :mod:`roborock.map.b01_q10_render`. Saved-map list/detail +operations remain on ``MapsTrait``. """ import logging @@ -107,20 +107,13 @@ def __init__( self._map_dps.add_update_listener(self._map_dps_updated) async def refresh(self) -> None: - """Request content for the first map in the latest saved-map list.""" - if (map_id := self._maps.current_map_id) is None: - raise RoborockException("Cannot request Q10 map content before the map list is available") - # Map lists and map content can change at different times. Reuse the - # stored ID so a content refresh does not also refresh the list. - await self._command.send( - B01_Q10_DP.COMMON, - { - str(B01_Q10_DP.MULTI_MAP.code): { - "op": "get", - "id": map_id, - } - }, - ) + """Request a safe asynchronous current-map/status push. + + Some ss07 firmware treats ``dpMultiMap op:get`` as an active + cleaning/relocation command. ``REQUEST_DPS`` is the device's read-only + current-map request and does not depend on a saved-map ID. + """ + await self._command.send(B01_Q10_DP.REQUEST_DPS, params={}) @property def image_content(self) -> bytes | None: diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 972123fe..ef6aa18f 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -140,34 +140,6 @@ async def refresh_map() -> None: self.map.refresh = refresh_map # type: ignore[method-assign] -class _FakeQ10PropertiesWithoutMapId: - def __init__(self) -> None: - command = cast(CommandTrait, Mock(spec=CommandTrait)) - self.maps = MapsTrait(command) - self.map = MapContentTrait(MapDpsTrait(), self.maps, command) - self.maps_refresh_count = 0 - self.map_refresh_count = 0 - - async def refresh_maps() -> None: - self.maps_refresh_count += 1 - self.maps.update_from_dps( - { - B01_Q10_DP.MULTI_MAP: { - "data": [{"id": "12345"}], - "op": "list", - "result": 1, - } - } - ) - - async def refresh_map() -> None: - self.map_refresh_count += 1 - self.map.update_from_trace_packet(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes())) - - self.maps.refresh = refresh_maps # type: ignore[method-assign] - self.map.refresh = refresh_map # type: ignore[method-assign] - - async def test_await_q10_map_push_waits_for_fresh_update() -> None: """A cached trace alone is not treated as a successful new map push.""" properties = _FakeQ10Properties() @@ -196,21 +168,6 @@ async def test_await_q10_map_push_returns_true_after_update() -> None: assert len(properties.map.path) == 14 -async def test_await_q10_map_push_requests_map_list_only_on_first_use() -> None: - """Content gets the list first only when no stored map ID is available.""" - properties = _FakeQ10PropertiesWithoutMapId() - - got_trace = await _await_q10_map_push( - cast(Q10PropertiesApi, properties), - lambda: bool(properties.map.path), - timeout=0.01, - ) - - assert got_trace is True - assert properties.maps_refresh_count == 1 - assert properties.map_refresh_count == 1 - - async def test_await_q10_map_push_can_fall_back_to_cached_map_on_timeout() -> None: properties = _FakeQ10Properties() properties.map.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) @@ -286,7 +243,7 @@ async def test_subscribe_loop_routes_trace_push( assert q10_api.map.robot_position is not None -async def test_map_list_and_content_refresh_are_independent( +async def test_map_list_and_current_content_refresh_are_independent( q10_api: Q10PropertiesApi, mock_channel: FakeB01Q10Channel, message_queue: asyncio.Queue[Q10Message], @@ -320,15 +277,7 @@ async def test_map_list_and_content_refresh_are_independent( await q10_api.map.refresh() - assert mock_channel.published_commands[1] == ( - B01_Q10_DP.COMMON, - { - str(B01_Q10_DP.MULTI_MAP.code): { - "op": "get", - "id": "12345", - } - }, - ) + assert mock_channel.published_commands[1] == (B01_Q10_DP.REQUEST_DPS, {}) assert q10_api.maps.current_map_id == "12345" @@ -355,10 +304,14 @@ async def test_empty_map_list_does_not_request_content( assert mock_channel.published_commands == [] -async def test_map_content_refresh_requires_stored_map_id(q10_api: Q10PropertiesApi) -> None: - """Content cannot be requested until the map list supplies an ID.""" - with pytest.raises(RoborockException, match="map list is available"): - await q10_api.map.refresh() +async def test_map_content_refresh_does_not_require_stored_map_id( + q10_api: Q10PropertiesApi, + mock_channel: FakeB01Q10Channel, +) -> None: + """Current-map refresh is read-only and independent of saved-map state.""" + await q10_api.map.refresh() + + assert mock_channel.published_commands == [(B01_Q10_DP.REQUEST_DPS, {})] async def test_map_content_refresh_requests_are_not_rate_limited(q10_api: Q10PropertiesApi) -> None: From 95d268e5373e44466cfff20bc815774ac4a431ff Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Mon, 31 Aug 2026 15:03:14 +0100 Subject: [PATCH 2/5] feat: parse Q10 archived map packets --- roborock/map/b01_q10_map_parser.py | 216 ++++++++++++++++++++--- roborock/map/b01_q10_render.py | 8 +- roborock/protocols/b01_q10_protocol.py | 17 +- tests/map/test_b01_q10_map_parser.py | 135 ++++++++++++++ tests/map/test_b01_q10_render.py | 11 +- tests/protocols/test_b01_q10_protocol.py | 18 +- 6 files changed, 362 insertions(+), 43 deletions(-) diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index 2f162c64..2c105e22 100644 --- a/roborock/map/b01_q10_map_parser.py +++ b/roborock/map/b01_q10_map_parser.py @@ -1,8 +1,9 @@ """Parser for Roborock Q10 (B01/ss07) map packets. -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: +Q10 devices deliver map data as protocol-301 ``MAP_RESPONSE`` pushes. Current +maps follow a read-only status request, while saved-map and clean-record detail +packets follow their respective ``select`` requests. 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 ``u16be`` dimensions: grid width (bytes 7-8) and grid height (bytes 9-10). @@ -22,6 +23,7 @@ import io import math import statistics +import struct from dataclasses import dataclass, field, replace from PIL import Image @@ -29,6 +31,7 @@ from vacuum_map_parser_base.config.image_config import ImageConfig from vacuum_map_parser_base.map_data import ImageData, MapData, Point +from roborock.data.code_mappings import RoborockEnum from roborock.data.containers import RoborockBase from roborock.exceptions import RoborockException @@ -66,9 +69,6 @@ def classify_q10_cell(value: int) -> str: return LAYER_FLOOR -MAP_PACKET_MARKER = b"\x01\x01" -TRACE_PACKET_MARKER = b"\x02\x01" - _MAP_ID_OFFSET = 2 # Width and height are two consecutive u16be fields. An earlier revision read the # width as u16le at offset 8; that high byte is actually the height's high byte, @@ -83,6 +83,7 @@ def classify_q10_cell(value: int) -> str: _ROOM_RECORD_LENGTH = 47 _ROOM_NAME_LENGTH_OFFSET = 26 _MAX_ROOMS = 32 +_MAX_GRID_CELLS = 16_000_000 # Sanity bound for the erase-zone vector section's vertices-per-polygon field. _MAX_ERASE_ZONE_VERTICES = 16 @@ -194,10 +195,30 @@ def charger_pixels(self) -> tuple[float, float] | None: ) +class Q10MapPacketKind(RoborockEnum): + """Semantic kind identified by a Q10 map packet's two-byte marker.""" + + CURRENT = 1 + TRACE = 2 + CLEAN_RECORD_DETAIL = 3 + SAVED_MAP_DETAIL = 4 + + @property + def marker(self) -> bytes: + """Return the two-byte wire marker for this packet kind.""" + return bytes((self.value, 1)) + + @classmethod + def from_payload(cls, payload: bytes) -> "Q10MapPacketKind | None": + """Return the recognized kind for a payload marker.""" + return next((kind for kind in cls if payload[:2] == kind.marker), None) + + @dataclass class Q10MapPacket: - """Decoded contents of a Q10 ``01 01`` map packet.""" + """Decoded contents of a Q10 current or archived map packet.""" + kind: Q10MapPacketKind map_id: int width: int height: int @@ -211,6 +232,8 @@ class Q10MapPacket: """Carpet mask decoded from the packet tail: a full ``width*height`` grid in the same (top-down) pixel space as :attr:`grid`, where a non-zero cell is carpet (the value is the carpet kind). ``None`` if the packet carried none.""" + historical_trace: "Q10HistoricalTracePacket | None" = None + """Cleaning path embedded in a clean-record detail packet, if present.""" @property def layers(self) -> GridLayers: @@ -266,6 +289,27 @@ def robot_position(self) -> Q10Point | None: return self.points[-1] if self.points else None +@dataclass +class Q10HistoricalTracePacket: + """Cleaning path embedded in a Q10 ``03 01`` clean-record detail packet. + + This is a different wire layout from the live ``02 01`` trace. Its header + carries a 16-bit format version, a 32-bit opaque value, a 32-bit + point count, a signed heading, and a zero reserved word. Points use the same + signed big-endian ``(x, y)`` coordinate pairs as the live trace. + """ + + points: list[Q10Point] = field(default_factory=list) + version: int = 0 + opaque_value: int = 0 + heading: int = 0 + + @property + def robot_position(self) -> Q10Point | None: + """The final recorded position, if the historical path is non-empty.""" + return self.points[-1] if self.points else None + + # Trace packet (``02 01``): a 14-byte header followed by big-endian int16 (x, y) # point pairs forming the accumulated session path. Header layout confirmed # against live ss07 captures and cross-checked by @andrewlyeats: @@ -275,18 +319,27 @@ def robot_position(self) -> Q10Point | None: # - bytes 10-11: the 0201 SLAM heading (s16be degrees; 0 = +x, +90 = +y, # +-180 = -x, -90 = -y) -- the robot's current orientation. # - bytes 12-13: a constant (0x0000). -# - byte 14 onward: the path points. +# - byte 14 onward: exactly ``point_count`` path points. # An earlier revision used a 10-byte header, which folded the heading word into # a phantom leading point ``(heading, 0)`` -- that is the "stray point" the # heuristic below was papering over, and why the count read "one high". The -# parser reads all 4-byte pairs in the body rather than trusting the count -# field, so a truncated tail can't desync it. +# parser requires the declared point count to match the complete body, so a +# truncated or extended tail cannot be silently interpreted as path data. # NOTE: the format documented by roborock-qseries-map-bridge (18-byte header) # did not match this firmware -- this 14-byte layout is what the device sent. _TRACE_HEADER_LENGTH = 14 _TRACE_SEQUENCE_OFFSET = 3 +_TRACE_POINT_COUNT_OFFSET = 8 _TRACE_HEADING_OFFSET = 10 +_HISTORICAL_TRACE_HEADER_LENGTH = 14 +_HISTORICAL_TRACE_PREFIX_LENGTH = 1 +_HISTORICAL_TRACE_VERSION = 1 +_HISTORICAL_TRACE_OPAQUE_VALUE_OFFSET = 2 +_HISTORICAL_TRACE_POINT_COUNT_OFFSET = 6 +_HISTORICAL_TRACE_HEADING_OFFSET = 10 +_HISTORICAL_TRACE_RESERVED_OFFSET = 12 + # Some cleans still prepend a single near-origin sentinel as the first real # point (e.g. ~(5, 76) / (-3, 0) when the path proper starts near (-1700, -800)); # it skews the rendered start/bounding box and any path-based calibration. (This @@ -301,12 +354,22 @@ def robot_position(self) -> Q10Point | None: def is_map_packet(payload: bytes) -> bool: """Return True if the payload is a Q10 full-map (``01 01``) packet.""" - return payload[:2] == MAP_PACKET_MARKER + return Q10MapPacketKind.from_payload(payload) is Q10MapPacketKind.CURRENT + + +def is_clean_record_map_packet(payload: bytes) -> bool: + """Return True for a Q10 clean-record detail (``03 01``) packet.""" + return Q10MapPacketKind.from_payload(payload) is Q10MapPacketKind.CLEAN_RECORD_DETAIL + + +def is_saved_map_packet(payload: bytes) -> bool: + """Return True for a Q10 saved-map detail (``04 01``) packet.""" + return Q10MapPacketKind.from_payload(payload) is Q10MapPacketKind.SAVED_MAP_DETAIL def is_trace_packet(payload: bytes) -> bool: """Return True if the payload is a Q10 live trace (``02 01``) packet.""" - return payload[:2] == TRACE_PACKET_MARKER + return Q10MapPacketKind.from_payload(payload) is Q10MapPacketKind.TRACE def parse_trace_packet(payload: bytes) -> Q10TracePacket: @@ -318,6 +381,9 @@ def parse_trace_packet(payload: bytes) -> Q10TracePacket: body = payload[_TRACE_HEADER_LENGTH:] if len(body) % 4: raise RoborockException("Q10 trace points are not 4-byte (x, y) pairs") + declared_point_count = int.from_bytes(payload[_TRACE_POINT_COUNT_OFFSET : _TRACE_POINT_COUNT_OFFSET + 2], "big") + if declared_point_count != len(body) // 4: + raise RoborockException("Q10 trace point count does not match its payload") heading = int.from_bytes(payload[_TRACE_HEADING_OFFSET : _TRACE_HEADING_OFFSET + 2], "big", signed=True) points = [ @@ -348,11 +414,13 @@ def _drop_stray_leading_point(points: list[Q10Point]) -> list[Q10Point]: return points -def lz4_block_decompress(data: bytes) -> bytes: +def lz4_block_decompress(data: bytes, max_output_size: int | None = None) -> bytes: """Decompress a raw LZ4 *block* (no frame header). The Q10 map grid is stored as a single LZ4 block. This implements the - standard LZ4 block format so we don't add a native dependency. + standard LZ4 block format so we don't add a native dependency. When + ``max_output_size`` is supplied, expansion beyond it is rejected before + allocating the excess output. """ index = 0 output = bytearray() @@ -380,6 +448,8 @@ def read_length(value: int) -> int: end = index + literal_length if end > len(data): raise RoborockException("Truncated LZ4 block while reading literals") + if max_output_size is not None and len(output) + literal_length > max_output_size: + raise RoborockException("LZ4 block exceeds maximum output size") output.extend(data[index:end]) index = end @@ -394,6 +464,8 @@ def read_length(value: int) -> int: raise RoborockException("Invalid LZ4 back-reference offset") match_length = read_length(token & 0x0F) + 4 + if max_output_size is not None and len(output) + match_length > max_output_size: + raise RoborockException("LZ4 block exceeds maximum output size") for _ in range(match_length): output.append(output[-offset]) @@ -458,8 +530,9 @@ def _parse_rooms(room_data: bytes, grid: bytes) -> list[Q10Room]: def parse_map_packet(payload: bytes) -> Q10MapPacket: - """Parse a Q10 ``01 01`` map packet into grid + room metadata.""" - if len(payload) < _LAYOUT_COMPRESSED_OFFSET or not is_map_packet(payload): + """Parse a Q10 current or archived map into typed source data.""" + kind = Q10MapPacketKind.from_payload(payload) + if len(payload) < _LAYOUT_COMPRESSED_OFFSET or kind is None or kind is Q10MapPacketKind.TRACE: raise RoborockException("Payload is not a Q10 map packet") map_id = int.from_bytes(payload[_MAP_ID_OFFSET : _MAP_ID_OFFSET + 4], "big") @@ -467,6 +540,8 @@ def parse_map_packet(payload: bytes) -> Q10MapPacket: height = int.from_bytes(payload[_HEIGHT_OFFSET : _HEIGHT_OFFSET + 2], "big") if width <= 0: raise RoborockException("Q10 map packet has invalid width") + if height > 0 and width * height > _MAX_GRID_CELLS: + raise RoborockException("Q10 map packet dimensions exceed the supported grid size") compressed_length = int.from_bytes( payload[_COMPRESSED_LAYOUT_LENGTH_OFFSET : _COMPRESSED_LAYOUT_LENGTH_OFFSET + 2], "big" @@ -475,7 +550,10 @@ def parse_map_packet(payload: bytes) -> Q10MapPacket: if compressed_length <= 0 or layout_end > len(payload): raise RoborockException("Q10 map packet has invalid layout block length") - decoded = lz4_block_decompress(payload[_LAYOUT_COMPRESSED_OFFSET:layout_end]) + decoded = lz4_block_decompress( + payload[_LAYOUT_COMPRESSED_OFFSET:layout_end], + max_output_size=_MAX_GRID_CELLS + 2 + _MAX_ROOMS * _ROOM_RECORD_LENGTH, + ) # Prefer the header height; fall back to inference if it doesn't line up # (e.g. older captures/fixtures that don't populate the height field). split = _split_with_dims(decoded, width, height) if height > 0 else None @@ -486,9 +564,14 @@ def parse_map_packet(payload: bytes) -> Q10MapPacket: rooms = _parse_rooms(room_data, grid) tail = payload[layout_end:] erase_zones = _parse_erase_zones(tail) - carpet_mask = _parse_carpet_mask(tail, width, height) + carpet_mask, carpet_end = _parse_carpet_block(tail, width, height) + if kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL and carpet_end is not None: + historical_trace, _ = _parse_clean_record_trace(tail, carpet_end) + else: + historical_trace = None header_calibration = _parse_header_calibration(payload) return Q10MapPacket( + kind=kind, map_id=map_id, width=width, height=height, @@ -497,6 +580,7 @@ def parse_map_packet(payload: bytes) -> Q10MapPacket: erase_zones=erase_zones, header_calibration=header_calibration, carpet_mask=carpet_mask, + historical_trace=historical_trace, ) @@ -569,7 +653,18 @@ def _carpet_offset(tail: bytes) -> int: return 2 + count * vertices_per * 4 -def _parse_carpet_mask(tail: bytes, width: int, height: int) -> bytes | None: +def _erase_section_end(tail: bytes) -> int: + """Return the end of a complete, structurally valid erase section.""" + if len(tail) < 2: + return 0 + count, vertices_per = tail[0], tail[1] + if count and not 1 <= vertices_per <= _MAX_ERASE_ZONE_VERTICES: + return 0 + end = _carpet_offset(tail) + return end if end <= len(tail) else 0 + + +def _parse_carpet_block(tail: bytes, width: int, height: int) -> tuple[bytes | None, int | None]: """Decode the carpet mask that follows the erase section in the packet tail. Framing matches the main grid block: ``[u32 uncompressed_len]`` @@ -578,23 +673,88 @@ def _parse_carpet_mask(tail: bytes, width: int, height: int) -> bytes | None: non-zero cell is carpet (the value is the carpet kind). Confirmed byte-exact on live ss07 captures (R1 / RDC), where ``uncompressed_len == width*height``. - Returns the decompressed mask, or ``None`` if the section is absent or does - not line up (the ``uncompressed_len == width*height`` invariant is used as the - guard so a mis-located section yields no carpet rather than garbage). + Returns the decompressed mask and its end offset. Both are ``None`` if the + section is absent or does not line up. The end offset is used to anchor + optional later sections without scanning arbitrary trailing bytes. """ - offset = _carpet_offset(tail) + offset = _erase_section_end(tail) + if offset == 0: + return None, None if offset + 6 > len(tail): - return None + return None, None uncompressed_len = int.from_bytes(tail[offset : offset + 4], "big") compressed_len = int.from_bytes(tail[offset + 4 : offset + 6], "big") block_end = offset + 6 + compressed_len if uncompressed_len != width * height or compressed_len <= 0 or block_end > len(tail): - return None + return None, None try: - mask = lz4_block_decompress(tail[offset + 6 : block_end]) + mask = lz4_block_decompress(tail[offset + 6 : block_end], max_output_size=width * height) except RoborockException: - return None - return mask if len(mask) == width * height else None + return None, None + if len(mask) != width * height: + return None, None + return mask, block_end + + +def _parse_carpet_mask(tail: bytes, width: int, height: int) -> bytes | None: + """Decode only the optional carpet mask (compatibility helper).""" + return _parse_carpet_block(tail, width, height)[0] + + +def _parse_clean_record_trace( + tail: bytes, + offset: int, +) -> tuple[Q10HistoricalTracePacket | None, int | None]: + """Decode the bounded historical path following a ``03 01`` carpet block. + + The header and declared point count were validated against a physical ss07 + clean-record response and its point bytes match captured prefixes of the + corresponding live trace exactly. One observed zero byte precedes the path; + its meaning is unknown, so a non-zero value makes the entire section opaque. + Any unsupported version, non-zero reserved word, or truncated point table is + likewise left completely opaque. Bytes after the declared points are + deliberately not consumed: the observed 12-byte suffix appears structured, + but there is not enough controlled evidence to name or decode it safely. + """ + if offset >= len(tail) or tail[offset] != 0: + return None, None + offset += _HISTORICAL_TRACE_PREFIX_LENGTH + header_end = offset + _HISTORICAL_TRACE_HEADER_LENGTH + if header_end > len(tail): + return None, None + version = int.from_bytes(tail[offset : offset + 2], "big") + reserved = int.from_bytes( + tail[offset + _HISTORICAL_TRACE_RESERVED_OFFSET : offset + _HISTORICAL_TRACE_RESERVED_OFFSET + 2], + "big", + ) + if version != _HISTORICAL_TRACE_VERSION or reserved != 0: + return None, None + point_count = int.from_bytes( + tail[offset + _HISTORICAL_TRACE_POINT_COUNT_OFFSET : offset + _HISTORICAL_TRACE_POINT_COUNT_OFFSET + 4], + "big", + ) + points_end = header_end + point_count * 4 + if points_end > len(tail): + return None, None + coordinates = struct.iter_unpack(">hh", memoryview(tail)[header_end:points_end]) + return ( + Q10HistoricalTracePacket( + points=[Q10Point(x=x, y=y) for x, y in coordinates], + version=version, + opaque_value=int.from_bytes( + tail[ + offset + _HISTORICAL_TRACE_OPAQUE_VALUE_OFFSET : offset + _HISTORICAL_TRACE_OPAQUE_VALUE_OFFSET + 4 + ], + "big", + ), + heading=int.from_bytes( + tail[offset + _HISTORICAL_TRACE_HEADING_OFFSET : offset + _HISTORICAL_TRACE_HEADING_OFFSET + 2], + "big", + signed=True, + ), + ), + points_end, + ) def erased_packet(packet: "Q10MapPacket", cells: set[int]) -> "Q10MapPacket": diff --git a/roborock/map/b01_q10_render.py b/roborock/map/b01_q10_render.py index 0e61452f..a24df43c 100644 --- a/roborock/map/b01_q10_render.py +++ b/roborock/map/b01_q10_render.py @@ -34,6 +34,7 @@ B01Q10MapParser, B01Q10MapParserConfig, Q10EraseZone, + Q10HistoricalTracePacket, Q10MapPacket, Q10TracePacket, erased_packet, @@ -62,7 +63,6 @@ # a much shorter path suffices to confirm it (early in a clean, not just a dense # one). See :func:`solve_calibration_with_origin`. _MIN_HEADER_CALIBRATION_POINTS = 4 - _Q10_DRAWABLE_TYPES = { Drawable.CHARGER, Drawable.NO_GO_AREAS, @@ -84,7 +84,7 @@ class Q10MapOverlays: def render_q10_map( packet: Q10MapPacket, - trace: Q10TracePacket | None, + trace: Q10TracePacket | Q10HistoricalTracePacket | None, overlays: Q10MapOverlays, *, config: B01Q10MapParserConfig, @@ -133,7 +133,7 @@ def render_q10_map( def solve_q10_calibration( packet: Q10MapPacket, - trace: Q10TracePacket | None, + trace: Q10TracePacket | Q10HistoricalTracePacket | None, ) -> GridCalibration | None: """Derive world-to-pixel calibration from a map and its current trace. @@ -232,7 +232,7 @@ def _erased_cells( def _place_trace( map_data: MapData, calibration: GridCalibration, - trace: Q10TracePacket, + trace: Q10TracePacket | Q10HistoricalTracePacket, *, charger_heading: int | None = None, ) -> None: diff --git a/roborock/protocols/b01_q10_protocol.py b/roborock/protocols/b01_q10_protocol.py index c0cf6b9b..864ca2b8 100644 --- a/roborock/protocols/b01_q10_protocol.py +++ b/roborock/protocols/b01_q10_protocol.py @@ -9,9 +9,8 @@ from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import ( Q10MapPacket, + Q10MapPacketKind, Q10TracePacket, - is_map_packet, - is_trace_packet, parse_map_packet, parse_trace_packet, ) @@ -113,18 +112,18 @@ class Q10DpsUpdate: def decode_message(message: RoborockMessage) -> Q10Message | None: """Decode a pushed Q10 ``RoborockMessage`` into a typed message. - ``MAP_RESPONSE`` (protocol 301) payloads carry the binary map (``01 01``) or - trace (``02 01``) packets, which are parsed by the map parser; any other - ``MAP_RESPONSE`` marker is unrecognized and yields ``None``. Every other - protocol is treated as a DPS status update. + ``MAP_RESPONSE`` (protocol 301) payloads carry binary current-map (``01 + 01``), trace (``02 01``), clean-record detail (``03 01``), or saved-map + detail (``04 01``) packets. Any other marker is unrecognized and yields + ``None``. Every other protocol is treated as a DPS status update. Raises ``RoborockException`` if a recognized payload fails to parse. """ if message.protocol == RoborockMessageProtocol.MAP_RESPONSE: payload = message.payload or b"" - if is_map_packet(payload): - return parse_map_packet(payload) - if is_trace_packet(payload): + if Q10MapPacketKind.from_payload(payload) is Q10MapPacketKind.TRACE: return parse_trace_packet(payload) + if Q10MapPacketKind.from_payload(payload) is not None: + return parse_map_packet(payload) return None return Q10DpsUpdate(dps=decode_rpc_response(message)) diff --git a/tests/map/test_b01_q10_map_parser.py b/tests/map/test_b01_q10_map_parser.py index c0200ff5..8b7a216c 100644 --- a/tests/map/test_b01_q10_map_parser.py +++ b/tests/map/test_b01_q10_map_parser.py @@ -2,6 +2,7 @@ import io from pathlib import Path +from typing import Any import pytest from PIL import Image @@ -10,9 +11,13 @@ from roborock.map.b01_grid_layers import LAYER_BACKGROUND, LAYER_FLOOR, LAYER_WALL from roborock.map.b01_q10_map_parser import ( B01Q10MapParser, + Q10MapPacketKind, + Q10Point, Q10Room, classify_q10_cell, + is_clean_record_map_packet, is_map_packet, + is_saved_map_packet, is_trace_packet, lz4_block_decompress, parse_map_packet, @@ -107,6 +112,13 @@ def test_lz4_block_back_reference() -> None: assert lz4_block_decompress(block) == b"A" * 9 +def test_lz4_block_rejects_output_over_limit() -> None: + block = bytes([0x14, ord("A"), 0x01, 0x00, 0x00]) + + with pytest.raises(RoborockException, match="maximum output size"): + lz4_block_decompress(block, max_output_size=8) + + def test_is_map_packet() -> None: assert is_map_packet(b"\x01\x01rest") assert not is_map_packet(b"\x02\x01rest") # trace packet @@ -356,6 +368,16 @@ def test_parse_trace_rejects_misaligned_points() -> None: parse_trace_packet(b"\x02\x01" + b"\x00" * 12 + b"\x01\x02\x03") +@pytest.mark.parametrize("declared_count", [0, 2]) +def test_parse_trace_rejects_declared_count_mismatch(declared_count: int) -> None: + """An aligned body cannot silently disagree with the firmware count.""" + payload = bytearray(_trace_payload([(10, 20)])) + payload[8:10] = declared_count.to_bytes(2, "big") + + with pytest.raises(RoborockException, match="point count"): + parse_trace_packet(bytes(payload)) + + def test_parse_rejects_bad_layout_length() -> None: payload = bytearray(_payload()) payload[27:29] = (0xFFFF).to_bytes(2, "big") # compressed length past the buffer @@ -363,6 +385,15 @@ def test_parse_rejects_bad_layout_length() -> None: parse_map_packet(bytes(payload)) +def test_parse_rejects_unreasonable_header_dimensions() -> None: + payload = bytearray(_payload()) + payload[7:9] = (65535).to_bytes(2, "big") + payload[9:11] = (65535).to_bytes(2, "big") + + with pytest.raises(RoborockException, match="supported grid size"): + parse_map_packet(bytes(payload)) + + def test_parse_erase_zones_from_map_packet_tail() -> None: """Erase rectangles appended after the grid decode to world polygons.""" rects = [ @@ -387,6 +418,32 @@ def _carpet_tail(width: int, height: int, carpet: bytes, erase: bytes = bytes([0 return erase + (width * height).to_bytes(4, "big") + len(block).to_bytes(2, "big") + block +def _map_detail_payload( + marker: bytes, + points: list[tuple[int, int]], + *, + version: int = 1, + opaque_value: int = 2, + heading: int = 3, + reserved: int = 0, + prefix: int = 0, + trailing: bytes = b"", +) -> bytes: + """Build a neutral synthetic detail packet from the existing map fixture.""" + header = ( + version.to_bytes(2, "big") + + opaque_value.to_bytes(4, "big") + + len(points).to_bytes(4, "big") + + heading.to_bytes(2, "big", signed=True) + + reserved.to_bytes(2, "big") + ) + point_table = b"".join(x.to_bytes(2, "big", signed=True) + y.to_bytes(2, "big", signed=True) for x, y in points) + history = bytes([prefix]) + header + point_table + payload = bytearray(FIXTURE.read_bytes() + _carpet_tail(8, 6, bytes(48)) + history + trailing) + payload[:2] = marker + return bytes(payload) + + def test_parse_carpet_mask_from_map_packet_tail() -> None: """A carpet mask after the erase section decodes to a same-dims grid. @@ -415,6 +472,84 @@ def test_parse_map_packet_without_carpet() -> None: assert parse_map_packet(FIXTURE.read_bytes()).carpet_mask is None +def test_classify_current_clean_record_and_saved_map_packets() -> None: + """All known map markers retain an explicit semantic kind.""" + current = FIXTURE.read_bytes() + clean_record = _map_detail_payload(b"\x03\x01", [(10, -20)]) + saved_map = _map_detail_payload(b"\x04\x01", [(10, -20)]) + + assert is_map_packet(current) + assert is_clean_record_map_packet(clean_record) + assert is_saved_map_packet(saved_map) + assert parse_map_packet(current).kind is Q10MapPacketKind.CURRENT + assert parse_map_packet(clean_record).kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL + assert parse_map_packet(saved_map).kind is Q10MapPacketKind.SAVED_MAP_DETAIL + + +def test_parse_clean_record_historical_trace_with_unknown_tail() -> None: + """The bounded historical path is decoded without interpreting later bytes.""" + packet = parse_map_packet(_map_detail_payload(b"\x03\x01", [(10, -20), (-30, 40)], trailing=b"future-section")) + + assert packet.historical_trace is not None + assert [(point.x, point.y) for point in packet.historical_trace.points] == [(10, -20), (-30, 40)] + assert packet.historical_trace.version == 1 + assert packet.historical_trace.opaque_value == 2 + assert packet.historical_trace.heading == 3 + assert packet.historical_trace.robot_position == Q10Point(-30, 40) + + +def test_zero_point_historical_trace_with_following_section() -> None: + """A zero-point path remains valid when a later section follows it.""" + packet = parse_map_packet(_map_detail_payload(b"\x03\x01", [], trailing=b"recorded-path")) + + assert packet.historical_trace is not None + assert packet.historical_trace.points == [] + + +def test_historical_trace_is_not_inferred_for_other_packet_kinds() -> None: + """The validated ``03 01`` layout is not assumed for current or saved maps.""" + current = parse_map_packet(_map_detail_payload(b"\x01\x01", [(10, -20)])) + saved_map = parse_map_packet(_map_detail_payload(b"\x04\x01", [(10, -20)])) + + assert current.historical_trace is None + assert saved_map.historical_trace is None + + +@pytest.mark.parametrize( + "kwargs", + [ + {"version": 2}, + {"reserved": 1}, + {"prefix": 1}, + ], +) +def test_unsupported_historical_trace_header_is_ignored(kwargs: dict[str, Any]) -> None: + payload = _map_detail_payload(b"\x03\x01", [(10, -20)], **kwargs) + packet = parse_map_packet(payload) + + assert packet.historical_trace is None + + +def test_truncated_historical_trace_is_ignored() -> None: + payload = _map_detail_payload(b"\x03\x01", [(10, -20)])[:-2] + packet = parse_map_packet(payload) + + assert packet.historical_trace is None + + +def test_invalid_erase_section_is_ignored() -> None: + """An invalid erase header cannot become an anchor for later sections.""" + tail = b"\x01\xffopaque-tail" + payload = bytearray(FIXTURE.read_bytes() + tail) + payload[:2] = b"\x03\x01" + + packet = parse_map_packet(bytes(payload)) + + assert packet.erase_zones == [] + assert packet.carpet_mask is None + assert packet.historical_trace is None + + def test_carpet_mask_ignored_when_uncompressed_len_mismatches() -> None: """If the section doesn't line up (uncompressed_len != w*h) carpet is dropped.""" carpet = bytes([4] * 48) diff --git a/tests/map/test_b01_q10_render.py b/tests/map/test_b01_q10_render.py index c8436aa2..4dd9013d 100644 --- a/tests/map/test_b01_q10_render.py +++ b/tests/map/test_b01_q10_render.py @@ -18,6 +18,7 @@ B01Q10MapParserConfig, Q10EraseZone, Q10HeaderCalibration, + Q10HistoricalTracePacket, Q10MapPacket, Q10Point, Q10TracePacket, @@ -58,7 +59,7 @@ def _packet() -> Q10MapPacket: def _render( packet: Q10MapPacket | None = None, *, - trace: Q10TracePacket | None = None, + trace: Q10TracePacket | Q10HistoricalTracePacket | None = None, overlays: Q10MapOverlays | None = None, ) -> bytes: return render_q10_map( @@ -117,6 +118,14 @@ def test_render_draws_path_and_position() -> None: assert rendered.getpixel(image_position) == (255, 255, 255, 255) +def test_render_accepts_historical_trace() -> None: + """A validated clean-record path uses the same calibrated drawing path.""" + packet, live_trace = _calibrated_inputs() + historical = Q10HistoricalTracePacket(points=live_trace.points, heading=live_trace.heading) + + assert _render(packet, trace=historical) == _render(packet, trace=live_trace) + + def test_render_draws_zones_and_virtual_walls() -> None: """Decoded DPS overlays are included in the composed image.""" packet, trace = _calibrated_inputs() diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index 5fddc41a..adf62f8c 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -13,7 +13,7 @@ from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXWaterLevel from roborock.data.code_mappings import completed_warnings from roborock.exceptions import RoborockException -from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket +from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10MapPacketKind, Q10TracePacket from roborock.protocols.b01_q10_protocol import ( Q10DpsUpdate, decode_message, @@ -69,6 +69,22 @@ def test_decode_message_map_packet() -> None: assert {room.id: room.name for room in decoded.rooms} == {2: "Living Room", 3: "Bedroom"} +@pytest.mark.parametrize( + ("marker", "kind"), + [ + (b"\x03\x01", Q10MapPacketKind.CLEAN_RECORD_DETAIL), + (b"\x04\x01", Q10MapPacketKind.SAVED_MAP_DETAIL), + ], +) +def test_decode_message_archived_map_packet(marker: bytes, kind: Q10MapPacketKind) -> None: + """The decoder recognizes both archived map-detail markers.""" + fixture = MAP_FIXTURE.read_bytes() + decoded = decode_message(_message(marker + fixture[2:], RoborockMessageProtocol.MAP_RESPONSE)) + + assert isinstance(decoded, Q10MapPacket) + assert decoded.kind is kind + + def test_decode_message_trace_packet() -> None: """A MAP_RESPONSE 02 01 payload decodes into a Q10TracePacket.""" message = _message(TRACE_FIXTURE.read_bytes(), RoborockMessageProtocol.MAP_RESPONSE) From eaac14189e00d8612825e831820b89d25579b764 Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Mon, 31 Aug 2026 15:06:07 +0100 Subject: [PATCH 3/5] feat: expose Q10 map archives --- roborock/cli.py | 15 +- roborock/data/b01_q10/b01_q10_containers.py | 4 +- roborock/devices/device_manager.py | 8 +- roborock/devices/traits/b01/q10/__init__.py | 45 +++- .../devices/traits/b01/q10/clean_history.py | 84 ++++++- roborock/devices/traits/b01/q10/map.py | 33 ++- roborock/devices/traits/b01/q10/maps.py | 84 ++++++- .../traits/b01/q10/test_clean_history.py | 53 ++++ tests/devices/traits/b01/q10/test_map.py | 229 ++++++++++++++++-- 9 files changed, 505 insertions(+), 50 deletions(-) diff --git a/roborock/cli.py b/roborock/cli.py index a83caa15..f33a1a6b 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -597,12 +597,13 @@ async def maps(ctx, device_id: str): # The Q10 publishes its current map asynchronously after a REQUEST_DPS. 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 +_Q10_MAP_PUSH_TIMEOUT = 75.0 async def _await_q10_map_push( properties: Q10PropertiesApi, predicate: Callable[[], bool], + revision: Callable[[], int], *, timeout: float = _Q10_MAP_PUSH_TIMEOUT, allow_cached_on_timeout: bool = False, @@ -615,9 +616,10 @@ async def _await_q10_map_push( """ loop = asyncio.get_running_loop() updated: asyncio.Future[None] = loop.create_future() + initial_revision = revision() def on_update() -> None: - if predicate() and not updated.done(): + if revision() > initial_revision and predicate() and not updated.done(): updated.set_result(None) unsub = properties.map.add_update_listener(on_update) @@ -647,6 +649,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, + lambda: properties.map.map_revision, allow_cached_on_timeout=True, ) image_content = properties.map.image_content @@ -695,8 +698,8 @@ async def map_data(ctx, device_id: str, include_path: bool): async def q10_position(ctx, device_id: str, include_path: bool): """Get the current Q10 robot position and live cleaning path. - The Q10 only streams its position/path while it is actively cleaning, so this - will report that no live trace is available for an idle/docked robot. + The Q10 normally streams position/path while it is actively cleaning, so an + idle device may report that no fresh live trace is available. """ context: RoborockContext = ctx.obj device_manager = await context.get_device_manager() @@ -708,9 +711,10 @@ async def q10_position(ctx, device_id: str, include_path: bool): got_trace = await _await_q10_map_push( properties, lambda: bool(properties.map.path), + lambda: properties.map.trace_revision, ) if not got_trace: - click.echo("No live trace available (the robot only reports position while cleaning).") + click.echo("No fresh live trace available.") return map_trait = properties.map position = map_trait.robot_position @@ -873,6 +877,7 @@ async def rooms(ctx, device_id: str): await _await_q10_map_push( properties, lambda: properties.map.image_content is not None, + lambda: properties.map.map_revision, allow_cached_on_timeout=True, ) click.echo(dump_json({room.id: room.name for room in properties.map.rooms})) diff --git a/roborock/data/b01_q10/b01_q10_containers.py b/roborock/data/b01_q10/b01_q10_containers.py index 00384e09..c88aa4b7 100644 --- a/roborock/data/b01_q10/b01_q10_containers.py +++ b/roborock/data/b01_q10/b01_q10_containers.py @@ -88,7 +88,9 @@ class Q10MapInfo(RoborockBase): """A saved map reported by ``dpMultiMap``. Q10 firmware represents the map identifier as a string on the wire. The - value is sent back unchanged in a subsequent ``{"op": "get"}`` request. + value is sent back unchanged in a subsequent ``{"op": "select"}`` detail + request. On Q10 firmware, ``select`` previews a saved map without applying + it as the active map. """ id: str diff --git a/roborock/devices/device_manager.py b/roborock/devices/device_manager.py index cdc8b0ae..0858d210 100644 --- a/roborock/devices/device_manager.py +++ b/roborock/devices/device_manager.py @@ -19,6 +19,7 @@ from roborock.devices.device import DeviceReadyCallback, RoborockDevice from roborock.diagnostics import Diagnostics, redact_device_data from roborock.exceptions import RoborockException +from roborock.map.b01_q10_map_parser import B01Q10MapParserConfig from roborock.map.map_parser import MapParserConfig from roborock.mqtt.roborock_session import create_lazy_mqtt_session from roborock.mqtt.session import MqttSession, SessionUnauthorizedHook @@ -262,7 +263,12 @@ def device_creator(home_data: HomeData, device: HomeDataDevice, product: HomeDat if "ss" in model_part: b01_q10_channel = create_b01_q10_channel(mqtt_channel) channel = b01_q10_channel - trait = b01.q10.create(channel) + trait = b01.q10.create( + channel, + map_parser_config=( + B01Q10MapParserConfig(map_scale=map_parser_config.map_scale) if map_parser_config else None + ), + ) elif "sc" in model_part: # Q7 devices start with 'sc' in their model naming. b01_q7_channel = create_b01_q7_channel(device, product, mqtt_channel) diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index c2b66f71..d675b44b 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -8,7 +8,12 @@ from roborock.data.containers import RoborockBase from roborock.devices.rpc.b01_q10_channel import B01Q10Channel from roborock.devices.traits import Trait -from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket +from roborock.map.b01_q10_map_parser import ( + B01Q10MapParserConfig, + Q10MapPacket, + Q10MapPacketKind, + Q10TracePacket, +) from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message from .button_light import ButtonLightTrait @@ -92,7 +97,12 @@ class Q10PropertiesApi(Trait): clean_history: CleanHistoryTrait """Trait for fetching the device clean-record history (``dpCleanRecord``).""" - def __init__(self, channel: B01Q10Channel) -> None: + def __init__( + self, + channel: B01Q10Channel, + *, + map_parser_config: B01Q10MapParserConfig, + ) -> None: """Initialize the B01Props API.""" self._channel = channel self.command = CommandTrait(channel) @@ -107,9 +117,16 @@ def __init__(self, channel: B01Q10Channel) -> None: self.network_info = NetworkInfoTrait() self.consumable = ConsumableTrait() self._map_dps = MapDpsTrait() - self.maps = MapsTrait(self.command) - self.map = MapContentTrait(self._map_dps, self.maps, self.command) - self.clean_history = CleanHistoryTrait(self.command) + self.maps = MapsTrait(self.command, map_parser_config=map_parser_config) + self.map = MapContentTrait( + self._map_dps, + self.command, + map_parser_config=map_parser_config, + ) + self.clean_history = CleanHistoryTrait( + self.command, + map_parser_config=map_parser_config, + ) # Read-model traits updated from the device's DPS push stream. self._updatable_traits = [ self.status, @@ -157,7 +174,12 @@ def _handle_message(self, message: Q10Message) -> None: Map-list DPS responses and other DPS updates feed the read-model traits. """ if isinstance(message, Q10MapPacket): - self.map.update_from_map_packet(message) + if message.kind is Q10MapPacketKind.CURRENT: + self.map.update_from_map_packet(message) + elif message.kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL: + self.clean_history.update_from_map_packet(message) + elif message.kind is Q10MapPacketKind.SAVED_MAP_DETAIL: + self.maps.update_from_map_packet(message) elif isinstance(message, Q10TracePacket): self.map.update_from_trace_packet(message) elif isinstance(message, Q10DpsUpdate): @@ -178,6 +200,13 @@ def as_dict(self) -> dict[str, Any]: return result -def create(channel: B01Q10Channel) -> Q10PropertiesApi: +def create( + channel: B01Q10Channel, + *, + map_parser_config: B01Q10MapParserConfig | None = None, +) -> Q10PropertiesApi: """Create traits for B01 devices.""" - return Q10PropertiesApi(channel) + return Q10PropertiesApi( + channel, + map_parser_config=map_parser_config or B01Q10MapParserConfig(), + ) diff --git a/roborock/devices/traits/b01/q10/clean_history.py b/roborock/devices/traits/b01/q10/clean_history.py index fde2bb2f..224af7b6 100644 --- a/roborock/devices/traits/b01/q10/clean_history.py +++ b/roborock/devices/traits/b01/q10/clean_history.py @@ -22,6 +22,15 @@ YXStartMethod, ) from roborock.data.b01_q10.b01_q10_containers import Q10CleanRecord +from roborock.exceptions import RoborockException +from roborock.map.b01_q10_map_parser import ( + B01Q10MapParserConfig, + Q10HistoricalTracePacket, + Q10MapPacket, + Q10MapPacketKind, + Q10Point, +) +from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map from .command import CommandTrait from .common import UpdatableTrait @@ -115,12 +124,28 @@ class CleanHistoryTrait(UpdatableTrait): or a single ``op:"notify"`` record) rather than a flat data-point-to-field map. """ - def __init__(self, command: CommandTrait) -> None: + _command: CommandTrait + + def __init__( + self, + command: CommandTrait, + *, + map_parser_config: B01Q10MapParserConfig | None = None, + ) -> None: """Initialize the clean history trait.""" UpdatableTrait.__init__(self, command, _LOGGER) + self._command = command self._converter = CleanRecordConverter() + self._map_parser_config = map_parser_config or B01Q10MapParserConfig() self.records: list[Q10CleanRecord] = [] """Decoded clean records, most recent first.""" + self.detail_packet: Q10MapPacket | None = None + """Most recently pushed ``03 01`` clean-record map detail.""" + self.detail_record: Q10CleanRecord | None = None + """Record associated with :attr:`detail_packet`, when requested here.""" + self.detail_image_content: bytes | None = None + """Rendered clean-record detail image, if decoding succeeded.""" + self._pending_detail_record: Q10CleanRecord | None = None @property def last_record(self) -> Q10CleanRecord | None: @@ -134,13 +159,47 @@ async def refresh(self) -> None: asynchronously on the device stream and populate :attr:`records` once :meth:`update_from_dps` processes the ``dpCleanRecord`` push. """ - if self._command is None: - raise ValueError("Trait is read-only; no command channel was provided") await self._command.send( B01_Q10_DP.COMMON, params={str(B01_Q10_DP.CLEAN_RECORD.code): {"op": "list"}}, ) + async def refresh_detail(self, record: Q10CleanRecord) -> None: + """Request the saved map and path for one clean record. + + The complete 12-field raw record is the firmware's detail identifier; + the shorter human-facing record ID is not accepted. Only one request + may be outstanding because ``03 01`` responses carry no correlation ID. + """ + if not record.raw or not record.map_len: + raise RoborockException("The Q10 clean record has no saved map detail") + if self._pending_detail_record is not None: + raise RoborockException("A Q10 clean-record detail request is already pending") + self._pending_detail_record = record + try: + await self._command.send( + B01_Q10_DP.COMMON, + params={ + str(B01_Q10_DP.CLEAN_RECORD.code): { + "op": "select", + "id": record.raw, + } + }, + ) + except RoborockException: + self._pending_detail_record = None + raise + + @property + def detail_trace(self) -> Q10HistoricalTracePacket | None: + """Historical path embedded in the selected clean-record detail.""" + return self.detail_packet.historical_trace if self.detail_packet else None + + @property + def detail_path(self) -> list[Q10Point]: + """Historical path points for the selected clean record.""" + return self.detail_trace.points if self.detail_trace else [] + def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: """Apply a ``dpCleanRecord`` push (a full list reply or a single notify).""" envelope = decoded_dps.get(B01_Q10_DP.CLEAN_RECORD) @@ -151,6 +210,25 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: return self._apply(push) + def update_from_map_packet(self, packet: Q10MapPacket) -> None: + """Store and render a pushed clean-record detail map.""" + if packet.kind is not Q10MapPacketKind.CLEAN_RECORD_DETAIL: + raise ValueError(f"Expected a Q10 clean-record detail packet, got {packet.kind.value}") + self.detail_record = self._pending_detail_record + self._pending_detail_record = None + self.detail_packet = packet + try: + self.detail_image_content = render_q10_map( + packet, + packet.historical_trace, + Q10MapOverlays(), + config=self._map_parser_config, + ) + except RoborockException: + _LOGGER.debug("Failed to render Q10 clean-record detail", exc_info=True) + self.detail_image_content = None + self._notify_update() + def _apply(self, push: CleanRecordPush) -> None: """Merge or replace the records from ``push``, then sort newest-first and notify.""" if push.replace: diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index 5890e688..d7b52010 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -24,6 +24,7 @@ from roborock.map.b01_q10_map_parser import ( B01Q10MapParserConfig, Q10MapPacket, + Q10MapPacketKind, Q10Point, Q10Room, Q10TracePacket, @@ -33,7 +34,6 @@ from .command import CommandTrait from .common import UpdatableTrait -from .maps import MapsTrait _LOGGER = logging.getLogger(__name__) _DOCKED_STATES = {YXDeviceState.CHARGING, YXDeviceState.EMPTYING_THE_BIN} @@ -83,15 +83,13 @@ class MapContentTrait(TraitUpdateListener): """High-level composed Q10 map view. The latest map and trace packets are combined with the injected - :class:`MapDpsTrait` whenever a source changes. The - :class:`MapsTrait` supplies a stored ID only when this trait requests - content. + :class:`MapDpsTrait` whenever a source changes. Current-map acquisition is + independent of the saved-map list. """ def __init__( self, map_dps: MapDpsTrait, - maps: MapsTrait, command: CommandTrait, *, map_parser_config: B01Q10MapParserConfig | None = None, @@ -99,11 +97,12 @@ def __init__( TraitUpdateListener.__init__(self, logger=_LOGGER) self._config = map_parser_config or B01Q10MapParserConfig() self._map_dps = map_dps - self._maps = maps self._command = command self._map_packet: Q10MapPacket | None = None self._trace_packet: Q10TracePacket | None = None self._image_content: bytes | None = None + self._map_revision = 0 + self._trace_revision = 0 self._map_dps.add_update_listener(self._map_dps_updated) async def refresh(self) -> None: @@ -120,6 +119,16 @@ def image_content(self) -> bytes | None: """The composed map PNG, if the latest map rendered successfully.""" return self._image_content + @property + def map_revision(self) -> int: + """Monotonic revision incremented only by current-map packets.""" + return self._map_revision + + @property + def trace_revision(self) -> int: + """Monotonic revision incremented only by live-trace state changes.""" + return self._trace_revision + @property def rooms(self) -> list[Q10Room]: """Rooms reported by the device.""" @@ -142,18 +151,28 @@ def robot_heading(self) -> int | None: def update_from_map_packet(self, packet: Q10MapPacket) -> None: """Store a map-protocol update and render the latest sources.""" + if packet.kind is not Q10MapPacketKind.CURRENT: + raise ValueError(f"Expected a current Q10 map packet, got {packet.kind.value}") self._map_packet = packet + self._map_revision += 1 self._render() self._notify_update() def update_from_trace_packet(self, packet: Q10TracePacket) -> None: """Store a trace-protocol update and render the latest sources.""" - self._trace_packet = packet + self._trace_packet = None if self._map_dps.robot_at_dock else packet + self._trace_revision += 1 self._render() self._notify_update() def _map_dps_updated(self) -> None: """Render after the low-level map DPS source changes.""" + if self._map_dps.robot_at_dock and self._trace_packet is not None: + # A completed cleaning trace is not the current robot position once + # the device is docked. Clear the public live-path state even if the + # firmware does not send its usual zero-point trace. + self._trace_packet = None + self._trace_revision += 1 if self._map_packet is None: return self._render() diff --git a/roborock/devices/traits/b01/q10/maps.py b/roborock/devices/traits/b01/q10/maps.py index d0945776..9683d711 100644 --- a/roborock/devices/traits/b01/q10/maps.py +++ b/roborock/devices/traits/b01/q10/maps.py @@ -6,8 +6,11 @@ from roborock.data import RoborockBase from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP -from roborock.data.b01_q10.b01_q10_containers import dpMultiMap +from roborock.data.b01_q10.b01_q10_containers import Q10MapInfo, dpMultiMap from roborock.devices.traits.common import DpsDataConverter +from roborock.exceptions import RoborockException +from roborock.map.b01_q10_map_parser import B01Q10MapParserConfig, Q10MapPacket, Q10MapPacketKind +from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map from .command import CommandTrait from .common import UpdatableTrait @@ -28,6 +31,13 @@ def current_map_id(self) -> str | None: return None return self.multi_map.current_map_id + @property + def map_list(self) -> list[Q10MapInfo]: + """Return a copy of the successfully reported saved-map list.""" + if self.multi_map is None or self.multi_map.op != "list" or self.multi_map.result != 1: + return [] + return list(self.multi_map.data) + class MapsTrait(Maps, UpdatableTrait): """Request and store the Q10 saved-map list.""" @@ -35,11 +45,24 @@ class MapsTrait(Maps, UpdatableTrait): _CONVERTER = DpsDataConverter.from_dataclass(Maps) _command: CommandTrait - def __init__(self, command: CommandTrait) -> None: + def __init__( + self, + command: CommandTrait, + *, + map_parser_config: B01Q10MapParserConfig | None = None, + ) -> None: """Initialize the saved-map list trait.""" Maps.__init__(self) UpdatableTrait.__init__(self, command, _LOGGER) self._command = command + self._map_parser_config = map_parser_config or B01Q10MapParserConfig() + self.detail_packet: Q10MapPacket | None = None + """Most recently pushed ``04 01`` saved-map detail.""" + self.detail_map_id: str | None = None + """Saved-map ID associated with :attr:`detail_packet`.""" + self.detail_image_content: bytes | None = None + """Rendered saved-map detail image, if decoding succeeded.""" + self._pending_detail_map_id: str | None = None async def refresh(self) -> None: """Request a new saved-map list from the device.""" @@ -48,6 +71,36 @@ async def refresh(self) -> None: {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}}, ) + async def refresh_detail(self, map_id: str | None = None) -> None: + """Request a read-only preview for one saved map. + + The device delivers the result asynchronously as a ``04 01`` map + response, which :meth:`update_from_map_packet` stores separately from + the live map. + """ + if map_id is None: + map_id = self.current_map_id + if map_id is None: + raise RoborockException("Cannot request Q10 saved-map detail before the map list is available") + if map_id not in {map_info.id for map_info in self.map_list}: + raise RoborockException(f"Unknown Q10 saved-map ID: {map_id}") + if self._pending_detail_map_id is not None: + raise RoborockException("A Q10 saved-map detail request is already pending") + self._pending_detail_map_id = map_id + try: + await self._command.send( + B01_Q10_DP.COMMON, + { + str(B01_Q10_DP.MULTI_MAP.code): { + "op": "select", + "id": map_id, + } + }, + ) + except RoborockException: + self._pending_detail_map_id = None + raise + def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: """Store a successful saved-map list response.""" response = decoded_dps.get(B01_Q10_DP.MULTI_MAP) @@ -56,3 +109,30 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: if not isinstance(response, dict) or response.get("op") != "list" or response.get("result") != 1: return super().update_from_dps(decoded_dps) + + def update_from_map_packet(self, packet: Q10MapPacket) -> None: + """Store and render a pushed saved-map detail packet.""" + if packet.kind is not Q10MapPacketKind.SAVED_MAP_DETAIL: + raise ValueError(f"Expected a Q10 saved-map detail packet, got {packet.kind.value}") + packet_map_id = str(packet.map_id) + if self._pending_detail_map_id is not None and packet_map_id != self._pending_detail_map_id: + _LOGGER.debug( + "Ignoring Q10 saved-map detail for map %s while waiting for %s", + packet_map_id, + self._pending_detail_map_id, + ) + return + self.detail_map_id = packet_map_id + self._pending_detail_map_id = None + self.detail_packet = packet + try: + self.detail_image_content = render_q10_map( + packet, + None, + Q10MapOverlays(), + config=self._map_parser_config, + ) + except RoborockException: + _LOGGER.debug("Failed to render Q10 saved-map detail", exc_info=True) + self.detail_image_content = None + self._notify_update() diff --git a/tests/devices/traits/b01/q10/test_clean_history.py b/tests/devices/traits/b01/q10/test_clean_history.py index 10fce2b6..fb4c78da 100644 --- a/tests/devices/traits/b01/q10/test_clean_history.py +++ b/tests/devices/traits/b01/q10/test_clean_history.py @@ -1,3 +1,5 @@ +from pathlib import Path + import pytest from roborock.data.b01_q10.b01_q10_code_mappings import ( @@ -10,6 +12,8 @@ from roborock.data.b01_q10.b01_q10_containers import Q10CleanRecord from roborock.devices.traits.b01.q10 import Q10PropertiesApi from roborock.devices.traits.b01.q10.clean_history import CleanHistoryTrait, CleanRecordConverter +from roborock.exceptions import RoborockException +from roborock.map.b01_q10_map_parser import parse_map_packet from .conftest import FakeB01Q10Channel @@ -175,3 +179,52 @@ async def test_refresh_sends_op_list(q10_api: Q10PropertiesApi, fake_channel: Fa B01_Q10_DP.COMMON, {"52": {"op": "list"}}, ) + + +async def test_refresh_detail_sends_full_raw_record( + clean_history: CleanHistoryTrait, + fake_channel: FakeB01Q10Channel, +) -> None: + record = CleanRecordConverter.parse_record(RECORD_A) + assert record is not None + + await clean_history.refresh_detail(record) + + assert fake_channel.published_commands == [ + ( + B01_Q10_DP.COMMON, + {"52": {"op": "select", "id": RECORD_A}}, + ) + ] + + +async def test_refresh_detail_rejects_record_without_map(clean_history: CleanHistoryTrait) -> None: + record = CleanRecordConverter.parse_record("x_1781226271_1_1_0_0_0_0_2_1_1_0") + assert record is not None + + with pytest.raises(RoborockException, match="no saved map detail"): + await clean_history.refresh_detail(record) + + +async def test_refresh_detail_rejects_parallel_request(clean_history: CleanHistoryTrait) -> None: + record = CleanRecordConverter.parse_record(RECORD_A) + assert record is not None + await clean_history.refresh_detail(record) + + with pytest.raises(RoborockException, match="already pending"): + await clean_history.refresh_detail(record) + + +async def test_detail_response_is_associated_with_pending_record( + clean_history: CleanHistoryTrait, +) -> None: + record = CleanRecordConverter.parse_record(RECORD_A) + assert record is not None + await clean_history.refresh_detail(record) + fixture = Path("tests/map/testdata/b01_q10_map.bin").read_bytes() + packet = parse_map_packet(b"\x03\x01" + fixture[2:]) + + clean_history.update_from_map_packet(packet) + + assert clean_history.detail_record is record + assert clean_history.detail_packet is packet diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index ef6aa18f..828ac540 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -1,9 +1,8 @@ """Tests for the Q10 B01 map content trait. -Map list data and map content have independent refresh schedules. Content -requests use a stored map ID, and the device sends the data later in a -``MAP_RESPONSE`` packet. These tests cover that state management. The render -details are tested in ``tests/map/test_b01_q10_render.py``. +Map list data and current-map content have independent refresh schedules. The +device sends map data later in a ``MAP_RESPONSE`` packet. These tests cover +that state management; rendering is tested separately. """ import asyncio @@ -23,6 +22,8 @@ from roborock.devices.traits.b01.q10.maps import MapsTrait from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import ( + B01Q10MapParserConfig, + Q10MapPacketKind, Q10Point, Q10TracePacket, parse_map_packet, @@ -38,19 +39,9 @@ def _map_trait(map_dps: MapDpsTrait | None = None) -> MapContentTrait: - """Create map content with a stored map ID for tests that do not perform I/O.""" + """Create map content for tests that do not perform I/O.""" command = cast(CommandTrait, Mock(spec=CommandTrait)) - maps = MapsTrait(command) - maps.update_from_dps( - { - B01_Q10_DP.MULTI_MAP: { - "data": [{"id": "12345"}], - "op": "list", - "result": 1, - } - } - ) - return MapContentTrait(map_dps or MapDpsTrait(), maps, command) + return MapContentTrait(map_dps or MapDpsTrait(), command) def _zone_blob() -> str: @@ -83,6 +74,15 @@ def test_update_from_map_packet_populates_image_and_rooms() -> None: assert len(updates) == 1 +def test_live_map_trait_rejects_archived_packet() -> None: + """Direct callers cannot bypass API routing and replace live map state.""" + payload = FIXTURE.read_bytes() + archived = parse_map_packet(b"\x03\x01" + payload[2:]) + + with pytest.raises(ValueError, match="Expected a current Q10 map packet"): + _map_trait().update_from_map_packet(archived) + + def test_update_from_trace_packet_populates_path_and_position() -> None: """A pushed 02 01 trace packet populates the path, position and heading.""" trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) @@ -120,7 +120,7 @@ def __init__(self) -> None: } } ) - self.map = MapContentTrait(MapDpsTrait(), self.maps, command) + self.map = MapContentTrait(MapDpsTrait(), command) self.refresh_count = 0 async def refresh_map() -> None: @@ -148,6 +148,7 @@ async def test_await_q10_map_push_waits_for_fresh_update() -> None: got_trace = await _await_q10_map_push( cast(Q10PropertiesApi, properties), lambda: bool(properties.map.path), + lambda: properties.map.trace_revision, timeout=0.01, ) @@ -161,6 +162,7 @@ async def test_await_q10_map_push_returns_true_after_update() -> None: got_trace = await _await_q10_map_push( cast(Q10PropertiesApi, properties), lambda: bool(properties.map.path), + lambda: properties.map.trace_revision, timeout=0.01, ) @@ -175,6 +177,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, + lambda: properties.map.map_revision, timeout=0.01, allow_cached_on_timeout=True, ) @@ -230,6 +233,86 @@ async def test_subscribe_loop_routes_map_push( assert {room.id: room.name for room in q10_api.map.rooms} == {2: "Living Room", 3: "Bedroom"} +async def test_archived_map_pushes_cannot_overwrite_live_map( + q10_api: Q10PropertiesApi, + message_queue: asyncio.Queue[Q10Message], +) -> None: + """03/04 detail packets are isolated from the current live-map trait.""" + current_bytes = FIXTURE.read_bytes() + current = parse_map_packet(current_bytes) + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + clean_record = parse_map_packet(b"\x03\x01" + current_bytes[2:]) + saved_map = parse_map_packet(b"\x04\x01" + current_bytes[2:]) + + message_queue.put_nowait(current) + message_queue.put_nowait(trace) + await _wait_for(lambda: q10_api.map.image_content is not None and bool(q10_api.map.path)) + live_image = q10_api.map.image_content + live_rooms = list(q10_api.map.rooms) + live_path = list(q10_api.map.path) + live_position = q10_api.map.robot_position + live_heading = q10_api.map.robot_heading + live_updates: list[None] = [] + clean_record_updates: list[None] = [] + saved_map_updates: list[None] = [] + q10_api.map.add_update_listener(lambda: live_updates.append(None)) + q10_api.clean_history.add_update_listener(lambda: clean_record_updates.append(None)) + q10_api.maps.add_update_listener(lambda: saved_map_updates.append(None)) + + message_queue.put_nowait(clean_record) + message_queue.put_nowait(saved_map) + await _wait_for(lambda: q10_api.maps.detail_packet is not None) + + assert q10_api.map.image_content == live_image + assert q10_api.map.rooms == live_rooms + assert q10_api.map.path == live_path + assert q10_api.map.robot_position == live_position + assert q10_api.map.robot_heading == live_heading + assert live_updates == [] + assert q10_api.clean_history.detail_packet is not None + assert q10_api.clean_history.detail_packet.kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL + assert q10_api.clean_history.detail_image_content is not None + assert q10_api.maps.detail_packet is not None + assert q10_api.maps.detail_packet.kind is Q10MapPacketKind.SAVED_MAP_DETAIL + assert q10_api.maps.detail_image_content is not None + assert clean_record_updates == [None] + assert saved_map_updates == [None] + + +def test_archive_owners_reject_wrong_packet_kinds(q10_api: Q10PropertiesApi) -> None: + """Semantic archive traits reject packets owned by another map stream.""" + current = parse_map_packet(FIXTURE.read_bytes()) + + with pytest.raises(ValueError, match="clean-record detail"): + q10_api.clean_history.update_from_map_packet(current) + with pytest.raises(ValueError, match="saved-map detail"): + q10_api.maps.update_from_map_packet(current) + + +def test_all_q10_map_views_share_the_injected_render_config( + fake_channel: FakeB01Q10Channel, +) -> None: + config = B01Q10MapParserConfig(map_scale=2) + api = Q10PropertiesApi(fake_channel, map_parser_config=config) + payload = FIXTURE.read_bytes() + + with ( + patch("roborock.devices.traits.b01.q10.map.render_q10_map", return_value=b"map") as live_render, + patch( + "roborock.devices.traits.b01.q10.clean_history.render_q10_map", + return_value=b"history", + ) as history_render, + patch("roborock.devices.traits.b01.q10.maps.render_q10_map", return_value=b"saved") as saved_render, + ): + api._handle_message(parse_map_packet(payload)) + api._handle_message(parse_map_packet(b"\x03\x01" + payload[2:])) + api._handle_message(parse_map_packet(b"\x04\x01" + payload[2:])) + + assert live_render.call_args.kwargs["config"] is config + assert history_render.call_args.kwargs["config"] is config + assert saved_render.call_args.kwargs["config"] is config + + async def test_subscribe_loop_routes_trace_push( q10_api: Q10PropertiesApi, message_queue: asyncio.Queue[Q10Message], @@ -332,6 +415,106 @@ async def test_map_content_refresh_requests_are_not_rate_limited(q10_api: Q10Pro assert send.await_count == 2 +async def test_saved_map_detail_refresh_uses_current_map_id(q10_api: Q10PropertiesApi) -> None: + """Saved-map detail uses the independently validated select request.""" + q10_api.maps.update_from_dps( + { + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": "12345"}], + "op": "list", + "result": 1, + } + } + ) + with patch.object(q10_api.command, "send") as send: + await q10_api.maps.refresh_detail() + + send.assert_awaited_once_with( + B01_Q10_DP.COMMON, + {str(B01_Q10_DP.MULTI_MAP.code): {"op": "select", "id": "12345"}}, + ) + + +async def test_saved_map_detail_refresh_accepts_any_listed_map_id(q10_api: Q10PropertiesApi) -> None: + q10_api.maps.update_from_dps( + { + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": "12345"}, {"id": "67890"}], + "op": "list", + "result": 1, + } + } + ) + assert [map_info.id for map_info in q10_api.maps.map_list] == ["12345", "67890"] + + with patch.object(q10_api.command, "send") as send: + await q10_api.maps.refresh_detail("67890") + + send.assert_awaited_once_with( + B01_Q10_DP.COMMON, + {str(B01_Q10_DP.MULTI_MAP.code): {"op": "select", "id": "67890"}}, + ) + + +async def test_saved_map_detail_refresh_rejects_unknown_or_parallel_request( + q10_api: Q10PropertiesApi, +) -> None: + q10_api.maps.update_from_dps( + { + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": "12345"}], + "op": "list", + "result": 1, + } + } + ) + with pytest.raises(RoborockException, match="Unknown Q10 saved-map ID"): + await q10_api.maps.refresh_detail("67890") + + await q10_api.maps.refresh_detail("12345") + with pytest.raises(RoborockException, match="already pending"): + await q10_api.maps.refresh_detail("12345") + + +async def test_saved_map_detail_correlates_pending_map_id(q10_api: Q10PropertiesApi) -> None: + requested_id = str(parse_map_packet(FIXTURE.read_bytes()).map_id) + q10_api.maps.update_from_dps( + { + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": requested_id}, {"id": "999"}], + "op": "list", + "result": 1, + } + } + ) + + await q10_api.maps.refresh_detail("999") + packet = parse_map_packet(b"\x04\x01" + FIXTURE.read_bytes()[2:]) + q10_api.maps.update_from_map_packet(packet) + assert q10_api.maps.detail_packet is None + + matching = MapsTrait(q10_api.command) + matching.update_from_dps( + { + B01_Q10_DP.MULTI_MAP: { + "data": [{"id": requested_id}], + "op": "list", + "result": 1, + } + } + ) + await matching.refresh_detail(requested_id) + matching.update_from_map_packet(packet) + assert matching.detail_packet is packet + assert matching.detail_map_id == requested_id + + +async def test_saved_map_detail_refresh_requires_stored_map_id(q10_api: Q10PropertiesApi) -> None: + """Detail cannot be requested until the saved-map list supplies an ID.""" + with pytest.raises(RoborockException, match="map list is available"): + await q10_api.maps.refresh_detail() + + def test_map_get_ack_does_not_replace_saved_map_list(q10_api: Q10PropertiesApi) -> None: """A content acknowledgement cannot remove the stored map ID.""" q10_api._handle_message( @@ -489,8 +672,8 @@ async def test_charging_status_renders_robot_at_dock(render_map: Mock) -> None: assert render_map.call_args.kwargs["robot_at_dock"] is True -def test_docked_state_hides_trace_only_from_rendering(render_map: Mock) -> None: - """A docked render omits the valid trace without deleting source data.""" +def test_docked_state_clears_stale_live_trace(render_map: Mock) -> None: + """A docked update removes a completed path from public live state.""" map_dps = MapDpsTrait() trait = _map_trait(map_dps) packet = parse_map_packet(FIXTURE.read_bytes()) @@ -503,13 +686,13 @@ def test_docked_state_hides_trace_only_from_rendering(render_map: Mock) -> None: map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code}) - assert trait.path == trace.points + assert trait.path == [] assert render_map.call_args.args[1] is None assert render_map.call_args.kwargs["robot_at_dock"] is True -def test_late_trace_is_retained_but_hidden_while_docked(render_map: Mock) -> None: - """A late trace stays available but is not part of a docked render.""" +def test_late_trace_is_ignored_while_docked(render_map: Mock) -> None: + """A delayed trace cannot repopulate public state while docked.""" map_dps = MapDpsTrait() map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code}) trait = _map_trait(map_dps) @@ -519,7 +702,7 @@ def test_late_trace_is_retained_but_hidden_while_docked(render_map: Mock) -> Non trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) trait.update_from_trace_packet(trace) - assert trait.path == trace.points + assert trait.path == [] assert render_map.call_args.args[1] is None From 2161763b52a0356a926e16163ec458bbe07328f5 Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Sat, 29 Aug 2026 22:39:58 +0100 Subject: [PATCH 4/5] feat: render Q10 map obstacle markers --- roborock/devices/device_manager.py | 7 +- .../devices/traits/b01/q10/clean_history.py | 6 ++ roborock/devices/traits/b01/q10/map.py | 7 ++ roborock/devices/traits/b01/q10/maps.py | 12 ++- roborock/map/b01_q10_map_parser.py | 90 ++++++++++++++----- roborock/map/b01_q10_render.py | 35 ++++++-- tests/devices/test_device_manager.py | 10 ++- .../b01/q10/__snapshots__/test_status.ambr | 6 ++ .../traits/b01/q10/test_clean_history.py | 18 +++- tests/devices/traits/b01/q10/test_map.py | 37 ++++++++ tests/map/test_b01_q10_map_parser.py | 83 +++++++++++++++-- tests/map/test_b01_q10_render.py | 61 +++++++++++++ 12 files changed, 338 insertions(+), 34 deletions(-) diff --git a/roborock/devices/device_manager.py b/roborock/devices/device_manager.py index 0858d210..15816dd9 100644 --- a/roborock/devices/device_manager.py +++ b/roborock/devices/device_manager.py @@ -266,7 +266,12 @@ def device_creator(home_data: HomeData, device: HomeDataDevice, product: HomeDat trait = b01.q10.create( channel, map_parser_config=( - B01Q10MapParserConfig(map_scale=map_parser_config.map_scale) if map_parser_config else None + B01Q10MapParserConfig( + map_scale=map_parser_config.map_scale, + drawables=map_parser_config.drawables, + ) + if map_parser_config + else None ), ) elif "sc" in model_part: diff --git a/roborock/devices/traits/b01/q10/clean_history.py b/roborock/devices/traits/b01/q10/clean_history.py index 224af7b6..95003f0b 100644 --- a/roborock/devices/traits/b01/q10/clean_history.py +++ b/roborock/devices/traits/b01/q10/clean_history.py @@ -28,6 +28,7 @@ Q10HistoricalTracePacket, Q10MapPacket, Q10MapPacketKind, + Q10Obstacle, Q10Point, ) from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map @@ -200,6 +201,11 @@ def detail_path(self) -> list[Q10Point]: """Historical path points for the selected clean record.""" return self.detail_trace.points if self.detail_trace else [] + @property + def detail_obstacles(self) -> list[Q10Obstacle]: + """Obstacle markers embedded in the selected clean-record map.""" + return list(self.detail_packet.obstacles) if self.detail_packet else [] + def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: """Apply a ``dpCleanRecord`` push (a full list reply or a single notify).""" envelope = decoded_dps.get(B01_Q10_DP.CLEAN_RECORD) diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index d7b52010..267e0b14 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -25,6 +25,7 @@ B01Q10MapParserConfig, Q10MapPacket, Q10MapPacketKind, + Q10Obstacle, Q10Point, Q10Room, Q10TracePacket, @@ -139,6 +140,11 @@ def path(self) -> list[Q10Point]: """Full path for live status and callers drawing their own map overlay.""" return self._trace_packet.points if self._trace_packet else [] + @property + def obstacles(self) -> list[Q10Obstacle]: + """Position-only obstacle markers reported by the current map.""" + return list(self._map_packet.obstacles) if self._map_packet else [] + @property def robot_position(self) -> Q10Point | None: """Current position for live status and caller-rendered map overlays.""" @@ -199,6 +205,7 @@ def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]: exclude_set = exclude or set() data = { "rooms": [room.as_dict() for room in self.rooms], + "obstacles": [obstacle.as_dict() for obstacle in self.obstacles], "path": [point.as_dict() for point in self.path], "robotPosition": self.robot_position.as_dict() if self.robot_position is not None else None, "robotHeading": self.robot_heading, diff --git a/roborock/devices/traits/b01/q10/maps.py b/roborock/devices/traits/b01/q10/maps.py index 9683d711..c6ded0ad 100644 --- a/roborock/devices/traits/b01/q10/maps.py +++ b/roborock/devices/traits/b01/q10/maps.py @@ -9,7 +9,12 @@ from roborock.data.b01_q10.b01_q10_containers import Q10MapInfo, dpMultiMap from roborock.devices.traits.common import DpsDataConverter from roborock.exceptions import RoborockException -from roborock.map.b01_q10_map_parser import B01Q10MapParserConfig, Q10MapPacket, Q10MapPacketKind +from roborock.map.b01_q10_map_parser import ( + B01Q10MapParserConfig, + Q10MapPacket, + Q10MapPacketKind, + Q10Obstacle, +) from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map from .command import CommandTrait @@ -64,6 +69,11 @@ def __init__( """Rendered saved-map detail image, if decoding succeeded.""" self._pending_detail_map_id: str | None = None + @property + def detail_obstacles(self) -> list[Q10Obstacle]: + """Obstacle markers embedded in the selected saved-map preview.""" + return list(self.detail_packet.obstacles) if self.detail_packet else [] + async def refresh(self) -> None: """Request a new saved-map list from the device.""" await self._command.send( diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index 2c105e22..0760218e 100644 --- a/roborock/map/b01_q10_map_parser.py +++ b/roborock/map/b01_q10_map_parser.py @@ -25,9 +25,11 @@ import statistics import struct from dataclasses import dataclass, field, replace +from typing import TypeVar from PIL import Image from vacuum_map_parser_base.config.color import ColorsPalette, SupportedColor +from vacuum_map_parser_base.config.drawable import Drawable from vacuum_map_parser_base.config.image_config import ImageConfig from vacuum_map_parser_base.map_data import ImageData, MapData, Point @@ -234,6 +236,10 @@ class Q10MapPacket: carpet (the value is the carpet kind). ``None`` if the packet carried none.""" historical_trace: "Q10HistoricalTracePacket | None" = None """Cleaning path embedded in a clean-record detail packet, if present.""" + obstacles: list["Q10Obstacle"] = field(default_factory=list) + """Obstacle markers embedded after the carpet block (50 raw units/pixel).""" + skip_cleaning_points: list["Q10Point"] = field(default_factory=list) + """Firmware skip-clean markers embedded after obstacles (10 raw units/pixel).""" @property def layers(self) -> GridLayers: @@ -251,6 +257,19 @@ class Q10Point(RoborockBase): y: int +@dataclass +class Q10Obstacle(Q10Point): + """A Q10 map obstacle marker in its raw map-package coordinate frame. + + The map package supplies positions only: there is no validated type, + confidence, or photo identifier on this model. Fifty raw units equal one + occupancy-grid pixel; placement is anchored by the map header origin. + """ + + +_PointType = TypeVar("_PointType", bound=Q10Point) + + @dataclass class Q10TracePacket: """Decoded contents of a Q10 ``02 01`` cleaning-path packet. @@ -332,13 +351,12 @@ def robot_position(self) -> Q10Point | None: _TRACE_POINT_COUNT_OFFSET = 8 _TRACE_HEADING_OFFSET = 10 -_HISTORICAL_TRACE_HEADER_LENGTH = 14 -_HISTORICAL_TRACE_PREFIX_LENGTH = 1 +_HISTORICAL_TRACE_HEADER_LENGTH = 13 _HISTORICAL_TRACE_VERSION = 1 -_HISTORICAL_TRACE_OPAQUE_VALUE_OFFSET = 2 -_HISTORICAL_TRACE_POINT_COUNT_OFFSET = 6 -_HISTORICAL_TRACE_HEADING_OFFSET = 10 -_HISTORICAL_TRACE_RESERVED_OFFSET = 12 +_HISTORICAL_TRACE_OPAQUE_VALUE_OFFSET = 1 +_HISTORICAL_TRACE_POINT_COUNT_OFFSET = 5 +_HISTORICAL_TRACE_HEADING_OFFSET = 9 +_HISTORICAL_TRACE_RESERVED_OFFSET = 11 # Some cleans still prepend a single near-origin sentinel as the first real # point (e.g. ~(5, 76) / (-3, 0) when the path proper starts near (-1700, -800)); @@ -565,10 +583,18 @@ def parse_map_packet(payload: bytes) -> Q10MapPacket: tail = payload[layout_end:] erase_zones = _parse_erase_zones(tail) carpet_mask, carpet_end = _parse_carpet_block(tail, width, height) - if kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL and carpet_end is not None: - historical_trace, _ = _parse_clean_record_trace(tail, carpet_end) - else: - historical_trace = None + obstacles: list[Q10Obstacle] = [] + skip_cleaning_points: list[Q10Point] = [] + historical_trace = None + if carpet_end is not None: + parsed_obstacles, obstacle_end = _parse_counted_points(tail, carpet_end, Q10Obstacle) + if obstacle_end is not None: + parsed_skip_points, skip_end = _parse_counted_points(tail, obstacle_end, Q10Point) + if skip_end is not None: + obstacles = parsed_obstacles + skip_cleaning_points = parsed_skip_points + if kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL: + historical_trace, _ = _parse_clean_record_trace(tail, skip_end) header_calibration = _parse_header_calibration(payload) return Q10MapPacket( kind=kind, @@ -580,6 +606,8 @@ def parse_map_packet(payload: bytes) -> Q10MapPacket: erase_zones=erase_zones, header_calibration=header_calibration, carpet_mask=carpet_mask, + obstacles=obstacles, + skip_cleaning_points=skip_cleaning_points, historical_trace=historical_trace, ) @@ -701,6 +729,29 @@ def _parse_carpet_mask(tail: bytes, width: int, height: int) -> bytes | None: return _parse_carpet_block(tail, width, height)[0] +def _parse_counted_points( + tail: bytes, + offset: int, + point_type: type[_PointType], +) -> tuple[list[_PointType], int | None]: + """Decode one bounded ``u8 count`` + signed-BE ``(x, y)`` point table. + + Obstacle and skip-clean sections use the same framing but different + coordinate scales. The caller owns those semantics; this helper only + validates and decodes the table atomically. A truncated table returns no + points and no end offset, preventing later sections from being misaligned. + """ + if offset >= len(tail): + return [], None + count = tail[offset] + points_start = offset + 1 + points_end = points_start + count * 4 + if points_end > len(tail): + return [], None + coordinates = struct.iter_unpack(">hh", memoryview(tail)[points_start:points_end]) + return ([point_type(x=x, y=y) for x, y in coordinates], points_end) + + def _parse_clean_record_trace( tail: bytes, offset: int, @@ -709,20 +760,17 @@ def _parse_clean_record_trace( The header and declared point count were validated against a physical ss07 clean-record response and its point bytes match captured prefixes of the - corresponding live trace exactly. One observed zero byte precedes the path; - its meaning is unknown, so a non-zero value makes the entire section opaque. - Any unsupported version, non-zero reserved word, or truncated point table is - likewise left completely opaque. Bytes after the declared points are - deliberately not consumed: the observed 12-byte suffix appears structured, - but there is not enough controlled evidence to name or decode it safely. + corresponding live trace exactly. The caller first consumes the obstacle + and skip-clean point tables; ``offset`` therefore starts at the one-byte + path version. Any unsupported version, non-zero reserved word, or truncated + point table is left completely opaque. Bytes after the declared points are + deliberately not consumed: the observed invariant 12-byte suffix appears + structured, but controlled captures disprove it as per-clean obstacles. """ - if offset >= len(tail) or tail[offset] != 0: - return None, None - offset += _HISTORICAL_TRACE_PREFIX_LENGTH header_end = offset + _HISTORICAL_TRACE_HEADER_LENGTH if header_end > len(tail): return None, None - version = int.from_bytes(tail[offset : offset + 2], "big") + version = tail[offset] reserved = int.from_bytes( tail[offset + _HISTORICAL_TRACE_RESERVED_OFFSET : offset + _HISTORICAL_TRACE_RESERVED_OFFSET + 2], "big", @@ -778,6 +826,8 @@ class B01Q10MapParserConfig: map_scale: int = 4 """Scale factor for the rendered map image.""" + drawables: list[Drawable] | None = None + """Enabled map overlays, or ``None`` for the Q10 defaults.""" class B01Q10MapParser: diff --git a/roborock/map/b01_q10_render.py b/roborock/map/b01_q10_render.py index a24df43c..1e35fdf5 100644 --- a/roborock/map/b01_q10_render.py +++ b/roborock/map/b01_q10_render.py @@ -20,7 +20,7 @@ from vacuum_map_parser_base.config.drawable import Drawable from vacuum_map_parser_base.config.size import Size, Sizes -from vacuum_map_parser_base.map_data import Area, MapData, Path, Point, Wall +from vacuum_map_parser_base.map_data import Area, MapData, Obstacle, ObstacleDetails, Path, Point, Wall from roborock.exceptions import RoborockException @@ -67,6 +67,7 @@ Drawable.CHARGER, Drawable.NO_GO_AREAS, Drawable.NO_MOPPING_AREAS, + Drawable.OBSTACLES, Drawable.PATH, Drawable.VACUUM_POSITION, Drawable.VIRTUAL_WALLS, @@ -114,7 +115,7 @@ def render_q10_map( raise RoborockException("Failed to render Q10 map image") map_data = parsed.map_data - has_drawables = False + has_drawables = _place_obstacles(map_data, packet) if trace_calibration is not None and trace is not None: charger_heading = packet.header_calibration.charger_phi if packet.header_calibration is not None else None _place_trace(map_data, trace_calibration, trace, charger_heading=charger_heading) @@ -131,6 +132,30 @@ def render_q10_map( return parsed.image_content +def _obstacle_calibration(packet: Q10MapPacket) -> GridCalibration | None: + """Build the obstacle-table transform from the map header. + + Q10 obstacle positions use 50 raw units per occupancy-grid pixel, unlike + the 5 mm restriction vectors and 2.5 mm cleaning traces. + """ + header = packet.header_calibration + if header is None or (origin := header.origin_pixels()) is None: + return None + return GridCalibration(resolution=50.0, origin_x=origin[0], origin_y=origin[1], y_sign=1) + + +def _place_obstacles(map_data: MapData, packet: Q10MapPacket) -> bool: + """Project position-only Q10 obstacle markers into shared ``MapData``.""" + calibration = _obstacle_calibration(packet) + if calibration is None or not packet.obstacles: + return False + map_data.obstacles = [ + Obstacle(*calibration.world_to_pixel(obstacle.x, obstacle.y), ObstacleDetails()) + for obstacle in packet.obstacles + ] + return True + + def solve_q10_calibration( packet: Q10MapPacket, trace: Q10TracePacket | Q10HistoricalTracePacket | None, @@ -326,10 +351,10 @@ def _draw_map_content( """Draw Q10 content with the shared V1 image generator.""" if map_data.image is None: raise RoborockException("Failed to render Q10 map image") - generator = _create_image_generator( - MapParserConfig(map_scale=config.map_scale), - drawables=_Q10_DRAWABLES, + drawables = ( + _Q10_DRAWABLES if config.drawables is None else [d for d in config.drawables if d in _Q10_DRAWABLE_TYPES] ) + generator = _create_image_generator(MapParserConfig(map_scale=config.map_scale), drawables=drawables) generator.draw_map(map_data) buffer = io.BytesIO() map_data.image.data.save(buffer, format="PNG") diff --git a/tests/devices/test_device_manager.py b/tests/devices/test_device_manager.py index 734c2065..d7272a73 100644 --- a/tests/devices/test_device_manager.py +++ b/tests/devices/test_device_manager.py @@ -8,13 +8,16 @@ import pytest import syrupy +from vacuum_map_parser_base.config.drawable import Drawable from roborock.data import HomeData, UserData from roborock.data.containers import HomeDataDevice, HomeDataProduct, RoborockCategory from roborock.devices.cache import InMemoryCache from roborock.devices.device import RoborockDevice from roborock.devices.device_manager import UserParams, create_device_manager, create_web_api_wrapper +from roborock.devices.traits.b01.q10 import create as create_q10 from roborock.exceptions import RoborockException, RoborockInvalidCredentials +from roborock.map.map_parser import MapParserConfig from roborock.testing import FakeRoborockCloud, Q10VacuumSimulator, V1VacuumSimulator from tests import mock_data @@ -85,7 +88,12 @@ async def test_with_q10_device(cloud: FakeRoborockCloud, patch_device_manager: N ) cloud.add_device(q10_sim) - device_manager = await create_device_manager(USER_PARAMS) + map_parser_config = MapParserConfig(drawables=[Drawable.OBSTACLES], map_scale=2) + with patch("roborock.devices.device_manager.b01.q10.create", wraps=create_q10) as q10_create: + device_manager = await create_device_manager(USER_PARAMS, map_parser_config=map_parser_config) + q10_config = q10_create.call_args.kwargs["map_parser_config"] + assert q10_config.map_scale == 2 + assert q10_config.drawables == [Drawable.OBSTACLES] devices = await device_manager.get_devices() # The setup includes fake_device (V1) by default because of the fake_device fixture diff --git a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr index daac7f55..01e8b2d5 100644 --- a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr +++ b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr @@ -24,6 +24,8 @@ 'dustSwitch': True, }), 'map': dict({ + 'obstacles': list([ + ]), 'path': list([ ]), 'robotHeading': None, @@ -404,6 +406,8 @@ 'dustSwitch': True, }), 'map': dict({ + 'obstacles': list([ + ]), 'path': list([ ]), 'robotHeading': None, @@ -480,6 +484,8 @@ 'dust_collection': dict({ }), 'map': dict({ + 'obstacles': list([ + ]), 'path': list([ ]), 'robotHeading': None, diff --git a/tests/devices/traits/b01/q10/test_clean_history.py b/tests/devices/traits/b01/q10/test_clean_history.py index fb4c78da..5ae0d7c8 100644 --- a/tests/devices/traits/b01/q10/test_clean_history.py +++ b/tests/devices/traits/b01/q10/test_clean_history.py @@ -1,3 +1,4 @@ +from dataclasses import replace from pathlib import Path import pytest @@ -13,7 +14,7 @@ from roborock.devices.traits.b01.q10 import Q10PropertiesApi from roborock.devices.traits.b01.q10.clean_history import CleanHistoryTrait, CleanRecordConverter from roborock.exceptions import RoborockException -from roborock.map.b01_q10_map_parser import parse_map_packet +from roborock.map.b01_q10_map_parser import Q10Obstacle, parse_map_packet from .conftest import FakeB01Q10Channel @@ -228,3 +229,18 @@ async def test_detail_response_is_associated_with_pending_record( assert clean_history.detail_record is record assert clean_history.detail_packet is packet + + +def test_clean_record_detail_exposes_obstacles(clean_history: CleanHistoryTrait) -> None: + fixture = Path("tests/map/testdata/b01_q10_map.bin").read_bytes() + packet = replace( + parse_map_packet(b"\x03\x01" + fixture[2:]), + obstacles=[Q10Obstacle(100, -200), Q10Obstacle(-300, 400)], + ) + + clean_history.update_from_map_packet(packet) + + assert clean_history.detail_obstacles == packet.obstacles + exposed = clean_history.detail_obstacles + exposed.clear() + assert clean_history.detail_obstacles == packet.obstacles diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 828ac540..a339f35f 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -8,6 +8,7 @@ import asyncio import base64 from collections.abc import AsyncGenerator, Generator +from dataclasses import replace from pathlib import Path from typing import cast from unittest.mock import Mock, patch @@ -24,6 +25,7 @@ from roborock.map.b01_q10_map_parser import ( B01Q10MapParserConfig, Q10MapPacketKind, + Q10Obstacle, Q10Point, Q10TracePacket, parse_map_packet, @@ -74,6 +76,26 @@ def test_update_from_map_packet_populates_image_and_rooms() -> None: assert len(updates) == 1 +def test_update_from_map_packet_exposes_obstacles() -> None: + """Live callers receive the same decoded obstacles used by rendering.""" + packet = replace( + parse_map_packet(FIXTURE.read_bytes()), + obstacles=[Q10Obstacle(250, -300), Q10Obstacle(-50, 100)], + ) + trait = _map_trait() + + trait.update_from_map_packet(packet) + + assert trait.obstacles == packet.obstacles + assert trait.as_dict()["obstacles"] == [ + {"x": 250, "y": -300}, + {"x": -50, "y": 100}, + ] + exposed = trait.obstacles + exposed.clear() + assert trait.obstacles == packet.obstacles + + def test_live_map_trait_rejects_archived_packet() -> None: """Direct callers cannot bypass API routing and replace live map state.""" payload = FIXTURE.read_bytes() @@ -289,6 +311,21 @@ def test_archive_owners_reject_wrong_packet_kinds(q10_api: Q10PropertiesApi) -> q10_api.maps.update_from_map_packet(current) +def test_saved_map_detail_exposes_obstacles(q10_api: Q10PropertiesApi) -> None: + payload = FIXTURE.read_bytes() + packet = replace( + parse_map_packet(b"\x04\x01" + payload[2:]), + obstacles=[Q10Obstacle(100, -200)], + ) + + q10_api.maps.update_from_map_packet(packet) + + assert q10_api.maps.detail_obstacles == packet.obstacles + exposed = q10_api.maps.detail_obstacles + exposed.clear() + assert q10_api.maps.detail_obstacles == packet.obstacles + + def test_all_q10_map_views_share_the_injected_render_config( fake_channel: FakeB01Q10Channel, ) -> None: diff --git a/tests/map/test_b01_q10_map_parser.py b/tests/map/test_b01_q10_map_parser.py index 8b7a216c..6236848c 100644 --- a/tests/map/test_b01_q10_map_parser.py +++ b/tests/map/test_b01_q10_map_parser.py @@ -12,6 +12,7 @@ from roborock.map.b01_q10_map_parser import ( B01Q10MapParser, Q10MapPacketKind, + Q10Obstacle, Q10Point, Q10Room, classify_q10_cell, @@ -422,24 +423,35 @@ def _map_detail_payload( marker: bytes, points: list[tuple[int, int]], *, + obstacles: list[tuple[int, int]] | None = None, + skip_cleaning_points: list[tuple[int, int]] | None = None, version: int = 1, opaque_value: int = 2, heading: int = 3, reserved: int = 0, - prefix: int = 0, trailing: bytes = b"", ) -> bytes: """Build a neutral synthetic detail packet from the existing map fixture.""" + obstacle_points = obstacles or [] + skipped_points = skip_cleaning_points or [] + obstacle_table = bytes([len(obstacle_points)]) + b"".join( + x.to_bytes(2, "big", signed=True) + y.to_bytes(2, "big", signed=True) for x, y in obstacle_points + ) + skip_table = bytes([len(skipped_points)]) + b"".join( + x.to_bytes(2, "big", signed=True) + y.to_bytes(2, "big", signed=True) for x, y in skipped_points + ) header = ( - version.to_bytes(2, "big") + version.to_bytes(1, "big") + opaque_value.to_bytes(4, "big") + len(points).to_bytes(4, "big") + heading.to_bytes(2, "big", signed=True) + reserved.to_bytes(2, "big") ) point_table = b"".join(x.to_bytes(2, "big", signed=True) + y.to_bytes(2, "big", signed=True) for x, y in points) - history = bytes([prefix]) + header + point_table - payload = bytearray(FIXTURE.read_bytes() + _carpet_tail(8, 6, bytes(48)) + history + trailing) + history = header + point_table + payload = bytearray( + FIXTURE.read_bytes() + _carpet_tail(8, 6, bytes(48)) + obstacle_table + skip_table + history + trailing + ) payload[:2] = marker return bytes(payload) @@ -472,6 +484,68 @@ def test_parse_map_packet_without_carpet() -> None: assert parse_map_packet(FIXTURE.read_bytes()).carpet_mask is None +def test_parse_obstacles_and_skip_cleaning_points_before_historical_trace() -> None: + """Each post-carpet point table is bounded before the 03 path header.""" + packet = parse_map_packet( + _map_detail_payload( + b"\x03\x01", + [(11, -12), (13, -14)], + obstacles=[(250, -300), (-32768, 32767)], + skip_cleaning_points=[(-40, 50)], + ) + ) + + assert packet.obstacles == [Q10Obstacle(250, -300), Q10Obstacle(-32768, 32767)] + assert packet.skip_cleaning_points == [Q10Point(-40, 50)] + assert packet.historical_trace is not None + assert packet.historical_trace.points == [Q10Point(11, -12), Q10Point(13, -14)] + + +@pytest.mark.parametrize("marker", [b"\x01\x01", b"\x04\x01"]) +def test_obstacles_are_decoded_for_current_and_saved_maps(marker: bytes) -> None: + """Obstacle metadata belongs to the map package, not only clean history.""" + packet = parse_map_packet( + _map_detail_payload( + marker, + [], + obstacles=[(100, 200)], + skip_cleaning_points=[(30, -40)], + ) + ) + + assert packet.obstacles == [Q10Obstacle(100, 200)] + assert packet.skip_cleaning_points == [Q10Point(30, -40)] + assert packet.historical_trace is None + + +def test_empty_obstacle_sections_do_not_consume_historical_header() -> None: + packet = parse_map_packet(_map_detail_payload(b"\x03\x01", [(10, -20)])) + + assert packet.obstacles == [] + assert packet.skip_cleaning_points == [] + assert packet.historical_trace is not None + assert packet.historical_trace.points == [Q10Point(10, -20)] + + +@pytest.mark.parametrize( + "section", + [ + b"\x02\x00\x01\x00\x02", # two obstacles declared, only one present + b"\x00\x02\x00\x01\x00\x02", # two skip points declared, only one present + ], +) +def test_truncated_obstacle_sections_are_ignored(section: bytes) -> None: + """A partial point table is never reinterpreted as a later section.""" + payload = bytearray(FIXTURE.read_bytes() + _carpet_tail(8, 6, bytes(48)) + section) + payload[:2] = b"\x03\x01" + + packet = parse_map_packet(bytes(payload)) + + assert packet.obstacles == [] + assert packet.skip_cleaning_points == [] + assert packet.historical_trace is None + + def test_classify_current_clean_record_and_saved_map_packets() -> None: """All known map markers retain an explicit semantic kind.""" current = FIXTURE.read_bytes() @@ -520,7 +594,6 @@ def test_historical_trace_is_not_inferred_for_other_packet_kinds() -> None: [ {"version": 2}, {"reserved": 1}, - {"prefix": 1}, ], ) def test_unsupported_historical_trace_header_is_ignored(kwargs: dict[str, Any]) -> None: diff --git a/tests/map/test_b01_q10_render.py b/tests/map/test_b01_q10_render.py index 4dd9013d..ea44f5ad 100644 --- a/tests/map/test_b01_q10_render.py +++ b/tests/map/test_b01_q10_render.py @@ -10,6 +10,7 @@ from pathlib import Path from PIL import Image +from vacuum_map_parser_base.config.drawable import Drawable from vacuum_map_parser_base.config.size import Size, Sizes from vacuum_map_parser_base.map_data import MapData, Point @@ -20,6 +21,7 @@ Q10HeaderCalibration, Q10HistoricalTracePacket, Q10MapPacket, + Q10Obstacle, Q10Point, Q10TracePacket, parse_map_packet, @@ -35,7 +37,9 @@ Q10MapOverlays, _calibration_from_header_metadata, _erased_cells, + _obstacle_calibration, _place_docked_robot, + _place_obstacles, _vector_calibration, render_q10_map, solve_q10_calibration, @@ -126,6 +130,63 @@ def test_render_accepts_historical_trace() -> None: assert _render(packet, trace=historical) == _render(packet, trace=live_trace) +def test_place_obstacles_uses_its_validated_coordinate_scale() -> None: + """Obstacle coordinates use 50 raw units per grid pixel around the origin.""" + packet = replace( + _packet(), + header_calibration=replace(HEADER, charger_x=0, charger_y=0), + obstacles=[Q10Obstacle(50, 100), Q10Obstacle(-50, -50)], + ) + map_data = MapData() + + calibration = _obstacle_calibration(packet) + assert calibration == GridCalibration(resolution=50.0, origin_x=0.0, origin_y=5.0, y_sign=1) + assert _place_obstacles(map_data, packet) + assert map_data.obstacles is not None + assert [(obstacle.x, obstacle.y) for obstacle in map_data.obstacles] == [(1.0, 3.0), (-1.0, 6.0)] + assert all(obstacle.details.type is None for obstacle in map_data.obstacles) + + +def test_place_obstacles_requires_header_calibration() -> None: + """Unanchored obstacle points remain exposed but cannot be drawn safely.""" + packet = replace(_packet(), header_calibration=None, obstacles=[Q10Obstacle(50, 100)]) + map_data = MapData() + + assert _obstacle_calibration(packet) is None + assert not _place_obstacles(map_data, packet) + assert map_data.obstacles is None + + +def test_render_obstacles_respects_drawables_config() -> None: + packet = replace( + _packet(), + header_calibration=replace(HEADER, charger_x=0, charger_y=0), + obstacles=[Q10Obstacle(100, 100)], + ) + + hidden = render_q10_map(packet, None, Q10MapOverlays(), config=B01Q10MapParserConfig(drawables=[])) + visible = render_q10_map( + packet, + None, + Q10MapOverlays(), + config=B01Q10MapParserConfig(drawables=[Drawable.OBSTACLES]), + ) + + assert visible != hidden + + +def test_skip_cleaning_points_are_not_rendered_as_obstacles() -> None: + """The distinct skip-clean table must never gain an obstacle glyph.""" + packet = replace( + _packet(), + header_calibration=replace(HEADER, charger_x=0, charger_y=0), + skip_cleaning_points=[Q10Point(100, 100)], + ) + config = B01Q10MapParserConfig(drawables=[Drawable.OBSTACLES]) + + assert _render(packet) == render_q10_map(packet, None, Q10MapOverlays(), config=config) + + def test_render_draws_zones_and_virtual_walls() -> None: """Decoded DPS overlays are included in the composed image.""" packet, trace = _calibrated_inputs() From cd2621d287c6594d60f0094ec55a159a9de771b4 Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Sat, 29 Aug 2026 23:37:03 +0100 Subject: [PATCH 5/5] feat: expose Q10 current cleaning room --- roborock/devices/traits/b01/q10/map.py | 41 +++- roborock/diagnostics.py | 1 + roborock/map/b01_q10_map_parser.py | 2 + roborock/map/b01_q10_render.py | 141 +++++++++-- .../b01/q10/__snapshots__/test_status.ambr | 3 + tests/devices/traits/b01/q10/test_map.py | 225 +++++++++++++++++- tests/map/test_b01_q10_map_parser.py | 8 +- tests/map/test_b01_q10_render.py | 183 +++++++++++++- tests/protocols/test_b01_q10_protocol.py | 2 +- tests/test_diagnostics.py | 19 +- 10 files changed, 584 insertions(+), 41 deletions(-) diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index 267e0b14..99fab451 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -31,13 +31,21 @@ Q10TracePacket, ) from roborock.map.b01_q10_overlays import parse_virtual_wall_blob, parse_zone_blob -from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map +from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map, resolve_q10_current_room from .command import CommandTrait from .common import UpdatableTrait _LOGGER = logging.getLogger(__name__) _DOCKED_STATES = {YXDeviceState.CHARGING, YXDeviceState.EMPTYING_THE_BIN} +_CURRENT_ROOM_STATES = { + YXDeviceState.CLEANING, + YXDeviceState.PAUSED, + YXDeviceState.SWEEPING, + YXDeviceState.MOPPING, + YXDeviceState.SWEEP_AND_MOP, + YXDeviceState.TRANSITIONING, +} @dataclass @@ -101,6 +109,7 @@ def __init__( self._command = command self._map_packet: Q10MapPacket | None = None self._trace_packet: Q10TracePacket | None = None + self._current_room_trace_fresh = False self._image_content: bytes | None = None self._map_revision = 0 self._trace_revision = 0 @@ -155,6 +164,25 @@ def robot_heading(self) -> int | None: """Current heading for orienting a robot marker on a caller-rendered map.""" return self._trace_packet.heading if self._trace_packet else None + @property + def current_room(self) -> Q10Room | None: + """Room currently occupied during an active or paused cleaning task. + + No authoritative active-segment field has been identified or observed + on ss07 firmware. This value is inferred conservatively from the latest + live robot position and segmented occupancy grid, and is ``None`` + outside cleaning states or whenever the position cannot be mapped + unambiguously. + """ + if ( + self._map_dps.status not in _CURRENT_ROOM_STATES + or self._map_packet is None + or self._trace_packet is None + or not self._current_room_trace_fresh + ): + return None + return resolve_q10_current_room(self._map_packet, self._trace_packet) + def update_from_map_packet(self, packet: Q10MapPacket) -> None: """Store a map-protocol update and render the latest sources.""" if packet.kind is not Q10MapPacketKind.CURRENT: @@ -167,17 +195,26 @@ def update_from_map_packet(self, packet: Q10MapPacket) -> None: def update_from_trace_packet(self, packet: Q10TracePacket) -> None: """Store a trace-protocol update and render the latest sources.""" self._trace_packet = None if self._map_dps.robot_at_dock else packet + # A trace can precede the first status push during startup, but a trace + # received while an explicitly inactive state is known must never be + # reused as the position for a later cleaning session. + self._current_room_trace_fresh = self._trace_packet is not None and ( + self._map_dps.status is None or self._map_dps.status in _CURRENT_ROOM_STATES + ) self._trace_revision += 1 self._render() self._notify_update() def _map_dps_updated(self) -> None: """Render after the low-level map DPS source changes.""" + if self._map_dps.status is not None and self._map_dps.status not in _CURRENT_ROOM_STATES: + self._current_room_trace_fresh = False if self._map_dps.robot_at_dock and self._trace_packet is not None: # A completed cleaning trace is not the current robot position once # the device is docked. Clear the public live-path state even if the # firmware does not send its usual zero-point trace. self._trace_packet = None + self._current_room_trace_fresh = False self._trace_revision += 1 if self._map_packet is None: return @@ -203,12 +240,14 @@ def _render(self) -> None: def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]: """Return the trait data as a dictionary, excluding large binary data.""" exclude_set = exclude or set() + current_room = self.current_room data = { "rooms": [room.as_dict() for room in self.rooms], "obstacles": [obstacle.as_dict() for obstacle in self.obstacles], "path": [point.as_dict() for point in self.path], "robotPosition": self.robot_position.as_dict() if self.robot_position is not None else None, "robotHeading": self.robot_heading, + "currentRoom": current_room.as_dict() if current_room is not None else None, } for key in exclude_set: data.pop(key, None) diff --git a/roborock/diagnostics.py b/roborock/diagnostics.py index 0c761572..3b1202a1 100644 --- a/roborock/diagnostics.py +++ b/roborock/diagnostics.py @@ -108,6 +108,7 @@ def reset(self) -> None: "wifiName", "lat", "long", + "rawName", } KEEP_KEYS = { # Product information not unique per user diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index 0760218e..1df45ff2 100644 --- a/roborock/map/b01_q10_map_parser.py +++ b/roborock/map/b01_q10_map_parser.py @@ -128,6 +128,8 @@ class Q10Room(RoborockBase): @property def name(self) -> str: """User friendly room name (firmware ``rr_`` defaults are normalized).""" + if not self.raw_name.startswith("rr_"): + return self.raw_name return self.raw_name.removeprefix("rr_").replace("_", " ").strip().title() diff --git a/roborock/map/b01_q10_render.py b/roborock/map/b01_q10_render.py index 1e35fdf5..c1e81a9a 100644 --- a/roborock/map/b01_q10_render.py +++ b/roborock/map/b01_q10_render.py @@ -16,7 +16,7 @@ import io import math from collections.abc import Sequence -from dataclasses import dataclass +from dataclasses import dataclass, replace from vacuum_map_parser_base.config.drawable import Drawable from vacuum_map_parser_base.config.size import Size, Sizes @@ -28,7 +28,6 @@ GridCalibration, GridLayers, solve_calibration, - solve_calibration_with_origin, ) from .b01_q10_map_parser import ( B01Q10MapParser, @@ -36,13 +35,16 @@ Q10EraseZone, Q10HistoricalTracePacket, Q10MapPacket, + Q10Point, + Q10Room, Q10TracePacket, erased_packet, ) from .b01_q10_overlays import ZONE_TYPE_NO_GO, ZONE_TYPE_NO_MOP, Q10Zone from .map_parser import DEFAULT_DRAWABLES, MapParserConfig, _create_image_generator -# Path-units-per-pixel candidates for calibration. A dense ss07 path lands a +# Path-units-per-pixel candidates for packets without usable header metadata. +# A dense ss07 path lands a # best fit of 20.0 around the header origin -- ground-truthed June 2026 on the # R1: a corridor drive registered at 20 (matching the format author's # independent "20 path-units/px"), and the dock->corridor span lined up with the @@ -56,13 +58,13 @@ # erase/restriction vectors use 5 mm units: one header unit therefore equals # two vector units. Trace points use a separate 2.5 mm coordinate scale. _Q10_VECTOR_UNITS_PER_HEADER_RESOLUTION_UNIT = 2.0 +# The grid header's resolution is centimetres per pixel, while live trace +# coordinates are 2.5 mm units. One header unit therefore spans four trace +# units (5 cm/pixel -> 20 trace units/pixel for the observed resolution 5). +_Q10_TRACE_UNITS_PER_HEADER_RESOLUTION_UNIT = 4.0 # A path needs enough shape to constrain a full (origin + resolution) fit; a few # points cannot. _MIN_CALIBRATION_POINTS = 20 -# When the grid-frame header supplies the origin, only the resolution is fit, so -# a much shorter path suffices to confirm it (early in a clean, not just a dense -# one). See :func:`solve_calibration_with_origin`. -_MIN_HEADER_CALIBRATION_POINTS = 4 _Q10_DRAWABLE_TYPES = { Drawable.CHARGER, Drawable.NO_GO_AREAS, @@ -162,30 +164,125 @@ def solve_q10_calibration( ) -> GridCalibration | None: """Derive world-to-pixel calibration from a map and its current trace. - When the map packet's grid-frame header carries a calibration origin (ss07), - only the resolution is fit -- around that fixed origin -- so a short path - suffices and the origin is exact rather than recovered by a slide. Otherwise - the full origin + resolution fit is used, which needs a reasonably dense - cleaning path. Returns ``None`` if the path is too short/featureless to fit. + When the map packet carries usable ss07 header metadata, its fixed origin + and resolution are authoritative and work from the first live pose. Older + or incomplete packets fall back to a full origin/resolution fit, which needs + a reasonably dense cleaning path. Returns ``None`` when neither is usable. """ - if trace is None: + if trace is None or not trace.points: return None + if calibration := _trace_calibration_from_header_metadata(packet, trace.points): + return calibration points: list[tuple[float, float]] = [(point.x, point.y) for point in trace.points] - return _calibration_from_header(packet, points) or _calibration_from_fit(packet.layers, points) + return _calibration_from_fit(packet.layers, points) -def _calibration_from_header( +def _trace_calibration_from_header_metadata( packet: Q10MapPacket, - points: list[tuple[float, float]], + points: Sequence[Q10Point], ) -> GridCalibration | None: - """Calibrate around the header-supplied origin (resolution fit to a path).""" - header_calibration = packet.header_calibration - if header_calibration is None or len(points) < _MIN_HEADER_CALIBRATION_POINTS: + """Build the live-trace transform directly from validated header metadata.""" + header = packet.header_calibration + if header is None or header.resolution <= 0 or (origin := header.origin_pixels()) is None: + return None + resolution = header.resolution * _Q10_TRACE_UNITS_PER_HEADER_RESOLUTION_UNIT + if not min(_Q10_RESOLUTIONS) <= resolution <= max(_Q10_RESOLUTIONS): + return None + calibration = GridCalibration( + resolution=resolution, + origin_x=origin[0], + origin_y=origin[1], + y_sign=1, + ) + return calibration if _trace_projects_onto_floor(packet.layers, points, calibration) else None + + +def _trace_projects_onto_floor( + layers: GridLayers, + points: Sequence[Q10Point], + calibration: GridCalibration, +) -> bool: + """Validate header calibration against a bounded sample of trace points.""" + if not points: + return False + stride = max(1, len(points) // 256) + sample = points[::stride] + on_floor = 0 + for point in sample: + pixel_x, pixel_y = calibration.world_to_pixel(point.x, point.y) + if not math.isfinite(pixel_x) or not math.isfinite(pixel_y): + continue + column, row = math.floor(pixel_x), math.floor(pixel_y) + if not (0 <= column < layers.width and 0 <= row < layers.height): + continue + if layers.cell_class(layers.grid[row * layers.width + column]) == "floor": + on_floor += 1 + return on_floor >= len(sample) * 0.5 + + +def _room_at_position( + packet: Q10MapPacket, + position: Q10Point, + calibration: GridCalibration, + *, + search_radius: int = 2, +) -> Q10Room | None: + """Resolve a live position to a segmented room in the occupancy grid. + + The exact cell wins. A robot centre can briefly land on a wall or doorway + pixel, so a small expanding neighbourhood is used only when the exact cell + is not segmented. Ambiguous ties stay unknown rather than naming the wrong + room. + """ + pixel_x, pixel_y = calibration.world_to_pixel(position.x, position.y) + if not math.isfinite(pixel_x) or not math.isfinite(pixel_y): + return None + column, row = math.floor(pixel_x), math.floor(pixel_y) + if not (0 <= column < packet.width and 0 <= row < packet.height): + return None + + rooms_by_value = {room.pixel_value: room for room in packet.rooms} + + def room_at(column: int, row: int) -> Q10Room | None: + if not (0 <= column < packet.width and 0 <= row < packet.height): + return None + return rooms_by_value.get(packet.grid[row * packet.width + column]) + + if room := room_at(column, row): + return room + + for radius in range(1, max(0, search_radius) + 1): + room_ids = { + candidate.id + for nearby_row in range(row - radius, row + radius + 1) + for nearby_column in range(column - radius, column + radius + 1) + if (candidate := room_at(nearby_column, nearby_row)) is not None + } + if not room_ids: + continue + if len(room_ids) != 1: + return None + room_id = room_ids.pop() + return next((room for room in packet.rooms if room.id == room_id), None) + return None + + +def resolve_q10_current_room(packet: Q10MapPacket, trace: Q10TracePacket | None) -> Q10Room | None: + """Infer the robot's current room from its latest live pose and map grid. + + No authoritative active-segment field has been identified or observed on + ss07 firmware. The live trace's final point is the robot pose, and the full + map labels every segmented floor cell. Header metadata supplies the fixed + transform from trace coordinates to that grid; older/incomplete packets + fall back to the path-fit calibration used by rendering. + """ + if trace is None or (position := trace.robot_position) is None: return None - origin = header_calibration.origin_pixels() - if origin is None: # keepalive frame -- no usable origin + calibration = solve_q10_calibration(packet, trace) + if calibration is None: return None - return solve_calibration_with_origin(packet.layers, points, origin, resolutions=_Q10_RESOLUTIONS) + room = _room_at_position(packet, position, calibration) + return replace(room) if room is not None else None def _calibration_from_header_metadata( diff --git a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr index 01e8b2d5..2c6e991d 100644 --- a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr +++ b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr @@ -24,6 +24,7 @@ 'dustSwitch': True, }), 'map': dict({ + 'currentRoom': None, 'obstacles': list([ ]), 'path': list([ @@ -406,6 +407,7 @@ 'dustSwitch': True, }), 'map': dict({ + 'currentRoom': None, 'obstacles': list([ ]), 'path': list([ @@ -484,6 +486,7 @@ 'dust_collection': dict({ }), 'map': dict({ + 'currentRoom': None, 'obstacles': list([ ]), 'path': list([ diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index a339f35f..e23fb845 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -72,7 +72,7 @@ def test_update_from_map_packet_populates_image_and_rooms() -> None: assert trait.image_content is not None assert trait.image_content[:8] == b"\x89PNG\r\n\x1a\n" - assert {room.id: room.name for room in trait.rooms} == {2: "Living Room", 3: "Bedroom"} + assert {room.id: room.name for room in trait.rooms} == {2: "Living Room", 3: "bedroom"} assert len(updates) == 1 @@ -122,6 +122,184 @@ def test_update_from_trace_packet_populates_path_and_position() -> None: assert len(updates) == 1 +@pytest.mark.parametrize( + "status", + [ + YXDeviceState.CLEANING, + YXDeviceState.SWEEPING, + YXDeviceState.MOPPING, + YXDeviceState.SWEEP_AND_MOP, + YXDeviceState.PAUSED, + YXDeviceState.TRANSITIONING, + ], +) +def test_current_room_is_available_during_active_cleaning_states(status: YXDeviceState) -> None: + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = parse_map_packet(FIXTURE.read_bytes()) + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + map_dps.update_from_dps({B01_Q10_DP.STATUS: status.code}) + trait.update_from_map_packet(packet) + trait.update_from_trace_packet(trace) + + with patch( + "roborock.devices.traits.b01.q10.map.resolve_q10_current_room", + return_value=packet.rooms[0], + ) as resolve: + assert trait.current_room == packet.rooms[0] + resolve.assert_called_once_with(packet, trace) + + +@pytest.mark.parametrize( + "status", + [ + None, + YXDeviceState.UNKNOWN, + YXDeviceState.IDLE, + YXDeviceState.RETURNING_HOME, + YXDeviceState.RELOCATING, + YXDeviceState.CHARGING, + YXDeviceState.EMPTYING_THE_BIN, + ], +) +def test_current_room_is_unavailable_outside_active_cleaning( + status: YXDeviceState | None, +) -> None: + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = parse_map_packet(FIXTURE.read_bytes()) + if status is not None: + map_dps.update_from_dps({B01_Q10_DP.STATUS: status.code}) + trait.update_from_map_packet(packet) + trait.update_from_trace_packet(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes())) + + with patch("roborock.devices.traits.b01.q10.map.resolve_q10_current_room") as resolve: + assert trait.current_room is None + resolve.assert_not_called() + + +def test_current_room_handles_status_arriving_after_map_and_trace() -> None: + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = parse_map_packet(FIXTURE.read_bytes()) + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + trait.update_from_map_packet(packet) + trait.update_from_trace_packet(trace) + + with patch( + "roborock.devices.traits.b01.q10.map.resolve_q10_current_room", + side_effect=lambda *_: replace(packet.rooms[0]), + ): + assert trait.current_room is None + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + assert trait.current_room == packet.rooms[0] + + +def test_current_room_handles_map_arriving_after_status_and_trace() -> None: + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = parse_map_packet(FIXTURE.read_bytes()) + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + trait.update_from_trace_packet(trace) + + with patch( + "roborock.devices.traits.b01.q10.map.resolve_q10_current_room", + side_effect=lambda *_: replace(packet.rooms[0]), + ): + assert trait.current_room is None + trait.update_from_map_packet(packet) + assert trait.current_room == packet.rooms[0] + + +@pytest.mark.parametrize( + "ending_status", + [YXDeviceState.UNKNOWN, YXDeviceState.IDLE, YXDeviceState.RETURNING_HOME], +) +def test_leaving_cleaning_clears_current_room_without_reusing_stale_trace( + ending_status: YXDeviceState, +) -> None: + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = parse_map_packet(FIXTURE.read_bytes()) + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + trait.update_from_map_packet(packet) + trait.update_from_trace_packet(trace) + + with patch( + "roborock.devices.traits.b01.q10.map.resolve_q10_current_room", + side_effect=lambda *_: replace(packet.rooms[0]), + ) as resolve: + assert trait.current_room == packet.rooms[0] + map_dps.update_from_dps({B01_Q10_DP.STATUS: ending_status.code}) + assert trait.current_room is None + assert trait.path == trace.points + assert resolve.call_count == 1 + + +def test_new_cleaning_session_waits_for_a_fresh_trace() -> None: + """A status-first new session cannot reuse the prior session's position.""" + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = parse_map_packet(FIXTURE.read_bytes()) + old_trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + new_trace = replace(old_trace, heading=old_trace.heading + 1) + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + trait.update_from_map_packet(packet) + trait.update_from_trace_packet(old_trace) + + with patch( + "roborock.devices.traits.b01.q10.map.resolve_q10_current_room", + side_effect=lambda *_: replace(packet.rooms[0]), + ) as resolve: + assert trait.current_room == packet.rooms[0] + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.IDLE.code}) + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + assert trait.current_room is None + trait.update_from_trace_packet(new_trace) + assert trait.current_room == packet.rooms[0] + assert resolve.call_count == 2 + + +def test_current_room_requires_both_map_and_trace() -> None: + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + + assert trait.current_room is None + trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) + assert trait.current_room is None + + +def test_current_room_is_a_defensive_copy_and_serializes_for_consumers() -> None: + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = parse_map_packet(FIXTURE.read_bytes()) + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + trait.update_from_map_packet(packet) + trait.update_from_trace_packet(trace) + + with patch( + "roborock.devices.traits.b01.q10.map.resolve_q10_current_room", + side_effect=lambda *_: replace(packet.rooms[0]), + ): + exposed = trait.current_room + assert exposed is not None + assert exposed is not packet.rooms[0] + exposed.raw_name = "mutated" + assert trait.current_room is not None + assert trait.current_room.raw_name == "rr_living_room" + assert trait.current_room.name == "Living Room" + assert trait.as_dict()["currentRoom"] == packet.rooms[0].as_dict() + assert "currentRoom" not in trait.as_dict({"currentRoom"}) + + +def test_current_room_serializes_none_when_unknown() -> None: + assert _map_trait().as_dict()["currentRoom"] is None + + def test_q10_position_is_available_as_top_level_cli_command() -> None: assert "q10-position" in cli.commands @@ -252,7 +430,7 @@ async def test_subscribe_loop_routes_map_push( message_queue.put_nowait(parse_map_packet(FIXTURE.read_bytes())) await _wait_for(lambda: q10_api.map.image_content is not None) - assert {room.id: room.name for room in q10_api.map.rooms} == {2: "Living Room", 3: "Bedroom"} + assert {room.id: room.name for room in q10_api.map.rooms} == {2: "Living Room", 3: "bedroom"} async def test_archived_map_pushes_cannot_overwrite_live_map( @@ -326,6 +504,27 @@ def test_saved_map_detail_exposes_obstacles(q10_api: Q10PropertiesApi) -> None: assert q10_api.maps.detail_obstacles == packet.obstacles +def test_archived_map_details_do_not_expose_or_replace_current_room(q10_api: Q10PropertiesApi) -> None: + payload = FIXTURE.read_bytes() + current = parse_map_packet(payload) + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + q10_api._map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + q10_api.map.update_from_map_packet(current) + q10_api.map.update_from_trace_packet(trace) + + with patch( + "roborock.devices.traits.b01.q10.map.resolve_q10_current_room", + return_value=current.rooms[0], + ): + assert q10_api.map.current_room == current.rooms[0] + q10_api._handle_message(parse_map_packet(b"\x03\x01" + payload[2:])) + q10_api._handle_message(parse_map_packet(b"\x04\x01" + payload[2:])) + assert q10_api.map.current_room == current.rooms[0] + + assert not hasattr(q10_api.clean_history, "current_room") + assert not hasattr(q10_api.maps, "current_room") + + def test_all_q10_map_views_share_the_injected_render_config( fake_channel: FakeB01Q10Channel, ) -> None: @@ -728,6 +927,28 @@ def test_docked_state_clears_stale_live_trace(render_map: Mock) -> None: assert render_map.call_args.kwargs["robot_at_dock"] is True +def test_docked_state_clears_current_room_and_ignores_late_trace() -> None: + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = parse_map_packet(FIXTURE.read_bytes()) + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + trait.update_from_map_packet(packet) + trait.update_from_trace_packet(trace) + + with patch( + "roborock.devices.traits.b01.q10.map.resolve_q10_current_room", + return_value=packet.rooms[0], + ) as resolve: + assert trait.current_room == packet.rooms[0] + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code}) + assert trait.current_room is None + trait.update_from_trace_packet(trace) + assert trait.current_room is None + assert trait.path == [] + assert resolve.call_count == 1 + + def test_late_trace_is_ignored_while_docked(render_map: Mock) -> None: """A delayed trace cannot repopulate public state while docked.""" map_dps = MapDpsTrait() diff --git a/tests/map/test_b01_q10_map_parser.py b/tests/map/test_b01_q10_map_parser.py index 6236848c..2bcc4675 100644 --- a/tests/map/test_b01_q10_map_parser.py +++ b/tests/map/test_b01_q10_map_parser.py @@ -154,7 +154,7 @@ def test_packet_layers_decompose_q10_fixture() -> None: """The Q10 synthetic fixture splits into floor + per-room layers.""" layers = parse_map_packet(_payload()).layers assert layers.class_counts.get(LAYER_FLOOR) == 26 - assert {room.id: room.name for room in layers.rooms} == {2: "Living Room", 3: "Bedroom"} + assert {room.id: room.name for room in layers.rooms} == {2: "Living Room", 3: "bedroom"} living = layers.render_room(2, (255, 0, 0, 255)) image = Image.open(io.BytesIO(living)) @@ -234,9 +234,9 @@ def test_parse_map_packet_dimensions_straddling_256() -> None: def test_room_name_normalization() -> None: - """Firmware ``rr_`` default names are normalized; custom names are titled.""" + """Firmware ``rr_`` defaults are normalized; custom names stay verbatim.""" assert Q10Room(id=2, raw_name="rr_living_room", pixel_value=8, pixel_count=9).name == "Living Room" - assert Q10Room(id=3, raw_name="bedroom", pixel_value=12, pixel_count=9).name == "Bedroom" + assert Q10Room(id=3, raw_name="Owner’s bedroom", pixel_value=12, pixel_count=9).name == "Owner’s bedroom" def test_room_pixel_count_matches_grid() -> None: @@ -251,7 +251,7 @@ def test_parser_renders_png_and_room_names() -> None: assert parsed.image_content is not None assert parsed.image_content[:8] == b"\x89PNG\r\n\x1a\n" # PNG magic assert parsed.map_data is not None - assert parsed.map_data.additional_parameters["room_names"] == {2: "Living Room", 3: "Bedroom"} + assert parsed.map_data.additional_parameters["room_names"] == {2: "Living Room", 3: "bedroom"} def test_parse_packet_preserves_decoded_packet_api() -> None: diff --git a/tests/map/test_b01_q10_render.py b/tests/map/test_b01_q10_render.py index ea44f5ad..08f9d375 100644 --- a/tests/map/test_b01_q10_render.py +++ b/tests/map/test_b01_q10_render.py @@ -8,7 +8,10 @@ import io from dataclasses import replace from pathlib import Path +from typing import cast +from unittest.mock import patch +import pytest from PIL import Image from vacuum_map_parser_base.config.drawable import Drawable from vacuum_map_parser_base.config.size import Size, Sizes @@ -21,8 +24,10 @@ Q10HeaderCalibration, Q10HistoricalTracePacket, Q10MapPacket, + Q10MapPacketKind, Q10Obstacle, Q10Point, + Q10Room, Q10TracePacket, parse_map_packet, ) @@ -33,15 +38,16 @@ Q10Zone, ) from roborock.map.b01_q10_render import ( - _Q10_RESOLUTIONS, Q10MapOverlays, _calibration_from_header_metadata, _erased_cells, _obstacle_calibration, _place_docked_robot, _place_obstacles, + _room_at_position, _vector_calibration, render_q10_map, + resolve_q10_current_room, solve_q10_calibration, ) @@ -93,6 +99,21 @@ def _world_vertices(calibration: GridCalibration, pixels: list[tuple[int, int]]) return vertices +def _room_packet(grid: list[int], width: int) -> Q10MapPacket: + """Build a minimal segmented map for deterministic room lookup tests.""" + return Q10MapPacket( + kind=Q10MapPacketKind.CURRENT, + map_id=1, + width=width, + height=len(grid) // width, + grid=bytes(grid), + rooms=[ + Q10Room(2, "rr_kitchen", 8, grid.count(8)), + Q10Room(3, "hall", 12, grid.count(12)), + ], + ) + + def _calibrated_inputs(*, heading: int = 0) -> tuple[Q10MapPacket, Q10TracePacket]: packet = replace(_packet(), header_calibration=HEADER) pixels = [(1, 1), (6, 1), (1, 2), (6, 2), (2, 3), (5, 3)] @@ -187,6 +208,126 @@ def test_skip_cleaning_points_are_not_rendered_as_obstacles() -> None: assert _render(packet) == render_q10_map(packet, None, Q10MapOverlays(), config=config) +def test_room_at_position_exact_segment_wins_over_nearby_majority() -> None: + packet = _room_packet([12, 12, 12, 12, 8, 12, 12, 12, 12], width=3) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + assert _room_at_position(packet, Q10Point(10, 10), calibration) == packet.rooms[0] + + +def test_room_at_position_uses_half_open_cell_boundaries() -> None: + packet = _room_packet([8, 8, 12, 12], width=4) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + assert _room_at_position(packet, Q10Point(19, 0), calibration) == packet.rooms[0] + assert _room_at_position(packet, Q10Point(20, 0), calibration) == packet.rooms[1] + + +@pytest.mark.parametrize("position", [Q10Point(-1, 10), Q10Point(30, 10), Q10Point(10, -1), Q10Point(10, 30)]) +def test_room_at_position_rejects_fractional_and_exact_out_of_bounds(position: Q10Point) -> None: + packet = _room_packet([8] * 9, width=3) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + assert _room_at_position(packet, position, calibration) is None + + +@pytest.mark.parametrize("coordinate", [float("nan"), float("inf"), float("-inf")]) +def test_room_at_position_rejects_nonfinite_coordinates(coordinate: float) -> None: + packet = _room_packet([8], width=1) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + assert _room_at_position(packet, Q10Point(cast(int, coordinate), 0), calibration) is None + + +def test_room_at_position_recovers_unique_nearby_room() -> None: + packet = _room_packet([8, 8, 8, 8, 0, 8, 8, 8, 8], width=3) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + assert _room_at_position(packet, Q10Point(10, 10), calibration) == packet.rooms[0] + + +def test_room_at_position_expands_when_first_radius_has_no_room_cells() -> None: + grid = [8, 0, 0, 0, 8] + [0] * 15 + [8, 0, 0, 0, 8] + packet = _room_packet(grid, width=5) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + # Radius one has no segmented cells; radius two has only Kitchen cells. + assert _room_at_position(packet, Q10Point(20, 20), calibration) == packet.rooms[0] + + +def test_room_at_position_returns_none_for_any_nearby_room_conflict() -> None: + packet = _room_packet([8, 0, 12, 8, 0, 8, 8, 8, 8], width=3) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + # Kitchen has the numerical majority, but Hall is also present in the first + # nonempty radius, so assigning a room would be ambiguous. + assert _room_at_position(packet, Q10Point(10, 10), calibration) is None + + +def test_room_at_position_can_disable_nearby_fallback() -> None: + packet = _room_packet([8, 8, 8, 8, 0, 8, 8, 8, 8], width=3) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + assert _room_at_position(packet, Q10Point(10, 10), calibration, search_radius=0) is None + + +def test_room_at_position_requires_matching_room_metadata() -> None: + packet = replace(_room_packet([8], width=1), rooms=[]) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + assert _room_at_position(packet, Q10Point(0, 0), calibration) is None + + +def test_resolve_current_room_uses_latest_live_position() -> None: + packet = _room_packet([8, 12], width=2) + trace = Q10TracePacket(points=[Q10Point(0, 0), Q10Point(10, 0)]) + calibration = GridCalibration(resolution=10, origin_x=0, origin_y=0, y_sign=-1) + + with patch("roborock.map.b01_q10_render.solve_q10_calibration", return_value=calibration): + room = resolve_q10_current_room(packet, trace) + + assert room == packet.rooms[1] + assert room is not packet.rooms[1] + assert room is not None + room.raw_name = "mutated" + assert packet.rooms[1].raw_name == "hall" + + +def test_resolve_current_room_requires_position_and_calibration() -> None: + packet = _room_packet([8], width=1) + + assert resolve_q10_current_room(packet, None) is None + assert resolve_q10_current_room(packet, Q10TracePacket()) is None + assert resolve_q10_current_room(packet, Q10TracePacket(points=[Q10Point(0, 0)])) is None + + +def test_single_point_uses_exact_header_transform_for_room_and_rendering() -> None: + packet = replace( + _packet(), + header_calibration=replace(HEADER, charger_x=0, charger_y=0), + ) + trace = Q10TracePacket(points=[Q10Point(100, 60)], heading=45) + + calibration = solve_q10_calibration(packet, trace) + assert calibration == TRACE_CALIBRATION + assert calibration.world_to_pixel(100, 60) == (5.0, 2.0) + assert resolve_q10_current_room(packet, trace) == packet.rooms[1] + + rendered = Image.open(io.BytesIO(_render(packet, trace=trace))).convert("RGBA") + position = (5 * CONFIG.map_scale, 2 * CONFIG.map_scale) + assert rendered.getpixel(position) == (255, 255, 255, 255) + + +def test_empty_trace_does_not_force_a_map_redraw() -> None: + """A metadata-only zero-point trace contributes no drawable content.""" + packet = replace( + _packet(), + header_calibration=replace(HEADER, charger_x=0, charger_y=0), + ) + + assert _render(packet, trace=Q10TracePacket()) == _render(packet) + + def test_render_draws_zones_and_virtual_walls() -> None: """Decoded DPS overlays are included in the composed image.""" packet, trace = _calibrated_inputs() @@ -264,7 +405,9 @@ def test_zero_degree_dock_places_robot_to_its_right() -> None: def test_render_applies_erase_zones() -> None: """With a calibration, erase-zone cells are blanked from the image.""" packet, trace = _calibrated_inputs() - base = _render(packet, trace=trace) + assert packet.header_calibration is not None + packet = replace(packet, header_calibration=replace(packet.header_calibration, charger_x=0, charger_y=0)) + base = _render(packet) trace_calibration = solve_q10_calibration(packet, trace) calibration = _vector_calibration(packet, trace_calibration) assert calibration == VECTOR_CALIBRATION @@ -273,7 +416,7 @@ def test_render_applies_erase_zones() -> None: corners = [(-1, -1), (8, -1), (8, 6), (-1, 6)] erase_zone = Q10EraseZone(vertices=_world_vertices(calibration, corners)) cells = _erased_cells(packet.layers, [erase_zone], calibration) - render = _render(replace(packet, erase_zones=[erase_zone]), trace=trace) + render = _render(replace(packet, erase_zones=[erase_zone])) assert len(cells) == packet.layers.width * packet.layers.height assert render != base @@ -306,7 +449,9 @@ def test_header_vector_calibration_is_independent_of_trace_orientation() -> None def test_render_partial_erase() -> None: """An erase rectangle only blanks the cells it covers, leaving the rest.""" packet, trace = _calibrated_inputs() - base = _render(packet, trace=trace) + assert packet.header_calibration is not None + packet = replace(packet, header_calibration=replace(packet.header_calibration, charger_x=0, charger_y=0)) + base = _render(packet) trace_calibration = solve_q10_calibration(packet, trace) calibration = _vector_calibration(packet, trace_calibration) assert calibration == VECTOR_CALIBRATION @@ -315,7 +460,7 @@ def test_render_partial_erase() -> None: corners = [(-1, -1), (8, -1), (8, 2), (-1, 2)] erase_zone = Q10EraseZone(vertices=_world_vertices(calibration, corners)) cells = _erased_cells(packet.layers, [erase_zone], calibration) - render = _render(replace(packet, erase_zones=[erase_zone]), trace=trace) + render = _render(replace(packet, erase_zones=[erase_zone])) assert 0 < len(cells) < packet.layers.width * packet.layers.height assert render != base @@ -330,16 +475,34 @@ def test_render_robot_marker_reflects_heading() -> None: def test_solve_q10_calibration_uses_header_origin_with_short_path() -> None: - """A grid-frame header origin lets a short path calibrate (origin is exact).""" + """Validated header metadata calibrates a short path exactly.""" packet, trace = _calibrated_inputs() assert len(trace.points) < 20 # far too short for the full origin+resolution fit cal = solve_q10_calibration(packet, trace) assert cal is not None - # Origin comes straight from the header (exact); only the resolution is fit, - # so it lands on one of the candidates (the exact pick is grid-quantized). - assert (cal.origin_x, cal.origin_y) == (0.0, 5.0) - assert cal.resolution in _Q10_RESOLUTIONS + assert cal == TRACE_CALIBRATION + + +@pytest.mark.parametrize( + "header", + [ + replace(HEADER, resolution=100), + replace(HEADER, origin_x=100_000, origin_y=100_000), + ], +) +def test_solve_q10_calibration_rejects_implausible_header_and_falls_back( + header: Q10HeaderCalibration, +) -> None: + """Corrupt metadata cannot displace an otherwise fit-able cleaning path.""" + packet = replace(_packet(), header_calibration=header) + trace = Q10TracePacket(points=[Q10Point(index, index) for index in range(20)]) + fallback = GridCalibration(resolution=17, origin_x=3, origin_y=4, y_sign=-1) + + with patch("roborock.map.b01_q10_render._calibration_from_fit", return_value=fallback) as fit: + assert solve_q10_calibration(packet, trace) == fallback + + fit.assert_called_once_with(packet.layers, [(point.x, point.y) for point in trace.points]) def test_solve_q10_calibration_short_path_without_header_returns_none() -> None: diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index adf62f8c..31536411 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -66,7 +66,7 @@ def test_decode_message_map_packet() -> None: message = _message(MAP_FIXTURE.read_bytes(), RoborockMessageProtocol.MAP_RESPONSE) decoded = decode_message(message) assert isinstance(decoded, Q10MapPacket) - assert {room.id: room.name for room in decoded.rooms} == {2: "Living Room", 3: "Bedroom"} + assert {room.id: room.name for room in decoded.rooms} == {2: "Living Room", 3: "bedroom"} @pytest.mark.parametrize( diff --git a/tests/test_diagnostics.py b/tests/test_diagnostics.py index 2fe8aff5..f46229cb 100644 --- a/tests/test_diagnostics.py +++ b/tests/test_diagnostics.py @@ -2,7 +2,7 @@ import pytest -from roborock.diagnostics import Diagnostics, redact_device_uid, redact_topic_name +from roborock.diagnostics import REDACTED, Diagnostics, redact_device_data, redact_device_uid, redact_topic_name def test_empty_diagnostics(): @@ -12,6 +12,23 @@ def test_empty_diagnostics(): assert diag.as_dict() == {} +def test_redact_q10_raw_room_name() -> None: + """Custom Q10 room labels are private in exported diagnostics.""" + assert redact_device_data( + { + "map": { + "rooms": [{"rawName": "Owner’s bedroom"}], + "currentRoom": {"rawName": "Owner’s bedroom"}, + } + } + ) == { + "map": { + "rooms": [{"rawName": REDACTED}], + "currentRoom": {"rawName": REDACTED}, + } + } + + def test_increment_counter(): """Test incrementing counters in Diagnostics."""