31class AuthApi(RoborockApiBase):
 32    """Email/password and email-code login flows."""
 33
 34    _LOGIN_RATES = [
 35        Rate(1, Duration.SECOND),
 36        Rate(3, Duration.MINUTE),
 37        Rate(10, Duration.HOUR),
 38        Rate(20, Duration.DAY),
 39    ]
 40    _login_limiter = Limiter(_LOGIN_RATES)
 41
 42    async def request_code(self) -> None:
 43        if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1):
 44            raise RoborockRateLimit("Reached maximum requests for login. Please try again later.")
 45        code_request = await self._iot_request()
 46
 47        code_response = await code_request.request(
 48            "post",
 49            "/api/v1/sendEmailCode",
 50            params={
 51                "username": self._username,
 52                "type": "auth",
 53            },
 54        )
 55        if code_response is None:
 56            raise RoborockException("Failed to get a response from send email code")
 57        response_code = code_response.get("code")
 58        if response_code != 200:
 59            _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response)
 60            if response_code == 2008:
 61                raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.")
 62            elif response_code == 9002:
 63                raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later")
 64            else:
 65                raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}")
 66
 67    async def request_code_v4(self) -> None:
 68        """Request a code using the v4 endpoint."""
 69        if await self.country_code is None or await self.country is None:
 70            _LOGGER.info("No country code or country found, trying old version of request code.")
 71            return await self.request_code()
 72        if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1):
 73            raise RoborockRateLimit("Reached maximum requests for login. Please try again later.")
 74        code_request = await self._iot_request(
 75            {
 76                "Content-Type": "application/x-www-form-urlencoded",
 77                "header_clientlang": "en",
 78            }
 79        )
 80
 81        code_response = await code_request.request(
 82            "post",
 83            "/api/v4/email/code/send",
 84            data={"email": self._username, "type": "login", "platform": ""},
 85        )
 86        if code_response is None:
 87            raise RoborockException("Failed to get a response from send email code")
 88        response_code = code_response.get("code")
 89        if response_code != 200:
 90            _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response)
 91            if response_code == 2008:
 92                raise RoborockAccountDoesNotExist("Account does not exist - check your login and try again.")
 93            elif response_code == 9002:
 94                raise RoborockTooFrequentCodeRequests("You have attempted to request too many codes. Try again later")
 95            elif response_code == 3030 and len(self._base_urls) > 1:
 96                self._base_urls = self._base_urls[1:]
 97                self._iot_login_info = None
 98                return await self.request_code_v4()
 99            else:
