feat: add basic schedule getting (#675)

* feat: add basic schedule getting

* chore: apply suggestions from code review

Co-authored-by: Allen Porter <allen.porter@gmail.com>

* chore: exception

---------

Co-authored-by: Allen Porter <allen.porter@gmail.com>

Author: Lash-L

roborock/data/containers.py

@@ -284,6 +284,15 @@ class HomeDataScene(RoborockBase):
     name: str
 
 
+@dataclass
+class HomeDataSchedule(RoborockBase):
+    id: int
+    cron: str
+    repeated: bool
+    enabled: bool
+    param: dict | None = None
+
+
 @dataclass
 class HomeData(RoborockBase):
     id: int

roborock/web_api.py

@@ -14,6 +14,7 @@
 from aiohttp import ContentTypeError, FormData
 from pyrate_limiter import BucketFullException, Duration, Limiter, Rate
 
+from roborock import HomeDataSchedule
 from roborock.data import HomeData, HomeDataRoom, HomeDataScene, ProductResponse, RRiot, UserData
 from roborock.exceptions import (
     RoborockAccountDoesNotExist,
@@ -607,6 +608,28 @@ async def execute_scene(self, user_data: UserData, scene_id: int) -> None:
         if not execute_scene_response.get("success"):
             raise RoborockException(execute_scene_response)
 
+    async def get_schedules(self, user_data: UserData, device_id: str) -> list[HomeDataSchedule]:
+        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")
+        schedules_request = PreparedRequest(
+            rriot.r.a,
+            self.session,
+            {
+                "Authorization": _get_hawk_authentication(rriot, f"/user/devices/{device_id}/jobs"),
+            },
+        )
+        schedules_response = await schedules_request.request("get", f"/user/devices/{str(device_id)}/jobs")
+        if not schedules_response.get("success"):
+            raise RoborockException(schedules_response)
+        schedules = schedules_response.get("result")
+        if isinstance(schedules, list):
+            return [HomeDataSchedule.from_dict(schedule) for schedule in schedules]
+        else:
+            raise RoborockException(f"schedule_response result was an unexpected type: {schedules}")
+
     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.base_url

tests/test_web_api.py

@@ -203,3 +203,52 @@ async def test_missing_country_login(mock_rest) -> None:
     assert ud is not None
     # Ensure we have no surprise REST calls.
     assert len(mock_rest.requests) == 3
+
+
+async def test_get_schedules(mock_rest) -> None:
+    """Test that we can get schedules."""
+    api = RoborockApiClient(username="test_user@gmail.com")
+    ud = await api.pass_login("password")
+
+    # Mock the response
+    mock_rest.get(
+        "https://api-us.roborock.com/user/devices/123456/jobs",
+        status=200,
+        payload={
+            "api": None,
+            "result": [
+                {
+                    "id": 3878757,
+                    "cron": "03 13 15 12 ?",
+                    "repeated": False,
+                    "enabled": True,
+                    "param": {
+                        "id": 1,
+                        "method": "server_scheduled_start",
+                        "params": [
+                            {
+                                "repeat": 1,
+                                "water_box_mode": 202,
+                                "segments": "0",
+                                "fan_power": 102,
+                                "mop_mode": 300,
+                                "clean_mop": 1,
+                                "map_index": -1,
+                                "name": "1765735413736",
+                            }
+                        ],
+                    },
+                }
+            ],
+            "status": "ok",
+            "success": True,
+        },
+    )
+
+    schedules = await api.get_schedules(ud, "123456")
+    assert len(schedules) == 1
+    schedule = schedules[0]
+    assert schedule.id == 3878757
+    assert schedule.cron == "03 13 15 12 ?"
+    assert schedule.repeated is False
+    assert schedule.enabled is True