diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index 2f162c64..8d8ac683 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 @@ -220,6 +241,18 @@ def layers(self) -> GridLayers: return decompose_grid(self.width, self.height, self.grid, rooms, classify_q10_cell, flip=False) +@dataclass +class Q10CleanRecordMapPacket(Q10MapPacket): + """A clean-record map with its embedded historical cleaning path. + + Current, saved and clean-record maps share the grid, rooms, erase zones, + carpet and calibration layout. Only clean-record packets carry this path; + it is not a separately received live trace. + """ + + historical_trace: "Q10HistoricalTracePacket | None" = None + + @dataclass class Q10Point(RoborockBase): """A single point in Q10 map/trace coordinate space.""" @@ -266,6 +299,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 +329,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 +364,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 +391,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 +424,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) -> 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. Expansion beyond + ``max_output_size`` is rejected before + allocating the excess output. """ index = 0 output = bytearray() @@ -380,6 +458,8 @@ def read_length(value: int) -> int: end = index + literal_length if end > len(data): raise RoborockException("Truncated LZ4 block while reading literals") + if len(output) + literal_length > max_output_size: + raise RoborockException("LZ4 block exceeds maximum output size") output.extend(data[index:end]) index = end @@ -394,6 +474,8 @@ def read_length(value: int) -> int: raise RoborockException("Invalid LZ4 back-reference offset") match_length = read_length(token & 0x0F) + 4 + if 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 +540,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 +550,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 +560,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 +574,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( + packet = Q10MapPacket( + kind=kind, map_id=map_id, width=width, height=height, @@ -498,6 +591,9 @@ def parse_map_packet(payload: bytes) -> Q10MapPacket: header_calibration=header_calibration, carpet_mask=carpet_mask, ) + if kind is Q10MapPacketKind.CLEAN_RECORD_DETAIL: + return Q10CleanRecordMapPacket(**vars(packet), historical_trace=historical_trace) + return packet def _parse_header_calibration(payload: bytes) -> Q10HeaderCalibration | None: @@ -569,7 +665,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 +685,83 @@ 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, 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, +) -> Q10HistoricalTracePacket | 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 - return mask if len(mask) == width * height else None + offset += _HISTORICAL_TRACE_PREFIX_LENGTH + header_end = offset + _HISTORICAL_TRACE_HEADER_LENGTH + if header_end > len(tail): + return 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 + 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 + 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, + ), + ) 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..70686243 100644 --- a/roborock/map/b01_q10_render.py +++ b/roborock/map/b01_q10_render.py @@ -33,7 +33,9 @@ from .b01_q10_map_parser import ( B01Q10MapParser, B01Q10MapParserConfig, + Q10CleanRecordMapPacket, Q10EraseZone, + Q10HistoricalTracePacket, Q10MapPacket, Q10TracePacket, erased_packet, @@ -62,7 +64,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, @@ -97,8 +98,10 @@ def render_q10_map( available trace and DPS overlays are projected and drawn in pixel space. Raises :class:`RoborockException` if map rendering fails. """ + # An archived map owns its path; a live trace must never replace it. + render_trace = packet.historical_trace if isinstance(packet, Q10CleanRecordMapPacket) else trace parser = B01Q10MapParser(config) - trace_calibration = solve_q10_calibration(packet, trace) + trace_calibration = solve_q10_calibration(packet, render_trace) vector_calibration = _vector_calibration(packet, trace_calibration) render_packet = packet @@ -115,9 +118,9 @@ def render_q10_map( map_data = parsed.map_data has_drawables = False - if trace_calibration is not None and trace is not None: + if trace_calibration is not None and render_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) + _place_trace(map_data, trace_calibration, render_trace, charger_heading=charger_heading) has_drawables = True has_drawables = _place_charger_from_header(map_data, packet) or has_drawables if robot_at_dock: @@ -133,7 +136,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 +235,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..f2796f32 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,19 @@ 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): + kind = Q10MapPacketKind.from_payload(payload) + if kind is Q10MapPacketKind.TRACE: return parse_trace_packet(payload) + if kind 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..b284f672 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,14 @@ from roborock.map.b01_grid_layers import LAYER_BACKGROUND, LAYER_FLOOR, LAYER_WALL from roborock.map.b01_q10_map_parser import ( B01Q10MapParser, + Q10CleanRecordMapPacket, + 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, @@ -96,7 +102,7 @@ def test_lz4_block_roundtrip_all_literals() -> None: block.append(0x0F << 4) block.append(len(original) - 15) block += original - assert lz4_block_decompress(bytes(block)) == original + assert lz4_block_decompress(bytes(block), max_output_size=len(original)) == original def test_lz4_block_back_reference() -> None: @@ -104,7 +110,14 @@ def test_lz4_block_back_reference() -> None: # seq1: 1 literal 'A', then match (offset 1, length 4+4=8) -> 'A' x9. # seq2: final literals-only token (0 literals) ends the block per LZ4 spec. block = bytes([0x14, ord("A"), 0x01, 0x00, 0x00]) - assert lz4_block_decompress(block) == b"A" * 9 + assert lz4_block_decompress(block, max_output_size=9) == 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: @@ -356,6 +369,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 +386,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 +419,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 +473,89 @@ 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 isinstance(packet, Q10CleanRecordMapPacket) + 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 isinstance(packet, Q10CleanRecordMapPacket) + 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 not isinstance(current, Q10CleanRecordMapPacket) + assert not isinstance(saved_map, Q10CleanRecordMapPacket) + + +@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 isinstance(packet, Q10CleanRecordMapPacket) + 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 isinstance(packet, Q10CleanRecordMapPacket) + 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 isinstance(packet, Q10CleanRecordMapPacket) + 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..429182fe 100644 --- a/tests/map/test_b01_q10_render.py +++ b/tests/map/test_b01_q10_render.py @@ -16,9 +16,12 @@ from roborock.map.b01_grid_layers import GridCalibration from roborock.map.b01_q10_map_parser import ( B01Q10MapParserConfig, + Q10CleanRecordMapPacket, Q10EraseZone, Q10HeaderCalibration, + Q10HistoricalTracePacket, Q10MapPacket, + Q10MapPacketKind, Q10Point, Q10TracePacket, parse_map_packet, @@ -117,6 +120,18 @@ 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) + + archived = Q10CleanRecordMapPacket( + **vars(replace(packet, kind=Q10MapPacketKind.CLEAN_RECORD_DETAIL)), historical_trace=historical + ) + assert _render(archived) == _render(packet, trace=live_trace) + assert _render(archived, trace=Q10TracePacket(points=[])) == _render(archived) + + 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)