100                raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}")
101
102    async def _sign_key_v3(self, s: str) -> str:
103        """Sign a randomly generated string."""
104        code_request = await self._iot_request()
105
106        code_response = await code_request.request(
107            "post",
108            "/api/v3/key/sign",
109            params={"s": s},
110        )
111
112        if not code_response or "data" not in code_response or "k" not in code_response["data"]:
113            raise RoborockException("Failed to get a response from sign key")
114        response_code = code_response.get("code")
115
116        if response_code != 200:
117            _LOGGER.info("Request code failed for %s with the following context: %s", self._username, code_response)
118            raise RoborockException(f"{code_response.get('msg')} - response code: {code_response.get('code')}")
119
120        return code_response["data"]["k"]
121
122    async def _get_agreement_version(self, country: str) -> dict[str, int]:
123        """Get the latest user agreement version for the given country.
124
125        The login endpoint rejects a stale agreement version with code 3006. The
126        current version differs per server and per country, so it must be looked
127        up rather than hardcoded. Falls back to the previously hardcoded values
128        if the lookup fails so that login is never blocked by this request.
129        """
130        try:
131            base_url = await self.base_url
132            agreement_request = PreparedRequest(base_url, self.session, {"header_clientlang": "en"})
133            response = await agreement_request.request(
134                "get",
135                "/api/v3/app/agreement/latest",
136                params={"country": country},
137            )
138            if response is not None and response.get("code") == 200:
139                data = response.get("data") or {}
140                major = data.get("majorVersion")
141                minor = data.get("minorVersion")
142                if isinstance(major, int) and isinstance(minor, int):
143                    _LOGGER.debug("Using user agreement version %s.%s for %s", major, minor, country)
144                    return {"majorVersion": major, "minorVersion": minor}
145            _LOGGER.debug("Unexpected agreement version response: %s", response)
146        except RoborockException as err:
147            _LOGGER.debug("Could not fetch latest user agreement version: %s", err)
148        return {"majorVersion": DEFAULT_AGREEMENT_MAJOR_VERSION, "minorVersion": DEFAULT_AGREEMENT_MINOR_VERSION}
149
150    async def code_login_v4(
151        self, code: int | str, country: str | None = None, country_code: int | None = None
152    ) -> UserData:
153        """
154        Login via code authentication.
155        :param code: The code from the email.
156        :param country: The two-character representation of the country, i.e. "US"
157        :param country_code: the country phone number code i.e. 1 for US.
158        """
159        if country is None:
160            country = await self.country
161        if country_code is None:
162            country_code = await self.country_code
163        if country_code is None or country is None:
164            _LOGGER.info("No country code or country found, trying old version of code login.")
165            return await self.code_login(code)
166        x_mercy_ks = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(16))
167        x_mercy_k = await self._sign_key_v3(x_mercy_ks)
168        login_request = await self._iot_request(
169            {
170                "x-mercy-ks": x_mercy_ks,
171                "x-mercy-k": x_mercy_k,
172                "Content-Type": "application/x-www-form-urlencoded",
173                "header_clientlang": "en",
174                "header_appversion": "4.54.02",
175                "header_phonesystem": "iOS",
176                "header_phonemodel": "iPhone16,1",
177            }
178        )
179        login_response = await login_request.request(
180            "post",
181            "/api/v4/auth/email/login/code",
182            data={
183                "country": country,
184                "countryCode": country_code,
185                "email": self._username,
186                "code": code,
187                **await self._get_agreement_version(country),
188            },
189        )
190        if login_response is None:
191            raise RoborockException("Login request response is None")
192        response_code = login_response.get("code")
193        if response_code != 200:
194            _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response)
195            if response_code == 2018:
196                raise RoborockInvalidCode("Invalid code - check your code and try again.")
197            if response_code == 3009:
198                raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.")
199            if response_code == 3006:
200                raise RoborockInvalidUserAgreement(
201                    "User agreement must be accepted again - or you are attempting to use the Mi Home app account."
202                )
203            if response_code == 3039:
204                raise RoborockAccountDoesNotExist(
205                    "This account does not exist - please ensure that you selected the right region and email."
206                )
207            raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}")
208        user_data = login_response.get("data")
209        if not isinstance(user_data, dict):
210            raise RoborockException("Got unexpected data type for user_data")
211        return UserData.from_dict(user_data)
212
213    async def pass_login(self, password: str) -> UserData:
214        if not await self._login_limiter.try_acquire_async("login", blocking=True, timeout=1):
215            raise RoborockRateLimit("Reached maximum requests for login. Please try again later.")
216        login_request = await self._iot_request()
217        login_response = await login_request.request(
218            "post",
219            "/api/v1/login",
220            params={
221                "username": self._username,
222                "password": password,
223                "needtwostepauth": "false",
224            },
225        )
226        if login_response is None:
227            raise RoborockException("Login response is none")
228        if login_response.get("code") != 200:
229            _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response)
230            raise RoborockException(f"{login_response.get('msg')} - response code: {login_response.get('code')}")
231        user_data = login_response.get("data")
232        if not isinstance(user_data, dict):
233            raise RoborockException("Got unexpected data type for user_data")
234        return UserData.from_dict(user_data)
235
236    async def pass_login_v3(self, password: str) -> UserData:
237        """Seemingly it follows the format below, but password is encrypted in some manner.
238        # login_response = await login_request.request(
239        #     "post",
240        #     "/api/v3/auth/email/login",
241        #     params={
242        #         "email": self._username,
243        #         "password": password,
244        #         "twoStep": 1,
245        #         "version": 0
246        #     },
247        # )
248        """
249        raise NotImplementedError("Pass_login_v3 has not yet been implemented")
250
251    async def code_login(self, code: int | str) -> UserData:
252        login_request = await self._iot_request()
253        login_response = await login_request.request(
254            "post",
255            "/api/v1/loginWithCode",
256            params={
257                "username": self._username,
258                "verifycode": code,
259                "verifycodetype": "AUTH_EMAIL_CODE",
260            },
261        )
262        if login_response is None:
263            raise RoborockException("Login request response is None")
264        response_code = login_response.get("code")
265        if response_code != 200:
266            _LOGGER.info("Login failed for %s with the following context: %s", self._username, login_response)
267            if response_code == 2018:
268                raise RoborockInvalidCode("Invalid code - check your code and try again.")
269            if response_code == 3009:
270                raise RoborockNoUserAgreement("You must accept the user agreement in the Roborock app to continue.")
271            if response_code == 3006:
272                raise RoborockInvalidUserAgreement(
273                    "User agreement must be accepted again - or you are attempting to use the Mi Home app account."
274                )
275            raise RoborockException(f"{login_response.get('msg')} - response code: {response_code}")
276        user_data = login_response.get("data")
277        if not isinstance(user_data, dict):
278            raise RoborockException("Got unexpected data type for user_data")
279        return UserData.from_dict(user_data)

