feat: render Q10 map obstacle markers

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,16 @@ 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, drawables=map_parser_config.drawables
+                            )
+                            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)

roborock/devices/traits/b01/q10/__init__.py

@@ -8,7 +8,7 @@
 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, Q10TracePacket
 from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message
 
 from .button_light import ButtonLightTrait
@@ -92,7 +92,7 @@ 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 = None) -> None:
         """Initialize the B01Props API."""
         self._channel = channel
         self.command = CommandTrait(channel)
@@ -107,7 +107,7 @@ def __init__(self, channel: B01Q10Channel) -> None:
         self.consumable = ConsumableTrait()
         self._map_dps = MapDpsTrait()
         self.maps = MapsTrait(self.command)
-        self.map = MapContentTrait(self._map_dps, self.maps, self.command)
+        self.map = MapContentTrait(self._map_dps, self.maps, self.command, map_parser_config=map_parser_config)
         self.vacuum = VacuumTrait(self.command, self.status, self.map)
         self.clean_history = CleanHistoryTrait(self.command)
         # Read-model traits updated from the device's DPS push stream.
@@ -179,6 +179,6 @@ 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)

roborock/map/b01_q10_map_parser.py

@@ -25,6 +25,8 @@
 import struct
 from dataclasses import dataclass, field, replace
 
+from vacuum_map_parser_base.config.drawable import Drawable
+
 from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint
 from roborock.data.code_mappings import RoborockEnum
 from roborock.data.containers import RoborockBase
@@ -209,6 +211,30 @@ def from_payload(cls, payload: bytes) -> "Q10MapPacketKind | None":
         return None if kind is cls.unknown else kind
 
 
+@dataclass
+class Q10Point(RoborockBase):
+    """A point in the Q10 firmware's dock-relative trace coordinate space."""
+
+    x: int
+    y: int
+
+    def to_roborock(self) -> Q10RoborockPoint:
+        """Convert this trace point to common Roborock coordinates."""
+        return Q10RoborockPoint.from_trace(self.x, self.y)
+
+
+@dataclass
+class Q10Obstacle(RoborockBase):
+    """A position-only obstacle in the map package's coordinate frame.
+
+    Fifty raw units equal one grid pixel. These coordinates use the map header
+    origin and carry no validated obstacle type, confidence, or photo identifier.
+    """
+
+    x: int
+    y: int
+
+
 @dataclass
 class Q10MapPacket:
     """Decoded contents of a Q10 current or archived map packet."""
