Merge branches 'feat/q10-cli-commands' and 'feat/q10-cli-commands' of https://github.com/lboue/python-roborock into feat/q10-cli-commands

# Conflicts:
#	CHANGELOG.md
#	roborock/cli.py
#	roborock/data/b01_q10/b01_q10_code_mappings.py

Author: lboue

CHANGELOG.md

@@ -15,6 +15,9 @@
 - Add Q10 vacuum CLI commands
   ([`6122647`](https://github.com/Python-roborock/python-roborock/commit/6122647a0e5e5b5c5e5e5e5e5e5e5e5e5e5e5e5e))
 
+- Add clean record for Q7 ([#745](https://github.com/Python-roborock/python-roborock/pull/745),
+  [`329e52b`](https://github.com/Python-roborock/python-roborock/commit/329e52bc34b1a5de2685b94002deae025eb0bd1c))
+
 ### Bug Fixes
 
 - Register Q10 commands in session shell

roborock/cli.py

@@ -50,6 +50,7 @@
 from roborock.devices.device import RoborockDevice
 from roborock.devices.device_manager import DeviceManager, UserParams, create_device_manager
 from roborock.devices.traits import Trait
+from roborock.devices.traits.b01.q10.vacuum import VacuumTrait
 from roborock.devices.traits.v1 import V1TraitMixin
 from roborock.devices.traits.v1.consumeable import ConsumableAttribute
 from roborock.devices.traits.v1.map_content import MapContentTrait
@@ -438,7 +439,7 @@ async def _display_v1_trait(context: RoborockContext, device_id: str, display_fu
     click.echo(dump_json(trait.as_dict()))
 
 
-async def _q10_vacuum_trait(context: RoborockContext, device_id: str):
+async def _q10_vacuum_trait(context: RoborockContext, device_id: str) -> VacuumTrait:
     """Get VacuumTrait from Q10 device."""
     device_manager = await context.get_device_manager()
     device = await device_manager.get_device(device_id)
@@ -1150,11 +1151,11 @@ def write_markdown_table(product_features: dict[str, dict[str, any]], all_featur
     click.echo("Done.")
 
 
-@click.command()
+@session.command()
 @click.option("--device_id", required=True, help="Device ID")
 @click.pass_context
 @async_command
-async def q10_vacuum_start(ctx, device_id):
+async def q10_vacuum_start(ctx: click.Context, device_id: str) -> None:
     """Start vacuum cleaning on Q10 device."""
     context: RoborockContext = ctx.obj
     try:
@@ -1167,11 +1168,11 @@ async def q10_vacuum_start(ctx, device_id):
         click.echo(f"Error: {e}")
 
 
-@click.command()
+@session.command()
 @click.option("--device_id", required=True, help="Device ID")
 @click.pass_context
 @async_command
-async def q10_vacuum_pause(ctx, device_id):
+async def q10_vacuum_pause(ctx: click.Context, device_id: str) -> None:
     """Pause vacuum cleaning on Q10 device."""
     context: RoborockContext = ctx.obj
     try:
@@ -1184,11 +1185,11 @@ async def q10_vacuum_pause(ctx, device_id):
         click.echo(f"Error: {e}")
 
 
-@click.command()
+@session.command()
 @click.option("--device_id", required=True, help="Device ID")
 @click.pass_context
 @async_command
-async def q10_vacuum_resume(ctx, device_id):
+async def q10_vacuum_resume(ctx: click.Context, device_id: str) -> None:
     """Resume vacuum cleaning on Q10 device."""
     context: RoborockContext = ctx.obj
     try:
@@ -1201,11 +1202,11 @@ async def q10_vacuum_resume(ctx, device_id):
         click.echo(f"Error: {e}")
 
 
-@click.command()
+@session.command()
 @click.option("--device_id", required=True, help="Device ID")
 @click.pass_context
 @async_command
-async def q10_vacuum_stop(ctx, device_id):
+async def q10_vacuum_stop(ctx: click.Context, device_id: str) -> None:
     """Stop vacuum cleaning on Q10 device."""
     context: RoborockContext = ctx.obj
     try:
@@ -1218,11 +1219,11 @@ async def q10_vacuum_stop(ctx, device_id):
         click.echo(f"Error: {e}")
 
 
-@click.command()
+@session.command()
 @click.option("--device_id", required=True, help="Device ID")
 @click.pass_context
 @async_command
-async def q10_vacuum_dock(ctx, device_id):
+async def q10_vacuum_dock(ctx: click.Context, device_id: str) -> None:
     """Return vacuum to dock on Q10 device."""
     context: RoborockContext = ctx.obj
     try:

roborock/data/b01_q7/b01_q7_containers.py

@@ -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/code_mappings.py

@@ -77,7 +77,8 @@ def from_code(cls, code: int) -> Self:
         raise ValueError(message)
 
     @classmethod
-    def from_code_optional(cls, code: int) -> RoborockModeEnum | None:
+    def from_code_optional(cls, code: int) -> Self | None:
+        """Gracefully return None if the code does not exist."""
         try:
             return cls.from_code(code)
         except ValueError:

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

@@ -1,6 +1,6 @@
 """Traits for Q10 B01 devices."""
 
-from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXCleanType
 
 from .command import CommandTrait
 
@@ -54,3 +54,17 @@ async def return_to_dock(self) -> None:
             command=B01_Q10_DP.START_DOCK_TASK,
             params={},
         )
+
+    async def empty_dustbin(self) -> None:
+        """Empty the dustbin at the dock."""
+        await self._command.send(
+            command=B01_Q10_DP.START_DOCK_TASK,
+            params=2,
+        )
+
+    async def set_clean_mode(self, mode: YXCleanType) -> None:
+        """Set the cleaning mode (vacuum, mop, or both)."""
+        await self._command.send(
+            command=B01_Q10_DP.CLEAN_MODE,
+            params=mode.code,
+        )

roborock/devices/traits/b01/q7/__init__.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/clean_summary.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/protocols/b01_q10_protocol.py

@@ -45,8 +45,7 @@ def _convert_datapoints(datapoints: dict[str, Any], message: RoborockMessage) ->
         except ValueError as e:
             raise ValueError(f"dps key is not a valid integer: {e} for {message.payload!r}") from e
         if (dps := B01_Q10_DP.from_code_optional(code)) is not None:
-            # Update from_code to use `Self` on newer python version to remove this type ignore
-            result[dps] = value  # type: ignore[index]
+            result[dps] = value
     return result