feat: add a clean summary trait

roborock/cli.py

@@ -388,16 +388,32 @@ async def status(ctx, device_id: str):
     device_manager = await context.get_device_manager()
     device = await device_manager.get_device(device_id)
 
-    click.echo(f"Getting status for device {device_id}")
     if not (status_trait := device.traits.get("status")):
         click.echo(f"Device {device.name} does not have a status trait")
         return
 
     status_result = await status_trait.get_status()
-    click.echo(f"Device {device_id} status:")
     click.echo(dump_json(status_result.as_dict()))
 
 
+@session.command()
+@click.option("--device_id", required=True)
+@click.pass_context
+@async_command
+async def clean_summary(ctx, device_id: str):
+    """Get device clean summary."""
+    context: RoborockContext = ctx.obj
+
+    device_manager = await context.get_device_manager()
+    device = await device_manager.get_device(device_id)
+    if not (clean_summary_trait := device.traits.get("clean_summary")):
+        click.echo(f"Device {device.name} does not have a clean summary trait")
+        return
+
+    clean_summary_result = await clean_summary_trait.get_clean_summary()
+    click.echo(dump_json(clean_summary_result.as_dict()))
+
+
 @click.command()
 @click.option("--device_id", required=True)
 @click.option("--cmd", required=True)
@@ -636,6 +652,7 @@ def write_markdown_table(product_features: dict[str, dict[str, any]], all_featur
 cli.add_command(session)
 cli.add_command(get_device_info)
 cli.add_command(update_docs)
+cli.add_command(clean_summary)
 
 
 def main():

roborock/devices/device_manager.py

@@ -22,6 +22,7 @@
 from .channel import Channel
 from .mqtt_channel import create_mqtt_channel
 from .traits.b01.props import B01PropsApi
+from .traits.clean_summary import CleanSummaryTrait
 from .traits.dnd import DoNotDisturbTrait
 from .traits.dyad import DyadApi
 from .traits.status import StatusTrait
@@ -154,6 +155,7 @@ def device_creator(device: HomeDataDevice, product: HomeDataProduct) -> Roborock
                 channel = create_v1_channel(user_data, mqtt_params, mqtt_session, device, cache)
                 traits.append(StatusTrait(product, channel.rpc_channel))
                 traits.append(DoNotDisturbTrait(channel.rpc_channel))
+                traits.append(CleanSummaryTrait(channel.rpc_channel))
             case DeviceVersion.A01:
                 mqtt_channel = create_mqtt_channel(user_data, mqtt_params, mqtt_session, device)
                 match product.category:

roborock/devices/traits/clean_summary.py

@@ -0,0 +1,52 @@
+"""Module for Roborock V1 devices.
+
+This interface is experimental and subject to breaking changes without notice
+until the API is stable.
+"""
+
+import logging
+
+from roborock.containers import (
+    CleanSummary,
+)
+from roborock.devices.v1_rpc_channel import V1RpcChannel
+from roborock.roborock_typing import RoborockCommand
+from roborock.util import unpack_list
+
+from .trait import Trait
+
+_LOGGER = logging.getLogger(__name__)
+
+__all__ = [
+    "CleanSummaryTrait",
+]
+
+
+class CleanSummaryTrait(Trait):
+    """Trait for managing the clean summary of Roborock devices."""
+
+    name = "clean_summary"
+
+    def __init__(self, rpc_channel: V1RpcChannel) -> None:
+        """Initialize the CleanSummaryTrait."""
+        self._rpc_channel = rpc_channel
+
+    async def get_clean_summary(self) -> CleanSummary:
+        """Get the current clean summary of the device.
+
+        This is a placeholder command and will likely be changed/moved in the future.
+        """
+        clean_summary = await self._rpc_channel.send_command(RoborockCommand.GET_CLEAN_SUMMARY)
+        if isinstance(clean_summary, dict):
+            return CleanSummary.from_dict(clean_summary)
+        elif isinstance(clean_summary, list):
+            clean_time, clean_area, clean_count, records = unpack_list(clean_summary, 4)
+            return CleanSummary(
+                clean_time=clean_time,
+                clean_area=clean_area,
+                clean_count=clean_count,
+                records=records,
+            )
+        elif isinstance(clean_summary, int):
+            return CleanSummary(clean_time=clean_summary)
+        raise ValueError(f"Unexpected clean summary format: {clean_summary!r}")

roborock/devices/traits/status.py

@@ -5,6 +5,7 @@
 """
 
 import logging
+from typing import Any
 
 from roborock.containers import (
     HomeDataProduct,
@@ -40,4 +41,9 @@ async def get_status(self) -> Status:
         This is a placeholder command and will likely be changed/moved in the future.
         """
         status_type: type[Status] = ModelStatus.get(self._product_info.model, S7MaxVStatus)
-        return await self._rpc_channel.send_command(RoborockCommand.GET_STATUS, response_type=status_type)
+        status: dict[str, Any] | list = await self._rpc_channel.send_command(RoborockCommand.GET_STATUS)
+        if isinstance(status, list):
+            status = status[0]
+        if not isinstance(status, dict):
+            raise ValueError(f"Unexpected status format: {status!r}")
+        return status_type.from_dict(status)

roborock/devices/v1_rpc_channel.py

@@ -17,6 +17,7 @@
     CommandType,
     ParamsType,
     RequestMessage,
+    ResponseData,
     SecurityData,
     decode_rpc_response,
 )