@@ -228,6 +254,11 @@ class Q10MapPacket:
     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."""
 
+    obstacles: list[Q10Obstacle] = field(default_factory=list)
+    """Obstacle positions embedded after the carpet mask (50 raw units/pixel)."""
+    skip_cleaning_points: list[Q10Point] = field(default_factory=list)
+    """Skip-clean markers in their distinct firmware coordinate frame."""
+
     @property
     def layers(self) -> GridLayers:
         """Split the occupancy grid into separable grid-pixel layers."""
@@ -236,18 +267,6 @@ def layers(self) -> GridLayers:
         return decompose_grid(self.width, self.height, self.grid, rooms, classify_q10_cell, flip=False)
 
 
-@dataclass
-class Q10Point(RoborockBase):
-    """A point in the Q10 firmware's dock-relative trace coordinate space."""
-
-    x: int
-    y: int
-
-    def to_roborock(self) -> Q10RoborockPoint:
-        """Convert this trace point to common Roborock coordinates."""
-        return Q10RoborockPoint.from_trace(self.x, self.y)
-
-
 @dataclass
 class Q10TracePacket:
     """Decoded contents of a Q10 ``02 01`` cleaning-path packet.
@@ -335,12 +354,11 @@ class Q10CleanRecordDetail:
 _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_POINT_COUNT_OFFSET = 6
-_HISTORICAL_TRACE_HEADING_OFFSET = 10
-_HISTORICAL_TRACE_RESERVED_OFFSET = 12
+_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));
@@ -572,6 +590,17 @@ def _parse_map_layout(payload: bytes) -> tuple[Q10MapPacket, bytes, int | None]:
     tail = payload[layout_end:]
     erase_zones = _parse_erase_zones(tail)
     carpet_mask, carpet_end = _parse_carpet_block(tail, width, height)
+    obstacles: list[Q10Obstacle] = []
+    skip_cleaning_points: list[Q10Point] = []
+    trace_offset = None
+    if carpet_end is not None:
+        obstacle_coords, obstacle_end = _parse_counted_points(tail, carpet_end)
+        if obstacle_end is not None:
+            skip_coords, skip_end = _parse_counted_points(tail, obstacle_end)
+            if skip_end is not None:
+                obstacles = [Q10Obstacle(x, y) for x, y in obstacle_coords]
+                skip_cleaning_points = [Q10Point(x, y) for x, y in skip_coords]
+                trace_offset = skip_end
     header_calibration = _parse_header_calibration(payload)
     packet = Q10MapPacket(
         kind=kind,
@@ -583,8 +612,10 @@ def _parse_map_layout(payload: bytes) -> tuple[Q10MapPacket, bytes, int | None]:
         erase_zones=erase_zones,
         header_calibration=header_calibration,
         carpet_mask=carpet_mask,
+        obstacles=obstacles,
+        skip_cleaning_points=skip_cleaning_points,
     )
-    return packet, tail, carpet_end
+    return packet, tail, trace_offset
 
 
 def _parse_header_calibration(payload: bytes) -> Q10HeaderCalibration | None:
@@ -699,6 +730,16 @@ def _parse_carpet_block(tail: bytes, width: int, height: int) -> tuple[bytes | N
     return mask, block_end
 
 
+def _parse_counted_points(tail: bytes, offset: int) -> tuple[list[tuple[int, int]], int | None]:
+    """Decode a complete u8-counted signed-BE coordinate table atomically."""
+    if offset >= len(tail):
+        return [], None
+    end = offset + 1 + tail[offset] * 4
+    if end > len(tail):
+        return [], None
+    return list(struct.iter_unpack(">hh", memoryview(tail)[offset + 1 : end])), end
+
+
 def _parse_clean_record_trace(
     tail: bytes,
     offset: int,
@@ -707,20 +748,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
+    corresponding live trace exactly. The obstacle and skip-clean tables precede
+    the one-byte path version. 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
-    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")
+    version = tail[offset]
     reserved = int.from_bytes(
         tail[offset + _HISTORICAL_TRACE_RESERVED_OFFSET : offset + _HISTORICAL_TRACE_RESERVED_OFFSET + 2],
         "big",
@@ -766,3 +804,5 @@ class B01Q10MapParserConfig:
 
     map_scale: int = 4
     """Scale factor for the rendered map image."""
+    drawables: list[Drawable] | None = None
+    """Enabled overlays, or ``None`` for the Q10 defaults."""

roborock/map/b01_q10_render.py

@@ -23,7 +23,7 @@
 from vacuum_map_parser_base.config.drawable import Drawable
 from vacuum_map_parser_base.config.image_config import ImageConfig
 from vacuum_map_parser_base.config.size import Size, Sizes
-from vacuum_map_parser_base.map_data import Area, ImageData, MapData, Path, Point, Wall
+from vacuum_map_parser_base.map_data import Area, ImageData, MapData, Obstacle, ObstacleDetails, Path, Point, Wall
 
 from roborock.exceptions import RoborockException
 
@@ -73,6 +73,7 @@
     Drawable.CHARGER,
     Drawable.NO_GO_AREAS,
     Drawable.NO_MOPPING_AREAS,
+    Drawable.OBSTACLES,
     Drawable.PATH,
     Drawable.VACUUM_POSITION,
     Drawable.VIRTUAL_WALLS,
@@ -118,7 +119,7 @@ def render_q10_map(
     if map_data.image is None:
         raise RoborockException("Failed to render Q10 map image")
 
-    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)
@@ -137,6 +138,19 @@ def render_q10_map(
     return buffer.getvalue()
 
 
+def _place_obstacles(map_data: MapData, packet: Q10MapPacket) -> bool:
+    """Project map-package obstacles with their own header-anchored scale."""
+    header = packet.header_calibration
+    if header is None or (origin := header.origin_pixels()) is None or not packet.obstacles:
+        return False
+    calibration = GridCalibration(resolution=50.0, origin_x=origin[0], origin_y=origin[1], y_sign=1)
+    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,
@@ -332,10 +346,12 @@ 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 [drawable for drawable in config.drawables if drawable 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")

tests/devices/traits/b01/q10/test_map.py

@@ -9,11 +9,13 @@
 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
 
 import pytest
+from vacuum_map_parser_base.config.drawable import Drawable
 
 from roborock.cli import _await_q10_map_push, cli
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXDeviceState
@@ -23,6 +25,9 @@
 from roborock.devices.traits.b01.q10.maps import MapsTrait
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import (
+    B01Q10MapParserConfig,
+    Q10HeaderCalibration,
+    Q10Obstacle,
     Q10Point,
     Q10TracePacket,
     parse_map_packet,
@@ -586,3 +591,17 @@ def test_map_content_trait_as_dict_camelizes_child_keys() -> None:
     }
     assert data["path"] == [{"x": 100, "y": 200}, {"x": 150, "y": 250}]
     assert data["robotPosition"] == {"x": 25875, "y": 26125}
+
+
+async def test_create_forwards_obstacle_drawables_to_rendering(fake_channel: FakeB01Q10Channel) -> None:
+    """The public Q10 factory config changes the rendered image."""
+    enabled = create(fake_channel, map_parser_config=B01Q10MapParserConfig(drawables=[Drawable.OBSTACLES]))
+    disabled = create(fake_channel, map_parser_config=B01Q10MapParserConfig(drawables=[]))
+    packet = replace(
+        parse_map_packet(FIXTURE.read_bytes()),
+        header_calibration=Q10HeaderCalibration(0, 50, 5, -1, -1, 0),
+        obstacles=[Q10Obstacle(150, 100)],
+    )
+    enabled.map.update_from_map_packet(packet)
+    disabled.map.update_from_map_packet(packet)
+    assert enabled.map.image_content != disabled.map.image_content

tests/map/test_b01_q10_map_parser.py

@@ -1,6 +1,7 @@
 """Tests for the Roborock Q10 (B01/ss07) map parser."""
 
 import io
+import struct
 from pathlib import Path
 from typing import Any
 
@@ -11,6 +12,7 @@
 from roborock.map.b01_grid_layers import LAYER_BACKGROUND, LAYER_FLOOR, LAYER_WALL
 from roborock.map.b01_q10_map_parser import (
     Q10MapPacketKind,
+    Q10Obstacle,
     Q10Point,
     Q10Room,
     classify_q10_cell,
@@ -585,3 +587,43 @@ def test_clean_record_path_stray_point_handling(points: list[tuple[int, int]], e
 def test_clean_record_parser_rejects_other_packet_kinds(marker: bytes) -> None:
     with pytest.raises(RoborockException, match="not a Q10 clean-record"):
         parse_clean_record_detail(marker + FIXTURE.read_bytes()[2:])
+
+
+@pytest.mark.parametrize("marker", [b"\x01\x01", b"\x03\x01", b"\x04\x01"])
+def test_obstacle_tables_preserve_signed_coordinates(marker: bytes) -> None:
+    """Complete counted tables retain signed coordinates for every map kind."""
+    payload = _map_detail_payload(marker, [(10, -20)])
+    offset = len(FIXTURE.read_bytes()) + len(_carpet_tail(8, 6, bytes(48)))
+    tables = b"\x02" + struct.pack(">hhhh", 250, -300, -32768, 32767) + b"\x01" + struct.pack(">hh", -40, 50)
+    payload = payload[:offset] + tables + payload[offset + 2 :]
+    packet = parse_map_packet(payload)
+    assert packet.obstacles == [Q10Obstacle(250, -300), Q10Obstacle(-32768, 32767)]
+    assert packet.skip_cleaning_points == [Q10Point(-40, 50)]
+
+
+def test_nonempty_obstacle_tables_preserve_historical_trace_boundary() -> None:
+    """Historical points start after the complete obstacle and skip tables."""
+    payload = _map_detail_payload(b"\x03\x01", [(10, -20)])
+    offset = len(FIXTURE.read_bytes()) + len(_carpet_tail(8, 6, bytes(48)))
+    tables = b"\x01" + struct.pack(">hh", 250, -300) + b"\x01" + struct.pack(">hh", -40, 50)
+    detail = parse_clean_record_detail(payload[:offset] + tables + payload[offset + 2 :])
+    assert detail.map.obstacles == [Q10Obstacle(250, -300)]
+    assert detail.map.skip_cleaning_points == [Q10Point(-40, 50)]
+    assert detail.trace is not None
+    assert detail.trace.points == [Q10Point(10, -20)]
+
+
+@pytest.mark.parametrize("table", [b"", b"\x02\x00\x01\x00\x02", b"\x00", b"\x00\x02\x00\x01\x00\x02"])
+def test_truncated_obstacle_tables_are_ignored_atomically(table: bytes) -> None:
+    payload = FIXTURE.read_bytes() + _carpet_tail(8, 6, bytes(48)) + table
+    packet = parse_map_packet(payload)
+    assert packet.obstacles == []
+    assert packet.skip_cleaning_points == []
+    assert packet.carpet_mask == bytes(48)
+
+
+def test_maximum_counted_obstacle_table_is_bounded() -> None:
+    table = b"\xff" + struct.pack(">hh", -10, 20) * 255 + b"\x00future"
+    packet = parse_map_packet(FIXTURE.read_bytes() + _carpet_tail(8, 6, bytes(48)) + table)
+    assert packet.obstacles == [Q10Obstacle(-10, 20)] * 255
+    assert packet.skip_cleaning_points == []