feat(web_api): add firmware/OTA info, update trigger and silent-OTA toggle

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: ximex

roborock/data/containers.py

@@ -328,6 +328,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 timestamp of the available firmware."""
+    force_update: bool | None = None
+    """Whether the update is mandatory (cannot be skipped)."""
+
+
 @dataclass
 class HomeDataSchedule(RoborockBase):
     id: int

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,81 @@ 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).
+
+        Uses the rriot API host (``user_data.rriot.r.a``, e.g. ``api-eu.roborock.com``)
+        with Hawk-signed authentication, identical to the home-data endpoints. The
+        IoT host (``usiot.roborock.com``) returns an empty body for this path.
+        """
+        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.
+
+        Empty-body POST to the rriot API host with Hawk auth. The device then
+        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.
+
+        Form-encoded PUT to the rriot API host with Hawk auth (the form field is
+        part of the signed MAC). 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:

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,

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")