1import base64
  2import hashlib
  3import hmac
  4import json
  5import logging
  6import math
  7import secrets
  8import string
  9import time
 10from collections.abc import Callable
 11from dataclasses import dataclass
 12
 13import aiohttp
 14from aiohttp import ContentTypeError, FormData
 15from pyrate_limiter import Duration, Limiter, Rate
 16
 17from roborock import HomeDataSchedule
 18from roborock.data import FirmwareInfo, HomeData, HomeDataRoom, HomeDataScene, ProductResponse, RRiot, UserData
 19from roborock.exceptions import (
 20    RoborockAccountDoesNotExist,
 21    RoborockException,
 22    RoborockInvalidCode,
 23    RoborockInvalidCredentials,
 24    RoborockInvalidEmail,
 25    RoborockInvalidUserAgreement,
 26    RoborockMissingParameters,
 27    RoborockNoResponseFromBaseURL,
 28    RoborockNoUserAgreement,
 29    RoborockRateLimit,
 30    RoborockTooFrequentCodeRequests,
 31)
 32
 33_LOGGER = logging.getLogger(__name__)
 34BASE_URLS = [
 35    "https://usiot.roborock.com",
 36    "https://euiot.roborock.com",
 37    "https://cniot.roborock.com",
 38    "https://ruiot.roborock.com",
 39]
 40
 41# Fallback user agreement version, used only if the latest version cannot be fetched.
 42DEFAULT_AGREEMENT_MAJOR_VERSION = 14
 43DEFAULT_AGREEMENT_MINOR_VERSION = 0
 44
 45
 46@dataclass
 47class IotLoginInfo:
 48    """Information about the login to the iot server."""
 49
 50    base_url: str
 51    country_code: str
 52    country: str
 53
 54
 55class RoborockApiClient:
 56    _LOGIN_RATES = [
 57        Rate(1, Duration.SECOND),
 58        Rate(3, Duration.MINUTE),
 59        Rate(10, Duration.HOUR),
 60        Rate(20, Duration.DAY),
 61    ]
 62    _HOME_DATA_RATES = [
 63        Rate(1, Duration.SECOND),
 64        Rate(3, Duration.MINUTE),
 65        Rate(5, Duration.HOUR),
 66        Rate(40, Duration.DAY),
 67    ]
 68
 69    _login_limiter = Limiter(_LOGIN_RATES)
 70    _home_data_limiter = Limiter(_HOME_DATA_RATES)
 71
 72    def __init__(
 73        self, username: str, base_url: str | None = None, session: aiohttp.ClientSession | None = None
 74    ) -> None:
 75        """Sample API Client."""
 76        self._username = username
 77        self._base_url = base_url
 78        self._device_identifier = secrets.token_urlsafe(16)
 79        self.session = session
 80        self._iot_login_info: IotLoginInfo | None = None
 81        self._base_urls = BASE_URLS if base_url is None else [base_url]
 82
 83    async def _get_iot_login_info(self) -> IotLoginInfo:
 84        if self._iot_login_info is None:
 85            for iot_url in self._base_urls:
 86                url_request = PreparedRequest(iot_url, self.session)
 87                response = await url_request.request(
 88                    "post",
 89                    "/api/v1/getUrlByEmail",
 90                    params={"email": self._username, "needtwostepauth": "false"},
 91                )
 92                if response is None:
 93                    continue
 94                response_code = response.get("code")
 95                if response_code != 200:
 96                    if response_code == 2003:
 97                        raise RoborockInvalidEmail("Your email was incorrectly formatted.")
 98                    elif response_code == 1001:
 99                        raise RoborockMissingParameters(
100                            "You are missing parameters for this request, are you sure you entered your username?"
101                        )
102                    else:
103                        raise RoborockException(f"{response.get('msg')} - response code: {response_code}")
104                country_code = response["data"]["countrycode"]
105                country = response["data"]["country"]
106                if country_code is not None or country is not None:
107                    self._iot_login_info = IotLoginInfo(
108                        base_url=response["data"]["url"],
109                        country=country,
110                        country_code=country_code,
111                    )
112                    _LOGGER.debug("Country determined to be %s and code is %s", country, country_code)
113                    return self._iot_login_info
114            raise RoborockNoResponseFromBaseURL(
115                "No account was found for any base url we tried. Either your email is incorrect or we do not have a"
116                " record of the roborock server your device is on."
117            )
118        return self._iot_login_info
119
120    @property
121    async def base_url(self):
122        if self._base_url is not None:
123            return self._base_url
124        return (await self._get_iot_login_info()).base_url
125
126    @property
127    async def country(self):
128        return (await self._get_iot_login_info()).country
129
130    @property
131    async def country_code(self):
132        return (await self._get_iot_login_info()).country_code
133
134    def _get_header_client_id(self):
135        md5 = hashlib.md5()
136        md5.update(self._username.encode())
137        md5.update(self._device_identifier.encode())
138        return base64.b64encode(md5.digest()).decode()
139
140    async def nc_prepare(self, user_data: UserData, timezone: str) -> dict:
141        """This gets a few critical parameters for adding a device to your account."""
142        if (
143            user_data.rriot is None
144            or user_data.rriot.r is None
145            or user_data.rriot.u is None
146            or user_data.rriot.r.a is None
147        ):
148            raise RoborockException("Your userdata is missing critical attributes.")
149        base_url = user_data.rriot.r.a
150        prepare_request = PreparedRequest(base_url, self.session)
151        hid = await self._get_home_id(user_data)
152
153        data = FormData()
154        data.add_field("hid", hid)
155        data.add_field("tzid", timezone)
156
157        prepare_response = await prepare_request.request(
158            "post",
159            "/nc/prepare",
160            headers={
161                "Authorization": _get_hawk_authentication(
162                    user_data.rriot, "/nc/prepare", {"hid": hid, "tzid": timezone}
163                ),
164            },
165            data=data,
166        )
167
168        if prepare_response is None:
169            raise RoborockException("prepare_response is None")
170        if not prepare_response.get("success"):
171            raise RoborockException(f"{prepare_response.get('msg')} - response code: {prepare_response.get('code')}")
172
173        return prepare_response["result"]
174
175    async def add_device(self, user_data: UserData, s: str, t: str) -> dict:
176        """This will add a new device to your account
177        it is recommended to only use this during a pairing cycle with a device.
178        Please see here: https://github.com/Python-roborock/Roborockmitmproxy/blob/main/handshake_protocol.md
179        """
180        if (
181            user_data.rriot is None
182            or user_data.rriot.r is None
183            or user_data.rriot.u is None
184            or user_data.rriot.r.a is None
185        ):
186            raise RoborockException("Your userdata is missing critical attributes.")
187        base_url = user_data.rriot.r.a
188        add_device_request = PreparedRequest(base_url, self.session)
189
190        add_device_response = await add_device_request.request(
191            "GET",
192            "/user/devices/newadd",
193            headers={
194                "Authorization": _get_hawk_authentication(
195                    user_data.rriot, "/user/devices/newadd", params={"s": s, "t": t}
196                ),
197            },
198            params={"s": s, "t": t},
199        )
200
201        if add_device_response is None:
202            raise RoborockException("add_device is None")
203        if not add_device_response.get("success"):
204            raise RoborockException(
205                f"{add_device_response.get('msg')} - response code: {add_device_response.get('code')}"
206            )
207
208        return add_device_response["result"]
209
210    async def request_code(self) -> None:
211        if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1):
212            raise RoborockRateLimit("Reached maximum requests for login. Please try again later.")
213        base_url = await self.base_url
214        header_clientid = self._get_header_client_id()
215        code_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
216
217        code_response = await code_request.request(
218            "post",
219            "/api/v1/sendEmailCode",
220            params={
221                "username": self._username,
222                "type": "auth",
223            },
224        )
225        if code_response is None:
226            raise RoborockException("Failed to get a response from send email code")
227        response_code = code_response.get("code")
228        if response_code != 200:
229            _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response)
230            if response_code == 2008:
231                raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.")
232            elif response_code == 9002:
233                raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later")
234            else:
235                raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}")
236
237    async def request_code_v4(self) -> None:
238        """Request a code using the v4 endpoint."""
239        if await self.country_code is None or await self.country is None:
240            _LOGGER.info("No country code or country found, trying old version of request code.")
241            return await self.request_code()
242        if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1):
243            raise RoborockRateLimit("Reached maximum requests for login. Please try again later.")
244        base_url = await self.base_url
245        header_clientid = self._get_header_client_id()
246        code_request = PreparedRequest(
247            base_url,
248            self.session,
249            {
250                "header_clientid": header_clientid,
251                "Content-Type": "application/x-www-form-urlencoded",
252                "header_clientlang": "en",
253            },
254        )
255
256        code_response = await code_request.request(
257            "post",
258            "/api/v4/email/code/send",
259            data={"email": self._username, "type": "login", "platform": ""},
260        )
261        if code_response is None:
262            raise RoborockException("Failed to get a response from send email code")
263        response_code = code_response.get("code")
264        if response_code != 200:
265            _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response)
266            if response_code == 2008:
267                raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.")
268            elif response_code == 9002:
269                raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later")
270            elif response_code == 3030 and len(self._base_urls) > 1:
271                self._base_urls = self._base_urls[1:]
272                self._iot_login_info = None
273                return await self.request_code_v4()
274            else:
275                raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}")
276
277    async def _sign_key_v3(self, s: str) -> str:
278        """Sign a randomly generated string."""
279        base_url = await self.base_url
280        header_clientid = self._get_header_client_id()
281        code_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
282
283        code_response = await code_request.request(
284            "post",
285            "/api/v3/key/sign",
286            params={"s": s},
287        )
288
289        if not code_response or "data" not in code_response or "k" not in code_response["data"]:
290            raise RoborockException("Failed to get a response from sign key")
291        response_code = code_response.get("code")
292
293        if response_code != 200:
294            _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response)
295            raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}")
296
297        return code_response["data"]["k"]
298
299    async def _get_agreement_version(self, country: str) -> dict[str, int]:
300        """Get the latest user agreement version for the given country.
301
302        The login endpoint rejects a stale agreement version with code 3006. The
303        current version differs per server and per country, so it must be looked
304        up rather than hardcoded. Falls back to the previously hardcoded values
305        if the lookup fails so that login is never blocked by this request.
306        """
307        try:
308            base_url = await self.base_url
309            agreement_request = PreparedRequest(base_url, self.session, {"header_clientlang": "en"})
310            response = await agreement_request.request(
311                "get",
312                "/api/v3/app/agreement/latest",
313                params={"country": country},
314            )
315            if response is not None and response.get("code") == 200:
316                data = response.get("data") or {}
317                major = data.get("majorVersion")
318                minor = data.get("minorVersion")
319                if isinstance(major, int) and isinstance(minor, int):
320                    _LOGGER.debug("Using user agreement version %s.%s for %s", major, minor, country)
321                    return {"majorVersion": major, "minorVersion": minor}
322            _LOGGER.debug("Unexpected agreement version response: %s", response)
323        except RoborockException as err:
324            _LOGGER.debug("Could not fetch latest user agreement version: %s", err)
325        return {"majorVersion": DEFAULT_AGREEMENT_MAJOR_VERSION, "minorVersion": DEFAULT_AGREEMENT_MINOR_VERSION}
326
327    async def code_login_v4(
328        self, code: int | str, country: str | None = None, country_code: int | None = None
329    ) -> UserData:
330        """
331        Login via code authentication.
332        :param code: The code from the email.
333        :param country: The two-character representation of the country, i.e. "US"
334        :param country_code: the country phone number code i.e. 1 for US.
335        """
336        base_url = await self.base_url
337        if country is None:
338            country = await self.country
339        if country_code is None:
340            country_code = await self.country_code
341        if country_code is None or country is None:
342            _LOGGER.info("No country code or country found, trying old version of code login.")
343            return await self.code_login(code)
344        header_clientid = self._get_header_client_id()
345        x_mercy_ks = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(16))
346        x_mercy_k = await self._sign_key_v3(x_mercy_ks)
347        login_request = PreparedRequest(
348            base_url,
349            self.session,
350            {
351                "header_clientid": header_clientid,
352                "x-mercy-ks": x_mercy_ks,
353                "x-mercy-k": x_mercy_k,
354                "Content-Type": "application/x-www-form-urlencoded",
355                "header_clientlang": "en",
356                "header_appversion": "4.54.02",
357                "header_phonesystem": "iOS",
358                "header_phonemodel": "iPhone16,1",
359            },
360        )
361        login_response = await login_request.request(
362            "post",
363            "/api/v4/auth/email/login/code",
364            data={
365                "country": country,
366                "countryCode": country_code,
367                "email": self._username,
368                "code": code,
369                **await self._get_agreement_version(country),
370            },
371        )
372        if login_response is None:
373            raise RoborockException("Login request response is None")
374        response_code = login_response.get("code")
375        if response_code != 200:
376            _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response)
377            if response_code == 2018:
378                raise RoborockInvalidCode("Invalid code - check your code and try again.")
379            if response_code == 3009:
380                raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.")
381            if response_code == 3006:
382                raise RoborockInvalidUserAgreement(
383                    "User agreement must be accepted again - or you are attempting to use the Mi Home app account."
384                )
385            if response_code == 3039:
386                raise RoborockAccountDoesNotExist(
387                    "This account does not exist - please ensure that you selected the right region and email."
388                )
389            raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}")
390        user_data = login_response.get("data")
391        if not isinstance(user_data, dict):
392            raise RoborockException("Got unexpected data type for user_data")
393        return UserData.from_dict(user_data)
394
395    async def pass_login(self, password: str) -> UserData:
396        if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1):
397            raise RoborockRateLimit("Reached maximum requests for login. Please try again later.")
398        base_url = await self.base_url
399        header_clientid = self._get_header_client_id()
400
401        login_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
402        login_response = await login_request.request(
403            "post",
404            "/api/v1/login",
405            params={
406                "username": self._username,
407                "password": password,
408                "needtwostepauth": "false",
409            },
410        )
411        if login_response is None:
412            raise RoborockException("Login response is none")
413        if login_response.get("code") != 200:
414            _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response)
415            raise RoborockException(f"{login_response.get('msg')} - response code: {login_response.get('code')}")
416        user_data = login_response.get("data")
417        if not isinstance(user_data, dict):
418            raise RoborockException("Got unexpected data type for user_data")
419        return UserData.from_dict(user_data)
420
421    async def pass_login_v3(self, password: str) -> UserData:
422        """Seemingly it follows the format below, but password is encrypted in some manner.
423        # login_response = await login_request.request(
424        #     "post",
425        #     "/api/v3/auth/email/login",
426        #     params={
427        #         "email": self._username,
428        #         "password": password,
429        #         "twoStep": 1,
430        #         "version": 0
431        #     },
432        # )
433        """
434        raise NotImplementedError("Pass_login_v3 has not yet been implemented")
435
436    async def code_login(self, code: int | str) -> UserData:
437        base_url = await self.base_url
438        header_clientid = self._get_header_client_id()
439
440        login_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
441        login_response = await login_request.request(
442            "post",
443            "/api/v1/loginWithCode",
444            params={
445                "username": self._username,
446                "verifycode": code,
447                "verifycodetype": "AUTH_EMAIL_CODE",
448            },
449        )
450        if login_response is None:
451            raise RoborockException("Login request response is None")
452        response_code = login_response.get("code")
453        if response_code != 200:
454            _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response)
455            if response_code == 2018:
456                raise RoborockInvalidCode("Invalid code - check your code and try again.")
457            if response_code == 3009:
458                raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.")
459            if response_code == 3006:
460                raise RoborockInvalidUserAgreement(
461                    "User agreement must be accepted again - or you are attempting to use the Mi Home app account."
462                )
463            raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}")
464        user_data = login_response.get("data")
465        if not isinstance(user_data, dict):
466            raise RoborockException("Got unexpected data type for user_data")
467        return UserData.from_dict(user_data)
468
469    async def _get_home_id(self, user_data: UserData):
470        base_url = await self.base_url
471        header_clientid = self._get_header_client_id()
472        home_id_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
473        home_id_response = await home_id_request.request(
474            "get",
475            "/api/v1/getHomeDetail",
476            headers={"Authorization": user_data.token},
477        )
478        if home_id_response is None:
479            raise RoborockException("home_id_response is None")
480        if home_id_response.get("code") != 200:
481            _LOGGER.info("Get Home Id failed with the following context: %s", home_id_response)
482            if home_id_response.get("code") == 2010:
483                raise RoborockInvalidCredentials(
484                    f"Invalid credentials ({home_id_response.get('msg')}) - check your login and try again."
485                )
486            raise RoborockException(f"{home_id_response.get('msg')} - response code: {home_id_response.get('code')}")
487
488        return home_id_response["data"]["rrHomeId"]
489
490    async def get_home_data(self, user_data: UserData) -> HomeData:
491        if not self._home_data_limiter.try_acquire("home_data", blocking=False):
492            raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.")
493        rriot = user_data.rriot
494        if rriot is None:
495            raise RoborockException("rriot is none")
496        home_id = await self._get_home_id(user_data)
497        if rriot.r.a is None:
498            raise RoborockException("Missing field 'a' in rriot reference")
499        home_request = PreparedRequest(
500            rriot.r.a,
501            self.session,
502            {
503                "Authorization": _get_hawk_authentication(rriot, f"/user/homes/{str(home_id)}"),
504            },
505        )
506        home_response = await home_request.request("get", "/user/homes/" + str(home_id))
507        if not home_response.get("success"):
508            raise RoborockException(home_response)
509        home_data = home_response.get("result")
510        if isinstance(home_data, dict):
511            return HomeData.from_dict(home_data)
512        else:
513            raise RoborockException("home_response result was an unexpected type")
514
515    async def get_home_data_v2(self, user_data: UserData) -> HomeData:
516        """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums."""
517        if not self._home_data_limiter.try_acquire("home_data", blocking=False):
518            raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.")
519        rriot = user_data.rriot
520        if rriot is None:
521            raise RoborockException("rriot is none")
522        home_id = await self._get_home_id(user_data)
523        if rriot.r.a is None:
524            raise RoborockException("Missing field 'a' in rriot reference")
525        home_request = PreparedRequest(
526            rriot.r.a,
527            self.session,
528            {
529                "Authorization": _get_hawk_authentication(rriot, "/v2/user/homes/" + str(home_id)),
530            },
531        )
532        home_response = await home_request.request("get", "/v2/user/homes/" + str(home_id))
533        if not home_response.get("success"):
534            raise RoborockException(home_response)
535        home_data = home_response.get("result")
536        if isinstance(home_data, dict):
537            return HomeData.from_dict(home_data)
538        else:
539            raise RoborockException("home_response result was an unexpected type")
540
541    async def get_home_data_v3(self, user_data: UserData) -> HomeData:
542        """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums."""
543        if not self._home_data_limiter.try_acquire("home_data", blocking=False):
544            raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.")
545        rriot = user_data.rriot
546        home_id = await self._get_home_id(user_data)
547        if rriot.r.a is None:
548            raise RoborockException("Missing field 'a' in rriot reference")
549        home_request = PreparedRequest(
550            rriot.r.a,
551            self.session,
552            {
553                "Authorization": _get_hawk_authentication(rriot, "/v3/user/homes/" + str(home_id)),
554            },
555        )
556        home_response = await home_request.request("get", "/v3/user/homes/" + str(home_id))
557        if not home_response.get("success"):
558            raise RoborockException(home_response)
559        home_data = home_response.get("result")
560        if isinstance(home_data, dict):
561            return HomeData.from_dict(home_data)
562        raise RoborockException(f"home_response result was an unexpected type: {home_data}")
563
564    async def get_rooms(self, user_data: UserData, home_id: int | None = None) -> list[HomeDataRoom]:
565        rriot = user_data.rriot
566        if rriot is None:
567            raise RoborockException("rriot is none")
568        if home_id is None:
569            home_id = await self._get_home_id(user_data)
570        if rriot.r.a is None:
571            raise RoborockException("Missing field 'a' in rriot reference")
572        room_request = PreparedRequest(
573            rriot.r.a,
574            self.session,
575            {
576                "Authorization": _get_hawk_authentication(rriot, f"/user/homes/{home_id}/rooms"),
577            },
578        )
579        room_response = await room_request.request("get", f"/user/homes/{home_id}/rooms")
580        if not room_response.get("success"):
581            raise RoborockException(room_response)
582        rooms = room_response.get("result")
583        if isinstance(rooms, list):
584            output_list = []
585            for room in rooms:
586                output_list.append(HomeDataRoom.from_dict(room))
587            return output_list
588        else:
589            raise RoborockException("home_response result was an unexpected type")
590
591    async def get_shared_device_rooms(self, user_data: UserData, device_id: str) -> list[HomeDataRoom]:
592        """Fetch room names for a shared (received) device."""
593        rriot = user_data.rriot
594        if rriot is None:
595            raise RoborockException("rriot is none")
596        if rriot.r.a is None:
597            raise RoborockException("Missing field 'a' in rriot reference")
598        path = f"/user/deviceshare/query/{device_id}/rooms"
599        room_request = PreparedRequest(
600            rriot.r.a,
601            self.session,
602            {"Authorization": _get_hawk_authentication(rriot, path)},
603        )
604        room_response = await room_request.request("get", path)
605        if not room_response.get("success"):
606            raise RoborockException(room_response)
607        rooms = room_response.get("result")
608        if isinstance(rooms, list):
609            output_list = []
610            for room in rooms:
611                normalized_room = room
612                if isinstance(room, dict) and "id" not in room and "roomId" in room:
613                    normalized_room = {**room, "id": room["roomId"]}
614                output_list.append(HomeDataRoom.from_dict(normalized_room))
615            return output_list
616        raise RoborockException("get_shared_device_rooms result was an unexpected type")
617
618    async def get_scenes(self, user_data: UserData, device_id: str) -> list[HomeDataScene]:
619        rriot = user_data.rriot
620        if rriot is None:
621            raise RoborockException("rriot is none")
622        if rriot.r.a is None:
623            raise RoborockException("Missing field 'a' in rriot reference")
624        scenes_request = PreparedRequest(
625            rriot.r.a,
626            self.session,
627            {
628                "Authorization": _get_hawk_authentication(rriot, f"/user/scene/device/{str(device_id)}"),
629            },
630        )
631        scenes_response = await scenes_request.request("get", f"/user/scene/device/{str(device_id)}")
632        if not scenes_response.get("success"):
633            raise RoborockException(scenes_response)
634        scenes = scenes_response.get("result")
635        if isinstance(scenes, list):
636            return [HomeDataScene.from_dict(scene) for scene in scenes]
637        else:
638            raise RoborockException("scene_response result was an unexpected type")
639
640    async def get_firmware_info(self, user_data: UserData, device_id: str) -> FirmwareInfo:
641        """Get firmware/OTA info for a device (latest version + updatable flag)."""
642        rriot = user_data.rriot
643        if rriot is None:
644            raise RoborockException("rriot is none")
645        if rriot.r.a is None:
646            raise RoborockException("Missing field 'a' in rriot reference")
647        path = f"/ota/firmware/{device_id}/updatev2"
648        params = {"lang": "en"}
649        firmware_request = PreparedRequest(
650            rriot.r.a,
651            self.session,
652            {
653                "Authorization": _get_hawk_authentication(rriot, path, params=params),
654            },
655        )
656        firmware_response = await firmware_request.request("get", path, params=params)
657        if not firmware_response or not firmware_response.get("success"):
658            raise RoborockException(firmware_response)
659        return FirmwareInfo.from_dict(firmware_response.get("result") or {})
660
661    async def start_firmware_update(self, user_data: UserData, device_id: str) -> None:
662        """Trigger the (irreversible) firmware update for a device.
663
664        The device downloads and flashes the latest firmware reported by
665        :meth:`get_firmware_info`.
666        """
667        rriot = user_data.rriot
668        if rriot is None:
669            raise RoborockException("rriot is none")
670        if rriot.r.a is None:
671            raise RoborockException("Missing field 'a' in rriot reference")
672        path = f"/ota/device/{device_id}/upgrade"
673        upgrade_request = PreparedRequest(
674            rriot.r.a,
675            self.session,
676            {
677                "Authorization": _get_hawk_authentication(rriot, path),
678            },
679        )
680        upgrade_response = await upgrade_request.request("post", path)
681        if not upgrade_response or not upgrade_response.get("success"):
682            raise RoborockException(upgrade_response)
683
684    async def set_silent_ota(self, user_data: UserData, device_id: str, enabled: bool) -> None:
685        """Enable/disable automatic (silent) firmware updates for a device.
686
687        The current value is reported as ``silent_ota_switch`` on the home-data device.
688        """
689        rriot = user_data.rriot
690        if rriot is None:
691            raise RoborockException("rriot is none")
692        if rriot.r.a is None:
693            raise RoborockException("Missing field 'a' in rriot reference")
694        path = f"/user/devices/{device_id}"
695        formdata = {"silentOtaSwitch": "true" if enabled else "false"}
696        silent_ota_request = PreparedRequest(
697            rriot.r.a,
698            self.session,
699            {
700                "Authorization": _get_hawk_authentication(rriot, path, formdata=formdata),
701            },
702        )
703        silent_ota_response = await silent_ota_request.request("put", path, data=formdata)
704        if not silent_ota_response or not silent_ota_response.get("success"):
705            raise RoborockException(silent_ota_response)
706
707    async def execute_scene(self, user_data: UserData, scene_id: int) -> None:
708        rriot = user_data.rriot
709        if rriot is None:
710            raise RoborockException("rriot is none")
711        if rriot.r.a is None:
712            raise RoborockException("Missing field 'a' in rriot reference")
713        execute_scene_request = PreparedRequest(
714            rriot.r.a,
715            self.session,
716            {
717                "Authorization": _get_hawk_authentication(rriot, f"/user/scene/{str(scene_id)}/execute"),
718            },
719        )
720        execute_scene_response = await execute_scene_request.request("POST", f"/user/scene/{str(scene_id)}/execute")
721        if not execute_scene_response.get("success"):
722            raise RoborockException(execute_scene_response)
723
724    async def get_schedules(self, user_data: UserData, device_id: str) -> list[HomeDataSchedule]:
725        rriot = user_data.rriot
726        if rriot is None:
727            raise RoborockException("rriot is none")
728        if rriot.r.a is None:
729            raise RoborockException("Missing field 'a' in rriot reference")
730        schedules_request = PreparedRequest(
731            rriot.r.a,
732            self.session,
733            {
734                "Authorization": _get_hawk_authentication(rriot, f"/user/devices/{device_id}/jobs"),
735            },
736        )
737        schedules_response = await schedules_request.request("get", f"/user/devices/{str(device_id)}/jobs")
738        if not schedules_response.get("success"):
739            raise RoborockException(schedules_response)
740        schedules = schedules_response.get("result")
741        if isinstance(schedules, list):
742            return [HomeDataSchedule.from_dict(schedule) for schedule in schedules]
743        else:
744            raise RoborockException(f"schedule_response result was an unexpected type: {schedules}")
745
746    async def create_job(self, user_data: UserData, device_id: str, job: dict) -> dict:
747        """Create a /jobs entry (schedule or one-time room clean) on a B01 device.
748
749        Body-bearing writes must sign the request body in the Hawk payload slot and send those same
750        compact bytes via ``data=``; ``json=`` would re-serialize with spaces and break the MAC.
751        """
752        rriot = user_data.rriot
753        if rriot is None:
754            raise RoborockException("rriot is none")
755        if rriot.r.a is None:
756            raise RoborockException("Missing field 'a' in rriot reference")
757        path = f"/user/devices/{device_id}/jobs"
758        job_request = PreparedRequest(
759            rriot.r.a,
760            self.session,
761            {
762                "Authorization": _get_hawk_authentication(rriot, path, body=job),
763                "Content-Type": "application/json",
764            },
765        )
766        response = await job_request.request("post", path, data=_compact_json(job).encode())
767        if not response.get("success"):
768            raise RoborockException(response)
769        return response
770
771    async def get_products(self, user_data: UserData) -> ProductResponse:
772        """Gets all products and their schemas, good for determining status codes and model numbers."""
773        base_url = await self.base_url
774        header_clientid = self._get_header_client_id()
775        product_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
776        product_response = await product_request.request(
777            "get",
778            "/api/v4/product",
779            headers={"Authorization": user_data.token},
780        )
781        if product_response is None:
782            raise RoborockException("home_id_response is None")
783        if product_response.get("code") != 200:
784            raise RoborockException(f"{product_response.get('msg')} - response code: {product_response.get('code')}")
785        result = product_response.get("data")
786        if isinstance(result, dict):
787            return ProductResponse.from_dict(result)
788        raise RoborockException("product result was an unexpected type")
789
790    async def download_code(self, user_data: UserData, product_id: int):
791        base_url = await self.base_url
792        header_clientid = self._get_header_client_id()
793        product_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
794        request = {"apilevel": 99999, "productids": [product_id], "type": 2}
795        response = await product_request.request(
796            "post",
797            "/api/v1/appplugin",
798            json=request,
799            headers={"Authorization": user_data.token, "Content-Type": "application/json"},
800        )
801        return response["data"][0]["url"]
802
803    async def download_category_code(self, user_data: UserData):
804        base_url = await self.base_url
805        header_clientid = self._get_header_client_id()
806        product_request = PreparedRequest(base_url, self.session, {"header_clientid": header_clientid})
807        response = await product_request.request(
808            "get",
809            "api/v1/plugins?apiLevel=99999&type=2",
810            headers={
811                "Authorization": user_data.token,
812            },
813        )
814        return {r["category"]: r["url"] for r in response["data"]["categoryPluginList"]}
815
816
817class PreparedRequest:
818    def __init__(
819        self, base_url: str, session: aiohttp.ClientSession | None = None, base_headers: dict | None = None
820    ) -> None:
821        self.base_url = base_url
822        self.base_headers = base_headers or {}
823        self.session = session
824
825    async def request(self, method: str, url: str, params=None, data=None, headers=None, json=None) -> dict:
826        _url = "/".join(s.strip("/") for s in [self.base_url, url])
827        _headers = {**self.base_headers, **(headers or {})}
828        close_session = self.session is None
829        session = self.session if self.session is not None else aiohttp.ClientSession()
830        try:
831            async with session.request(method, _url, params=params, data=data, headers=_headers, json=json) as resp:
832                return await resp.json()
833        except ContentTypeError as err:
834            """If we get an error, lets log everything for debugging."""
835            try:
836                resp_json = await resp.json(content_type=None)
837                _LOGGER.info("Resp: %s", resp_json)
838            except ContentTypeError as err_2:
839                _LOGGER.info(err_2)
840            resp_raw = await resp.read()
841            _LOGGER.info("Resp raw: %s", resp_raw)
842            # Still raise the err so that it's clear it failed.
843            raise err
844        except (aiohttp.ClientError, TimeoutError, OSError) as err:
845            raise RoborockException(f"Network error contacting {_url}: {err}") from err
846        finally:
847            if close_session:
848                await session.close()
849
850
851def _process_extra_hawk_values(values: dict | None) -> str:
852    if values is None:
853        return ""
854    else:
855        sorted_keys = sorted(values.keys())
856        result = []
857        for key in sorted_keys:
858            value = values.get(key)
859            result.append(f"{key}={value}")
860        return hashlib.md5("&".join(result).encode()).hexdigest()
861
862
863def _compact_json(body: dict) -> str:
864    """Serialize a JSON body to the exact compact bytes that are both signed and sent."""
865    return json.dumps(body, separators=(",", ":"))
866
867
868def _get_hawk_authentication(
869    rriot: RRiot,
870    url: str,
871    formdata: dict | None = None,
872    params: dict | None = None,
873    body: dict | None = None,
874) -> str:
875    timestamp = math.floor(time.time())
876    nonce = secrets.token_urlsafe(6)
877    params_str = _process_extra_hawk_values(params)
878    if body is not None:
879        payload_str = hashlib.md5(_compact_json(body).encode()).hexdigest()
880    else:
881        payload_str = _process_extra_hawk_values(formdata)
882
883    prestr = ":".join(
884        [
885            rriot.u,
886            rriot.s,
887            nonce,
888            str(timestamp),
889            hashlib.md5(url.encode()).hexdigest(),
890            params_str,
891            payload_str,
892        ]
893    )
894    mac = base64.b64encode(hmac.new(rriot.h.encode(), prestr.encode(), hashlib.sha256).digest()).decode()
895    return f'Hawk id="{rriot.u}",s="{rriot.s}",ts="{timestamp}",nonce="{nonce}",mac="{mac}"'
896
897
898class UserWebApiClient:
899    """Wrapper around RoborockApiClient to provide information for a specific user.
900
901    This binds a RoborockApiClient to a specific user context with the
902    provided UserData. This allows for easier access to user-specific data,
903    to avoid needing to pass UserData around and mock out the web API.
904    """
905
906    def __init__(
907        self, web_api: RoborockApiClient, user_data: UserData, unauthorized_hook: Callable[[], None] | None = None
908    ) -> None:
909        """Initialize the wrapper with the API client and user data."""
910        self._web_api = web_api
911        self._user_data = user_data
912        self._unauthorized_hook = unauthorized_hook
913
914    async def get_home_data(self) -> HomeData:
915        """Fetch home data using the API client."""
916        try:
917            return await self._web_api.get_home_data_v3(self._user_data)
918        except RoborockInvalidCredentials:
919            if self._unauthorized_hook:
920                self._unauthorized_hook()
921            raise
922
923    async def get_routines(self, device_id: str) -> list[HomeDataScene]:
924        """Fetch routines (scenes) for a specific device."""
925        try:
926            return await self._web_api.get_scenes(self._user_data, device_id)
927        except RoborockInvalidCredentials:
928            if self._unauthorized_hook:
929                self._unauthorized_hook()
930            raise
931
932    async def get_rooms(self) -> list[HomeDataRoom]:
933        """Fetch rooms using the API client."""
934        try:
935            return await self._web_api.get_rooms(self._user_data)
936        except RoborockInvalidCredentials:
937            if self._unauthorized_hook:
938                self._unauthorized_hook()
939            raise
940
941    async def get_shared_device_rooms(self, device_id: str) -> list[HomeDataRoom]:
942        """Fetch shared-device rooms using the API client."""
943        return await self._web_api.get_shared_device_rooms(self._user_data, device_id)
944
945    async def execute_routine(self, scene_id: int) -> None:
946        """Execute a specific routine (scene) by its ID."""
947        try:
948            await self._web_api.execute_scene(self._user_data, scene_id)
949        except RoborockInvalidCredentials:
950            if self._unauthorized_hook:
951                self._unauthorized_hook()
952            raise