feat: add clean record for Q7

roborock/data/b01_q7/b01_q7_containers.py

@@ -203,3 +203,49 @@ def wind_name(self) -> str | None:
     def work_mode_name(self) -> str | None:
         """Returns the name of the current work mode."""
         return self.work_mode.value if self.work_mode is not None else None
+
+
+@dataclass
+class CleanRecordDetail(RoborockBase):
+    """Represents a single clean record detail (from `record_list[].detail`)."""
+
+    record_start_time: int | None = None
+    method: int | None = None
+    record_use_time: int | None = None
+    clean_count: int | None = None
+    record_clean_area: int | None = None
+    record_clean_mode: int | None = None
+    record_clean_way: int | None = None
+    record_task_status: int | None = None
+    record_faultcode: int | None = None
+    record_dust_num: int | None = None
+    clean_current_map: int | None = None
+    record_map_url: str | None = None
+
+
+@dataclass
+class CleanRecordListItem(RoborockBase):
+    """Represents an entry in the clean record list returned by `service.get_record_list`."""
+
+    url: str | None = None
+    detail: str | dict | None = None
+
+
+@dataclass
+class CleanRecordList(RoborockBase):
+    """Represents the clean record list response from `service.get_record_list`."""
+
+    total_area: int | None = None
+    total_time: int | None = None
+    total_count: int | None = None
+    record_list: list[CleanRecordListItem] = field(default_factory=list)
+
+
+@dataclass
+class CleanRecordSummary(RoborockBase):
+    """Represents clean record totals for B01/Q7 devices."""
+
+    total_area: int | None = None
+    total_time: int | None = None
+    total_count: int | None = None
+    last_record_detail: CleanRecordDetail | None = None

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

@@ -17,17 +17,24 @@
 from roborock.roborock_message import RoborockB01Props
 from roborock.roborock_typing import RoborockB01Q7Methods
 
+from .clean_summary import CleanSummaryTrait
+
 __all__ = [
     "Q7PropertiesApi",
+    "CleanSummaryTrait",
 ]
 
 
 class Q7PropertiesApi(Trait):
     """API for interacting with B01 devices."""
 
+    clean_summary: CleanSummaryTrait
+    """Trait for clean records / clean summary (Q7 `service.get_record_list`)."""
+
     def __init__(self, channel: MqttChannel) -> None:
         """Initialize the B01Props API."""
         self._channel = channel
+        self.clean_summary = CleanSummaryTrait(channel)
 
     async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None:
         """Query the device for the values of the given Q7 properties."""

roborock/devices/traits/b01/q7/clean_summary.py

@@ -0,0 +1,82 @@
+"""Clean summary / clean records trait for B01 Q7 devices.
+
+For B01/Q7, the Roborock app uses `service.get_record_list` which returns totals
+and a `record_list` whose items contain a JSON string in `detail`.
+"""
+
+from __future__ import annotations
+
+import json
+
+from roborock import CleanRecordDetail, CleanRecordList, CleanRecordSummary
+from roborock.devices.rpc.b01_q7_channel import send_decoded_command
+from roborock.devices.traits import Trait
+from roborock.devices.transport.mqtt_channel import MqttChannel
+from roborock.exceptions import RoborockException
+from roborock.protocols.b01_q7_protocol import Q7RequestMessage
+from roborock.roborock_typing import RoborockB01Q7Methods
+
+__all__ = [
+    "CleanSummaryTrait",
+]
+
+
+class CleanSummaryTrait(CleanRecordSummary, Trait):
+    """B01/Q7 clean summary + clean record access (via record list service)."""
+
+    def __init__(self, channel: MqttChannel) -> None:
+        super().__init__()
+        self._channel = channel
+
+    async def refresh(self) -> None:
+        """Refresh totals and last record detail from the device."""
+        record_list = await self.get_record_list()
+
+        self.total_time = record_list.total_time
+        self.total_area = record_list.total_area
+        self.total_count = record_list.total_count
+
+        details = await self.get_clean_record_details(record_list=record_list)
+        self.last_record_detail = details[0] if details else None
+
+    async def get_record_list(self) -> CleanRecordList:
+        """Fetch the raw device clean record list (`service.get_record_list`)."""
+        result = await send_decoded_command(
+            self._channel,
+            Q7RequestMessage(dps=10000, command=RoborockB01Q7Methods.GET_RECORD_LIST, params={}),
+        )
+
+        if not isinstance(result, dict):
+            raise TypeError(f"Unexpected response type for GET_RECORD_LIST: {type(result).__name__}: {result!r}")
+        return CleanRecordList.from_dict(result)
+
+    @staticmethod
+    def _parse_record_detail(detail: dict | str | None) -> CleanRecordDetail | None:
+        if detail is None:
+            return None
+        if isinstance(detail, str):
+            try:
+                parsed = json.loads(detail)
+            except json.JSONDecodeError as ex:
+                raise RoborockException(f"Invalid B01 record detail JSON: {detail!r}") from ex
+            if not isinstance(parsed, dict):
+                raise RoborockException(f"Unexpected B01 record detail type: {type(parsed).__name__}: {parsed!r}")
+            return CleanRecordDetail.from_dict(parsed)
+        if isinstance(detail, dict):
+            return CleanRecordDetail.from_dict(detail)
+        raise TypeError(f"Unexpected B01 record detail type: {type(detail).__name__}: {detail!r}")
+
+    async def get_clean_record_details(self, *, record_list: CleanRecordList | None = None) -> list[CleanRecordDetail]:
+        """Return parsed record detail objects (newest-first)."""
+        if record_list is None:
+            record_list = await self.get_record_list()
+
+        details: list[CleanRecordDetail] = []
+        for item in record_list.record_list:
+            parsed = self._parse_record_detail(item.detail)
+            if parsed is not None:
+                details.append(parsed)
+
+        # App treats the newest record as the end of the list
+        details.reverse()
+        return details

tests/data/b01_q7/test_b01_q7_containers.py

@@ -1,11 +1,16 @@
 """Test cases for the containers module."""
 
+import json
+
 from roborock.data.b01_q7 import (
     B01Fault,
     B01Props,
+    CleanRecordDetail,
+    CleanRecordList,
     SCWindMapping,
     WorkStatusMapping,
 )
+from roborock.devices.traits.b01.q7.clean_summary import CleanSummaryTrait
 
 
 def test_b01props_deserialization():
@@ -102,3 +107,45 @@ def test_b01props_deserialization():
     assert deserialized.wind == SCWindMapping.STRONG
     assert deserialized.net_status is not None
     assert deserialized.net_status.ip == "192.168.1.102"
+
+
+def test_b01_q7_clean_record_list_parses_detail_fields():
+    payload = {
+        "total_time": 34980,
+        "total_area": 28540,
+        "total_count": 1,
+        "record_list": [
+            {
+                "url": "/userdata/record_map/1766368207_1766368283_0_clean_map.bin",
+                "detail": json.dumps(
+                    {
+                        "record_start_time": 1766368207,
+                        "method": 0,
+                        "record_use_time": 60,
+                        "clean_count": 1,
+                        "record_clean_area": 85,
+                        "record_clean_mode": 0,
+                        "record_clean_way": 0,
+                        "record_task_status": 20,
+                        "record_faultcode": 0,
+                        "record_dust_num": 0,
+                        "clean_current_map": 0,
+                        "record_map_url": "/userdata/record_map/1766368207_1766368283_0_clean_map.bin",
+                    }
+                ),
+            }
+        ],
+    }
+
+    parsed = CleanRecordList.from_dict(payload)
+    assert isinstance(parsed, CleanRecordList)
+    assert parsed.record_list[0].url == "/userdata/record_map/1766368207_1766368283_0_clean_map.bin"
+
+    detail = CleanSummaryTrait._parse_record_detail(parsed.record_list[0].detail)
+    assert isinstance(detail, CleanRecordDetail)
+    assert detail.method == 0
+    assert detail.clean_count == 1
+    assert detail.record_clean_way == 0
+    assert detail.record_faultcode == 0
+    assert detail.record_dust_num == 0
+    assert detail.clean_current_map == 0