add scenes support

Author: regevbr

roborock/cli.py

@@ -113,6 +113,38 @@ async def list_devices(ctx):
     click.echo(f"Known devices {device_name_id}")
 
 
+@click.command()
+@click.option("--device_id", required=True)
+@click.pass_context
+@run_sync()
+async def list_scenes(ctx, device_id):
+    context: RoborockContext = ctx.obj
+    login_data = context.login_data()
+    if not login_data.home_data:
+        await _discover(ctx)
+        login_data = context.login_data()
+    client = RoborockApiClient(login_data.email)
+    scenes = await client.get_scenes(login_data.user_data, device_id)
+    output_list = []
+    for scene in scenes:
+        output_list.append(scene.as_dict())
+    click.echo(json.dumps(output_list, indent = 4))
+
+
+@click.command()
+@click.option("--scene_id", required=True)
+@click.pass_context
+@run_sync()
+async def execute_scene(ctx, scene_id):
+    context: RoborockContext = ctx.obj
+    login_data = context.login_data()
+    if not login_data.home_data:
+        await _discover(ctx)
+        login_data = context.login_data()
+    client = RoborockApiClient(login_data.email)
+    await client.execute_scene(login_data.user_data, scene_id)
+
+
 @click.command()
 @click.option("--device_id", required=True)
 @click.option("--cmd", required=True)
@@ -192,6 +224,8 @@ def on_package(packet: Packet):
 cli.add_command(login)
 cli.add_command(discover)
 cli.add_command(list_devices)
+cli.add_command(list_scenes)
+cli.add_command(execute_scene)
 cli.add_command(command)
 cli.add_command(parser)
 

roborock/containers.py

@@ -438,6 +438,11 @@ class HomeDataRoom(RoborockBase):
     id: int
     name: str
 
+@dataclass
+class HomeDataScene(RoborockBase):
+    id: int
+    name: str
+
 
 @dataclass
 class HomeData(RoborockBase):

roborock/web_api.py

@@ -11,7 +11,7 @@
 import aiohttp
 from aiohttp import ContentTypeError
 
-from roborock.containers import HomeData, HomeDataRoom, ProductResponse, RRiot, UserData
+from roborock.containers import HomeData, HomeDataRoom, ProductResponse, RRiot, UserData, HomeDataScene
 from roborock.exceptions import (
     RoborockAccountDoesNotExist,
     RoborockException,
@@ -269,6 +269,46 @@ async def get_rooms(self, user_data: UserData, home_id: int | None = None) -> li
         else:
             raise RoborockException("home_response result was an unexpected type")
 
+    async def get_scenes(self, user_data: UserData, home_id: int) -> list[HomeDataRoom]:
+        rriot = user_data.rriot
+        if rriot is None:
+            raise RoborockException("rriot is none")
+        if rriot.r.a is None:
+            raise RoborockException("Missing field 'a' in rriot reference")
+        scenes_request = PreparedRequest(
+            rriot.r.a,
+            {
+                "Authorization": self._get_hawk_authentication(rriot, f"/user/scene/device/{str(home_id)}"),
+            },
+        )
+        scenes_response = await scenes_request.request("get", f"/user/scene/device/{str(home_id)}")
+        if not scenes_response.get("success"):
+            raise RoborockException(scenes_response)
+        scenes = scenes_response.get("result")
+        if isinstance(scenes, list):
+            output_list = []
+            for scene in scenes:
+                output_list.append(HomeDataScene.from_dict(scene))
+            return output_list
+        else:
+            raise RoborockException("home_response result was an unexpected type")
+
+    async def execute_scene(self, user_data: UserData, scene_id: int) -> list[HomeDataRoom]:
+        rriot = user_data.rriot
+        if rriot is None:
+            raise RoborockException("rriot is none")
+        if rriot.r.a is None:
+            raise RoborockException("Missing field 'a' in rriot reference")
+        execute_scene_request = PreparedRequest(
+            rriot.r.a,
+            {
+                "Authorization": self._get_hawk_authentication(rriot, f"/user/scene/{str(scene_id)}/execute"),
+            },
+        )
+        execute_scene_response = await execute_scene_request.request("post", f"/user/scene/{str(scene_id)}/execute")
+        if not execute_scene_response.get("success"):
+            raise RoborockException(execute_scene_response)
+
     async def get_products(self, user_data: UserData) -> ProductResponse:
         """Gets all products and their schemas, good for determining status codes and model numbers."""
         base_url = await self._get_base_url()