feat: Q10 (B01/ss07) clean-record history trait

roborock/data/b01_q10/b01_q10_code_mappings.py

@@ -226,6 +226,43 @@ class YXDeviceCleanTask(RoborockModeEnum):
     PART = "part", 5
 
 
+class YXCleanScope(RoborockModeEnum):
+    """Clean scope/type as stored in a *clean record* (``dpCleanRecord``, field 7).
+
+    This is the same conceptual axis as the live :class:`YXDeviceCleanTask`, but the
+    persisted record uses a different integer encoding -- e.g. a full clean records
+    ``0`` here vs ``1`` (``smart``) live, and a select-rooms clean records ``1`` here
+    vs ``2`` (``electoral``) live. Ground-truthed against the app's History labels;
+    code ``2`` was never observed on ss07 and is intentionally unmapped (so it
+    resolves to ``None`` rather than a guessed label).
+    """
+
+    UNKNOWN = "unknown", -1
+    FULL = "full", 0
+    SELECTIVE_ROOM = "selective_room", 1
+    ZONE = "zone", 3
+    SPOT = "spot", 4
+
+
+class YXCleaningResult(RoborockModeEnum):
+    """How a clean ended, as stored in a clean record (``dpCleanRecord``, field 9)."""
+
+    UNKNOWN = "unknown", -1
+    INTERRUPTED = "interrupted", 0  # ended on a fault
+    COMPLETED = "completed", 1
+    STOPPED = "stopped", 2  # ended early without a fault
+
+
+class YXStartMethod(RoborockModeEnum):
+    """What initiated a clean, as stored in a clean record (``dpCleanRecord``, field 10)."""
+
+    UNKNOWN = "unknown", -1
+    REMOTE = "remote", 0
+    APP = "app", 1
+    TIMER = "timer", 2  # schedule / timer
+    BUTTON = "button", 3  # device button
+
+
 class YXDeviceDustCollectionFrequency(RoborockModeEnum):
     # The app exposes "regular" (code 0) vs "frequent", where "frequent" selects
     # one of the every-N-cleans intervals below.

roborock/data/b01_q10/b01_q10_containers.py

