diff --git a/custom_components/xiaomi_cloud_map_extractor/vacuum_platforms/xiaomi_cloud_connector.py b/custom_components/xiaomi_cloud_map_extractor/vacuum_platforms/xiaomi_cloud_connector.py index 869efd9..b8475bb 100644 --- a/custom_components/xiaomi_cloud_map_extractor/vacuum_platforms/xiaomi_cloud_connector.py +++ b/custom_components/xiaomi_cloud_map_extractor/vacuum_platforms/xiaomi_cloud_connector.py @@ -6,6 +6,7 @@ import os import random import time +from typing import Any, Dict, NamedTuple, Optional, Tuple from Crypto.Cipher import ARC4 import requests @@ -15,6 +16,21 @@ CONF_AVAILABLE_COUNTRIES = ["cn", "de", "us", "ru", "tw", "sg", "in", "i2"] +class XiaomiHome(NamedTuple): + homeid: int + owner: int + + +class XiaomiDeviceInfo(NamedTuple): + device_id: str + name: str + model: str + token: str + country: str + home_id: int + user_id: int + + # noinspection PyBroadException class XiaomiCloudConnector: @@ -131,31 +147,78 @@ def get_raw_map_data(self, map_url: str | None) -> bytes | None: return response.content return None - def get_device_details(self, token: str, - country: str) -> tuple[str | None, str | None, str | None, str | None]: - countries_to_check = CONF_AVAILABLE_COUNTRIES - if country is not None: - countries_to_check = [country] - for country in countries_to_check: - devices = self.get_devices(country) - if devices is None: - continue - found = list(filter(lambda d: str(d["token"]).casefold() == str(token).casefold(), - devices["result"]["list"])) - if len(found) > 0: - user_id = found[0]["uid"] - device_id = found[0]["did"] - model = found[0]["model"] - self.country = country - return country, user_id, device_id, model - return None, None, None, None + def get_homes_iter(self, country: str): + url = self.get_api_url(country) + "/v2/homeroom/gethome" + params = { + "data": json.dumps( + { + "fg": True, + "fetch_share": True, + "fetch_share_dev": True, + "limit": 300, + "app_ver": 7, + } + ) + } + + if (response := self.execute_api_call_encrypted(url, params)) is None: + return None + + if homelist := response["result"]["homelist"]: + yield from (XiaomiHome(int(home["id"]), home["uid"]) for home in homelist) - def get_devices(self, country: str) -> any: - url = self.get_api_url(country) + "/home/device_list" + if homelist := response["result"]["share_home_list"]: + yield from (XiaomiHome(int(home["id"]), home["uid"]) for home in homelist) + + def get_devices_from_home_iter(self, country: str, home_id: int, owner_id: int): + url = self.get_api_url(country) + "/v2/home/home_device_list" params = { - "data": '{"getVirtualModel":false,"getHuamiDevices":0}' + "data": json.dumps( + { + "home_id": home_id, + "home_owner": owner_id, + "limit": 200, + "get_split_device": True, + "support_smart_home": True, + } + ) } - return self.execute_api_call_encrypted(url, params) + if (response := self.execute_api_call_encrypted(url, params)) is None: + return + + if (raw_devices := response["result"]["device_info"]) is None: + return + + yield from ( + XiaomiDeviceInfo( + device_id=device["did"], + name=device["name"], + model=device["model"], + token=device["token"], + country=country, + user_id=owner_id, + home_id=home_id, + ) + for device in raw_devices + ) + + def get_devices_iter(self, country: Optional[str] = None): + countries_to_check = CONF_AVAILABLE_COUNTRIES if country is None else [country] + for _country in countries_to_check: + homes = self.get_homes_iter(_country) + for home in homes: + devices = self.get_devices_from_home_iter( + _country, home.homeid, home.owner + ) + yield from devices + + def get_device_details(self, token: str, country: Optional[str] = None): + devices = self.get_devices_iter(country) + matching_token = filter(lambda device: device.token == token, devices) + if match := next(matching_token, None): + return match.country, match.user_id, match.device_id, match.model + + return None, None, None, None def get_other_info(self, device_id: str, method: str, parameters: dict) -> any: url = self.get_api_url('sg') + "/v2/home/rpc/" + device_id