diff --git a/roborock/data/containers.py b/roborock/data/containers.py index 4926b1c5..cf013990 100644 --- a/roborock/data/containers.py +++ b/roborock/data/containers.py @@ -348,6 +348,24 @@ class HomeDataScene(RoborockBase): name: str +@dataclass +class FirmwareInfo(RoborockBase): + """Firmware/OTA info from the cloud (`ota/firmware/{duid}/updatev2`).""" + + version: str | None = None + """Latest available firmware version.""" + current_version: str | None = None + """Currently installed firmware version.""" + updatable: bool | None = None + """Whether a newer firmware is available to install.""" + desc: str | None = None + """Release notes / description.""" + release_time: str | None = None + """Release date of the available firmware as an ISO-8601 date (``YYYY-MM-DD``).""" + force_update: bool | None = None + """Whether the update is mandatory (cannot be skipped).""" + + @dataclass class HomeDataSchedule(RoborockBase): id: int diff --git a/roborock/web_api.py b/roborock/web_api.py index b13b8e06..802c1021 100644 --- a/roborock/web_api.py +++ b/roborock/web_api.py @@ -15,7 +15,7 @@ from pyrate_limiter import Duration, Limiter, Rate from roborock import HomeDataSchedule -from roborock.data import HomeData, HomeDataRoom, HomeDataScene, ProductResponse, RRiot, UserData +from roborock.data import FirmwareInfo, HomeData, HomeDataRoom, HomeDataScene, ProductResponse, RRiot, UserData from roborock.exceptions import ( RoborockAccountDoesNotExist, RoborockException, @@ -608,6 +608,73 @@ async def get_scenes(self, user_data: UserData, device_id: str) -> list[HomeData else: raise RoborockException("scene_response result was an unexpected type") + async def get_firmware_info(self, user_data: UserData, device_id: str) -> FirmwareInfo: + """Get firmware/OTA info for a device (latest version + updatable flag).""" + 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") + path = f"/ota/firmware/{device_id}/updatev2" + params = {"lang": "en"} + firmware_request = PreparedRequest( + rriot.r.a, + self.session, + { + "Authorization": _get_hawk_authentication(rriot, path, params=params), + }, + ) + firmware_response = await firmware_request.request("get", path, params=params) + if not firmware_response or not firmware_response.get("success"): + raise RoborockException(firmware_response) + return FirmwareInfo.from_dict(firmware_response.get("result") or {}) + + async def start_firmware_update(self, user_data: UserData, device_id: str) -> None: + """Trigger the (irreversible) firmware update for a device. + + The device downloads and flashes the latest firmware reported by + :meth:`get_firmware_info`. + """ + 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") + path = f"/ota/device/{device_id}/upgrade" + upgrade_request = PreparedRequest( + rriot.r.a, + self.session, + { + "Authorization": _get_hawk_authentication(rriot, path), + }, + ) + upgrade_response = await upgrade_request.request("post", path) + if not upgrade_response or not upgrade_response.get("success"): + raise RoborockException(upgrade_response) + + async def set_silent_ota(self, user_data: UserData, device_id: str, enabled: bool) -> None: + """Enable/disable automatic (silent) firmware updates for a device. + + The current value is reported as ``silent_ota_switch`` on the home-data device. + """ + 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") + path = f"/user/devices/{device_id}" + formdata = {"silentOtaSwitch": "true" if enabled else "false"} + silent_ota_request = PreparedRequest( + rriot.r.a, + self.session, + { + "Authorization": _get_hawk_authentication(rriot, path, formdata=formdata), + }, + ) + silent_ota_response = await silent_ota_request.request("put", path, data=formdata) + if not silent_ota_response or not silent_ota_response.get("success"): + raise RoborockException(silent_ota_response) + async def execute_scene(self, user_data: UserData, scene_id: int) -> None: rriot = user_data.rriot if rriot is None: diff --git a/tests/fixtures/web_api_fixtures.py b/tests/fixtures/web_api_fixtures.py index f90d5b9f..c52c829a 100644 --- a/tests/fixtures/web_api_fixtures.py +++ b/tests/fixtures/web_api_fixtures.py @@ -132,11 +132,40 @@ def mock_rest_fixture(skip_rate_limit: Any, home_data: dict[str, Any]) -> aiores status=200, payload={"api": None, "code": 200, "result": HOME_DATA_SCENES_RAW, "status": "ok", "success": True}, ) + mocked.get( + re.compile(r"https://[^/]+\.roborock\.com/ota/firmware/[^/]+/updatev2(\?.*)?$"), + status=200, + payload={ + "api": None, + "code": 200, + "result": { + "version": "1.2.3", + "currentVersion": "1.2.0", + "updatable": True, + "desc": "Bug fixes and improvements", + "releaseTime": "2026-05-01", + "forceUpdate": False, + }, + "status": "ok", + "success": True, + }, + ) mocked.post( re.compile(r"https://api-.*\.roborock\.com/user/scene/.*/execute"), status=200, payload={"api": None, "code": 200, "result": None, "status": "ok", "success": True}, ) + mocked.post( + re.compile(r"https://[^/]+\.roborock\.com/ota/device/[^/]+/upgrade(\?.*)?$"), + status=200, + payload={"api": None, "result": None, "status": "ok", "success": True}, + ) + mocked.put( + re.compile(r"https://[^/]+\.roborock\.com/user/devices/[^/]+$"), + status=200, + payload={"api": "修改设备信息", "result": None, "status": "ok", "success": True}, + repeat=True, + ) mocked.post( re.compile(r"https://.*iot\.roborock\.com/api/v4/email/code/send.*"), status=200, diff --git a/tests/test_web_api.py b/tests/test_web_api.py index 3cb9d3d6..15bdf5af 100644 --- a/tests/test_web_api.py +++ b/tests/test_web_api.py @@ -81,6 +81,36 @@ async def test_get_scenes(): ] +async def test_get_firmware_info(): + """Test that we can get firmware/OTA info for a device.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + fw = await api.get_firmware_info(ud, "abc123") + assert fw.version == "1.2.3" + assert fw.current_version == "1.2.0" + assert fw.updatable is True + assert fw.desc == "Bug fixes and improvements" + assert fw.release_time == "2026-05-01" + assert fw.force_update is False + + +async def test_start_firmware_update(): + """Test that we can trigger a firmware update for a device.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + # Should not raise when the API reports success. + await api.start_firmware_update(ud, "abc123") + + +async def test_set_silent_ota(): + """Test that we can toggle automatic firmware updates for a device.""" + api = RoborockApiClient(username="test_user@gmail.com") + ud = await api.pass_login("password") + # Should not raise when the API reports success. + await api.set_silent_ota(ud, "abc123", True) + await api.set_silent_ota(ud, "abc123", False) + + async def test_execute_scene(mock_rest): """Test that we can execute a scene""" api = RoborockApiClient(username="test_user@gmail.com")