feat: Q10 map layers, calibration, path & no-go zones (+ Q7 layer reuse)

roborock/cli.py

@@ -607,14 +607,19 @@ async def _await_q10_map_push(
     timeout: float = _Q10_MAP_PUSH_TIMEOUT,
     allow_cached_on_timeout: bool = False,
 ) -> bool:
-    """Nudge a Q10 to push its map/trace and wait for a fresh update.
+    """Nudge a Q10 to push its map/trace and wait until ``predicate`` holds.
 
     The Q10 map API is entirely push-driven: there is no synchronous get-map
     request. A ``dpRequestDps`` causes the device to publish a ``MAP_RESPONSE``,
     which the device's subscribe loop feeds into the map trait. Here we register
-    an update listener, send the request, and wait for a newly pushed update to
-    satisfy ``predicate``. Returns whether it did within ``timeout``.
+    an update listener, send the request, and wait for the pushed data to satisfy
+    ``predicate``. Returns whether it did within ``timeout``. When the predicate
+    already holds against cached content we return immediately without nudging.
+    If ``allow_cached_on_timeout`` is set, a timeout still returns ``True`` when
+    the predicate holds against the previously cached content.
     """
+    if predicate():
+        return True
     loop = asyncio.get_running_loop()
     updated: asyncio.Future[None] = loop.create_future()
 
@@ -662,6 +667,59 @@ async def map_image(ctx, device_id: str, output_file: str):
         click.echo("No map image content available.")
 
 
+@session.command()
+@click.option("--device_id", required=True)
+@click.option("--output-dir", default=None, help="If set, write one transparent PNG per layer here.")
+@click.pass_context
+@async_command
+async def q10_map_layers(ctx, device_id: str, output_dir: str | None):
+    """List the Q10 map's separable layers (background/wall/floor/per-room).
+
+    With --output-dir, also exports each layer as a transparent PNG that can be
+    stacked in a frontend (background, then floor, then walls, then each room).
+    """
+    import os
+
+    context: RoborockContext = ctx.obj
+    device_manager = await context.get_device_manager()
+    device = await device_manager.get_device(device_id)
+    if device.b01_q10_properties is None:
+        click.echo("Feature not supported by device")
+        return
+    properties = device.b01_q10_properties
+    await _await_q10_map_push(properties, lambda: properties.map.layers is not None)
+    layers = properties.map.layers
+    if layers is None:
+        click.echo("No map layers available.")
+        return
+
+    summary = {
+        "size": {"width": layers.width, "height": layers.height},
+        "class_counts": layers.class_counts,
+        "rooms": [
+            {"id": r.id, "name": r.name, "pixel_count": r.pixel_count, "bbox": list(r.bbox)} for r in layers.rooms
+        ],
+    }
+    click.echo(dump_json(summary))
+
+    if output_dir:
+        os.makedirs(output_dir, exist_ok=True)
+        exports = {
+            "background": layers.render_class("background", (210, 210, 215, 255), scale=2),
+            "floor": layers.render_class("floor", (70, 170, 95, 200), scale=2),
+            "wall": layers.render_class("wall", (20, 20, 25, 255), scale=2),
+        }
+        for name, png in exports.items():
+            with open(os.path.join(output_dir, f"layer_{name}.png"), "wb") as f:
+                f.write(png)
+        for room in layers.rooms:
+            png = layers.render_room(room.id, (90, 140, 220, 200), scale=2)
+            safe = "".join(c if c.isalnum() else "_" for c in room.name) or f"room{room.id}"
+            with open(os.path.join(output_dir, f"room_{room.id}_{safe}.png"), "wb") as f:
+                f.write(png)
+        click.echo(f"Wrote {3 + len(layers.rooms)} layer PNGs to {output_dir}")
+
+
 @session.command()
 @click.option("--device_id", required=True)
 @click.option("--include_path", is_flag=True, default=False, help="Include path data in the output.")
@@ -721,6 +779,42 @@ async def q10_position(ctx, device_id: str, include_path: bool):
     click.echo(dump_json(summary))
 
 
+@session.command()
+@click.option("--device_id", required=True)
+@click.option("--output-file", required=True, help="Path to save the map image with the path drawn.")
+@click.pass_context
+@async_command
+async def q10_map_with_path(ctx, device_id: str, output_file: str):
+    """Render the Q10 map with the current cleaning path + robot position drawn.
+
+    Needs the robot to be actively cleaning (the path/calibration come from the
+    live trace). Fetches the map and the path, solves the world<->pixel
+    calibration, and writes the annotated PNG.
+    """
+    context: RoborockContext = ctx.obj
+    device_manager = await context.get_device_manager()
+    device = await device_manager.get_device(device_id)
+    if device.b01_q10_properties is None:
+        click.echo("Feature not supported by device")
+        return
+    properties = device.b01_q10_properties
+    map_trait = properties.map
+    await _await_q10_map_push(properties, lambda: map_trait.image_content is not None)
+    got_path = await _await_q10_map_push(properties, lambda: bool(map_trait.path))
+    if not got_path:
+        click.echo("No live path available (the robot only reports its path while cleaning).")
+        return
+    try:
+        image = map_trait.render_path_on_map()
+    except RoborockException as err:
+        click.echo(f"Could not render path on map: {err}")
+        return
+    with open(output_file, "wb") as f:
+        f.write(image)
+    cal = map_trait.calibration
+    click.echo(f"Saved map with {len(map_trait.path)}-point path to {output_file} (calibration: {cal})")
+
+
 @session.command()
 @click.option("--device_id", required=True)
 @click.pass_context

roborock/data/b01_q10/b01_q10_containers.py

@@ -117,6 +117,11 @@ class Q10Status(RoborockBase):
     cleaning_progress: int | None = field(default=None, metadata={"dps": B01_Q10_DP.CLEAN_PROGRESS})
     fault: int | None = field(default=None, metadata={"dps": B01_Q10_DP.FAULT})
 
+    # Raw base64 map-overlay blobs (decoded by roborock.map.b01_q10_overlays).
+    restricted_zone_up: str | None = field(default=None, metadata={"dps": B01_Q10_DP.RESTRICTED_ZONE_UP})
+    virtual_wall_up: str | None = field(default=None, metadata={"dps": B01_Q10_DP.VIRTUAL_WALL_UP})
+    zoned_up: str | None = field(default=None, metadata={"dps": B01_Q10_DP.ZONED_UP})
+
     # Additional state reported in the device's full status dump.
     clean_line: YXCleanLine | None = field(default=None, metadata={"dps": B01_Q10_DP.CLEAN_LINE})
     carpet_clean_type: YXCarpetCleanType | None = field(default=None, metadata={"dps": B01_Q10_DP.CARPET_CLEAN_TYPE})

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

@@ -13,6 +13,7 @@
 from .button_light import ButtonLightTrait
 from .child_lock import ChildLockTrait
 from .command import CommandTrait
+from .common import DpsUpdatable
 from .consumable import ConsumableTrait
 from .do_not_disturb import DoNotDisturbTrait
 from .dust_collection import DustCollectionTrait
@@ -94,14 +95,18 @@ def __init__(self, channel: MqttChannel) -> None:
         self.consumable = ConsumableTrait()
         self.map = MapContentTrait()
         # Read-model traits updated from the device's DPS push stream.
-        self._updatable_traits = [
+        self._updatable_traits: list[DpsUpdatable] = [
             self.status,
             self.volume,
             self.child_lock,
             self.do_not_disturb,
             self.dust_collection,
             self.network_info,
             self.consumable,
+            # The map trait owns the vector-overlay data points (no-go zones /
+            # virtual walls), which arrive as status DPs rather than in the map
+            # packet, so it updates from the DPS stream like any other read-model.
+            self.map,
         ]
         self._subscribe_task: asyncio.Task[None] | None = None
 
@@ -145,7 +150,8 @@ def _handle_message(self, message: Q10Message) -> None:
         elif isinstance(message, Q10DpsUpdate):
             _LOGGER.debug("Received Q10 status update: %s", message.dps)
             # Notify all read-model traits about the new message; each trait
-            # only updates the fields that it is responsible for.
+            # only updates the fields that it is responsible for (the map trait
+            # picks out the vector-overlay data points it owns).
             for trait in self._updatable_traits:
                 trait.update_from_dps(message.dps)
 

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

@@ -11,7 +11,7 @@
 """
 
 import logging
-from typing import Any, ClassVar, cast
+from typing import Any, ClassVar, Protocol, cast
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.data.containers import RoborockBase
@@ -20,6 +20,18 @@
 from .command import CommandTrait
 
 
+class DpsUpdatable(Protocol):
+    """A trait that updates itself from the Q10 DPS push stream.
+
+    Implemented by :class:`UpdatableTrait` (read-model traits) and by the map
+    trait, which owns the vector-overlay data points. The ``Q10PropertiesApi``
+    subscribe loop fans each push out to every such trait; each picks out the
+    data points it is responsible for and ignores the rest.
+    """
+
+    def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: ...
+
+
 class UpdatableTrait(TraitUpdateListener):
     """Base for Q10 traits backed by a read-model updated from the DPS stream.