Merge branch 'main' into feat/q10-cli-commands

Author: lboue

CHANGELOG.md

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "python-roborock"
-version = "2.25.0"
+version = "4.10.0"
 description = "A package to control Roborock vacuums."
 authors = ["humbertogontijo <humbertogontijo@users.noreply.github.com>"]
 license = "GPL-3.0-only"

pyproject.toml

@@ -1,5 +1,9 @@
+import datetime
+import json
 from dataclasses import dataclass, field
+from functools import cached_property
 
+from ...exceptions import RoborockException
 from ..containers import RoborockBase
 from .b01_q7_code_mappings import (
     B01Fault,
@@ -205,3 +209,82 @@ 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
+    # This is seemingly returned in meters (non-squared)
+    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
+
+    @property
+    def start_datetime(self) -> datetime.datetime | None:
+        """Convert the start datetime into a datetime object."""
+        if self.record_start_time is not None:
+            return datetime.datetime.fromtimestamp(self.record_start_time).astimezone(datetime.UTC)
+        return None
+
+    @property
+    def square_meters_area_cleaned(self) -> float | None:
+        """Returns the area cleaned in square meters."""
+        if self.record_clean_area is not None:
+            return self.record_clean_area / 100
+        return 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 | None = None
+
+    @cached_property
+    def detail_parsed(self) -> CleanRecordDetail | None:
+        """Parse and return the detail as a CleanRecordDetail object."""
+        if self.detail is None:
+            return None
+        try:
+            parsed = json.loads(self.detail)
+        except json.JSONDecodeError as ex:
+            raise RoborockException(f"Invalid B01 record detail JSON: {self.detail!r}") from ex
+        return CleanRecordDetail.from_dict(parsed)
+
+
+@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  # stored in seconds
+    total_count: int | None = None
+    record_list: list[CleanRecordListItem] = field(default_factory=list)
+
+    @property
+    def square_meters_area_cleaned(self) -> float | None:
+        """Returns the area cleaned in square meters."""
+        if self.total_area is not None:
+            return self.total_area / 100
+        return None
+
+
+@dataclass
+class CleanRecordSummary(RoborockBase):
+    """Represents clean record totals for B01/Q7 devices."""
+
+    total_time: int | None = None
+    total_area: int | None = None
+    total_count: int | None = None
+    last_record_detail: CleanRecordDetail | None = None

roborock/data/b01_q7/b01_q7_containers.py

@@ -18,17 +18,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/__init__.py

@@ -0,0 +1,75 @@
+"""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 logging
+
+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",
+]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class CleanSummaryTrait(CleanRecordSummary, Trait):
+    """B01/Q7 clean summary + clean record access (via record list service)."""
+
+    def __init__(self, channel: MqttChannel) -> None:
+        """Initialize the clean summary trait.
+
+        Args:
+            channel: MQTT channel used to communicate with the device.
+        """
+        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)
+
+    async def _get_clean_record_details(self, *, record_list: CleanRecordList) -> list[CleanRecordDetail]:
+        """Return parsed record detail objects (newest-first)."""
+        details: list[CleanRecordDetail] = []
+        for item in record_list.record_list:
+            try:
+                parsed = item.detail_parsed
+            except RoborockException as ex:
+                # Rather than failing if something goes wrong here, we should fail and log to tell the user.
+                _LOGGER.debug("Failed to parse record detail: %s", ex)
+                continue
+            if parsed is not None:
+                details.append(parsed)
+
+        # The server returns the newest record at the end of record_list; reverse so newest is first (index 0).
+        details.reverse()
+        return details

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

@@ -1,8 +1,12 @@
 """Test cases for the containers module."""
 
+import json
+
 from roborock.data.b01_q7 import (
     B01Fault,
     B01Props,
+    CleanRecordDetail,
+    CleanRecordList,
     SCWindMapping,
     WorkStatusMapping,
 )
@@ -102,3 +106,52 @@ 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_dict = json.loads(parsed.record_list[0].detail or "{}")
+    detail = CleanRecordDetail.from_dict(detail_dict)
+    assert isinstance(detail, CleanRecordDetail)
+    assert detail.record_start_time == 1766368207
+    assert detail.record_use_time == 60
+    assert detail.record_clean_area == 85
+    assert detail.record_clean_mode == 0
+    assert detail.record_task_status == 20
+    assert detail.record_map_url == "/userdata/record_map/1766368207_1766368283_0_clean_map.bin"
+    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

tests/data/b01_q7/test_b01_q7_containers.py

@@ -0,0 +1,41 @@
+import json
+from typing import Any
+
+from Crypto.Cipher import AES
+from Crypto.Util.Padding import pad
+
+from roborock.devices.traits.b01.q7 import Q7PropertiesApi
+from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol
+from tests.fixtures.channel_fixtures import FakeChannel
+
+
+class B01MessageBuilder:
+    """Helper class to build B01 RPC response messages for tests."""
+
+    def __init__(self) -> None:
+        self.msg_id = 123456789
+        self.seq = 2020
+
+    def build(self, data: dict[str, Any] | str, code: int | None = None) -> RoborockMessage:
+        """Build an encoded B01 RPC response message."""
+        message: dict[str, Any] = {
+            "msgId": str(self.msg_id),
+            "data": data,
+        }
+        if code is not None:
+            message["code"] = code
+        return self._build_dps(message)
+
+    def _build_dps(self, message: dict[str, Any] | str) -> RoborockMessage:
+        """Build an encoded B01 RPC response message."""
+        dps_payload = {"dps": {"10000": json.dumps(message)}}
+        self.seq += 1
+        return RoborockMessage(
+            protocol=RoborockMessageProtocol.RPC_RESPONSE,
+            payload=pad(
+                json.dumps(dps_payload).encode(),
+                AES.block_size,
+            ),
+            version=b"B01",
+            seq=self.seq,
+        )

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

@@ -0,0 +1,42 @@
+import math
+import time
+from collections.abc import Generator
+from unittest.mock import patch
+
+import pytest
+
+from roborock.devices.traits.b01.q7 import Q7PropertiesApi
+from tests.fixtures.channel_fixtures import FakeChannel
+
+from . import B01MessageBuilder
+
+
+@pytest.fixture(name="fake_channel")
+def fake_channel_fixture() -> FakeChannel:
+    return FakeChannel()
+
+
+@pytest.fixture(name="q7_api")
+def q7_api_fixture(fake_channel: FakeChannel) -> Q7PropertiesApi:
+    return Q7PropertiesApi(fake_channel)  # type: ignore[arg-type]
+
+
+@pytest.fixture(name="expected_msg_id", autouse=True)
+def next_message_id_fixture() -> Generator[int, None, None]:
+    """Fixture to patch get_next_int to return the expected message ID.
+
+    We pick an arbitrary number, but just need it to ensure we can craft a fake
+    response with the message id matched to the outgoing RPC.
+    """
+    expected_msg_id = math.floor(time.time())
+
+    # Patch get_next_int to return our expected msg_id so the channel waits for it
+    with patch("roborock.protocols.b01_q7_protocol.get_next_int", return_value=expected_msg_id):
+        yield expected_msg_id
+
+
+@pytest.fixture(name="message_builder")
+def message_builder_fixture(expected_msg_id: int) -> B01MessageBuilder:
+    builder = B01MessageBuilder()
+    builder.msg_id = expected_msg_id
+    return builder