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

roborock/data/b01_q10/b01_q10_containers.py

@@ -6,7 +6,10 @@
 automatically update objects from raw device responses.
 """
 
+import datetime
 from dataclasses import dataclass, field
+from enum import IntEnum
+from typing import TypeVar
 
 from ..containers import RoborockBase
 from .b01_q10_code_mappings import (
@@ -32,6 +35,146 @@ class dpCleanRecord(RoborockBase):
     data: list
 
 
+class CleanScope(IntEnum):
+    """The clean scope/type (clean record field 7), ground-truthed vs the app's History labels.
+
+    Value ``2`` is intentionally unmapped: it was not observed on ss07 across a large
+    record corpus, so it surfaces as the raw ``clean_mode`` int (``scope`` returns ``None``).
+    """
+
+    FULL = 0
+    SELECTIVE_ROOM = 1
+    ZONE = 3
+    SPOT = 4
+
+
+class CleanWorkMode(IntEnum):
+    """The actual work performed (clean record field 8) -- distinct from the clean-mode *setting*."""
+
+    VAC_AND_MOP = 1
+    VACUUM = 2
+    MOP = 3
+
+
+class CleaningResult(IntEnum):
+    """How a clean ended (clean record field 9)."""
+
+    INTERRUPTED = 0
+    COMPLETED = 1
+    STOPPED = 2
+
+
+class StartMethod(IntEnum):
+    """What initiated a clean (clean record field 10)."""
+
+    REMOTE = 0
+    APP = 1
+    TIMER = 2
+    BUTTON = 3
+
+
+_E = TypeVar("_E", bound=IntEnum)
+
+
+def _safe_clean_enum(enum_cls: type[_E], value: int | None) -> _E | None:
+    """Return the enum member for ``value``, or ``None`` if unset/unmapped (never raises)."""
+    if value is None:
+        return None
+    try:
+        return enum_cls(value)
+    except ValueError:
+        return None
+
+
+@dataclass
+class Q10CleanRecord(RoborockBase):
+    """A single Q10 (ss07) clean record decoded from a ``dpCleanRecord`` (DP 52) entry.
+
+    The device returns each record as an underscore-delimited string in the ``data``
+    list of a ``{"op": "list"}`` query. The field names and meanings here follow the
+    device's own app, which splits the string into these 12 values. The ``*_len``
+    values are internal length metrics whose units aren't confirmed; the ``raw``
+    string is always retained.
+    """
+
+    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: int | None = None
+    """Clean scope/type code (see :attr:`scope`): 0 full, 1 selective-room, 3 zone, 4 spot."""
+    work_mode: int | None = None
+    """Actual work performed (see :attr:`work`): 1 vac+mop, 2 vacuum, 3 mop."""
+    cleaning_result: int | None = None
+    """0 = interrupted (ended on a fault), 1 = completed, 2 = stopped before completing (no fault)."""
+    start_method: int | None = None
+    """0 = remote control, 1 = app, 2 = schedule/timer, 3 = device button."""
+    collect_dust_count: int | None = None
+    """Number of dock auto-empties during the clean."""
+
+    @classmethod
+    def from_record_string(cls, raw: str) -> "Q10CleanRecord | None":
+        """Decode one underscore-delimited record string, or ``None`` if malformed."""
+        parts = raw.split("_")
+        if len(parts) != 12:
+            return None
+        try:
+            return cls(
+                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=int(parts[7]),
+                work_mode=int(parts[8]),
+                cleaning_result=int(parts[9]),
+                start_method=int(parts[10]),
+                collect_dust_count=int(parts[11]),
+            )
+        except ValueError:
+            return None
+
+    @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
+
+    @property
+    def scope(self) -> CleanScope | None:
+        """The clean scope/type, or ``None`` if the raw ``clean_mode`` is unset/unmapped."""
+        return _safe_clean_enum(CleanScope, self.clean_mode)
+
+    @property
+    def work(self) -> CleanWorkMode | None:
+        """The actual work performed, or ``None`` if the raw ``work_mode`` is unset/unmapped."""
+        return _safe_clean_enum(CleanWorkMode, self.work_mode)
+
+    @property
+    def result(self) -> CleaningResult | None:
+        """How the clean ended, or ``None`` if the raw ``cleaning_result`` is unset/unmapped."""
+        return _safe_clean_enum(CleaningResult, self.cleaning_result)
+
+    @property
+    def started_by(self) -> StartMethod | None:
+        """What initiated the clean, or ``None`` if the raw ``start_method`` is unset/unmapped."""
+        return _safe_clean_enum(StartMethod, self.start_method)
+
+
 @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,104 @@
+"""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.
+"""
+
+import logging
+from typing import Any
+
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
+from roborock.data.b01_q10.b01_q10_containers import Q10CleanRecord
+
+from .command import CommandTrait
+from .common import UpdatableTrait
+
+__all__ = [
+    "CleanHistoryTrait",
+]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+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 record list is a structured
+    push (a list of underscore-delimited records / a single ``op:"notify"``
+    record) rather than a flat data-point-to-field mapping, so it has no
+    ``_CONVERTER``.
+    """
+
+    def __init__(self, command: CommandTrait) -> None:
+        """Initialize the clean history trait."""
+        UpdatableTrait.__init__(self, command, _LOGGER)
+        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:
+        """Decode a ``dpCleanRecord`` push into :attr:`records`.
+
+        Handles both the full-list reply (``{"op": "list", "data": [...]}``, sent in
+        response to :meth:`refresh`) and the single clean-finished push the device
+        emits when a clean ends (``{"op": "notify", "id": "<record>"}``), which upserts
+        that one record. Other / malformed pushes are ignored.
+        """
+        envelope = decoded_dps.get(B01_Q10_DP.CLEAN_RECORD)
+        if not isinstance(envelope, dict):
+            return
+        if envelope.get("op") == "notify":
+            self._apply_notify(envelope)
+            return
+        data = envelope.get("data")
+        if not isinstance(data, list):
+            return
+        records = [
+            record
+            for item in data
+            if isinstance(item, str) and (record := Q10CleanRecord.from_record_string(item)) is not None
+        ]
+        records.sort(key=lambda record: record.start_time or 0, reverse=True)
+        self.records = records
+        self._notify_update()
+
+    def _apply_notify(self, envelope: dict[str, Any]) -> None:
+        """Upsert the single record from an ``op:"notify"`` clean-finished push.
+
+        The just-finished record arrives in the ``id`` field (not a ``data`` list).
+        It replaces any existing record with the same ``record_id`` and the list is
+        kept newest-first. A malformed/missing record is ignored.
+        """
+        raw = envelope.get("id")
+        if not isinstance(raw, str):
+            return
+        record = Q10CleanRecord.from_record_string(raw)
+        if record is None:
+            return
+        merged = [existing for existing in self.records if existing.record_id != record.record_id]
+        merged.append(record)
+        merged.sort(key=lambda record: record.start_time or 0, reverse=True)
+        self.records = merged
+        self._notify_update()