🚀 request xiaomi cloud async (#1802)

custom_components/xiaomi_miot/__init__.py

@@ -685,6 +685,8 @@ def hardware_version(self):
 
 
 class MiotDevice(MiotDeviceBase):
+    hass = None
+
     def get_properties_for_mapping(self, *, max_properties=12, did=None, mapping=None) -> list:
         if mapping is None:
             mapping = self.mapping
@@ -698,6 +700,17 @@ def get_properties_for_mapping(self, *, max_properties=12, did=None, mapping=Non
             max_properties=max_properties,
         )
 
+    async def async_get_properties_for_mapping(self, *args, **kwargs) -> list:
+        if not self.hass:
+            return self.get_properties_for_mapping(*args, **kwargs)
+
+        return await self.hass.async_add_executor_job(
+            partial(
+                self.get_properties_for_mapping,
+                *args, **kwargs,
+            )
+        )
+
 
 class BaseEntity(Entity):
     _config = None
@@ -1256,6 +1269,7 @@ def miot_device(self):
             except ValueError as exc:
                 self.logger.warning('%s: Initializing with host %s failed: %s', host, self.name_model, exc)
             if device:
+                device.hass = self.hass
                 self._device = device
         return self._device
 
@@ -1413,13 +1427,10 @@ async def async_update(self):
                             10, 9, 9, 9, 9, 9, 10, 10, 10, 10,
                         ]
                         max_properties = 10 if idx >= len(chunks) else chunks[idx]
-                    results = await self.hass.async_add_executor_job(
-                        partial(
-                            self._device.get_properties_for_mapping,
-                            max_properties=max_properties,
-                            did=self.miot_did,
-                            mapping=local_mapping,
-                        )
+                    results = await self._device.async_get_properties_for_mapping(
+                        max_properties=max_properties,
+                        did=self.miot_did,
+                        mapping=local_mapping,
                     )
                 self._local_state = True
             except (DeviceException, OSError) as exc:
@@ -1440,9 +1451,7 @@ async def async_update(self):
             updater = 'cloud'
             try:
                 mic = self.xiaomi_cloud
-                results = await self.hass.async_add_executor_job(
-                    partial(mic.get_properties_for_mapping, self.miot_did, mapping)
-                )
+                results = await mic.async_get_properties_for_mapping(self.miot_did, mapping)
                 if self.custom_config_bool('check_lan'):
                     if self.miot_device:
                         await self.hass.async_add_executor_job(self.miot_device.info)
@@ -1788,7 +1797,7 @@ async def async_update_micloud_statistics(self, lst):
         if attrs:
             await self.async_update_attrs(attrs)
 
-    def get_properties(self, mapping, update_entity=False, throw=False, **kwargs):
+    async def async_get_properties(self, mapping, update_entity=False, throw=False, **kwargs):
         results = []
         if isinstance(mapping, list):
             new_mapping = {}
@@ -1805,9 +1814,9 @@ def get_properties(self, mapping, update_entity=False, throw=False, **kwargs):
             return
         try:
             if self._local_state:
-                results = self.miot_device.get_properties_for_mapping(mapping=mapping)
+                results = await self.miot_device.async_get_properties_for_mapping(mapping=mapping)
             elif self.miot_cloud:
-                results = self.miot_cloud.get_properties_for_mapping(self.miot_did, mapping)
+                results = await self.miot_cloud.async_get_properties_for_mapping(self.miot_did, mapping)
         except (ValueError, DeviceException) as exc:
             self.logger.error(
                 '%s: Got exception while get properties: %s, mapping: %s, miio: %s',
@@ -1821,15 +1830,10 @@ def get_properties(self, mapping, update_entity=False, throw=False, **kwargs):
         self.logger.info('%s: Get miot properties: %s', self.name_model, results)
 
         if attrs and update_entity:
-            self.update_attrs(attrs, update_subs=True)
+            await self.async_update_attrs(attrs, update_subs=True)
             self.schedule_update_ha_state()
         return attrs
 
-    async def async_get_properties(self, mapping, **kwargs):
-        return await self.hass.async_add_executor_job(
-            partial(self.get_properties, mapping, **kwargs)
-        )
-
     def set_property(self, field, value):
         if isinstance(field, MiotProperty):
             siid = field.siid

custom_components/xiaomi_miot/core/xiaomi_cloud.py

@@ -1,4 +1,6 @@
 import logging
+import aiohttp
+import asyncio
 import json
 import time
 import string
@@ -16,6 +18,7 @@
     CONF_USERNAME,
 )
 from homeassistant.helpers.storage import Store
+from homeassistant.helpers.aiohttp_client import async_create_clientsession
 from homeassistant.components import persistent_notification
 
 from .const import DOMAIN, CONF_XIAOMI_CLOUD
@@ -51,14 +54,15 @@ def __init__(self, hass, username, password, country=None, sid=None):
         self.useragent = UA % self.client_id
         self.http_timeout = int(hass.data[DOMAIN].get('config', {}).get('http_timeout') or 10)
         self.login_times = 0
+        self.async_session = None
         self.attrs = {}
 
     @property
     def unique_id(self):
         uid = self.user_id or self.username
         return f'{uid}-{self.default_server}-{self.sid}'
 
-    def get_properties_for_mapping(self, did, mapping: dict):
+    async def async_get_properties_for_mapping(self, did, mapping: dict):
         pms = []
         rmp = {}
         for k, v in mapping.items():
@@ -68,7 +72,7 @@ def get_properties_for_mapping(self, did, mapping: dict):
             p = v.get('piid')
             pms.append({'did': str(did), 'siid': s, 'piid': p})
             rmp[f'prop.{s}.{p}'] = k
-        rls = self.get_props(pms)
+        rls = await self.async_get_props(pms)
         if not rls:
             return None
         dls = []
@@ -85,12 +89,21 @@ def get_properties_for_mapping(self, did, mapping: dict):
     def get_props(self, params=None):
         return self.request_miot_spec('prop/get', params)
 
+    async def async_get_props(self, params=None):
+        return await self.async_request_miot_spec('prop/get', params)
+
     def set_props(self, params=None):
         return self.request_miot_spec('prop/set', params)
 
+    async def async_set_props(self, params=None):
+        return await self.async_request_miot_spec('prop/set', params)
+
     def do_action(self, params=None):
         return self.request_miot_spec('action', params)
 
+    async def async_do_action(self, params=None):
+        return await self.async_request_miot_spec('action', params)
+
     def request_miot_spec(self, api, params=None):
         rdt = self.request_miot_api('miotspec/' + api, {
             'params': params or [],
@@ -100,6 +113,15 @@ def request_miot_spec(self, api, params=None):
             raise MiCloudException(json.dumps(rdt))
         return rls
 
+    async def async_request_miot_spec(self, api, params=None):
+        rdt = await self.async_request_api('miotspec/' + api, {
+            'params': params or [],
+        }) or {}
+        rls = rdt.get('result')
+        if not rls and rdt.get('code'):
+            raise MiCloudException(json.dumps(rdt))
+        return rls
+
     async def async_get_user_device_data(self, *args, **kwargs):
         return await self.hass.async_add_executor_job(
             partial(self.get_user_device_data, *args, **kwargs)
@@ -173,12 +195,49 @@ async def async_check_auth(self, notify=False):
             _LOGGER.warning('Retry login xiaomi account failed: %s', self.username)
         return False
 
-    async def async_request_api(self, *args, **kwargs):
+    async def async_request_api(self, api, data, method='POST', crypt=True, debug=True, **kwargs):
         if not self.service_token:
             await self.async_login()
-        return await self.hass.async_add_executor_job(
-            partial(self.request_miot_api, *args, **kwargs)
-        )
+
+        params = {}
+        if data is not None:
+            params['data'] = self.json_encode(data)
+        raw = kwargs.pop('raw', self.sid != 'xiaomiio')
+        rsp = None
+        try:
+            if raw:
+                rsp = await self.hass.async_add_executor_job(
+                    partial(self.request_raw, api, data, method, **kwargs)
+                )
+            elif crypt:
+                rsp = await self.async_request_rc4_api(api, params, method, **kwargs)
+            else:
+                rsp = await self.hass.async_add_executor_job(
+                    partial(self.request, self.get_api_url(api), params, **kwargs)
+                )
+            rdt = json.loads(rsp)
+            if debug:
+                _LOGGER.debug(
+                    'Request miot api: %s %s result: %s',
+                    api, data, rsp,
+                )
+            self.attrs['timeouts'] = 0
+        except asyncio.TimeoutError as exc:
+            rdt = None
+            self.attrs.setdefault('timeouts', 0)
+            self.attrs['timeouts'] += 1
+            if 5 < self.attrs['timeouts'] <= 10:
+                _LOGGER.error('Request xiaomi api: %s %s timeout, exception: %s', api, data, exc)
+        except (TypeError, ValueError):
+            rdt = None
+        code = rdt.get('code') if rdt else None
+        if code == 3:
+            self._logout()
+            _LOGGER.warning('Unauthorized while request to %s, response: %s, logged out.', api, rsp)
+        elif code or not rdt:
+            fun = _LOGGER.info if rdt else _LOGGER.warning
+            fun('Request xiaomi api: %s %s failed, response: %s', api, data, rsp)
+        return rdt
 
     def request_miot_api(self, api, data, method='POST', crypt=True, debug=True, **kwargs):
         params = {}
@@ -228,20 +287,20 @@ async def async_get_device(self, mac=None, host=None):
                 return d
         return None
 
-    def get_device_list(self):
-        rdt = self.request_miot_api('home/device_list', {
+    async def get_device_list(self):
+        rdt = await self.async_request_api('home/device_list', {
             'getVirtualModel': True,
             'getHuamiDevices': 1,
             'get_split_device': False,
             'support_smart_home': True,
         }, debug=False, timeout=60) or {}
         if rdt and 'result' in rdt:
             return rdt['result']['list']
-        _LOGGER.warning('Got xiaomi cloud devices for %s failed: %s', self.username, rdt)
+        _LOGGER.warning('Got xiaomi devices for %s failed: %s', self.username, rdt)
         return None
 
-    def get_home_devices(self):
-        rdt = self.request_miot_api('homeroom/gethome', {
+    async def get_home_devices(self):
+        rdt = await self.async_request_api('homeroom/gethome', {
             'fetch_share_dev': True,
         }, debug=False, timeout=60) or {}
         rdt = rdt.get('result') or {}
@@ -276,9 +335,9 @@ async def async_get_devices(self, renew=False, return_all=False):
                 dvs = cds
         if not dvs:
             try:
-                dvs = await self.hass.async_add_executor_job(self.get_device_list)
+                hls = await self.get_home_devices()
+                dvs = await self.get_device_list()
                 if dvs:
-                    hls = await self.hass.async_add_executor_job(self.get_home_devices)
                     if hls:
                         hds = hls.get('devices') or {}
                         dvs = [
@@ -377,7 +436,7 @@ def _logout(self):
         self.service_token = None
 
     def _login_request(self, captcha=None):
-        self._init_session()
+        self._init_session(True)
         auth = self.attrs.pop('login_data', None)
         if captcha and auth:
             auth['captcha'] = captcha
@@ -513,6 +572,7 @@ async def from_token(hass, config: dict, login=None):
         mic.user_id = str(config.get('user_id') or '')
         if a := hass.data[DOMAIN].get('sessions', {}).get(mic.unique_id):
             mic = a
+            mic.async_session = None
             if mic.password != config.get(CONF_PASSWORD):
                 mic.password = config.get(CONF_PASSWORD)
                 mic.service_token = None
@@ -562,17 +622,33 @@ async def async_stored_auth(self, uid=None, save=False):
             return cfg
         return old
 
-    def api_session(self):
+    def api_session(self, **kwargs):
         if not self.service_token or not self.user_id:
             raise MiCloudException('Cannot execute request. service token or userId missing. Make sure to login.')
 
-        session = requests.Session()
-        session.headers.update({
+        if kwargs.get('async'):
+            if not (session := self.async_session):
+                session = async_create_clientsession(
+                    self.hass,
+                    headers=self.api_headers(),
+                    cookies=self.api_cookies(),
+                )
+                self.async_session = session
+        else:
+            session = requests.Session()
+            session.headers.update(self.api_headers())
+            session.cookies.update(self.api_cookies())
+        return session
+
+    def api_headers(self):
+        return {
             'X-XIAOMI-PROTOCAL-FLAG-CLI': 'PROTOCAL-HTTP2',
             'Content-Type': 'application/x-www-form-urlencoded',
             'User-Agent': self.useragent,
-        })
-        session.cookies.update({
+        }
+
+    def api_cookies(self):
+        return {
             'userId': str(self.user_id),
             'yetAnotherServiceToken': self.service_token,
             'serviceToken': self.service_token,
@@ -581,8 +657,7 @@ def api_session(self):
             'is_daylight': str(time.daylight),
             'dst_offset': str(time.localtime().tm_isdst * 60 * 60 * 1000),
             'channel': 'MI_APP_STORE',
-        })
-        return session
+        }
 
     def request(self, url, params, **kwargs):
         self.session = self.api_session()
@@ -632,6 +707,33 @@ def request_rc4_api(self, api, params: dict, method='POST', **kwargs):
         except MiCloudException as exc:
             _LOGGER.warning('Error while decrypting response of request to %s :%s', url, exc)
 
+    async def async_request_rc4_api(self, api, params: dict, method='POST', **kwargs):
+        url = self.get_api_url(api)
+        session = self.api_session(**{'async': True})
+        timeout = aiohttp.ClientTimeout(total=kwargs.get('timeout', self.http_timeout))
+        headers = {
+            'MIOT-ENCRYPT-ALGORITHM': 'ENCRYPT-RC4',
+            'Accept-Encoding': 'identity',
+        }
+        try:
+            params = self.rc4_params(method, url, params)
+            if method == 'GET':
+                response = await session.get(url, params=params, timeout=timeout, headers=headers)
+            else:
+                response = await session.post(url, data=params, timeout=timeout, headers=headers)
+            rsp = await response.text()
+            if not rsp or 'error' in rsp or 'invalid' in rsp:
+                _LOGGER.warning('Error while executing request to %s: %s', url, rsp or response.status)
+            elif 'message' not in rsp:
+                try:
+                    signed_nonce = self.signed_nonce(params['_nonce'])
+                    rsp = MiotCloud.decrypt_data(signed_nonce, rsp)
+                except ValueError:
+                    _LOGGER.warning('Error while decrypting response of request to %s :%s', url, rsp)
+            return rsp
+        except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
+            _LOGGER.warning('Error while executing request to %s: %s', url, exc)
+
     def request_raw(self, url, data=None, method='GET', **kwargs):
         self.session = self.api_session()
         url = self.get_api_url(url)