feat: expose Q10 map archives

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/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 B01Q10MapParserConfig, 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, *, map_parser_config: B01Q10MapParserConfig | None = None) -> 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,9 +118,13 @@ def __init__(self, channel: B01Q10Channel, *, map_parser_config: B01Q10MapParser
         self.consumable = ConsumableTrait()
         self._map_dps = MapDpsTrait()
         self.maps = MapsTrait(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.map = MapContentTrait(
+            self._map_dps,
+            self.command,
+            map_parser_config=map_parser_config,
+        )
         self.clean_history = CleanHistoryTrait(self.command)
+        self.vacuum = VacuumTrait(self.command, self.status, self.map)
         # Read-model traits updated from the device's DPS push stream.
         self._updatable_traits = [
             self.status,
@@ -158,7 +173,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 +199,13 @@ def as_dict(self) -> dict[str, Any]:
         return result
 
 
-def create(channel: B01Q10Channel, *, map_parser_config: B01Q10MapParserConfig | None = None) -> Q10PropertiesApi:
+def create(
+    channel: B01Q10Channel,
+    *,
+    map_parser_config: B01Q10MapParserConfig | None = None,
+) -> Q10PropertiesApi:
     """Create traits for B01 devices."""
-    return Q10PropertiesApi(channel, map_parser_config=map_parser_config)
+    return Q10PropertiesApi(
+        channel,
+        map_parser_config=map_parser_config,
+    )

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

@@ -14,6 +14,7 @@
 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 +23,20 @@
     YXStartMethod,
 )
 from roborock.data.b01_q10.b01_q10_containers import Q10CleanRecord
+from roborock.exceptions import RoborockException
+from roborock.map.b01_q10_map_parser import (
+    Q10CleanRecordDetail,
+    Q10HistoricalTracePacket,
+    Q10MapPacket,
+    Q10MapPacketKind,
+    Q10Point,
+)
 
 from .command import CommandTrait
 from .common import UpdatableTrait
 
 __all__ = [
+    "CleanHistory",
     "CleanHistoryTrait",
     "CleanRecordConverter",
     "CleanRecordPush",
@@ -107,40 +117,78 @@ 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_path: list[Q10Point] = field(default_factory=list)
+    """Historical path points from the latest received archive, included in diagnostics."""
 
     @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
+
+    def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:
+        """Serialize history metadata and path without binary map packets and grids."""
+        return super().as_dict(exclude=(exclude or set()) | {"detail"})
+
+
+class CleanHistoryTrait(CleanHistory, UpdatableTrait):
+    """Browse past cleaning runs and retrieve their parsed maps and historical paths.
+
+    Subscribe with ``add_update_listener`` to receive requested or unsolicited
+    archives. Archive pushes contain no record ID, so the latest detail cannot
+    be attributed to a particular selection.
+    """
+
+    _command: CommandTrait
+
+    def __init__(self, command: CommandTrait) -> None:
+        CleanHistory.__init__(self)
+        UpdatableTrait.__init__(self, command, _LOGGER)
+        self._command = command
+        self._converter = CleanRecordConverter()
+
     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:
+        """Request the saved map and path for a past cleaning run.
+
+        Returns after sending; listeners receive the archive separately. The
+        push contains no record ID to correlate it with this selection.
+        """
+        if not record.raw or not record.map_len:
+            raise RoborockException("The Q10 clean record has no saved map detail")
+        await self._command.send(
+            B01_Q10_DP.COMMON,
+            {str(B01_Q10_DP.CLEAN_RECORD.code): {"op": "select", "id": record.raw}},
+        )
+
     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 +199,14 @@ 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 a pushed clean-record map and path and notify listeners."""
+        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
+        self.detail_path = list(detail.trace.points) if detail.trace else []
+        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:

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

@@ -26,6 +26,7 @@
 from roborock.map.b01_q10_map_parser import (
     B01Q10MapParserConfig,
     Q10MapPacket,
+    Q10MapPacketKind,
     Q10Point,
     Q10Room,
     Q10TracePacket,
@@ -36,7 +37,6 @@
 
 from .command import CommandTrait
 from .common import UpdatableTrait
-from .maps import MapsTrait
 
 _LOGGER = logging.getLogger(__name__)
 _DOCKED_STATES = {YXDeviceState.CHARGING, YXDeviceState.EMPTYING_THE_BIN}
@@ -90,23 +90,20 @@ class MapContentTrait(TraitUpdateListener):
     """High-level composed Q10 map view.
 
     The latest map and trace packets are combined with the injected
-    :class:`MapDpsTrait` whenever a source changes. The
-    :class:`MapsTrait` supplies a stored ID only when this trait requests
-    content.
+    :class:`MapDpsTrait` whenever a source changes. Current-map acquisition is
+    independent of the saved-map list.
     """
 
     def __init__(
         self,
         map_dps: MapDpsTrait,
-        maps: MapsTrait,
         command: CommandTrait,
         *,
         map_parser_config: B01Q10MapParserConfig | None = None,
     ) -> None:
         TraitUpdateListener.__init__(self, logger=_LOGGER)
         self._config = map_parser_config or B01Q10MapParserConfig()
         self._map_dps = map_dps
-        self._maps = maps
         self._command = command
         self._map_packet: Q10MapPacket | None = None
         self._trace_packet: Q10TracePacket | None = None
@@ -214,6 +211,8 @@ def robot_heading(self) -> int | None:
 
     def update_from_map_packet(self, packet: Q10MapPacket) -> None:
         """Store a map-protocol update and render the latest sources."""
+        if packet.kind is not Q10MapPacketKind.CURRENT:
+            raise ValueError(f"Expected a current Q10 map packet, got {packet.kind.value}")
         self._map_packet = packet
         self._render()
         self._notify_update()

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

@@ -9,6 +9,7 @@
 from roborock.data.b01_q10.b01_q10_containers import Q10MapInfo, dpMultiMap
 from roborock.devices.traits.common import DpsDataConverter
 from roborock.exceptions import RoborockException
+from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10MapPacketKind
 
 from .command import CommandTrait
 from .common import UpdatableTrait
@@ -22,6 +23,15 @@ class Maps(RoborockBase):
 
     multi_map: dpMultiMap | None = field(default=None, metadata={"dps": B01_Q10_DP.MULTI_MAP})
 
+    detail_packet: Q10MapPacket | None = None
+    """Most recently received saved-map preview."""
+    detail_map_id: str | None = None
+    """Map ID supplied by the preview packet itself."""
+
+    def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:
+        """Serialize metadata without binary map packets and grids."""
+        return super().as_dict(exclude=(exclude or set()) | {"detail_packet"})
+
     @property
     def current_map_id(self) -> str | None:
         """Return the first saved-map ID for a content request, if available."""
@@ -38,7 +48,11 @@ def map_list(self) -> list[Q10MapInfo]:
 
 
 class MapsTrait(Maps, UpdatableTrait):
-    """Request and store the Q10 saved-map list."""
+    """Browse saved maps and retrieve their parsed previews independently of the live map.
+
+    Preview pushes identify their map, but carry no request correlation ID.
+    Consumers receive them through ``add_update_listener``.
+    """
 
     _CONVERTER = DpsDataConverter.from_dataclass(Maps)
     _command: CommandTrait
@@ -70,6 +84,24 @@ async def set_current_map(self, map_id: str) -> None:
             },
         )
 
+    async def refresh_detail(self, map_id: str | None = None) -> None:
+        """Request a read-only preview for one saved map.
+
+        Returns after publishing. Subscribe with ``add_update_listener`` to
+        receive previews separately from the live map. ``detail_map_id`` is
+        read from each pushed packet; it is not a request correlation ID.
+        """
+        if map_id is None:
+            map_id = self.current_map_id
+        if map_id is None:
+            raise RoborockException("Cannot request Q10 saved-map detail before the map list is available")
+        if map_id not in {map_info.id for map_info in self.map_list}:
+            raise RoborockException(f"Unknown Q10 saved-map ID: {map_id}")
+        await self._command.send(
+            B01_Q10_DP.COMMON,
+            {str(B01_Q10_DP.MULTI_MAP.code): {"op": "select", "id": map_id}},
+        )
+
     def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
         """Store a successful saved-map list response."""
         response = decoded_dps.get(B01_Q10_DP.MULTI_MAP)
@@ -78,3 +110,11 @@ def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
         if not isinstance(response, dict) or response.get("op") != "list" or response.get("result") != 1:
             return
         super().update_from_dps(decoded_dps)
+
+    def update_from_map_packet(self, packet: Q10MapPacket) -> None:
+        """Store a pushed saved-map detail packet and notify listeners."""
+        if packet.kind is not Q10MapPacketKind.SAVED_MAP_DETAIL:
+            raise ValueError(f"Expected a Q10 saved-map detail packet, got {packet.kind.value}")
+        self.detail_map_id = str(packet.map_id)
+        self.detail_packet = packet
+        self._notify_update()

tests/devices/traits/b01/q10/__snapshots__/test_status.ambr

@@ -4,6 +4,12 @@
     'child_lock': dict({
       'childLock': False,
     }),
+    'clean_history': dict({
+      'detailPath': list([
+      ]),
+      'records': list([
+      ]),
+    }),
     'consumable': dict({
       'filterLife': 0,
       'mainBrushLife': 0,
@@ -384,6 +390,12 @@
       'child_lock': dict({
         'childLock': False,
       }),
+      'clean_history': dict({
+        'detailPath': list([
+        ]),
+        'records': list([
+        ]),
+      }),
       'consumable': dict({
         'filterLife': 0,
         'mainBrushLife': 0,
@@ -473,6 +485,12 @@
   dict({
     'child_lock': dict({
     }),
+    'clean_history': dict({
+      'detailPath': list([
+      ]),
+      'records': list([
+      ]),
+    }),
     'consumable': dict({
     }),
     'do_not_disturb': dict({