From a412e9e5dde912b6afc7a77fbc1a9ca3fe133cf7 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 19:14:17 +0400 Subject: [PATCH 1/2] feat: decode Q10 map geometry metadata --- roborock/map/b01_grid_layers.py | 56 ++++ roborock/map/b01_q10_map_parser.py | 342 +++++++++++++++++++-- roborock/map/b01_q10_overlays.py | 75 ++++- tests/map/test_b01_grid_layers.py | 70 ++++- tests/map/test_b01_q10_map_parser.py | 170 ++++++++-- tests/map/test_b01_q10_overlays.py | 104 +++++++ tests/map/testdata/b01_q10_trace_multi.bin | Bin 22 -> 26 bytes tests/protocols/test_b01_q10_protocol.py | 5 +- 8 files changed, 760 insertions(+), 62 deletions(-) diff --git a/roborock/map/b01_grid_layers.py b/roborock/map/b01_grid_layers.py index 410d60fd..46026f17 100644 --- a/roborock/map/b01_grid_layers.py +++ b/roborock/map/b01_grid_layers.py @@ -200,6 +200,62 @@ def solve_calibration( return best[1] +def solve_calibration_with_origin( + layers: GridLayers, + points: list[tuple[float, float]], + origin: tuple[float, float], + *, + resolutions: Iterable[float], + y_signs: Iterable[int] = (1, -1), + min_on_floor: float = 0.5, +) -> GridCalibration | None: + """Fit resolution + Y orientation around a *known* grid-pixel origin. + + Unlike :func:`solve_calibration`, the pixel origin ``(ox, oy)`` is fixed -- + e.g. read straight from the Q10 grid-frame header -- so this only sweeps the + candidate ``resolutions`` and ``y_signs`` and keeps the placement landing the + most ``points`` on floor. With the expensive 2-D offset slide gone, far fewer + points are needed to confirm the fit, so it works from a short path rather + than a dense clean. Returns ``None`` if no candidate lands a ``min_on_floor`` + fraction of points on floor (e.g. the origin or points are bogus). + """ + if not points: + return None + w, h = layers.width, layers.height + ox, oy = origin + classify = layers.classifier + # 1 = floor, 2 = wall/background (blocked), 0 = other. Index by cell. + klass = bytes( + 1 if (c := classify(v)) == LAYER_FLOOR else 2 if c in (LAYER_WALL, LAYER_BACKGROUND) else 0 for v in layers.grid + ) + + best: tuple[float, GridCalibration] | None = None + for resolution in resolutions: + if resolution <= 0: + continue + for y_sign in y_signs: + on_floor = 0 + blocked = 0 + for x, y in points: + px = int(x / resolution + ox) + py = int(oy - y_sign * y / resolution) + if not (0 <= px < w and 0 <= py < h): + blocked += 1 + continue + k = klass[py * w + px] + if k == 1: + on_floor += 1 + elif k == 2: + blocked += 1 + score = on_floor - 1.5 * blocked + if best is None or score > best[0]: + best = (score, GridCalibration(float(resolution), float(ox), float(oy), y_sign)) + + if best is None or best[0] < len(points) * min_on_floor: + return None + return best[1] + + def decompose_grid( width: int, height: int, diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index 89dec083..52dfc76d 100644 --- a/roborock/map/b01_q10_map_parser.py +++ b/roborock/map/b01_q10_map_parser.py @@ -23,7 +23,7 @@ import io import math import statistics -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from PIL import Image from vacuum_map_parser_base.config.image_config import ImageConfig @@ -31,10 +31,46 @@ from roborock.exceptions import RoborockException +from .b01_grid_layers import ( + LAYER_BACKGROUND, + LAYER_FLOOR, + LAYER_UNKNOWN, + LAYER_WALL, + GridLayers, + decompose_grid, +) from .map_parser import ParsedMapData _MAP_FILE_FORMAT = "PNG" +# Semantic raster classes, confirmed against real ss07 captures (rendered and +# eyeballed): 243 is the background outside the home (~half the grid), 240 is +# scanned floor not yet assigned to a room, other values >= 240 are walls, and +# 0 < value < 240 are per-room floor cells (value == room_id * 4). +_BACKGROUND_VALUE = 243 +_UNSEGMENTED_FLOOR_VALUE = 240 + + +def classify_q10_cell(value: int) -> str: + """Map a Q10 grid cell value to a canonical layer class.""" + if value == 0: + return LAYER_UNKNOWN + if value == _BACKGROUND_VALUE: + return LAYER_BACKGROUND + if value == _UNSEGMENTED_FLOOR_VALUE: + return LAYER_FLOOR + if value >= _WALL_THRESHOLD: + return LAYER_WALL + return LAYER_FLOOR + + +def decompose_layers(packet: "Q10MapPacket") -> GridLayers: + """Split a parsed Q10 map packet into separable grid-pixel layers.""" + rooms = [(room.id, room.name, room.pixel_value, room.pixel_count) for room in packet.rooms] + # The ss07 grid is stored top-down (row 0 = top), so no display flip is applied. + return decompose_grid(packet.width, packet.height, packet.grid, rooms, classify_q10_cell, flip=False) + + MAP_PACKET_MARKER = b"\x01\x01" TRACE_PACKET_MARKER = b"\x02\x01" @@ -52,6 +88,28 @@ _ROOM_RECORD_LENGTH = 47 _ROOM_NAME_LENGTH_OFFSET = 26 _MAX_ROOMS = 32 +# Sanity bound for the erase-zone vector section's vertices-per-polygon field. +_MAX_ERASE_ZONE_VERTICES = 16 + +# The 01 01 grid-frame header also carries the map's calibration, so a +# GridCalibration can be derived without fitting a cleaning path (i.e. docked / +# pre-clean). Absolute byte offsets in the frame, reported and verified by +# @andrewlyeats across independent ss07 (fw 03.11.24) captures and cross-checked +# with the ioBroker roborock adapter: +# - 11-12 x_min, 13-14 y_min (s16be): map origin in 5 mm units. The grid is +# 50 mm/px, so dividing by 10 yields the origin in grid pixels -- the (ox, oy) +# that solve_calibration otherwise recovers by sliding the path. +# - 15-16 resolution (u16be): reads 5 (= 0.05 m/px = 50 mm/px) universally. +# - 17-18 charger x, 19-20 charger y (s16be, 5 mm units), 21-22 charger phi. +_ORIGIN_X_OFFSET = 11 +_ORIGIN_Y_OFFSET = 13 +_HEADER_RESOLUTION_OFFSET = 15 +_CHARGER_X_OFFSET = 17 +_CHARGER_Y_OFFSET = 19 +_CHARGER_PHI_OFFSET = 21 +# The header origin/charger are in 5 mm units and the grid is 50 mm/px, so a +# header coordinate maps to grid pixels by dividing by this. +_HEADER_UNITS_PER_PIXEL = 10 # Grid cell values >= this are walls / borders rather than room segments. _WALL_THRESHOLD = 240 @@ -72,6 +130,60 @@ def name(self) -> str: return self.raw_name.removeprefix("rr_").replace("_", " ").strip().title() +@dataclass +class Q10EraseZone: + """A user-drawn "erase" area (polygon) carried in the map packet. + + These are the app's *Erase* tool rectangles -- regions the user marked to be + removed from the map (e.g. phantom floor the lidar mapped through windows). + Coordinates are world units (millimetres), same frame as the path/zones. + + Confirmed by a controlled diff: removing the two erase zones on a live device + dropped this section's count from 2 to 0 while the grid and the trailing + raster were byte-identical. (Earlier revisions misidentified this section as + "carpets"; it is the erase-zone list.) + """ + + vertices: list[tuple[int, int]] = field(default_factory=list) + + +@dataclass +class Q10HeaderCalibration: + """Calibration carried in the ``01 01`` grid-frame header (ss07). + + Lets a :class:`~roborock.map.b01_grid_layers.GridCalibration` origin be read + straight from the map packet -- no cleaning path / fit required, so it works + docked or pre-clean. See :meth:`origin_pixels`. + + ``origin_x`` / ``origin_y`` and the charger coordinates are in 5 mm units; + ``resolution`` is the raw header field (5 == 50 mm/px). ``charger_phi`` is the + raw dock heading field. Reported and verified by @andrewlyeats (ss07). + """ + + origin_x: int + origin_y: int + resolution: int + charger_x: int + charger_y: int + charger_phi: int + + @property + def is_keepalive(self) -> bool: + """True for null/keepalive frames (``x_min == y_min == 0``), which carry + no usable origin -- callers should fall back to a path-fit calibration.""" + return self.origin_x == 0 and self.origin_y == 0 + + def origin_pixels(self) -> tuple[float, float] | None: + """The grid-pixel origin ``(ox, oy)`` for a ``GridCalibration``. + + The header origin is in 5 mm units and the grid is 50 mm/px, so the + pixel origin is the header value divided by 10. Returns ``None`` for a + keepalive frame (no origin to use).""" + if self.is_keepalive: + return None + return (self.origin_x / _HEADER_UNITS_PER_PIXEL, self.origin_y / _HEADER_UNITS_PER_PIXEL) + + @dataclass class Q10MapPacket: """Decoded contents of a Q10 ``01 01`` map packet.""" @@ -81,6 +193,14 @@ class Q10MapPacket: height: int grid: bytes rooms: list[Q10Room] = field(default_factory=list) + erase_zones: list[Q10EraseZone] = field(default_factory=list) + """Erase areas decoded from the packet tail (world coordinates).""" + header_calibration: Q10HeaderCalibration | None = None + """Calibration read straight from the grid-frame header (ss07), or ``None``.""" + carpet_mask: bytes | None = None + """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.""" @dataclass @@ -98,13 +218,14 @@ class Q10TracePacket: The robot accumulates the **full path of the current cleaning session** and serves it in a single packet: ``points`` holds the whole trajectory so far (oldest first), growing as the robot cleans. This was confirmed live -- a - corridor run produced packets of 1, then 3, then 15 points, each a strict - superset describing the path travelled. Because the robot keeps the path - server-side, a client that connects mid-session still receives the complete - path (this is how the app shows the trail even after a cold launch). - - The robot only emits these while a session is active, so an idle/docked robot - will not produce them. The most recent point is the current robot position. + corridor run produced packets of 0 (just a heading, docked), then 3, then 14 + points, each a strict superset describing the path travelled. Because the + robot keeps the path server-side, a client that connects mid-session still + receives the complete path (this is how the app shows the trail even after a + cold launch). + + A docked/idle robot can still emit a packet carrying only the ``heading`` + (zero points). The most recent point is the current robot position. """ points: list[Q10Point] = field(default_factory=list) @@ -112,30 +233,52 @@ class Q10TracePacket: """Session counter (byte 3); increments per cleaning session, tracking the device clean count. Not a per-packet sequence.""" + heading: int = 0 + """Robot heading from the 0201 SLAM field (bytes 10-11), in degrees: + ``0`` = +x, ``+90`` = +y, ``±180`` = −x, ``−90`` = −y. This is the current + orientation; pair it with :attr:`robot_position` to draw a facing robot. + + Convention (incl. the y-sign) ground-truthed on a live R1 clean: across + straight segments the reported heading equalled the direction of travel + ``atan2(dy, dx)`` -- +x read 0, −x read ±180, a slight −y drift read + negative.""" + @property def robot_position(self) -> Q10Point | None: """The current robot position (the most recent point).""" return self.points[-1] if self.points else None -# Trace packet (``02 01``): a 10-byte header followed by big-endian int16 (x, y) +# 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: byte 3 is a session counter (tracks the device -# clean count); bytes 8-9 are a u16be point count minus one (verified: a 15-point -# packet carried 0x000e == 14). The parser reads all 4-byte pairs in the body -# rather than trusting the count field, so a truncated tail can't desync it. +# against live ss07 captures and cross-checked by @andrewlyeats: +# - byte 3: a session counter (tracks the device clean count). +# - bytes 8-9: a u16be point count -- the exact number of (x, y) pairs from +# byte 14 (verified: captures of 1417 / 2462 points carried 0x0589 / 0x099e). +# - 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. +# 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. # NOTE: the format documented by roborock-qseries-map-bridge (18-byte header) -# did not match this firmware -- this 10-byte layout is what the device sent. -_TRACE_HEADER_LENGTH = 10 +# did not match this firmware -- this 14-byte layout is what the device sent. +_TRACE_HEADER_LENGTH = 14 _TRACE_SEQUENCE_OFFSET = 3 - -# Some cleans prepend a single stray point to the path, far outside the map -# (e.g. ~(0, -1907) when the real path starts near (-3760, -1920)); it skews the -# rendered start/bounding box and any path-based calibration. We drop points[0] -# only when its step to points[1] is a gross outlier (this multiple of the -# median step of the rest of the path), so a genuine first point is never lost. -# The current position (last point) is unaffected. Trigger and threshold -# reported and verified by @andrewlyeats across independent B01/Q10 captures. +_TRACE_HEADING_OFFSET = 10 + +# 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 +# is distinct from the heading word the old 10-byte header used to surface as a +# phantom point -- that is now consumed by the header.) We drop points[0] only +# when its step to points[1] is a gross outlier (this multiple of the median step +# of the rest of the path), so a genuine first point is never lost. The current +# position (last point) is unaffected. Trigger and threshold reported and +# verified by @andrewlyeats across independent B01/Q10 captures. _STRAY_POINT_STEP_RATIO = 20 @@ -159,6 +302,7 @@ def parse_trace_packet(payload: bytes) -> Q10TracePacket: if len(body) % 4: raise RoborockException("Q10 trace points are not 4-byte (x, y) pairs") + heading = int.from_bytes(payload[_TRACE_HEADING_OFFSET : _TRACE_HEADING_OFFSET + 2], "big", signed=True) points = [ Q10Point( x=int.from_bytes(body[offset : offset + 2], "big", signed=True), @@ -167,7 +311,7 @@ def parse_trace_packet(payload: bytes) -> Q10TracePacket: for offset in range(0, len(body), 4) ] points = _drop_stray_leading_point(points) - return Q10TracePacket(points=points, sequence=payload[_TRACE_SEQUENCE_OFFSET]) + return Q10TracePacket(points=points, sequence=payload[_TRACE_SEQUENCE_OFFSET], heading=heading) def _drop_stray_leading_point(points: list[Q10Point]) -> list[Q10Point]: @@ -323,7 +467,132 @@ def parse_map_packet(payload: bytes) -> Q10MapPacket: else: height, grid, room_data = _infer_layout(decoded, width) rooms = _parse_rooms(room_data, grid) - return Q10MapPacket(map_id=map_id, width=width, height=height, grid=grid, rooms=rooms) + tail = payload[layout_end:] + erase_zones = _parse_erase_zones(tail) + carpet_mask = _parse_carpet_mask(tail, width, height) + header_calibration = _parse_header_calibration(payload) + return Q10MapPacket( + map_id=map_id, + width=width, + height=height, + grid=grid, + rooms=rooms, + erase_zones=erase_zones, + header_calibration=header_calibration, + carpet_mask=carpet_mask, + ) + + +def _parse_header_calibration(payload: bytes) -> Q10HeaderCalibration | None: + """Read the calibration fields from the ``01 01`` grid-frame header. + + All fields sit inside the fixed 29-byte header (already length-checked by + the caller). See the offset constants for the byte layout. Returns the + decoded :class:`Q10HeaderCalibration` (callers skip keepalive frames via + :attr:`Q10HeaderCalibration.is_keepalive`).""" + + def s16(offset: int) -> int: + return int.from_bytes(payload[offset : offset + 2], "big", signed=True) + + return Q10HeaderCalibration( + origin_x=s16(_ORIGIN_X_OFFSET), + origin_y=s16(_ORIGIN_Y_OFFSET), + resolution=int.from_bytes(payload[_HEADER_RESOLUTION_OFFSET : _HEADER_RESOLUTION_OFFSET + 2], "big"), + charger_x=s16(_CHARGER_X_OFFSET), + charger_y=s16(_CHARGER_Y_OFFSET), + charger_phi=s16(_CHARGER_PHI_OFFSET), + ) + + +def _parse_erase_zones(tail: bytes) -> list[Q10EraseZone]: + """Decode erase areas from the bytes after the compressed grid block. + + The tail begins with a vector section ``[count: u8][vertices_per: u8]`` then + ``count`` polygons of ``vertices_per`` int16-BE (x, y) pairs (axis-aligned + rectangles in practice). Identified by a controlled diff on a live ss07 + device: deleting the two app *Erase* zones dropped ``count`` 2->0 with the + rest of the packet byte-identical. The remaining tail (a run-length raster + + signature) is unrelated to erase and is not decoded here. + """ + if len(tail) < 2: + return [] + count = tail[0] + vertices_per = tail[1] + if count == 0 or not 1 <= vertices_per <= _MAX_ERASE_ZONE_VERTICES: + return [] + + erase_zones: list[Q10EraseZone] = [] + offset = 2 + for _ in range(count): + end = offset + vertices_per * 4 + if end > len(tail): + break + vertices = [ + ( + int.from_bytes(tail[offset + j * 4 : offset + j * 4 + 2], "big", signed=True), + int.from_bytes(tail[offset + j * 4 + 2 : offset + j * 4 + 4], "big", signed=True), + ) + for j in range(vertices_per) + ] + erase_zones.append(Q10EraseZone(vertices=vertices)) + offset = end + return erase_zones + + +def _carpet_offset(tail: bytes) -> int: + """Byte offset within the post-grid tail where the carpet mask begins. + + The erase section is ``[u8 count][u8 vertices_per]`` then ``count`` polygons + of ``vertices_per`` int16 (x, y) pairs, so it always occupies + ``2 + count*vertices_per*4`` bytes -- just the 2-byte header when count is 0. + """ + if len(tail) < 2: + return len(tail) + count, vertices_per = tail[0], tail[1] + return 2 + count * vertices_per * 4 + + +def _parse_carpet_mask(tail: bytes, width: int, height: int) -> bytes | None: + """Decode the carpet mask that follows the erase section in the packet tail. + + Framing matches the main grid block: ``[u32 uncompressed_len]`` + ``[u16 compressed_len][LZ4 block]``. The decompressed mask is a full + ``width*height`` grid in the same (top-down) pixel space as the main grid; a + 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). + """ + offset = _carpet_offset(tail) + if offset + 6 > len(tail): + return 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 + try: + mask = lz4_block_decompress(tail[offset + 6 : block_end]) + except RoborockException: + return None + return mask if len(mask) == width * height else None + + +def erased_packet(packet: "Q10MapPacket", cells: set[int]) -> "Q10MapPacket": + """Return a copy of ``packet`` with ``cells`` (grid indices) set to background. + + Used to apply the app's erase zones: cells inside an erase rectangle are blanked + to the background class so they drop out of the rendered map and every layer. + """ + if not cells: + return packet + grid = bytearray(packet.grid) + for cell in cells: + if 0 <= cell < len(grid): + grid[cell] = _BACKGROUND_VALUE + return replace(packet, grid=bytes(grid)) @dataclass @@ -340,16 +609,17 @@ class B01Q10MapParser: def __init__(self, config: B01Q10MapParserConfig | None = None) -> None: self._config = config or B01Q10MapParserConfig() + @property + def config(self) -> B01Q10MapParserConfig: + """The parser configuration (image scale, ...).""" + return self._config + def parse(self, payload: bytes) -> ParsedMapData: """Parse a raw Q10 map packet into a rendered PNG + ``MapData``.""" - return self.parse_packet(parse_map_packet(payload)) - - def parse_packet(self, packet: Q10MapPacket) -> ParsedMapData: - """Render an already-parsed Q10 map packet into a PNG + ``MapData``. + return self.parsed_from_packet(parse_map_packet(payload)) - The protocol layer parses the wire bytes into a :class:`Q10MapPacket`; - this renders that packet without re-parsing it. - """ + def parsed_from_packet(self, packet: Q10MapPacket) -> ParsedMapData: + """Render a (possibly erase-modified) packet into a PNG + ``MapData``.""" image = self._render(packet) map_data = MapData() @@ -367,6 +637,11 @@ def parse_packet(self, packet: Q10MapPacket) -> ParsedMapData: if room_names: map_data.additional_parameters["room_names"] = room_names + # Carpet cells are flat grid indices (y*width + x) in the same top-down + # pixel space as the rendered raster, so they line up with the image. + if packet.carpet_mask is not None: + map_data.carpet_map = {i for i, value in enumerate(packet.carpet_mask) if value} + image_bytes = io.BytesIO() image.save(image_bytes, format=_MAP_FILE_FORMAT) return ParsedMapData(image_content=image_bytes.getvalue(), map_data=map_data) @@ -378,7 +653,8 @@ def _render(self, packet: Q10MapPacket) -> Image.Image: for value in packet.grid: rgb.extend(palette[value]) img = Image.frombytes("RGB", (packet.width, packet.height), bytes(rgb)) - img = img.transpose(Image.Transpose.FLIP_TOP_BOTTOM) + # The ss07 grid is stored top-down (row 0 = top of the home), so it is + # rendered as-is -- unlike the V1/Q7 convention, no vertical flip. scale = self._config.map_scale if scale > 1: img = img.resize((packet.width * scale, packet.height * scale), resample=Image.Resampling.NEAREST) diff --git a/roborock/map/b01_q10_overlays.py b/roborock/map/b01_q10_overlays.py index 13e67491..34fcf71c 100644 --- a/roborock/map/b01_q10_overlays.py +++ b/roborock/map/b01_q10_overlays.py @@ -4,13 +4,21 @@ the map raster; the device reports them as base64-encoded blobs in separate data points (``dpRestrictedZoneUp`` 55, ``dpVirtualWallUp`` 57, ``dpZonedUp`` 59). -The blob format was reverse-engineered from a live ss07 (confirmed against 7 -real no-go zones): +The restricted-zone / zoned blob format (DP 55 and DP 59) was reverse-engineered +from a live ss07 (confirmed against 7 real no-go zones): [version: u8][count: u8] then ``count`` fixed-size records, each: [type: u8][vertex_count: u8] then vertex_count (x, y) int16-BE pairs, zero-padded to the record size. +Use :func:`parse_zone_blob` for those. Virtual walls (DP 57) use a *different +framing* -- a bare ``[count]`` and 8-byte ``(x, y)`` records, no version/type/pad +-- so they have their own :func:`parse_virtual_wall_blob`; feeding DP 57 to +:func:`parse_zone_blob` mis-frames it (the leading byte is read as a version and +the next, a coordinate, as a record count). The coordinate order matches the +zones (first wire word = x), confirmed against the app. Provenance and the +byte-level breakdown are in PR #850's review thread. + Coordinates are in the device's world units (the same space as the cleaning path), so a :class:`~roborock.map.b01_grid_layers.GridCalibration` maps them to map pixels. ``type`` distinguishes the restriction kind (2 = no-mop, 3 = door @@ -83,12 +91,67 @@ def parse_zone_blob(data: bytes | str | None) -> list[Q10Zone]: return zones +_WALL_RECORD_SIZE = 8 # two (x, y) int16-BE endpoints + + +def parse_virtual_wall_blob(data: bytes | str | None) -> list[Q10Zone]: + """Decode a Q10 virtual-wall overlay blob (``dpVirtualWallUp`` 57). + + Virtual walls use a *different framing* from the restricted-zone DPs handled + by :func:`parse_zone_blob`: a single ``[count: u8]`` byte (no version, no + per-record type/pad) followed by ``count`` 8-byte records, each two + ``(x, y)`` int16-BE endpoints. The *coordinate order is the same* as the + restricted zones (first wire word = x), so a wall and a no-go zone drawn on + the same line decode parallel rather than transposed. + + Each wall is returned as a :class:`Q10Zone` of type + :data:`ZONE_TYPE_VIRTUAL_WALL` with its two ``(x, y)`` endpoints, so callers + can place them onto the map through the same + :class:`~roborock.map.b01_grid_layers.GridCalibration` as the zones. + + Accepts raw bytes or the base64 string straight from the data point. Returns + ``[]`` for empty/absent/unparsable blobs (the device sends a single ``0x00`` + byte -- base64 ``AA==`` -- when there are none). + + The axis order was confirmed against the app: a horizontal wall drawn below + a room reads back with x varying and y constant (and the wide RDC no-go zone + reads back wide), so DP 57 shares DP 55's order. An earlier revision swapped + the wall axes to ``(y, x)`` -- following a misreading of PR #850's notes -- + which placed every wall transposed 90 degrees from where it was drawn. + """ + raw = _as_bytes(data) + if len(raw) < 1: + return [] + count = raw[0] + if count <= 0: + return [] + + body = raw[1:] + walls: list[Q10Zone] = [] + for index in range(count): + record = body[index * _WALL_RECORD_SIZE : (index + 1) * _WALL_RECORD_SIZE] + if len(record) < _WALL_RECORD_SIZE: + break # truncated trailing record; stop rather than misread + vertices = [ + ( + int.from_bytes(record[p * 4 : p * 4 + 2], "big", signed=True), + int.from_bytes(record[p * 4 + 2 : p * 4 + 4], "big", signed=True), + ) + for p in range(2) + ] + walls.append(Q10Zone(type=ZONE_TYPE_VIRTUAL_WALL, vertices=vertices)) + return walls + + # Observed ``type`` values, confirmed against an ss07 Q10 (firmware 03.11.24) and # cross-checked with the ioBroker roborock adapter: 2 = no-mop, 3 = door -# threshold, 1 = virtual wall. Any other value (including 0) is a no-go zone. -# In practice virtual walls arrive on a separate DP (VIRTUAL_WALL_UP 57), so this -# restricted-zone DP normally only carries 0 / 2 / 3. The raw value is also kept -# on ``Q10Zone.type`` for callers that recognise it. +# threshold. Any other value (including 0) is a no-go zone. The raw value is also +# kept on ``Q10Zone.type`` for callers that recognise it. +# +# Virtual walls arrive on a separate DP (VIRTUAL_WALL_UP 57) with their own frame +# (see :func:`parse_virtual_wall_blob`), so this restricted-zone DP only carries +# 0 / 2 / 3 -- never a 1. ``ZONE_TYPE_VIRTUAL_WALL`` is kept here only to tag the +# walls that :func:`parse_virtual_wall_blob` produces. # # Corrected from an earlier reading that treated type 3 as no-mop -- 3 is the # door-threshold rectangle; the no-mop area reads back as type 2. Reported and diff --git a/tests/map/test_b01_grid_layers.py b/tests/map/test_b01_grid_layers.py index 2299ac3f..52954203 100644 --- a/tests/map/test_b01_grid_layers.py +++ b/tests/map/test_b01_grid_layers.py @@ -1,7 +1,9 @@ -"""Tests for the device-agnostic grid->layers decomposition + calibration.""" +"""Tests for the device-agnostic grid->layers decomposition + Q10 classifier.""" import io +from pathlib import Path +import pytest from PIL import Image from roborock.map.b01_grid_layers import ( @@ -11,7 +13,30 @@ GridCalibration, decompose_grid, solve_calibration, + solve_calibration_with_origin, ) +from roborock.map.b01_q10_map_parser import ( + classify_q10_cell, + decompose_layers, + parse_map_packet, +) + +FIXTURE = Path(__file__).resolve().parent / "testdata" / "b01_q10_map.bin" + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (0, "unknown"), + (8, LAYER_FLOOR), + (12, LAYER_FLOOR), + (240, LAYER_FLOOR), + (243, LAYER_BACKGROUND), + (249, LAYER_WALL), + ], +) +def test_classify_q10_cell(value: int, expected: str) -> None: + assert classify_q10_cell(value) == expected def test_decompose_grid_generic_classifier_and_bbox() -> None: @@ -72,6 +97,25 @@ def test_render_scale_upsamples() -> None: assert Image.open(io.BytesIO(png)).size == (6, 3) +def test_decompose_layers_on_q10_fixture() -> None: + """The Q10 synthetic fixture splits into floor + per-room layers.""" + layers = decompose_layers(parse_map_packet(FIXTURE.read_bytes())) + assert layers.class_counts.get(LAYER_FLOOR) == 26 + names = {room.id: room.name for room in layers.rooms} + assert names == {2: "Living Room", 3: "Bedroom"} + # Each room renders to a valid PNG and only its own pixels are opaque. + living = layers.render_room(2, (255, 0, 0, 255)) + img = Image.open(io.BytesIO(living)) + opaque = sum(1 for *_rgb, a in img.getdata() if a > 0) + assert opaque == next(r.pixel_count for r in layers.rooms if r.id == 2) + + +def test_render_room_unknown_id_raises() -> None: + layers = decompose_layers(parse_map_packet(FIXTURE.read_bytes())) + with pytest.raises(KeyError): + layers.render_room(999, (0, 0, 0, 255)) + + def test_calibration_roundtrip() -> None: cal = GridCalibration(resolution=2.0, origin_x=3.0, origin_y=8.0, y_sign=1) assert cal.world_to_pixel(0, 0) == (3.0, 8.0) @@ -117,3 +161,27 @@ def test_solve_calibration_returns_none_when_unfittable() -> None: # Points so far apart no resolution keeps them on the 6x6 floor block. points = [(0.0, 0.0), (1000.0, 0.0), (0.0, 1000.0)] assert solve_calibration(layers, points, resolutions=[2.0]) is None + + +def test_solve_calibration_with_origin_fits_resolution_from_short_path() -> None: + """With a known origin only resolution is fit, so a tiny path suffices.""" + layers = _floor_block_layers() + true = GridCalibration(2.0, 3.0, 8.0, 1) + points = [true.pixel_to_world(px, py) for px, py in [(4, 7), (6, 5)]] # two points + cal = solve_calibration_with_origin(layers, points, (true.origin_x, true.origin_y), resolutions=[1.0, 2.0, 3.0]) + assert cal is not None + assert (cal.resolution, cal.origin_x, cal.origin_y, cal.y_sign) == (2.0, 3.0, 8.0, 1) + + +def test_solve_calibration_with_origin_returns_none_off_floor() -> None: + """A wrong origin that lands the path off floor is rejected, not forced.""" + layers = _floor_block_layers() + true = GridCalibration(2.0, 3.0, 8.0, 1) + points = [true.pixel_to_world(px, py) for px, py in [(4, 7), (6, 5)]] + # Origin shoved into the background corner: every point lands off floor. + assert solve_calibration_with_origin(layers, points, (0.0, 0.0), resolutions=[2.0]) is None + + +def test_solve_calibration_with_origin_returns_none_without_points() -> None: + layers = _floor_block_layers() + assert solve_calibration_with_origin(layers, [], (3.0, 8.0), resolutions=[2.0]) is None diff --git a/tests/map/test_b01_q10_map_parser.py b/tests/map/test_b01_q10_map_parser.py index 53c3c63c..2ceb5e72 100644 --- a/tests/map/test_b01_q10_map_parser.py +++ b/tests/map/test_b01_q10_map_parser.py @@ -18,7 +18,7 @@ FIXTURE = Path(__file__).resolve().parent / "testdata" / "b01_q10_map.bin" TRACE_FIXTURE = Path(__file__).resolve().parent / "testdata" / "b01_q10_trace.bin" TRACE_MULTI_FIXTURE = Path(__file__).resolve().parent / "testdata" / "b01_q10_trace_multi.bin" -# Real 15-point packet captured from an R1 corridor run (full session path). +# Real 14-point packet captured from an R1 corridor run (full session path). TRACE_SESSION_FIXTURE = Path(__file__).resolve().parent / "testdata" / "b01_q10_trace_session.bin" @@ -75,10 +75,12 @@ def _full_header_map_payload(width: int, height: int, decoded_layout: bytes) -> return bytes(payload) -def _trace_payload(points: list[tuple[int, int]], sequence: int = 1) -> bytes: - header = bytearray(10) +def _trace_payload(points: list[tuple[int, int]], sequence: int = 1, heading: int = 0) -> bytes: + header = bytearray(14) # _TRACE_HEADER_LENGTH header[0:2] = b"\x02\x01" header[3] = sequence + header[8:10] = len(points).to_bytes(2, "big") # point count == number of (x, y) pairs + header[10:12] = heading.to_bytes(2, "big", signed=True) body = b"".join(x.to_bytes(2, "big", signed=True) + y.to_bytes(2, "big", signed=True) for x, y in points) return bytes(header) + body @@ -190,30 +192,37 @@ def test_packet_markers_are_distinct() -> None: def test_parse_trace_packet_real_single_point() -> None: - """A real ss07 packet captured early in a session has a single path point.""" + """A real ss07 packet captured while docked: a heading, but no path points. + + The 14-byte header carries point count 0 (bytes 8-9) and heading 169 (bytes + 10-11); the rest of the frame is empty. An earlier 10-byte-header revision + mis-decoded the heading word as a phantom path point ``(169, 0)``. + """ trace = parse_trace_packet(TRACE_FIXTURE.read_bytes()) assert trace.sequence == 9 - assert [(p.x, p.y) for p in trace.points] == [(169, 0)] - assert trace.robot_position is not None - assert (trace.robot_position.x, trace.robot_position.y) == (169, 0) + assert trace.heading == 169 + assert trace.points == [] + assert trace.robot_position is None def test_parse_trace_packet_real_session_path() -> None: - """A real 15-point packet (corridor run) decodes the full accumulated path. + """A real 14-point packet (corridor run) decodes the full accumulated path. Captured live from an R1: the same session emitted packets of 1, then 3, - then 15 points, proving the path accumulates rather than reporting only the - current position. The most recent point is the current robot position. + then 14 points, proving the path accumulates rather than reporting only the + current position. The most recent point is the current robot position, and + the header carries the robot heading (bytes 10-11). """ trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) points = [(p.x, p.y) for p in trace.points] - assert len(points) == 15 - assert points[0] == (-34, 0) # oldest + assert len(points) == 14 + assert trace.heading == -34 # robot orientation, not a path point + assert points[0] == (41, 64) # oldest real point (from byte 14) assert points[-1] == (276, -1) # most recent == current position - # After the initial repositioning, x marches steadily down the corridor. - tail_x = [p[0] for p in points[2:]] + # After the initial reposition, x marches steadily down the corridor. + tail_x = [p[0] for p in points[1:]] assert tail_x == sorted(tail_x) - assert points[-1][0] - points[0][0] > 300 # spans the corridor + assert points[-1][0] - points[0][0] > 200 # spans the corridor assert trace.robot_position is not None assert (trace.robot_position.x, trace.robot_position.y) == (276, -1) @@ -222,6 +231,7 @@ def test_parse_trace_packet_multi_point() -> None: """A multi-point packet decodes all points; position is the most recent.""" trace = parse_trace_packet(TRACE_MULTI_FIXTURE.read_bytes()) assert [(p.x, p.y) for p in trace.points] == [(100, 200), (150, 250), (-50, 300)] + assert trace.heading == 45 # Signed coordinates are supported (negative x). assert trace.robot_position is not None assert (trace.robot_position.x, trace.robot_position.y) == (-50, 300) @@ -243,18 +253,25 @@ def test_parse_trace_keeps_genuine_first_point() -> None: assert [(p.x, p.y) for p in trace.points] == points -def test_parse_trace_session_keeps_initial_reposition() -> None: - """The real corridor capture has a 4.8x first step -- well under the 20x cut.""" +def test_parse_trace_reads_heading_not_a_leading_point() -> None: + """The real corridor capture's heading word is decoded as orientation. + + Bytes 10-11 (``ff de`` == -34) are the SLAM heading, not a path point, so the + first genuine point ``(41, 64)`` is kept and no spurious near-origin point is + introduced (the old 10-byte header surfaced ``(-34, 0)`` here). + """ trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) - assert len(trace.points) == 15 - assert (trace.points[0].x, trace.points[0].y) == (-34, 0) + assert trace.heading == -34 + assert len(trace.points) == 14 + assert (trace.points[0].x, trace.points[0].y) == (41, 64) def test_parse_trace_empty_path_has_no_position() -> None: - header_only = b"\x02\x01" + b"\x00" * 8 # 10-byte header, no points + header_only = b"\x02\x01" + b"\x00" * 12 # 14-byte header, no points trace = parse_trace_packet(header_only) assert trace.points == [] assert trace.robot_position is None + assert trace.heading == 0 def test_parse_trace_rejects_non_trace_packet() -> None: @@ -264,7 +281,7 @@ def test_parse_trace_rejects_non_trace_packet() -> None: def test_parse_trace_rejects_misaligned_points() -> None: with pytest.raises(RoborockException, match="not 4-byte"): - parse_trace_packet(b"\x02\x01" + b"\x00" * 8 + b"\x01\x02\x03") + parse_trace_packet(b"\x02\x01" + b"\x00" * 12 + b"\x01\x02\x03") def test_parse_rejects_bad_layout_length() -> None: @@ -272,3 +289,114 @@ def test_parse_rejects_bad_layout_length() -> None: payload[27:29] = (0xFFFF).to_bytes(2, "big") # compressed length past the buffer with pytest.raises(RoborockException, match="invalid layout block length"): 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 = [ + [(100, 200), (300, 200), (300, 50), (100, 50)], + [(-40, -10), (10, -10), (10, -60), (-40, -60)], + ] + tail = bytes([len(rects), 4]) + for rect in rects: + for x, y in rect: + tail += int.to_bytes(x & 0xFFFF, 2, "big") + int.to_bytes(y & 0xFFFF, 2, "big") + packet = parse_map_packet(FIXTURE.read_bytes() + tail) + assert [z.vertices for z in packet.erase_zones] == rects # incl. signed coords + + +def test_parse_map_packet_without_erase_tail() -> None: + assert parse_map_packet(FIXTURE.read_bytes()).erase_zones == [] + + +def _carpet_tail(width: int, height: int, carpet: bytes, erase: bytes = bytes([0, 0])) -> bytes: + """Build a packet tail: an erase section followed by a carpet-mask section.""" + block = _literal_lz4_block(carpet) + return erase + (width * height).to_bytes(4, "big") + len(block).to_bytes(2, "big") + block + + +def test_parse_carpet_mask_from_map_packet_tail() -> None: + """A carpet mask after the erase section decodes to a same-dims grid. + + Framing confirmed byte-exact on live ss07 captures (R1 / RDC): the carpet + mask is a second LZ4 grid (``[u32 uncompressed][u16 compressed][block]``) + the same size as the main grid, with non-zero cells marking carpet. + """ + width, height = 8, 6 # the fixture's dimensions + carpet = bytearray(width * height) + carpet[10] = 4 # one carpet cell (kind 4) + carpet[20] = 3 # another carpet kind + # A non-empty erase section in front, so the carpet offset must skip it. + erase = bytes([1, 4]) + b"".join( + int.to_bytes(v & 0xFFFF, 2, "big") for x, y in [(0, 0), (4, 0), (4, 4), (0, 4)] for v in (x, y) + ) + packet = parse_map_packet(FIXTURE.read_bytes() + _carpet_tail(width, height, bytes(carpet), erase)) + + assert len(packet.erase_zones) == 1 # erase still parses + assert packet.carpet_mask == bytes(carpet) + map_data = B01Q10MapParser().parsed_from_packet(packet).map_data + assert map_data is not None + assert map_data.carpet_map == {10, 20} # flat grid indices of the carpet cells + + +def test_parse_map_packet_without_carpet() -> None: + assert parse_map_packet(FIXTURE.read_bytes()).carpet_mask 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) + block = _literal_lz4_block(carpet) + tail = bytes([0, 0]) + (999).to_bytes(4, "big") + len(block).to_bytes(2, "big") + block + assert parse_map_packet(FIXTURE.read_bytes() + tail).carpet_mask is None + + +def _calibrated_map_payload( + width: int, + height: int, + decoded_layout: bytes, + *, + origin: tuple[int, int], + resolution: int = 5, + charger: tuple[int, int, int] = (0, 0, 0), +) -> bytes: + """A map packet with the grid-frame header calibration fields populated.""" + payload = bytearray(_full_header_map_payload(width, height, decoded_layout)) + payload[11:13] = origin[0].to_bytes(2, "big", signed=True) + payload[13:15] = origin[1].to_bytes(2, "big", signed=True) + payload[15:17] = resolution.to_bytes(2, "big") + payload[17:19] = charger[0].to_bytes(2, "big", signed=True) + payload[19:21] = charger[1].to_bytes(2, "big", signed=True) + payload[21:23] = charger[2].to_bytes(2, "big", signed=True) + return bytes(payload) + + +def test_parse_header_calibration_fields() -> None: + """The 01 01 header's calibration fields decode to a usable origin (ss07).""" + grid = bytes([8]) * 6 + bytes([12]) * 6 # 4x3 grid, two rooms + layout = grid + b"\x01\x02" + _room_record(2, "rr_kitchen") + _room_record(3, "den") + payload = _calibrated_map_payload(4, 3, layout, origin=(-3760, 1920), resolution=5, charger=(-50, 30, 180)) + cal = parse_map_packet(payload).header_calibration + assert cal is not None + assert (cal.origin_x, cal.origin_y) == (-3760, 1920) + assert cal.resolution == 5 + assert (cal.charger_x, cal.charger_y, cal.charger_phi) == (-50, 30, 180) + assert not cal.is_keepalive + # 5 mm units / (50 mm/px) -> divide by 10 for grid pixels. + assert cal.origin_pixels() == (-376.0, 192.0) + + +def test_parse_header_calibration_keepalive_has_no_origin() -> None: + """A null/keepalive frame (x_min == y_min == 0) yields no usable origin.""" + grid = bytes([8]) * 6 + bytes([12]) * 6 + layout = grid + b"\x01\x02" + _room_record(2, "rr_kitchen") + _room_record(3, "den") + cal = parse_map_packet(_calibrated_map_payload(4, 3, layout, origin=(0, 0))).header_calibration + assert cal is not None + assert cal.is_keepalive + assert cal.origin_pixels() is None + + +def test_real_fixture_header_calibration_is_keepalive() -> None: + """The synthetic fixture carries no header origin, so callers fall back to a fit.""" + cal = parse_map_packet(FIXTURE.read_bytes()).header_calibration + assert cal is not None and cal.is_keepalive diff --git a/tests/map/test_b01_q10_overlays.py b/tests/map/test_b01_q10_overlays.py index 5af2acc8..1d13eb62 100644 --- a/tests/map/test_b01_q10_overlays.py +++ b/tests/map/test_b01_q10_overlays.py @@ -7,6 +7,7 @@ ZONE_TYPE_NO_MOP, ZONE_TYPE_THRESHOLD, ZONE_TYPE_VIRTUAL_WALL, + parse_virtual_wall_blob, parse_zone_blob, ) @@ -71,3 +72,106 @@ def test_parse_zone_blob_real_record_size_inferred() -> None: rect = _rect(ZONE_TYPE_NO_GO, [(100, 200), (300, 200), (300, 50), (100, 50)]) zones = parse_zone_blob(_blob(1, [rect], record_size=38)) assert len(zones) == 1 and zones[0].vertices[0] == (100, 200) + + +def test_parse_zone_blob_real_rdc_three_no_go() -> None: + """Real DP-55 read-back from our RDC ss07 with three No-Go Zones drawn. + + Captured live; exercises the 38-byte slot walk at count=3 against actual + device bytes (all type 0 = no-go). + """ + blob = ( + "AQMABP9A9ToAEPU6ABDzzv9A884AAAAAAAAAAAAAAAAAAAAAAAAAAAAE/Rb/mv5z/5r+c/3L/Rb9ywAAAA" + "AAAAAAAAAAAAAAAAAAAAAAAAQDpADEB3UAxAd1/GMDpPxjAAAAAAAAAAAAAAAAAAAAAAAAAAA=" + ) + zones = parse_zone_blob(blob) + assert [z.type for z in zones] == [ZONE_TYPE_NO_GO, ZONE_TYPE_NO_GO, ZONE_TYPE_NO_GO] + assert zones[2].vertices == [(932, 196), (1909, 196), (1909, -925), (932, -925)] + + +def test_parse_virtual_wall_blob_real_capture() -> None: + """Real DP-57 read-back from an ss07 (one wall drawn in the app). + + Frame is ``[count]`` + 8-byte ``(x, y)`` records -- no version byte, and the + same coordinate order as the restricted zones (first wire word = x). + Provenance: PR #850 review thread. + """ + walls = parse_virtual_wall_blob("Aflu+cX87PoO") + assert len(walls) == 1 + assert walls[0].type == ZONE_TYPE_VIRTUAL_WALL + assert walls[0].vertices == [(-1682, -1595), (-788, -1522)] + + +def test_parse_virtual_wall_blob_real_r1_horizontal() -> None: + """Real DP-57 read-back from the R1, ground-truthed against the app. + + The wall was drawn horizontally just below the Kids bedroom; it reads back + with x varying (377 -> 951) and y constant (-83), i.e. horizontal. This is + the capture that settled DP 57's axis order: an earlier revision swapped the + axes and would have placed this wall vertical (transposed 90 degrees). + """ + walls = parse_virtual_wall_blob("AQF5/60Dt/+t") + assert [(w.type, w.vertices) for w in walls] == [ + (ZONE_TYPE_VIRTUAL_WALL, [(377, -83), (951, -83)]), + ] + + +def test_parse_virtual_wall_blob_real_r1_mixed_orientation() -> None: + """Real DP-57 read-back from the R1 with one horizontal + one vertical wall. + + Ground-truthed against the app: a wall drawn horizontally and another drawn + (near-)vertically. They decode as such -- the first with y constant, the + second with x near-constant (the 4-unit x drift matches the wall not being + drawn perfectly vertical). This pins DP 57's axis order for *both* + orientations at once, so a mixed pair can't come back transposed. + """ + walls = parse_virtual_wall_blob("AgNyBGMFsARjAf8FAAH7CnE=") + assert [(w.type, w.vertices) for w in walls] == [ + (ZONE_TYPE_VIRTUAL_WALL, [(882, 1123), (1456, 1123)]), # horizontal: y constant + (ZONE_TYPE_VIRTUAL_WALL, [(511, 1280), (507, 2673)]), # vertical: x near-constant + ] + + +def test_parse_virtual_wall_blob_real_rdc_two_walls() -> None: + """Real DP-57 read-back from our RDC ss07 with two Invisible Walls drawn. + + Captured live after drawing the walls in the official app; the old + parse_zone_blob silently returned [] for this exact blob (the regression + this fix addresses). + """ + walls = parse_virtual_wall_blob("AgAz/QMCDv0D/83+Dv/O/OY=") + assert [(w.type, w.vertices) for w in walls] == [ + (ZONE_TYPE_VIRTUAL_WALL, [(51, -765), (526, -765)]), + (ZONE_TYPE_VIRTUAL_WALL, [(-51, -498), (-50, -794)]), + ] + + +def test_parse_virtual_wall_blob_empty_variants() -> None: + assert parse_virtual_wall_blob(None) == [] + assert parse_virtual_wall_blob(b"\x00") == [] # device's "no walls" sentinel + assert parse_virtual_wall_blob("AA==") == [] # base64 of 0x00 + + +def test_parse_virtual_wall_blob_multiple_walls() -> None: + """Two walls back-to-back; each is a separate 8-byte (x, y) record.""" + wall_a = bytes.fromhex("000a0014001e0028") # (x,y)=(10,20)->(30,40) + wall_b = bytes.fromhex("fffb0005fff6000a") # (x,y)=(-5,5)->(-10,10) + walls = parse_virtual_wall_blob(bytes([2]) + wall_a + wall_b) + assert [w.vertices for w in walls] == [[(10, 20), (30, 40)], [(-5, 5), (-10, 10)]] + + +def test_parse_virtual_wall_blob_truncated_record_dropped() -> None: + """A trailing record shorter than 8 bytes is dropped, not misread.""" + blob = bytes([2]) + bytes([0x00, 0x0A, 0x00, 0x14, 0x00, 0x1E, 0x00, 0x28]) + b"\x00\x00" + walls = parse_virtual_wall_blob(blob) + assert [w.vertices for w in walls] == [[(10, 20), (30, 40)]] + + +def test_zone_parser_misframes_virtual_wall_blob() -> None: + """The restricted-zone parser must NOT be used on DP 57 -- it mis-frames it. + + Regression guard for the original bug: a DP-57 blob fed to parse_zone_blob + reads the leading 0x01 as a version and the next coordinate byte as a record + count, yielding garbage (here: nothing), so DP 57 needs its own decoder. + """ + assert parse_zone_blob("Aflu+cX87PoO") != parse_virtual_wall_blob("Aflu+cX87PoO") diff --git a/tests/map/testdata/b01_q10_trace_multi.bin b/tests/map/testdata/b01_q10_trace_multi.bin index 8377e6c0d2b46c02bf0934e59c4c997027d424b3..c388ab0ceb45e322910fbf311834eb5883093157 100644 GIT binary patch literal 26 dcmZQ#WME|g0cHkWAeq8&f?*oNum9&5bpRfe1n&R< literal 22 bcmZQ#WME}rVgP{@h7%0a7=Haf$EX7U8@vR; diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index e3617709..f5d916d1 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -54,7 +54,10 @@ def test_decode_message_trace_packet() -> None: message = _message(TRACE_FIXTURE.read_bytes(), RoborockMessageProtocol.MAP_RESPONSE) decoded = decode_message(message) assert isinstance(decoded, Q10TracePacket) - assert [(p.x, p.y) for p in decoded.points] == [(169, 0)] + # This docked capture carries only a heading (no path points); see the + # parser tests for the full byte-level decode. + assert decoded.points == [] + assert decoded.heading == 169 def test_decode_message_unknown_map_marker_returns_none() -> None: From f119a75d1783208407207e7915360b66aa9234c7 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 19:41:00 +0400 Subject: [PATCH 2/2] fix: preserve Q10 decoded packet API --- roborock/map/b01_q10_map_parser.py | 8 ++++++++ tests/devices/traits/b01/q10/test_map.py | 8 ++++---- tests/map/test_b01_q10_map_parser.py | 10 ++++++++++ 3 files changed, 22 insertions(+), 4 deletions(-) diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index 52dfc76d..a3f7d1e7 100644 --- a/roborock/map/b01_q10_map_parser.py +++ b/roborock/map/b01_q10_map_parser.py @@ -618,6 +618,14 @@ def parse(self, payload: bytes) -> ParsedMapData: """Parse a raw Q10 map packet into a rendered PNG + ``MapData``.""" return self.parsed_from_packet(parse_map_packet(payload)) + def parse_packet(self, packet: Q10MapPacket) -> ParsedMapData: + """Render an already-decoded packet. + + This preserves the public API used by the existing map trait. Callers + rendering a modified packet can use :meth:`parsed_from_packet` directly. + """ + return self.parsed_from_packet(packet) + def parsed_from_packet(self, packet: Q10MapPacket) -> ParsedMapData: """Render a (possibly erase-modified) packet into a PNG + ``MapData``.""" image = self._render(packet) diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index fb07e585..65216496 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -22,7 +22,7 @@ from .conftest import FakeB01Q10Channel FIXTURE = Path("tests/map/testdata/b01_q10_map.bin") -TRACE_FIXTURE = Path("tests/map/testdata/b01_q10_trace.bin") +TRACE_FIXTURE = Path("tests/map/testdata/b01_q10_trace_multi.bin") def test_update_from_map_packet_populates_image_and_rooms() -> None: @@ -50,9 +50,9 @@ def test_update_from_trace_packet_populates_path_and_position() -> None: trait.update_from_trace_packet(packet) - assert [(p.x, p.y) for p in trait.path] == [(169, 0)] + assert [(p.x, p.y) for p in trait.path] == [(100, 200), (150, 250), (-50, 300)] assert trait.robot_position is not None - assert (trait.robot_position.x, trait.robot_position.y) == (169, 0) + assert (trait.robot_position.x, trait.robot_position.y) == (-50, 300) assert len(updates) == 1 @@ -99,7 +99,7 @@ async def test_await_q10_map_push_returns_true_after_update() -> None: ) assert got_trace is True - assert [(p.x, p.y) for p in properties.map.path] == [(169, 0)] + assert [(p.x, p.y) for p in properties.map.path] == [(100, 200), (150, 250), (-50, 300)] async def test_await_q10_map_push_can_fall_back_to_cached_map_on_timeout() -> None: diff --git a/tests/map/test_b01_q10_map_parser.py b/tests/map/test_b01_q10_map_parser.py index 2ceb5e72..43095c57 100644 --- a/tests/map/test_b01_q10_map_parser.py +++ b/tests/map/test_b01_q10_map_parser.py @@ -179,6 +179,16 @@ def test_parser_renders_png_and_room_names() -> None: assert parsed.map_data.additional_parameters["room_names"] == {2: "Living Room", 3: "Bedroom"} +def test_parse_packet_preserves_decoded_packet_api() -> None: + """The existing map trait can render a packet without re-decoding bytes.""" + packet = parse_map_packet(_payload()) + + parsed = B01Q10MapParser().parse_packet(packet) + + assert parsed.image_content is not None + assert parsed.map_data is not None + + def test_parse_rejects_non_map_packet() -> None: with pytest.raises(RoborockException, match="not a Q10 map packet"): parse_map_packet(b"\x02\x01" + b"\x00" * 40)