@@ -130,15 +131,15 @@ async def _send_raw_command(
         method: CommandType,
         *,
         params: ParamsType = None,
-    ) -> Any:
+    ) -> ResponseData:
         """Send a command and return a parsed response RoborockBase type."""
         request_message = RequestMessage(method, params=params)
         _LOGGER.debug(
             "Sending command (%s, request_id=%s): %s, params=%s", self._name, request_message.request_id, method, params
         )
         message = self._payload_encoder(request_message)
 
-        future: asyncio.Future[dict[str, Any]] = asyncio.Future()
+        future: asyncio.Future[ResponseData] = asyncio.Future()
 
         def find_response(response_message: RoborockMessage) -> None:
             try:

roborock/protocols/v1_protocol.py

@@ -95,15 +95,18 @@ def _as_payload(self, security_data: SecurityData | None) -> bytes:
         )
 
 
+ResponseData = dict[str, Any] | list | int
+
+
 @dataclass(kw_only=True, frozen=True)
 class ResponseMessage:
     """Data structure for v1 RoborockMessage responses."""
 
     request_id: int | None
     """The request ID of the response."""
 
-    data: dict[str, Any]
-    """The data of the response."""
+    data: ResponseData
+    """The data of the response, where the type depends on the command."""
 
 
 def decode_rpc_response(message: RoborockMessage) -> ResponseMessage:
@@ -139,12 +142,12 @@ def decode_rpc_response(message: RoborockMessage) -> ResponseMessage:
     if not (result := data_point_response.get("result")):
         raise RoborockException(f"Invalid V1 message format: missing 'result' in data point for {message.payload!r}")
     _LOGGER.debug("Decoded V1 message result: %s", result)
-    if isinstance(result, list) and result:
-        result = result[0]
     if isinstance(result, str) and result == "ok":
         result = {}
-    if not isinstance(result, dict):
-        raise RoborockException(f"Invalid V1 message format: 'result' should be a dictionary for {message.payload!r}")
+    if not isinstance(result, (dict, list, int)):
+        raise RoborockException(
+            f"Invalid V1 message format: 'result' was unexpected type {type(result)}. {message.payload!r}"
+        )
     return ResponseMessage(request_id=request_id, data=result)
 
 

tests/devices/test_v1_device.py

@@ -73,7 +73,7 @@ async def test_device_connection(device: RoborockDevice, channel: AsyncMock) ->
 async def test_device_get_status_command(device: RoborockDevice, rpc_channel: AsyncMock) -> None:
     """Test the device get_status command."""
     # Mock response for get_status command
-    rpc_channel.send_command.return_value = STATUS
+    rpc_channel.send_command.return_value = [STATUS.as_dict()]
 
     # Test get_status and verify the command was sent
     status_api = device.traits["status"]

tests/devices/traits/test_clean_summary.py

