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

roborock/cli.py

@@ -541,14 +541,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()
 
@@ -596,6 +601,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.")
@@ -655,6 +713,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

@@ -108,3 +108,7 @@ class Q10Status(RoborockBase):
     back_type: YXBackType | None = field(default=None, metadata={"dps": B01_Q10_DP.BACK_TYPE})
     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})

roborock/devices/rpc/b01_q10_channel.py

@@ -2,42 +2,41 @@
 
 import logging
 from collections.abc import AsyncGenerator
+from typing import Any
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.devices.transport.mqtt_channel import MqttChannel
 from roborock.exceptions import RoborockException
 from roborock.protocols.b01_q10_protocol import (
     ParamsType,
-    Q10Message,
-    decode_message,
+    decode_rpc_response,
     encode_mqtt_payload,
 )
 
 _LOGGER = logging.getLogger(__name__)
 
 
-async def stream_decoded_messages(
+async def stream_decoded_responses(
     mqtt_channel: MqttChannel,
-) -> AsyncGenerator[Q10Message, None]:
-    """Stream decoded Q10 messages received via MQTT.
+) -> AsyncGenerator[dict[B01_Q10_DP, Any], None]:
+    """Stream decoded DPS messages received via MQTT.
 
-    Each pushed ``RoborockMessage`` is decoded into a typed :data:`Q10Message`
-    (a DPS status update, a map packet, or a trace packet). Messages that fail
-    to decode or carry an unrecognized payload are skipped.
+    Messages that are not decodable DPS responses (e.g. protocol-301
+    ``MAP_RESPONSE`` map pushes) are skipped; callers that need the raw
+    messages should subscribe to :meth:`MqttChannel.subscribe_stream` directly.
     """
 
-    async for message in mqtt_channel.subscribe_stream():
+    async for response_message in mqtt_channel.subscribe_stream():
         try:
-            decoded = decode_message(message)
+            decoded_dps = decode_rpc_response(response_message)
         except RoborockException as ex:
             _LOGGER.debug(
-                "Failed to decode B01 Q10 message: %s: %s",
-                message,
+                "Failed to decode B01 Q10 RPC response: %s: %s",
+                response_message,
                 ex,
             )
             continue
-        if decoded is not None:
-            yield decoded
+        yield decoded_dps
 
 
 async def send_command(

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

@@ -4,11 +4,11 @@
 import logging
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
-from roborock.devices.rpc.b01_q10_channel import stream_decoded_messages
 from roborock.devices.traits import Trait
 from roborock.devices.transport.mqtt_channel import MqttChannel
-from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket
-from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message
+from roborock.exceptions import RoborockException
+from roborock.protocols.b01_q10_protocol import decode_rpc_response
+from roborock.roborock_message import RoborockMessage
 
 from .command import CommandTrait
 from .map import MapContentTrait
@@ -73,25 +73,40 @@ async def refresh(self) -> None:
         await self.command.send(B01_Q10_DP.REQUEST_DPS, params={})
 
     async def _subscribe_loop(self) -> None:
-        """Persistent loop dispatching decoded messages to the read-model traits."""
-        async for message in stream_decoded_messages(self._channel):
+        """Persistent loop dispatching pushed messages to the read-model traits."""
+        async for message in self._channel.subscribe_stream():
             self._handle_message(message)
 
-    def _handle_message(self, message: Q10Message) -> None:
-        """Route a single decoded message to the trait responsible for it.
+    def _handle_message(self, message: RoborockMessage) -> None:
+        """Route a single pushed message to the trait responsible for it.
 
-        Map and trace packets arrive as protocol-301 ``MAP_RESPONSE`` pushes (the
-        Q10 is entirely push-driven: there is no synchronous get-map request, a
-        ``dpRequestDps`` just nudges the device to publish its current map). DPS
-        updates feed the status trait. More traits can be dispatched here below.
+        Map/trace pushes arrive as protocol-301 ``MAP_RESPONSE`` messages (not
+        DPS), so they are handled separately from the status DPS stream. The Q10
+        is entirely push-driven: there is no synchronous get-map request, the
+        device just publishes its current map (a ``dpRequestDps`` nudges it to).
         """
-        if isinstance(message, Q10MapPacket):
-            self.map.update_from_map_packet(message)
-        elif isinstance(message, Q10TracePacket):
-            self.map.update_from_trace_packet(message)
-        elif isinstance(message, Q10DpsUpdate):
-            _LOGGER.debug("Received Q10 status update: %s", message.dps)
-            self.status.update_from_dps(message.dps)
+        if self.map.update_from_map_response(message):
+            return
+
+        try:
+            decoded_dps = decode_rpc_response(message)
+        except RoborockException as ex:
+            _LOGGER.debug("Failed to decode Q10 RPC response: %s: %s", message, ex)
+            return
+
+        _LOGGER.debug("Received Q10 status update: %s", decoded_dps)
+        # Notify all traits about a new message and each trait will
+        # only update what fields that it is responsible for.
+        # More traits can be added here below.
+        self.status.update_from_dps(decoded_dps)
+
+        # Feed the map's vector-overlay data points (no-go zones / virtual
+        # walls) to the map trait so they are decoded as they arrive.
+        if B01_Q10_DP.RESTRICTED_ZONE_UP in decoded_dps or B01_Q10_DP.VIRTUAL_WALL_UP in decoded_dps:
+            self.map.load_overlays(
+                restricted_zone_up=decoded_dps.get(B01_Q10_DP.RESTRICTED_ZONE_UP),
+                virtual_wall_up=decoded_dps.get(B01_Q10_DP.VIRTUAL_WALL_UP),
+            )
 
 
 def create(channel: MqttChannel) -> Q10PropertiesApi: