feat: expose Q10 map archives

roborock/cli.py

@@ -697,8 +697,8 @@ async def map_data(ctx, device_id: str, include_path: bool):
 async def q10_position(ctx, device_id: str, include_path: bool):
     """Get the current Q10 robot position and live cleaning path.
 
-    The Q10 only streams its position/path while it is actively cleaning, so this
-    will report that no live trace is available for an idle/docked robot.
+    The Q10 normally streams position/path while it is actively cleaning, so an
+    idle device may report that no fresh live trace is available.
     """
     context: RoborockContext = ctx.obj
     device_manager = await context.get_device_manager()
@@ -712,7 +712,7 @@ async def q10_position(ctx, device_id: str, include_path: bool):
         lambda: bool(properties.map.path),
     )
     if not got_trace:
-        click.echo("No live trace available (the robot only reports position while cleaning).")
+        click.echo("No fresh live trace available.")
         return
     map_trait = properties.map
     position = map_trait.robot_position

roborock/data/b01_q10/b01_q10_containers.py

@@ -144,7 +144,9 @@ class Q10MapInfo(RoborockBase):
     """A saved map reported by ``dpMultiMap``.
 
     Q10 firmware represents the map identifier as a string on the wire. The
-    value is sent back unchanged in a subsequent ``{"op": "get"}`` request.
+    value is sent back unchanged in a subsequent ``{"op": "select"}`` detail
+    request. On Q10 firmware, ``select`` previews a saved map without applying
+    it as the active map.
     """
 
     id: str

roborock/devices/README.md

@@ -39,3 +39,70 @@ Use `FileCache` or your own `Cache` implementation to persist:
 - `Device Capabilities`: What features your specific model supports.
 
 This speeds up startup time and reduces load on the Roborock cloud APIs.
+
+
+## Q10 archive updates
+
+Q10 traits are push driven. `refresh_detail()` publishes a selection and returns
+when the command has been sent. It does not wait for the map response. Register
+an update listener before requesting data, and read the trait when it notifies
+you. A history listener can also fire for record-list updates; compare the
+received detail with the previous detail to distinguish archive updates.
+
+```python
+import asyncio
+
+from roborock.devices.traits.b01.q10 import Q10PropertiesApi
+
+
+async def request_archive_preview(properties: Q10PropertiesApi) -> bytes | None:
+    history = properties.clean_history
+    previous_detail = history.detail
+    received = asyncio.Event()
+
+    def archive_updated() -> None:
+        if history.detail is not previous_detail:
+            received.set()
+
+    record = history.last_record  # Populated by an earlier history list push.
+    if record is None or not record.map_len:
+        return None
+    unsubscribe = history.add_update_listener(archive_updated)
+    try:
+        # This caller chooses to wait; the trait itself only publishes.
+        async with asyncio.timeout(30):
+            await history.refresh_detail(record)
+            await received.wait()
+        return history.detail_image_content
+    finally:
+        unsubscribe()
+```
+
+For a persistent archive view, keep its listener until the view closes. The
+example is a one-shot caller that waits for its own event with a deadline; that
+waiting policy belongs to the caller. A missing response does not block later
+requests. Publication failures and cancellation propagate to the caller, and
+the publication lock is released.
+The traits do not create response-waiting tasks or futures.
+
+Clean-record detail packets contain no record identifier. `detail` represents
+the latest received archive, including delayed or unsolicited pushes; it cannot
+be reliably attributed to the record passed to `refresh_detail()`. In
+particular, the API does not expose a `detail_record` association. Avoid
+presenting a requested record's label as confirmed metadata for the response.
+Concurrent selections and selections from another client have the same
+limitation.
+
+Saved-map previews follow the same listener pattern via `properties.maps`.
+Call `await properties.maps.refresh_detail(map_id)` for a listed map, or omit
+`map_id` to use the first map in the list. `detail_map_id` comes from the received
+packet itself, so a consumer can check it before displaying a preview. A delayed
+preview can replace the latest saved-map preview; no request ID is available to
+identify which selection produced it. Archive updates never replace the live
+map or trace in `properties.map`.
+
+`properties.as_dict()` includes clean-history records and path data and
+saved-map metadata. Binary map grids and PNG bytes are excluded; access images
+through `detail_image_content`. Live map and trace updates use the usual trait
+listener API without revision counters. The CLI subscribes before requesting a
+push, then waits for an update satisfying its predicate, with a 30-second limit.

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,12 @@ 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) 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,13 @@
 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,
+    Q10CleanRecordDetail,
+    Q10MapPacket,
+    Q10MapPacketKind,
+    Q10TracePacket,
+)
 from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message
 
 from .button_light import ButtonLightTrait
@@ -92,7 +98,12 @@ 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:
         """Initialize the B01Props API."""
         self._channel = channel
         self.command = CommandTrait(channel)
@@ -106,10 +117,17 @@ def __init__(self, channel: B01Q10Channel) -> None:
         self.network_info = NetworkInfoTrait()
         self.consumable = ConsumableTrait()
         self._map_dps = MapDpsTrait()
-        self.maps = MapsTrait(self.command)
-        self.map = MapContentTrait(self._map_dps, self.maps, self.command)
+        self.maps = MapsTrait(self.command, map_parser_config=map_parser_config)
+        self.map = MapContentTrait(
+            self._map_dps,
+            self.command,
+            map_parser_config=map_parser_config,
+        )
+        self.clean_history = CleanHistoryTrait(
+            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.
         self._updatable_traits = [
             self.status,
@@ -158,7 +176,12 @@ def _handle_message(self, message: Q10Message) -> None:
         Map-list DPS responses and other DPS updates feed the read-model traits.
         """
         if isinstance(message, Q10MapPacket):
-            self.map.update_from_map_packet(message)
+            if message.kind is Q10MapPacketKind.CURRENT:
+                self.map.update_from_map_packet(message)
+            elif message.kind is Q10MapPacketKind.SAVED_MAP_DETAIL:
+                self.maps.update_from_map_packet(message)
+        elif isinstance(message, Q10CleanRecordDetail):
+            self.clean_history.update_from_detail(message)
         elif isinstance(message, Q10TracePacket):
             self.map.update_from_trace_packet(message)
         elif isinstance(message, Q10DpsUpdate):
@@ -179,6 +202,13 @@ 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 or B01Q10MapParserConfig(),
+    )

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

@@ -10,10 +10,12 @@
 a ``dpCleanRecord`` envelope into a :class:`CleanRecordPush`, and the trait applies it.
 """
 
+import asyncio
 import logging
 from dataclasses import dataclass, field
 from typing import Any
 
+from roborock.data import RoborockBase
 from roborock.data.b01_q10.b01_q10_code_mappings import (
     B01_Q10_DP,
     YXCleaningResult,
@@ -22,11 +24,22 @@
     YXStartMethod,
 )
 from roborock.data.b01_q10.b01_q10_containers import Q10CleanRecord
+from roborock.exceptions import RoborockException
+from roborock.map.b01_q10_map_parser import (
+    B01Q10MapParserConfig,
+    Q10CleanRecordDetail,
+    Q10HistoricalTracePacket,
+    Q10MapPacket,
+    Q10MapPacketKind,
+    Q10Point,
+)
+from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map
 
 from .command import CommandTrait
 from .common import UpdatableTrait
 
 __all__ = [
+    "CleanHistory",
     "CleanHistoryTrait",
     "CleanRecordConverter",
     "CleanRecordPush",
@@ -107,40 +120,95 @@ def parse_record(raw: Any | None) -> Q10CleanRecord | None:
             return None
 
 
-class CleanHistoryTrait(UpdatableTrait):
-    """Access to the Q10 clean-record history (``dpCleanRecord``, DP 52).
-
-    A read-model trait updated from the DPS stream like the others, but it overrides
-    :meth:`update_from_dps` because the payload is a structured push (a record list,
-    or a single ``op:"notify"`` record) rather than a flat data-point-to-field map.
-    """
+@dataclass
+class CleanHistory(RoborockBase):
+    """Received clean-history data, independent of request and transport state."""
 
-    def __init__(self, command: CommandTrait) -> None:
-        """Initialize the clean history trait."""
-        UpdatableTrait.__init__(self, command, _LOGGER)
-        self._converter = CleanRecordConverter()
-        self.records: list[Q10CleanRecord] = []
-        """Decoded clean records, most recent first."""
+    records: list[Q10CleanRecord] = field(default_factory=list)
+    """Decoded clean records, most recent first."""
+    detail: Q10CleanRecordDetail | None = None
+    """Latest received archive. Its wire payload contains no record identifier."""
+    detail_image_content: bytes | None = None
+    """Rendered archive PNG, available separately from diagnostic serialization."""
 
     @property
     def last_record(self) -> Q10CleanRecord | None:
         """The most recent clean record, or ``None`` if there are none."""
         return self.records[0] if self.records else None
 
+    @property
+    def detail_packet(self) -> Q10MapPacket | None:
+        """The map from the most recently received clean-record detail."""
+        return self.detail.map if self.detail else None
+
+    @property
+    def detail_trace(self) -> Q10HistoricalTracePacket | None:
+        """Historical path embedded in the latest received clean-record detail."""
+        return self.detail.trace if self.detail else None
+
+    @property
+    def detail_path(self) -> list[Q10Point]:
+        """Historical path points from the latest received archive."""
+        return list(self.detail_trace.points) if self.detail_trace else []
+
+    def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:
+        """Serialize public history data without binary map grids or PNG bytes."""
+        excluded = exclude or set()
+        data: dict[str, Any] = {}
+        if "records" not in excluded:
+            data["records"] = [record.as_dict() for record in self.records]
+        if "detail_path" not in excluded:
+            data["detailPath"] = [point.as_dict() for point in self.detail_path]
+        return data
+
+
+class CleanHistoryTrait(CleanHistory, UpdatableTrait):
+    """Request history updates and notify listeners when device pushes arrive.
+
+    ``refresh_detail`` returns after publishing; it does not wait for an archive.
+    Consumers subscribe with ``add_update_listener`` and read ``detail`` there.
+    Clean-record archives cannot be correlated with selections: delayed pushes
+    and selections by other clients are indistinguishable on the wire.
+    """
+
+    _command: CommandTrait
+
+    def __init__(self, command: CommandTrait, *, map_parser_config: B01Q10MapParserConfig) -> None:
+        CleanHistory.__init__(self)
+        UpdatableTrait.__init__(self, command, _LOGGER)
+        self._command = command
+        self._converter = CleanRecordConverter()
+        self._map_parser_config = map_parser_config
+        self._detail_lock = asyncio.Lock()
+
     async def refresh(self) -> None:
         """Request the clean-record list from the device.
 
         This sends the query and returns immediately; the records arrive
         asynchronously on the device stream and populate :attr:`records` once
         :meth:`update_from_dps` processes the ``dpCleanRecord`` push.
         """
-        if self._command is None:
-            raise ValueError("Trait is read-only; no command channel was provided")
         await self._command.send(
             B01_Q10_DP.COMMON,
             params={str(B01_Q10_DP.CLEAN_RECORD.code): {"op": "list"}},
         )
 
+    async def refresh_detail(self, record: Q10CleanRecord) -> None:
+        """Publish a selection using the complete firmware record identifier.
+
+        Returns after sending. The resulting archive is received independently
+        through the update listener API, with no guaranteed record attribution.
+        The lock serializes publication only; missing pushes do not block retries.
+        """
+        if not record.raw or not record.map_len:
+            raise RoborockException("The Q10 clean record has no saved map detail")
+        record_id = record.raw
+        async with self._detail_lock:
+            await self._command.send(
+                B01_Q10_DP.COMMON,
+                {str(B01_Q10_DP.CLEAN_RECORD.code): {"op": "select", "id": record_id}},
+            )
+
     def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
         """Apply a ``dpCleanRecord`` push (a full list reply or a single notify)."""
         envelope = decoded_dps.get(B01_Q10_DP.CLEAN_RECORD)
@@ -151,6 +219,23 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
             return
         self._apply(push)
 
+    def update_from_detail(self, detail: Q10CleanRecordDetail) -> None:
+        """Store and render a pushed clean-record detail map."""
+        if detail.map.kind is not Q10MapPacketKind.CLEAN_RECORD_DETAIL:
+            raise ValueError(f"Expected a Q10 clean-record detail packet, got {detail.map.kind.value}")
+        self.detail = detail
+        try:
+            self.detail_image_content = render_q10_map(
+                detail.map,
+                detail.trace,
+                Q10MapOverlays(),
+                config=self._map_parser_config,
+            )
+        except RoborockException:
+            _LOGGER.debug("Failed to render Q10 clean-record detail", exc_info=True)
+            self.detail_image_content = None
+        self._notify_update()
+
     def _apply(self, push: CleanRecordPush) -> None:
         """Merge or replace the records from ``push``, then sort newest-first and notify."""
         if push.replace: