feat: Add a trait for working with routines

roborock/devices/device_manager.py

@@ -187,6 +187,7 @@ def device_creator(home_data: HomeData, device: HomeDataDevice, product: HomeDat
                     channel.rpc_channel,
                     channel.mqtt_rpc_channel,
                     channel.map_rpc_channel,
+                    web_api,
                     cache,
                     map_parser_config=map_parser_config,
                 )

roborock/devices/traits/v1/__init__.py

@@ -40,6 +40,7 @@
 from roborock.devices.traits import Trait
 from roborock.devices.v1_rpc_channel import V1RpcChannel
 from roborock.map.map_parser import MapParserConfig
+from roborock.web_api import UserWebApiClient
 
 from .child_lock import ChildLockTrait
 from .clean_summary import CleanSummaryTrait
@@ -56,6 +57,7 @@
 from .maps import MapsTrait
 from .network_info import NetworkInfoTrait
 from .rooms import RoomsTrait
+from .routines import RoutinesTrait
 from .smart_wash_params import SmartWashParamsTrait
 from .status import StatusTrait
 from .valley_electricity_timer import ValleyElectricityTimerTrait
@@ -85,6 +87,7 @@
     "WashTowelModeTrait",
     "SmartWashParamsTrait",
     "NetworkInfoTrait",
+    "RoutinesTrait",
 ]
 
 
@@ -108,6 +111,7 @@ class PropertiesApi(Trait):
     home: HomeTrait
     device_features: DeviceFeaturesTrait
     network_info: NetworkInfoTrait
+    routines: RoutinesTrait
 
     # Optional features that may not be supported on all devices
     child_lock: ChildLockTrait | None = None
@@ -126,6 +130,7 @@ def __init__(
         rpc_channel: V1RpcChannel,
         mqtt_rpc_channel: V1RpcChannel,
         map_rpc_channel: V1RpcChannel,
+        web_api: UserWebApiClient,
         cache: Cache,
         map_parser_config: MapParserConfig | None = None,
     ) -> None:
@@ -134,6 +139,7 @@ def __init__(
         self._rpc_channel = rpc_channel
         self._mqtt_rpc_channel = mqtt_rpc_channel
         self._map_rpc_channel = map_rpc_channel
+        self._web_api = web_api
         self._cache = cache
 
         self.status = StatusTrait(product)
@@ -144,6 +150,7 @@ def __init__(
         self.home = HomeTrait(self.status, self.maps, self.map_content, self.rooms, cache)
         self.device_features = DeviceFeaturesTrait(product.product_nickname, cache)
         self.network_info = NetworkInfoTrait(device_uid, cache)
+        self.routines = RoutinesTrait(device_uid, web_api)
 
         # Dynamically create any traits that need to be populated
         for item in fields(self):
@@ -267,6 +274,7 @@ def create(
     rpc_channel: V1RpcChannel,
     mqtt_rpc_channel: V1RpcChannel,
     map_rpc_channel: V1RpcChannel,
+    web_api: UserWebApiClient,
     cache: Cache,
     map_parser_config: MapParserConfig | None = None,
 ) -> PropertiesApi:
@@ -278,6 +286,7 @@ def create(
         rpc_channel,
         mqtt_rpc_channel,
         map_rpc_channel,
+        web_api,
         cache,
         map_parser_config,
     )

roborock/devices/traits/v1/routines.py

@@ -0,0 +1,26 @@
+"""Routines trait for V1 devices."""
+
+from roborock.data.containers import HomeDataScene
+from roborock.web_api import UserWebApiClient
+
+
+class RoutinesTrait:
+    """Trait for interacting with routines."""
+
+    def __init__(self, device_id: str, web_api: UserWebApiClient) -> None:
+        """Initialize the routines trait."""
+        self._device_id = device_id
+        self._web_api = web_api
+
+    async def get_routines(self) -> list[HomeDataScene]:
+        """Get available routines."""
+        return await self._web_api.get_routines(self._device_id)
+
+    async def execute_routine(self, routine_id: int) -> None:
+        """Execute a routine by its ID.
+
+        Technically, routines are per-device, but the API does not
+        require the device ID to execute them. This can execute a
+        routine for any device but it is exposed here for convenience.
+        """
+        await self._web_api.execute_routine(routine_id)

roborock/web_api.py

@@ -725,3 +725,11 @@ def __init__(self, web_api: RoborockApiClient, user_data: UserData) -> None:
     async def get_home_data(self) -> HomeData:
         """Fetch home data using the API client."""
         return await self._web_api.get_home_data_v3(self._user_data)
+
+    async def get_routines(self, device_id: str) -> list[HomeDataScene]:
+        """Fetch routines (scenes) for a specific device."""
+        return await self._web_api.get_scenes(self._user_data, device_id)
+
+    async def execute_routine(self, scene_id: int) -> None:
+        """Execute a specific routine (scene) by its ID."""
+        await self._web_api.execute_scene(self._user_data, scene_id)

tests/devices/test_v1_device.py

@@ -62,6 +62,7 @@ def device_fixture(channel: AsyncMock, rpc_channel: AsyncMock, mqtt_rpc_channel:
             rpc_channel,
             mqtt_rpc_channel,
             AsyncMock(),
+            AsyncMock(),
             NoCache(),
         ),
     )

tests/devices/traits/v1/fixtures.py

@@ -40,6 +40,12 @@ def map_rpc_channel_fixture() -> AsyncMock:
     return AsyncMock()
 
 
+@pytest.fixture(autouse=True, name="web_api_client")
+def web_api_client_fixture() -> AsyncMock:
+    """Fixture to set up the web API client for tests."""
+    return AsyncMock()
+
+
 @pytest.fixture(autouse=True, name="roborock_cache")
 def roborock_cache_fixture() -> Cache:
     """Fixture to provide a NoCache instance for tests."""
@@ -52,6 +58,7 @@ def device_fixture(
     mock_rpc_channel: AsyncMock,
     mock_mqtt_rpc_channel: AsyncMock,
     mock_map_rpc_channel: AsyncMock,
+    web_api_client: AsyncMock,
     roborock_cache: Cache,
 ) -> RoborockDevice:
     """Fixture to set up the device for tests."""
@@ -66,6 +73,7 @@ def device_fixture(
             mock_rpc_channel,
             mock_mqtt_rpc_channel,
             mock_map_rpc_channel,
+            web_api_client,
             roborock_cache,
         ),
     )

tests/devices/traits/v1/test_routines.py

@@ -0,0 +1,31 @@
+"""Tests for the RoutinesTrait."""
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from roborock.data.containers import HomeDataScene
+from roborock.devices.device import RoborockDevice
+from roborock.devices.traits.v1.routines import RoutinesTrait
+
+
+@pytest.fixture(name="routines_trait")
+def routines_trait_fixture(device: RoborockDevice) -> RoutinesTrait:
+    """Fixture for the routines trait."""
+    assert device.v1_properties
+    return device.v1_properties.routines
+
+
+async def test_get_routines(routines_trait: RoutinesTrait, web_api_client: AsyncMock) -> None:
+    """Test getting routines."""
+    web_api_client.get_routines.return_value = [HomeDataScene(id=1, name="test_scene")]
+    routines = await routines_trait.get_routines()
+    assert len(routines) == 1
+    assert routines[0].name == "test_scene"
+    web_api_client.get_routines.assert_called_once()
+
+
+async def test_execute_routine(routines_trait: RoutinesTrait, web_api_client: AsyncMock) -> None:
+    """Test executing a routine."""
+    await routines_trait.execute_routine(1)
+    web_api_client.execute_routine.assert_called_once_with(1)