@@ -0,0 +1,125 @@
+"""Tests for the CleanSummary class."""
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from roborock.containers import CleanSummary
+from roborock.devices.traits.clean_summary import CleanSummaryTrait
+from roborock.devices.v1_rpc_channel import V1RpcChannel
+from roborock.exceptions import RoborockException
+from roborock.roborock_typing import RoborockCommand
+
+CLEAN_SUMMARY_DATA = [
+    1442559,
+    24258125000,
+    296,
+    [
+        1756848207,
+        1754930385,
+        1753203976,
+        1752183435,
+        1747427370,
+        1746204046,
+        1745601543,
+        1744387080,
+        1743528522,
+        1742489154,
+        1741022299,
+        1740433682,
+        1739902516,
+        1738875106,
+        1738864366,
+        1738620067,
+        1736873889,
+        1736197544,
+        1736121269,
+        1734458038,
+    ],
+]
+
+
+@pytest.fixture
+def mock_rpc_channel() -> AsyncMock:
+    """Create a mock RPC channel."""
+    mock_channel = AsyncMock(spec=V1RpcChannel)
+    # Ensure send_command is an AsyncMock that returns awaitable coroutines
+    mock_channel.send_command = AsyncMock()
+    return mock_channel
+
+
+@pytest.fixture
+def clean_summary_trait(mock_rpc_channel: AsyncMock) -> CleanSummaryTrait:
+    """Create a CleanSummaryTrait instance with mocked dependencies."""
+    return CleanSummaryTrait(mock_rpc_channel)
+
+
+@pytest.fixture
+def sample_clean_summary() -> CleanSummary:
+    """Create a sample CleanSummary for testing."""
+    return CleanSummary(
+        clean_area=100,
+        clean_time=3600,
+    )
+
+
+def test_trait_name(clean_summary_trait: CleanSummaryTrait) -> None:
+    """Test that the trait has the correct name."""
+    assert clean_summary_trait.name == "clean_summary"
+
+
+async def test_get_clean_summary_success(
+    clean_summary_trait: CleanSummaryTrait, mock_rpc_channel: AsyncMock, sample_clean_summary: CleanSummary
+) -> None:
+    """Test successfully getting clean summary."""
+    # Setup mock to return the sample clean summary
+    mock_rpc_channel.send_command.return_value = CLEAN_SUMMARY_DATA
+
+    # Call the method
+    result = await clean_summary_trait.get_clean_summary()
+
+    # Verify the result
+    assert result.clean_area == 24258125000
+    assert result.clean_time == 1442559
+    assert result.square_meter_clean_area == 24258.1
+    assert result.clean_count == 296
+    assert result.records
+    assert len(result.records) == 20
+    assert result.records[0] == 1756848207
+
+    # Verify the RPC call was made correctly
+    mock_rpc_channel.send_command.assert_called_once_with(RoborockCommand.GET_CLEAN_SUMMARY)
+
+
+async def test_get_clean_summary_clean_time_only(
+    clean_summary_trait: CleanSummaryTrait, mock_rpc_channel: AsyncMock, sample_clean_summary: CleanSummary
+) -> None:
+    """Test successfully getting clean summary where the response only has the clean time."""
+
+    mock_rpc_channel.send_command.return_value = [1442559]
+
+    # Call the method
+    result = await clean_summary_trait.get_clean_summary()
+
+    # Verify the result
+    assert result.clean_area is None
+    assert result.clean_time == 1442559
+    assert result.square_meter_clean_area is None
+    assert result.clean_count is None
+    assert not result.records
+
+    # Verify the RPC call was made correctly
+    mock_rpc_channel.send_command.assert_called_once_with(RoborockCommand.GET_CLEAN_SUMMARY)
+
+
+async def test_get_clean_summary_propagates_exception(
+    clean_summary_trait: CleanSummaryTrait, mock_rpc_channel: AsyncMock
+) -> None:
+    """Test that exceptions from RPC channel are propagated in get_clean_summary."""
+
+    # Setup mock to raise an exception
+    mock_rpc_channel.send_command.side_effect = RoborockException("Communication error")
+
+    # Verify the exception is propagated
+    with pytest.raises(RoborockException, match="Communication error"):
+        await clean_summary_trait.get_clean_summary()