14class DevicesApi(RoborockApiBase):
15    """Adding devices to an account and editing per-device cloud settings."""
16
17    async def nc_prepare(self, user_data: UserData, timezone: str) -> dict:
18        """This gets a few critical parameters for adding a device to your account."""
19        if (
20            user_data.rriot is None
21            or user_data.rriot.r is None
22            or user_data.rriot.u is None
23            or user_data.rriot.r.a is None
24        ):
25            raise RoborockException("Your userdata is missing critical attributes.")
26        base_url = user_data.rriot.r.a
27        prepare_request = PreparedRequest(base_url, self.session)
28        hid = await self._get_home_id(user_data)
29
30        data = FormData()
31        data.add_field("hid", hid)
32        data.add_field("tzid", timezone)
33
34        prepare_response = await prepare_request.request(
35            "post",
36            "/nc/prepare",
37            headers={
38                "Authorization": _get_hawk_authentication(
39                    user_data.rriot, "/nc/prepare", {"hid": hid, "tzid": timezone}
40                ),
41            },
42            data=data,
43        )
44
45        if prepare_response is None:
46            raise RoborockException("prepare_response is None")
47        if not prepare_response.get("success"):
48            raise RoborockException(f"{prepare_response.get('msg')} - response code: {prepare_response.get('code')}")
49
50        return prepare_response["result"]
51
52    async def add_device(self, user_data: UserData, s: str, t: str) -> dict:
53        """This will add a new device to your account
54        it is recommended to only use this during a pairing cycle with a device.
55        Please see here: https://github.com/Python-roborock/Roborockmitmproxy/blob/main/handshake_protocol.md
56        """
57        if (
58            user_data.rriot is None
59            or user_data.rriot.r is None
60            or user_data.rriot.u is None
61            or user_data.rriot.r.a is None
62        ):
63            raise RoborockException("Your userdata is missing critical attributes.")
64        base_url = user_data.rriot.r.a
65        add_device_request = PreparedRequest(base_url, self.session)
66
67        add_device_response = await add_device_request.request(
68            "GET",
69            "/user/devices/newadd",
70            headers={
71                "Authorization": _get_hawk_authentication(
72                    user_data.rriot, "/user/devices/newadd", params={"s": s, "t": t}
73                ),
74            },
75            params={"s": s, "t": t},
76        )
77
78        if add_device_response is None:
79            raise RoborockException("add_device is None")
80        if not add_device_response.get("success"):
81            raise RoborockException(
82                f"{add_device_response.get('msg')} - response code: {add_device_response.get('code')}"
83            )
84
85        return add_device_response["result"]

