|
| 1 | +import os |
| 2 | +import time |
| 3 | +import base64 |
| 4 | +import hashlib |
| 5 | +import hmac |
| 6 | +import json |
| 7 | +from miaccount import MiAccount |
| 8 | + |
| 9 | +# REGIONS = ['cn', 'de', 'i2', 'ru', 'sg', 'us'] |
| 10 | + |
| 11 | + |
| 12 | +class MiIOService: |
| 13 | + |
| 14 | + def __init__(self, account: MiAccount, region=None): |
| 15 | + self.account = account |
| 16 | + self.server = 'https://' + ('' if region is None or region == 'cn' else region + '.') + 'api.io.mi.com/app' |
| 17 | + |
| 18 | + async def miio_request(self, uri, data): |
| 19 | + def prepare_data(token, cookies): |
| 20 | + cookies['PassportDeviceId'] = token['deviceId'] |
| 21 | + return MiIOService.sign_data(uri, data, token['xiaomiio'][0]) |
| 22 | + headers = {'User-Agent': 'iOS-14.4-6.0.103-iPhone12,3--D7744744F7AF32F0544445285880DD63E47D9BE9-8816080-84A3F44E137B71AE-iPhone', 'x-xiaomi-protocal-flag-cli': 'PROTOCAL-HTTP2'} |
| 23 | + resp = await self.account.mi_request('xiaomiio', self.server + uri, prepare_data, headers) |
| 24 | + if 'result' not in resp: |
| 25 | + raise Exception(f"Error {uri}: {resp}") |
| 26 | + return resp['result'] |
| 27 | + |
| 28 | + async def home_request(self, did, method, params): |
| 29 | + return await self.miio_request('/home/rpc/' + did, {'id': 1, 'method': method, "accessKey": "IOS00026747c5acafc2", 'params': params}) |
| 30 | + |
| 31 | + async def home_get_props(self, did, props): |
| 32 | + return await self.home_request(did, 'get_prop', props) |
| 33 | + |
| 34 | + async def home_set_props(self, did, props): |
| 35 | + return [await self.home_set_prop(did, i[0], i[1]) for i in props] |
| 36 | + |
| 37 | + async def home_get_prop(self, did, prop): |
| 38 | + return (await self.home_get_props(did, [prop]))[0] |
| 39 | + |
| 40 | + async def home_set_prop(self, did, prop, value): |
| 41 | + result = (await self.home_request(did, 'set_' + prop, value if isinstance(value, list) else [value]))[0] |
| 42 | + return 0 if result == 'ok' else result |
| 43 | + |
| 44 | + async def miot_request(self, cmd, params): |
| 45 | + return await self.miio_request('/miotspec/' + cmd, {'params': params}) |
| 46 | + |
| 47 | + async def miot_get_props(self, did, iids): |
| 48 | + params = [{'did': did, 'siid': i[0], 'piid': i[1]} for i in iids] |
| 49 | + result = await self.miot_request('prop/get', params) |
| 50 | + return [it.get('value') if it.get('code') == 0 else None for it in result] |
| 51 | + |
| 52 | + async def miot_set_props(self, did, props): |
| 53 | + params = [{'did': did, 'siid': i[0], 'piid': i[1], 'value': i[2]} for i in props] |
| 54 | + result = await self.miot_request('prop/set', params) |
| 55 | + return [it.get('code', -1) for it in result] |
| 56 | + |
| 57 | + async def miot_get_prop(self, did, iid): |
| 58 | + return (await self.miot_get_props(did, [iid]))[0] |
| 59 | + |
| 60 | + async def miot_set_prop(self, did, iid, value): |
| 61 | + return (await self.miot_set_props(did, [(iid[0], iid[1], value)]))[0] |
| 62 | + |
| 63 | + async def miot_action(self, did, iid, args=[]): |
| 64 | + result = await self.miot_request('action', {'did': did, 'siid': iid[0], 'aiid': iid[1], 'in': args}) |
| 65 | + return result.get('code', -1) |
| 66 | + |
| 67 | + async def device_list(self, name=None, getVirtualModel=False, getHuamiDevices=0): |
| 68 | + result = await self.miio_request('/home/device_list', {'getVirtualModel': bool(getVirtualModel), 'getHuamiDevices': int(getHuamiDevices)}) |
| 69 | + result = result['list'] |
| 70 | + return result if name == 'full' else [{'name': i['name'], 'model': i['model'], 'did': i['did'], 'token': i['token']} for i in result if not name or name in i['name']] |
| 71 | + |
| 72 | + async def miot_spec(self, type=None, format=None): |
| 73 | + if not type or not type.startswith('urn'): |
| 74 | + def get_spec(all): |
| 75 | + if not type: |
| 76 | + return all |
| 77 | + ret = {} |
| 78 | + for m, t in all.items(): |
| 79 | + if type == m: |
| 80 | + return {m: t} |
| 81 | + elif type in m: |
| 82 | + ret[m] = t |
| 83 | + return ret |
| 84 | + import tempfile |
| 85 | + path = os.path.join(tempfile.gettempdir(), 'miservice_miot_specs.json') |
| 86 | + try: |
| 87 | + with open(path) as f: |
| 88 | + result = get_spec(json.load(f)) |
| 89 | + except: |
| 90 | + result = None |
| 91 | + if not result: |
| 92 | + async with self.account.session.get('http://miot-spec.org/miot-spec-v2/instances?status=all') as r: |
| 93 | + all = {i['model']: i['type'] for i in (await r.json())['instances']} |
| 94 | + with open(path, 'w') as f: |
| 95 | + json.dump(all, f) |
| 96 | + result = get_spec(all) |
| 97 | + if len(result) != 1: |
| 98 | + return result |
| 99 | + type = list(result.values())[0] |
| 100 | + |
| 101 | + url = 'http://miot-spec.org/miot-spec-v2/instance?type=' + type |
| 102 | + async with self.account.session.get(url) as r: |
| 103 | + result = await r.json() |
| 104 | + |
| 105 | + def parse_desc(node): |
| 106 | + desc = node['description'] |
| 107 | + # pos = desc.find(' ') |
| 108 | + # if pos != -1: |
| 109 | + # return (desc[:pos], ' # ' + desc[pos + 2:]) |
| 110 | + name = '' |
| 111 | + for i in range(len(desc)): |
| 112 | + d = desc[i] |
| 113 | + if d in '-—{「[【(（<《': |
| 114 | + return (name, ' # ' + desc[i:]) |
| 115 | + name += '_' if d == ' ' else d |
| 116 | + return (name, '') |
| 117 | + |
| 118 | + def make_line(siid, iid, desc, comment, readable=False): |
| 119 | + value = f"({siid}, {iid})" if format == 'python' else iid |
| 120 | + return f" {'' if readable else '_'}{desc} = {value}{comment}\n" |
| 121 | + |
| 122 | + if format != 'json': |
| 123 | + STR_HEAD, STR_SRV, STR_VALUE = ('from enum import Enum\n\n', '\nclass {}(tuple, Enum):\n', '\nclass {}(int, Enum):\n') if format == 'python' else ('', '{} = {}\n', '{}\n') |
| 124 | + text = '# Generated by https://github.com/Yonsm/MiService\n# ' + url + '\n\n' + STR_HEAD |
| 125 | + svcs = [] |
| 126 | + vals = [] |
| 127 | + |
| 128 | + for s in result['services']: |
| 129 | + siid = s['iid'] |
| 130 | + svc = s['description'].replace(' ', '_') |
| 131 | + svcs.append(svc) |
| 132 | + text += STR_SRV.format(svc, siid) |
| 133 | + for p in s.get('properties', []): |
| 134 | + name, comment = parse_desc(p) |
| 135 | + access = p['access'] |
| 136 | + |
| 137 | + comment += ''.join([' # ' + k for k, v in [(p['format'], 'string'), (''.join([a[0] for a in access]), 'r')] if k and k != v]) |
| 138 | + text += make_line(siid, p['iid'], name, comment, 'read' in access) |
| 139 | + if 'value-range' in p: |
| 140 | + valuer = p['value-range'] |
| 141 | + length = min(3, len(valuer)) |
| 142 | + values = {['MIN', 'MAX', 'STEP'][i]: valuer[i] for i in range(length) if i != 2 or valuer[i] != 1} |
| 143 | + elif 'value-list' in p: |
| 144 | + values = {i['description'].replace(' ', '_') if i['description'] else str(i['value']): i['value'] for i in p['value-list']} |
| 145 | + else: |
| 146 | + continue |
| 147 | + vals.append((svc + '_' + name, values)) |
| 148 | + if 'actions' in s: |
| 149 | + text += '\n' |
| 150 | + for a in s['actions']: |
| 151 | + name, comment = parse_desc(a) |
| 152 | + comment += ''.join([f" # {io}={a[io]}" for io in ['in', 'out'] if a[io]]) |
| 153 | + text += make_line(siid, a['iid'], name, comment) |
| 154 | + text += '\n' |
| 155 | + for name, values in vals: |
| 156 | + text += STR_VALUE.format(name) |
| 157 | + for k, v in values.items(): |
| 158 | + text += f" {'_' + k if k.isdigit() else k} = {v}\n" |
| 159 | + text += '\n' |
| 160 | + if format == 'python': |
| 161 | + text += '\nALL_SVCS = (' + ', '.join(svcs) + ')\n' |
| 162 | + result = text |
| 163 | + return result |
| 164 | + |
| 165 | + @staticmethod |
| 166 | + def miot_decode(ssecurity, nonce, data, gzip=False): |
| 167 | + from Crypto.Cipher import ARC4 |
| 168 | + r = ARC4.new(base64.b64decode(MiIOService.sign_nonce(ssecurity, nonce))) |
| 169 | + r.encrypt(bytes(1024)) |
| 170 | + decrypted = r.encrypt(base64.b64decode(data)) |
| 171 | + if gzip: |
| 172 | + try: |
| 173 | + from io import BytesIO |
| 174 | + from gzip import GzipFile |
| 175 | + compressed = BytesIO() |
| 176 | + compressed.write(decrypted) |
| 177 | + compressed.seek(0) |
| 178 | + decrypted = GzipFile(fileobj=compressed, mode='rb').read() |
| 179 | + except: |
| 180 | + pass |
| 181 | + return json.loads(decrypted.decode()) |
| 182 | + |
| 183 | + @staticmethod |
| 184 | + def sign_nonce(ssecurity, nonce): |
| 185 | + m = hashlib.sha256() |
| 186 | + m.update(base64.b64decode(ssecurity)) |
| 187 | + m.update(base64.b64decode(nonce)) |
| 188 | + return base64.b64encode(m.digest()).decode() |
| 189 | + |
| 190 | + @staticmethod |
| 191 | + def sign_data(uri, data, ssecurity): |
| 192 | + if not isinstance(data, str): |
| 193 | + data = json.dumps(data) |
| 194 | + nonce = base64.b64encode(os.urandom(8) + int(time.time() / 60).to_bytes(4, 'big')).decode() |
| 195 | + snonce = MiIOService.sign_nonce(ssecurity, nonce) |
| 196 | + msg = '&'.join([uri, snonce, nonce, 'data=' + data]) |
| 197 | + sign = hmac.new(key=base64.b64decode(snonce), msg=msg.encode(), digestmod=hashlib.sha256).digest() |
| 198 | + return {'_nonce': nonce, 'data': data, 'signature': base64.b64encode(sign).decode()} |
0 commit comments