feat: add v1 api support for the list of maps

Author: allenporter

roborock/cli.py

@@ -439,6 +439,16 @@ async def set_volume(ctx, device_id: str, volume: int):
     click.echo(f"Set Device {device_id} volume to {volume}")
 
 
+@session.command()
+@click.option("--device_id", required=True)
+@click.pass_context
+@async_command
+async def maps(ctx, device_id: str):
+    """Get device maps info."""
+    context: RoborockContext = ctx.obj
+    await _display_v1_trait(context, device_id, lambda v1: v1.maps)
+
+
 @click.command()
 @click.option("--device_id", required=True)
 @click.option("--cmd", required=True)
@@ -680,6 +690,7 @@ def write_markdown_table(product_features: dict[str, dict[str, any]], all_featur
 cli.add_command(clean_summary)
 cli.add_command(volume)
 cli.add_command(set_volume)
+cli.add_command(maps)
 
 
 def main():

roborock/devices/traits/v1/__init__.py

@@ -2,13 +2,14 @@
 
 from dataclasses import dataclass, field, fields
 
-from roborock.containers import HomeDataProduct
+from roborock.containers import HomeData, HomeDataProduct
 from roborock.devices.traits import Trait
 from roborock.devices.v1_rpc_channel import V1RpcChannel
 
 from .clean_summary import CleanSummaryTrait
 from .common import V1TraitMixin
 from .do_not_disturb import DoNotDisturbTrait
+from .maps import MapsTrait
 from .status import StatusTrait
 from .volume import SoundVolumeTrait
 
@@ -19,6 +20,7 @@
     "DoNotDisturbTrait",
     "CleanSummaryTrait",
     "SoundVolumeTrait",
+    "MapsTrait",
 ]
 
 
@@ -34,12 +36,14 @@ class PropertiesApi(Trait):
     dnd: DoNotDisturbTrait
     clean_summary: CleanSummaryTrait
     sound_volume: SoundVolumeTrait
+    maps: MapsTrait
 
     # In the future optional fields can be added below based on supported features
 
     def __init__(self, product: HomeDataProduct, rpc_channel: V1RpcChannel) -> None:
         """Initialize the V1TraitProps with None values."""
         self.status = StatusTrait(product)
+        self.maps = MapsTrait(self.status)
 
         # This is a hack to allow setting the rpc_channel on all traits. This is
         # used so we can preserve the dataclass behavior when the values in the

roborock/devices/traits/v1/common.py

@@ -4,7 +4,7 @@
 """
 
 from abc import ABC
-from dataclasses import asdict, dataclass, fields
+from dataclasses import dataclass, fields
 from typing import ClassVar, Self
 
 from roborock.containers import RoborockBase
@@ -77,9 +77,11 @@ async def refresh(self) -> Self:
         """Refresh the contents of this trait."""
         response = await self.rpc_channel.send_command(self.command)
         new_data = self._parse_response(response)
-        for k, v in asdict(new_data).items():
-            if v is not None:
-                setattr(self, k, v)
+        if not isinstance(new_data, RoborockBase):
+            raise ValueError(f"Internal error, unexpected response type: {new_data!r}")
+        for field in fields(new_data):
+            new_value = getattr(new_data, field.name, None)
+            setattr(self, field.name, new_value)
         return self
 
 

roborock/devices/traits/v1/maps.py

@@ -0,0 +1,70 @@
+"""Trait for managing maps and room mappings on Roborock devices.
+
+New datatypes are introduced here to manage the additional information associated
+with maps and rooms, such as map names and room names. These override the
+base container datatypes to add additional fields.
+"""
+import logging
+from typing import Self
+
+from roborock.containers import MultiMapsList, MultiMapsListMapInfo
+from roborock.devices.traits.v1 import common
+from roborock.roborock_typing import RoborockCommand
+
+from .status import StatusTrait
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class MapsTrait(MultiMapsList, common.V1TraitMixin):
+    """Trait for managing the maps of Roborock devices.
+
+    A device may have multiple maps, each identified by a unique map_flag.
+    Each map can have multiple rooms associated with it, in a `RoomMapping`.
+    """
+
+    command = RoborockCommand.GET_MULTI_MAPS_LIST
+
+    def __init__(self, status_trait: StatusTrait) -> None:
+        """Initialize the MapsTrait.
+
+        We keep track of the StatusTrait to ensure we have the latest
+        status information when dealing with maps.
+        """
+        super().__init__()
+        self._status_trait = status_trait
+
+    @property
+    def current_map(self) -> int | None:
+        """Returns the currently active map (map_flag), if available."""
+        return self._status_trait.current_map
+
+    @property
+    def current_map_info(self) -> MultiMapsListMapInfo | None:
+        """Returns the currently active map info, if available."""
+        if (current_map := self.current_map) is None or self.map_info is None:
+            return None
+        for map_info in self.map_info:
+            if map_info.map_flag == current_map:
+                return map_info
+        return None
+
+    async def set_current_map(self, map_flag: int) -> None:
+        """Update the current map of the device by it's map_flag id."""
+        await self.rpc_channel.send_command(RoborockCommand.LOAD_MULTI_MAP, params=[map_flag])
+        # Refresh our status to make sure it reflects the new map
+        await self._status_trait.refresh()
+
+    def _parse_response(self, response: common.V1ResponseData) -> Self:
+        """Parse the response from the device into a MapsTrait instance.
+
+        This overrides the base implementation to handle the specific
+        response format for the multi maps list. This is needed because we have
+        a custom constructor that requires the StatusTrait.
+        """
+        if not isinstance(response, list):
+            raise ValueError(f"Unexpected MapsTrait response format: {response!r}")
+        response = response[0]
+        if not isinstance(response, dict):
+            raise ValueError(f"Unexpected MapsTrait response format: {response!r}")
+        return MultiMapsList.from_dict(response)

tests/devices/traits/v1/test_maps.py

@@ -0,0 +1,117 @@
+"""Tests for the Maps related functionality."""
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from roborock.devices.device import RoborockDevice
+from roborock.devices.traits.v1.maps import MapsTrait
+from roborock.devices.traits.v1.status import StatusTrait
+from roborock.roborock_typing import RoborockCommand
+from tests import mock_data
+
+UPDATED_STATUS = {
+    **mock_data.STATUS,
+    "map_status": 123 * 4 + 3,  # Set current map to 123
+}
+
+
+MULTI_MAP_LIST_DATA = [
+    {
+        "max_multi_map": 1,
+        "max_bak_map": 1,
+        "multi_map_count": 1,
+        "map_info": [
+            {
+                "mapFlag": 0,
+                "add_time": 1747132930,
+                "length": 0,
+                "name": "Map 1",
+                "bak_maps": [{"mapFlag": 4, "add_time": 1747132936}],
+            },
+            {
+                "mapFlag": 123,
+                "add_time": 1747132930,
+                "length": 0,
+                "name": "Map 1",
+                "bak_maps": [{"mapFlag": 4, "add_time": 1747132936}],
+            },
+        ],
+    }
+]
+
+
+@pytest.fixture
+def status_trait(device: RoborockDevice) -> StatusTrait:
+    """Create a MapsTrait instance with mocked dependencies."""
+    assert device.v1_properties
+    return device.v1_properties.status
+
+
+@pytest.fixture
+def maps_trait(device: RoborockDevice) -> MapsTrait:
+    """Create a MapsTrait instance with mocked dependencies."""
+    assert device.v1_properties
+    return device.v1_properties.maps
+
+
+async def test_refresh_maps_trait(
+    maps_trait: MapsTrait,
+    mock_rpc_channel: AsyncMock,
+) -> None:
+    """Test successfully getting multi maps list."""
+    # Setup mock to return the sample multi maps list
+    mock_rpc_channel.send_command.return_value = MULTI_MAP_LIST_DATA
+
+    # Call the method
+    await maps_trait.refresh()
+
+    assert maps_trait.max_multi_map == 1
+    assert maps_trait.max_bak_map == 1
+    assert maps_trait.multi_map_count == 1
+    assert maps_trait.map_info
+    assert len(maps_trait.map_info) == 2
+    map_infos = maps_trait.map_info
+    assert len(map_infos) == 2
+    assert map_infos[0].map_flag == 0
+    assert map_infos[0].name == "Map 1"
+    assert map_infos[0].add_time == 1747132930
+    assert map_infos[1].map_flag == 123
+    assert map_infos[1].name == "Map 1"
+    assert map_infos[1].add_time == 1747132930
+
+    # Verify the RPC call was made correctly
+    mock_rpc_channel.send_command.assert_called_once_with(RoborockCommand.GET_MULTI_MAPS_LIST)
+
+
+async def test_set_current_map(
+    status_trait: StatusTrait,
+    maps_trait: MapsTrait,
+    mock_rpc_channel: AsyncMock,
+) -> None:
+    """Test successfully setting the current map."""
+    mock_rpc_channel.send_command.side_effect = [
+        mock_data.STATUS,  # Initial status fetch
+        MULTI_MAP_LIST_DATA,  # Response for LOAD_MULTI_MAP
+        {},  # Response for setting the current map
+        UPDATED_STATUS,  # Response for refreshing status
+    ]
+    await status_trait.refresh()
+
+    # First refresh to populate initial state
+    await maps_trait.refresh()
+
+    # Call the method to set current map
+    await maps_trait.set_current_map(123)
+
+    # Verify the current map is updated
+    assert maps_trait.current_map == 123
+
+    # Verify the RPC call was made correctly to load the map
+    mock_rpc_channel.send_command.assert_any_call(RoborockCommand.LOAD_MULTI_MAP, params=[123])
+    # Command sent are:
+    # 1. GET_STATUS to get initial status
+    # 2. GET_MULTI_MAPS_LIST to get the map list
+    # 3. LOAD_MULTI_MAP to set the map
+    # 4. GET_STATUS to refresh the current map in status
+    assert mock_rpc_channel.send_command.call_count == 4

tests/protocols/__snapshots__/test_v1_protocol.ambr

@@ -49,6 +49,34 @@
   ]
   '''
 # ---
+# name: test_decode_rpc_payload[get_multi_maps_list]
+  20001
+# ---
+# name: test_decode_rpc_payload[get_multi_maps_list].1
+  '''
+  [
+    {
+      "max_multi_map": 1,
+      "max_bak_map": 1,
+      "multi_map_count": 1,
+      "map_info": [
+        {
+          "mapFlag": 0,
+          "add_time": 1747132930,
+          "length": 0,
+          "name": "",
+          "bak_maps": [
+            {
+              "mapFlag": 4,
+              "add_time": 1747132936
+            }
+          ]
+        }
+      ]
+    }
+  ]
+  '''
+# ---
 # name: test_decode_rpc_payload[get_status]
   20001
 # ---

tests/protocols/testdata/v1_protocol/get_multi_maps_list.json

@@ -0,0 +1 @@
+{"t":1758987228,"dps":{"102":"{\"id\":20001,\"result\":[{\"max_multi_map\":1,\"max_bak_map\":1,\"multi_map_count\":1,\"map_info\":[{\"mapFlag\":0,\"add_time\":1747132930,\"length\":0,\"name\":\"\",\"bak_maps\":[{\"mapFlag\":4,\"add_time\":1747132936}]}]}]}"}}