12class HomesApi(RoborockApiBase):
13    """Home data (devices, products, rooms) lookups."""
14
15    _HOME_DATA_RATES = [
16        Rate(1, Duration.SECOND),
17        Rate(3, Duration.MINUTE),
18        Rate(5, Duration.HOUR),
19        Rate(40, Duration.DAY),
20    ]
21    _home_data_limiter = Limiter(_HOME_DATA_RATES)
22
23    async def _get_home_data_version(self, user_data: UserData, path_prefix: str) -> HomeData:
24        """Fetch home data from ``{path_prefix}/user/homes/{home_id}`` with rate limiting."""
25        if not self._home_data_limiter.try_acquire("home_data", blocking=False):
26            raise RoborockRateLimit("Reached maximum requests for home data. Please try again later.")
27        if user_data.rriot is None:
28            raise RoborockException("rriot is none")
29        home_id = await self._get_home_id(user_data)
30        path = f"{path_prefix}/user/homes/{home_id}"
31        home_request = self._rriot_request(user_data, path)
32        home_response = await home_request.request("get", path)
33        if not home_response.get("success"):
34            raise RoborockException(home_response)
35        home_data = home_response.get("result")
36        if isinstance(home_data, dict):
37            return HomeData.from_dict(home_data)
38        raise RoborockException(f"home_response result was an unexpected type: {home_data}")
39
40    async def get_home_data(self, user_data: UserData) -> HomeData:
41        return await self._get_home_data_version(user_data, "")
42
43    async def get_home_data_v2(self, user_data: UserData) -> HomeData:
44        """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums."""
45        return await self._get_home_data_version(user_data, "/v2")
46
47    async def get_home_data_v3(self, user_data: UserData) -> HomeData:
48        """This is the same as get_home_data, but uses a different endpoint and includes non-robotic vacuums."""
49        return await self._get_home_data_version(user_data, "/v3")
50
51    async def get_rooms(self, user_data: UserData, home_id: int | None = None) -> list[HomeDataRoom]:
52        if user_data.rriot is None:
53            raise RoborockException("rriot is none")
54        if home_id is None:
55            home_id = await self._get_home_id(user_data)
56        path = f"/user/homes/{home_id}/rooms"
57        room_request = self._rriot_request(user_data, path)
58        room_response = await room_request.request("get", path)
59        if not room_response.get("success"):
60            raise RoborockException(room_response)
61        rooms = room_response.get("result")
62        if isinstance(rooms, list):
63            return [HomeDataRoom.from_dict(room) for room in rooms]
64        raise RoborockException("home_response result was an unexpected type")
65
66    async def get_shared_device_rooms(self, user_data: UserData, device_id: str) -> list[HomeDataRoom]:
67        """Fetch room names for a shared (received) device."""
68        path = f"/user/deviceshare/query/{device_id}/rooms"
69        room_request = self._rriot_request(user_data, path)
70        room_response = await room_request.request("get", path)
71        if not room_response.get("success"):
72            raise RoborockException(room_response)
73        rooms = room_response.get("result")
74        if isinstance(rooms, list):
75            output_list = []
76            for room in rooms:
77                normalized_room = room
78                if isinstance(room, dict) and "id" not in room and "roomId" in room:
79                    normalized_room = {**room, "id": room["roomId"]}
80                output_list.append(HomeDataRoom.from_dict(normalized_room))
81            return output_list
82        raise RoborockException("get_shared_device_rooms result was an unexpected type")

40@dataclass
41class IotLoginInfo:
42    """Information about the login to the iot server."""
43
44    base_url: str
45    country_code: str
46    country: str