@@ -6,6 +6,7 @@
 automatically update objects from raw device responses.
 """
 
+import datetime
 from dataclasses import dataclass, field
 
 from ..containers import RoborockBase
@@ -14,12 +15,15 @@
     YXAreaUnit,
     YXBackType,
     YXCarpetCleanType,
+    YXCleaningResult,
     YXCleanLine,
+    YXCleanScope,
     YXCleanType,
     YXDeviceCleanTask,
     YXDeviceDustCollectionFrequency,
     YXDeviceState,
     YXFanLevel,
+    YXStartMethod,
     YXWaterLevel,
 )
 
@@ -32,6 +36,52 @@ class dpCleanRecord(RoborockBase):
     data: list
 
 
+@dataclass
+class Q10CleanRecord(RoborockBase):
+    """A single Q10 (ss07) clean record decoded from a ``dpCleanRecord`` (DP 52) entry.
+
+    The device returns each record as a 12-field underscore-delimited string in the
+    ``data`` list of a ``{"op": "list"}`` query (or the ``id`` of an ``{"op": "notify"}``
+    push). The ``*_len`` values are internal blob-length metrics whose units aren't
+    confirmed; the original ``raw`` string is always retained. The enum fields resolve
+    an unmapped/unset code to ``None`` rather than guessing.
+    """
+
+    raw: str
+    record_id: str | None = None
+    start_time: int | None = None
+    """Clean start time, Unix seconds."""
+    clean_time: int | None = None
+    """Cleaning time, minutes."""
+    clean_area: int | None = None
+    """Cleaned area in square meters."""
+    map_len: int | None = None
+    """Length of the saved map blob for this record (0 = none stored)."""
+    path_len: int | None = None
+    """Length of the saved path blob for this record (0 = none stored)."""
+    virtual_len: int | None = None
+    """Length of the saved virtual-restriction blob for this record (0 = none stored)."""
+    clean_mode: YXCleanScope | None = None
+    """Clean scope/type (full / selective-room / zone / spot). Same axis as the live
+    :class:`YXDeviceCleanTask` but a different record encoding -- see :class:`YXCleanScope`."""
+    work_mode: YXCleanType | None = None
+    """Actual work performed (vac+mop / vacuum / mop) -- the same enum :class:`Q10Status`
+    uses for the live clean-mode DP. Records only ever carry 1/2/3 here."""
+    cleaning_result: YXCleaningResult | None = None
+    """How the clean ended: 0 interrupted (fault), 1 completed, 2 stopped (no fault)."""
+    start_method: YXStartMethod | None = None
+    """What initiated the clean: 0 remote, 1 app, 2 timer, 3 button."""
+    collect_dust_count: int | None = None
+    """Number of dock auto-empties during the clean."""
+
+    @property
+    def start_datetime(self) -> datetime.datetime | None:
+        """The start time as a timezone-aware (UTC) datetime."""
+        if self.start_time is not None:
+            return datetime.datetime.fromtimestamp(self.start_time).astimezone(datetime.UTC)
+        return None
+
+
 @dataclass
 class dpMultiMap(RoborockBase):
     op: str

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

@@ -12,6 +12,7 @@
 
 from .button_light import ButtonLightTrait
 from .child_lock import ChildLockTrait
+from .clean_history import CleanHistoryTrait
 from .command import CommandTrait
 from .consumable import ConsumableTrait
 from .do_not_disturb import DoNotDisturbTrait
@@ -27,6 +28,7 @@
     "Q10PropertiesApi",
     "ButtonLightTrait",
     "ChildLockTrait",
+    "CleanHistoryTrait",
     "ConsumableTrait",
     "DoNotDisturbTrait",
     "DustCollectionTrait",
@@ -78,6 +80,9 @@ class Q10PropertiesApi(Trait):
     map: MapContentTrait
     """Trait for fetching the current parsed map (image + rooms)."""
 
+    clean_history: CleanHistoryTrait
+    """Trait for fetching the device clean-record history (``dpCleanRecord``)."""
+
     def __init__(self, channel: MqttChannel) -> None:
         """Initialize the B01Props API."""
         self._channel = channel
@@ -93,6 +98,7 @@ def __init__(self, channel: MqttChannel) -> None:
         self.network_info = NetworkInfoTrait()
         self.consumable = ConsumableTrait()
         self.map = MapContentTrait()
+        self.clean_history = CleanHistoryTrait(self.command)
         # Read-model traits updated from the device's DPS push stream.
         self._updatable_traits = [
             self.status,
@@ -102,6 +108,7 @@ def __init__(self, channel: MqttChannel) -> None:
             self.dust_collection,
             self.network_info,
             self.consumable,
+            self.clean_history,
         ]
         self._subscribe_task: asyncio.Task[None] | None = None
 

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

@@ -0,0 +1,164 @@
+"""Clean history trait for Q10 B01 devices.
+
+Unlike the Q7 (which exposes a synchronous ``service.get_record_list`` RPC), the
+Q10 is push-driven: :meth:`CleanHistoryTrait.refresh` sends an ``{"op": "list"}``
+query for ``dpCleanRecord`` (DP 52) over the ``dpCommon`` channel, and the device
+publishes its clean-record list back on the subscribe stream, which
+:meth:`CleanHistoryTrait.update_from_dps` then decodes.
+
+Wire parsing is separated from state management: :class:`CleanRecordConverter` turns
+a ``dpCleanRecord`` envelope into a :class:`CleanRecordPush`, and the trait applies it.
+"""
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+
+from roborock.data.b01_q10.b01_q10_code_mappings import (
+    B01_Q10_DP,
+    YXCleaningResult,
+    YXCleanScope,
+    YXCleanType,
+    YXStartMethod,
+)
+from roborock.data.b01_q10.b01_q10_containers import Q10CleanRecord
+
+from .command import CommandTrait
+from .common import UpdatableTrait
+
+__all__ = [
+    "CleanHistoryTrait",
+    "CleanRecordConverter",
+    "CleanRecordPush",
+]
+
+_LOGGER = logging.getLogger(__name__)
+
+_RECORD_FIELD_COUNT = 12
+
+
+@dataclass
+class CleanRecordPush:
+    """A parsed ``dpCleanRecord`` push: the records it carries and how to apply them.
+
+    ``replace`` is ``True`` for an ``{"op": "list"}`` reply (the full history, which
+    replaces the current state) and ``False`` for an ``{"op": "notify"}`` push (a
+    single just-finished record, which is upserted).
+    """
+
+    records: list[Q10CleanRecord] = field(default_factory=list)
+    replace: bool = False
+
+
+class CleanRecordConverter:
+    """Converts a raw ``dpCleanRecord`` (DP 52) envelope into a :class:`CleanRecordPush`.
+
+    Mirrors the converter-per-object pattern used by the other Q10 traits: parsing the
+    wire payload lives here, separate from the trait that manages the record list.
+    """
+
+    def parse(self, envelope: dict[str, Any]) -> CleanRecordPush | None:
+        """Parse a decoded ``dpCleanRecord`` envelope into a :class:`CleanRecordPush`.
+
+        Returns ``None`` for an envelope that carries nothing usable (so the trait can
+        ignore it without changing state). Malformed individual records are skipped.
+        """
+        if envelope.get("op") == "notify":
+            record = CleanRecordConverter.parse_record(envelope.get("id"))
+            return CleanRecordPush([record], replace=False) if record is not None else None
+        data = envelope.get("data")
+        if not isinstance(data, list):
+            return None
+        records = [record for item in data if (record := CleanRecordConverter.parse_record(item)) is not None]
+        return CleanRecordPush(records, replace=True)
+
+    @staticmethod
+    def parse_record(raw: Any | None) -> Q10CleanRecord | None:
+        """Decode one underscore-delimited clean-record string into a :class:`Q10CleanRecord`.
+
+        The device joins 12 values with ``_``: recordId, startTime (Unix s), cleanTime
+        (min), cleanArea (m2), mapLen, pathLen, virtualLen, cleanMode, workMode,
+        cleaningResult, startMethod, collectDustCount. Returns ``None`` for anything but
+        a well-formed 12-field string; an unmapped enum code resolves to ``None`` on its
+        field (``raw`` keeps the original).
+        """
+        if not isinstance(raw, str):
+            return None
+        parts = raw.split("_")
+        if len(parts) != _RECORD_FIELD_COUNT:
+            return None
+        try:
+            return Q10CleanRecord(
+                raw=raw,
+                record_id=parts[0],
+                start_time=int(parts[1]),
+                clean_time=int(parts[2]),
+                clean_area=int(parts[3]),
+                map_len=int(parts[4]),
+                path_len=int(parts[5]),
+                virtual_len=int(parts[6]),
+                clean_mode=YXCleanScope.from_code_optional(int(parts[7])),
+                work_mode=YXCleanType.from_code_optional(int(parts[8])),
+                cleaning_result=YXCleaningResult.from_code_optional(int(parts[9])),
+                start_method=YXStartMethod.from_code_optional(int(parts[10])),
+                collect_dust_count=int(parts[11]),
+            )
+        except ValueError:
+            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.
+    """
+
+    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."""
+
+    @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
+
+    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"}},
+        )
+
+    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)
+        if not isinstance(envelope, dict):
+            return
+        push = self._converter.parse(envelope)
+        if push is None:
+            return
+        self._apply(push)
+
+    def _apply(self, push: CleanRecordPush) -> None:
+        """Merge or replace the records from ``push``, then sort newest-first and notify."""
+        if push.replace:
+            records = list(push.records)
+        else:
+            updated_ids = {record.record_id for record in push.records}
+            records = [record for record in self.records if record.record_id not in updated_ids]
+            records.extend(push.records)
+        records.sort(key=lambda record: record.start_time or 0, reverse=True)
+        self.records = records
+        self._notify_update()