11class JobsApi(RoborockApiBase):
12    """Cloud-side cleaning schedules and one-time jobs."""
13
14    async def get_schedules(self, user_data: UserData, device_id: str) -> list[HomeDataSchedule]:
15        path = f"/user/devices/{device_id}/jobs"
16        schedules_request = self._rriot_request(user_data, path)
17        schedules_response = await schedules_request.request("get", path)
18        if not schedules_response.get("success"):
19            raise RoborockException(schedules_response)
20        schedules = schedules_response.get("result")
21        if isinstance(schedules, list):
22            return [HomeDataSchedule.from_dict(schedule) for schedule in schedules]
23        raise RoborockException(f"schedule_response result was an unexpected type: {schedules}")
24
25    async def create_job(self, user_data: UserData, device_id: str, job: dict) -> dict:
26        """Create a /jobs entry (schedule or one-time room clean) on a B01 device.
27
28        Body-bearing writes must sign the request body in the Hawk payload slot and send those same
29        compact bytes via ``data=``; ``json=`` would re-serialize with spaces and break the MAC.
30        """
31        path = f"/user/devices/{device_id}/jobs"
32        job_request = self._rriot_request(user_data, path, body=job, extra_headers={"Content-Type": "application/json"})
33        response = await job_request.request("post", path, data=_compact_json(job).encode())
34        if not response.get("success"):
35            raise RoborockException(response)
36        return response
37
38    async def update_job(self, user_data: UserData, device_id: str, job: HomeDataSchedule) -> None:
39        """Replace an existing cloud job with the full supplied schedule.
40
41        Pass a schedule returned by :meth:`get_schedules` with the desired changes.
42        The job ID identifies the URL; all other fields are sent as the replacement body.
43        """
44        path = f"/user/devices/{device_id}/jobs/{job.id}"
45        body = job.as_dict(exclude={"id"})
46        job_request = self._rriot_request(
47            user_data, path, body=body, extra_headers={"Content-Type": "application/json"}
48        )
49        response = await job_request.request("put", path, data=_compact_json(body).encode())
50        if not response or not response.get("success"):
51            raise RoborockException(response)

10class OtaApi(RoborockApiBase):
11    """Firmware update discovery and control."""
12
13    async def get_firmware_info(self, user_data: UserData, device_id: str) -> FirmwareInfo:
14        """Get firmware/OTA info for a device (latest version + updatable flag)."""
15        path = f"/ota/firmware/{device_id}/updatev2"
16        params = {"lang": "en"}
17        firmware_request = self._rriot_request(user_data, path, params=params)
18        firmware_response = await firmware_request.request("get", path, params=params)
19        if not firmware_response or not firmware_response.get("success"):
20            raise RoborockException(firmware_response)
21        return FirmwareInfo.from_dict(firmware_response.get("result") or {})
22
23    async def start_firmware_update(self, user_data: UserData, device_id: str) -> None:
24        """Trigger the (irreversible) firmware update for a device.
25
26        The device downloads and flashes the latest firmware reported by
27        :meth:`get_firmware_info`.
28        """
29        path = f"/ota/device/{device_id}/upgrade"
30        upgrade_request = self._rriot_request(user_data, path)
31        upgrade_response = await upgrade_request.request("post", path)
32        if not upgrade_response or not upgrade_response.get("success"):
33            raise RoborockException(upgrade_response)
34
35    async def set_silent_ota(self, user_data: UserData, device_id: str, enabled: bool) -> None:
36        """Enable/disable automatic (silent) firmware updates for a device.
37
38        The current value is reported as ``silent_ota_switch`` on the home-data device.
39        """
40        path = f"/user/devices/{device_id}"
41        formdata = {"silentOtaSwitch": "true" if enabled else "false"}
42        silent_ota_request = self._rriot_request(user_data, path, formdata=formdata)
43        silent_ota_response = await silent_ota_request.request("put", path, data=formdata)
44        if not silent_ota_response or not silent_ota_response.get("success"):
45            raise RoborockException(silent_ota_response)