commit cf481e5f13ec685967b8d15309fa3c2cb57c621a Author: Andre Basche Date: Mon Feb 13 01:41:38 2023 +0100 Init commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6b7bc8d --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +.idea/ +venv/ +__pycache__/ +dist/ +**/*.egg-info/ +test.py diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..318db18 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2023 Andre Basche + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..fdb991f --- /dev/null +++ b/README.md @@ -0,0 +1 @@ +# pyhOn \ No newline at end of file diff --git a/pyhon/__init__.py b/pyhon/__init__.py new file mode 100644 index 0000000..17e63aa --- /dev/null +++ b/pyhon/__init__.py @@ -0,0 +1 @@ +from .api import HonConnection \ No newline at end of file diff --git a/pyhon/api.py b/pyhon/api.py new file mode 100644 index 0000000..d6172df --- /dev/null +++ b/pyhon/api.py @@ -0,0 +1,130 @@ +import json +import logging +import secrets +from datetime import datetime +from typing import List + +import aiohttp as aiohttp + +import const +from auth import HonAuth +from device import HonDevice + +_LOGGER = logging.getLogger() + + +class HonConnection: + def __init__(self, email, password) -> None: + super().__init__() + self._email = email + self._password = password + self._request_headers = {"Content-Type": "application/json"} + self._session = None + self._devices = [] + self._mobile_id = secrets.token_hex(8) + + async def __aenter__(self): + self._session = aiohttp.ClientSession() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self._session.close() + + @property + def devices(self) -> List[HonDevice]: + return self._devices + + @property + async def _headers(self): + if "cognito-token" not in self._request_headers or "id-token" not in self._request_headers: + auth = HonAuth() + if await auth.authorize(self._email, self._password, self._mobile_id): + self._request_headers["cognito-token"] = auth.cognito_token + self._request_headers["id-token"] = auth.id_token + else: + raise PermissionError("Can't Login") + return self._request_headers + + async def setup(self): + async with aiohttp.ClientSession() as session: + async with session.get(f"{const.API_URL}/commands/v1/appliance", + headers=await self._headers) as resp: + try: + appliances = (await resp.json())["payload"]["appliances"] + self._devices = [HonDevice(self, appliance) for appliance in appliances] + except json.JSONDecodeError: + _LOGGER.error("No JSON Data after GET: %s", await resp.text()) + return False + return True + + async def load_commands(self, device: HonDevice): + params = { + "applianceType": device.appliance_type_name, + "code": device.code, + "applianceModelId": device.appliance_model_id, + "firmwareId": "41", + "macAddress": device.mac_address, + "fwVersion": device.fw_version, + "os": const.OS, + "appVersion": const.APP_VERSION, + "series": device.series, + } + url = f"{const.API_URL}/commands/v1/retrieve" + async with self._session.get(url, params=params, headers=await self._headers) as response: + result = (await response.json()).get("payload", {}) + if not result or result.pop("resultCode") != "0": + return {} + return result + + async def load_attributes(self, device: HonDevice): + params = { + "macAddress": device.mac_address, + "applianceType": device.appliance_type_name, + "category": "CYCLE" + } + url = f"{const.API_URL}/commands/v1/context" + async with self._session.get(url, params=params, headers=await self._headers) as response: + return (await response.json()).get("payload", {}) + + async def load_statistics(self, device: HonDevice): + params = { + "macAddress": device.mac_address, + "applianceType": device.appliance_type_name + } + url = f"{const.API_URL}/commands/v1/statistics" + async with self._session.get(url, params=params, headers=await self._headers) as response: + return (await response.json()).get("payload", {}) + + async def send_command(self, device, command, parameters, ancillary_parameters): + now = datetime.utcnow().isoformat() + data = { + "macAddress": device.mac_address, + "timestamp": f"{now[:-3]}Z", + "commandName": command, + "transactionId": f"{device.mac_address}_{now[:-3]}Z", + "applianceOptions": device.commands_options, + "device": { + "mobileId": self._mobile_id, + "mobileOs": const.OS, + "osVersion": const.OS_VERSION, + "appVersion": const.APP_VERSION, + "deviceModel": const.DEVICE_MODEL + }, + "attributes": { + "channel": "mobileApp", + "origin": "standardProgram", + "energyLabel": "0" + }, + "ancillaryParameters": ancillary_parameters, + "parameters": parameters, + "applianceType": device.appliance_type_name + } + url = f"{const.API_URL}/commands/v1/send" + async with self._session.post(url, headers=await self._headers, json=data) as resp: + try: + json_data = await resp.json() + except json.JSONDecodeError: + return False + if json_data["payload"]["resultCode"] == "0": + return True + return False diff --git a/pyhon/auth.py b/pyhon/auth.py new file mode 100644 index 0000000..bfc6b0a --- /dev/null +++ b/pyhon/auth.py @@ -0,0 +1,128 @@ +import json +import logging +import re +import secrets +import urllib +from urllib import parse + +import aiohttp as aiohttp + +import const + +_LOGGER = logging.getLogger() + + +class HonAuth: + def __init__(self) -> None: + self._framework = "" + self._cognito_token = "" + self._id_token = "" + + @property + def cognito_token(self): + return self._cognito_token + + @property + def id_token(self): + return self._id_token + + async def _get_frontdoor_url(self, session, email, password): + data = { + "message": { + "actions": [ + { + "id": "79;a", + "descriptor": "apex://LightningLoginCustomController/ACTION$login", + "callingDescriptor": "markup://c:loginForm", + "params": { + "username": email, + "password": password, + "startUrl": "" + } + } + ] + }, + "aura.context": { + "mode": "PROD", + "fwuid": self._framework, + "app": "siteforce:loginApp2", + "loaded": {"APPLICATION@markup://siteforce:loginApp2": "YtNc5oyHTOvavSB9Q4rtag"}, + "dn": [], + "globals": {}, + "uad": False}, + "aura.pageURI": f"SmartHome/s/login/?language={const.LANGUAGE}", + "aura.token": None} + + params = {"r": 3, "other.LightningLoginCustom.login": 1} + async with session.post( + const.AUTH_API + "/s/sfsites/aura", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + data="&".join(f"{k}={json.dumps(v)}" for k, v in data.items()), + params=params + ) as response: + if response.status != 200: + _LOGGER.error("Unable to connect to the login service: %s\n%s", response.status, await response.text()) + return "" + try: + text = await response.text() + return (await response.json())["events"][0]["attributes"]["values"]["url"] + except json.JSONDecodeError: + if framework := re.findall('clientOutOfSync.*?Expected: ([\\w-]+?) Actual: (.*?)"', text): + self._framework, actual = framework[0] + _LOGGER.warning('Framework update from "%s" to "%s"', self._framework, actual) + return await self._get_frontdoor_url(session, email, password) + _LOGGER.error("Unable to retrieve the frontdoor URL. Message: " + text) + return "" + + async def _prepare_login(self, session, email, password): + if not (frontdoor_url := await self._get_frontdoor_url(session, email, password)): + return False + + async with session.get(frontdoor_url) as resp: + if resp.status != 200: + _LOGGER.error("Unable to connect to the login service: %s", resp.status) + return False + + params = {"retURL": "/SmartHome/apex/CustomCommunitiesLanding"} + async with session.get(f"{const.AUTH_API}/apex/ProgressiveLogin", params=params) as resp: + if resp.status != 200: + _LOGGER.error("Unable to connect to the login service: %s", resp.status) + return False + return True + + async def _login(self, session): + nonce = secrets.token_hex(16) + nonce = f"{nonce[:8]}-{nonce[8:12]}-{nonce[12:16]}-{nonce[16:20]}-{nonce[20:]}" + params = { + "response_type": "token+id_token", + "client_id": const.CLIENT_ID, + "redirect_uri": urllib.parse.quote(f"{const.APP}://mobilesdk/detect/oauth/done"), + "display": "touch", + "scope": "api openid refresh_token web", + "nonce": nonce + } + params = "&".join([f"{k}={v}" for k, v in params.items()]) + async with session.get(f"{const.AUTH_API}/services/oauth2/authorize?{params}") as resp: + if id_token := re.findall("id_token=(.*?)&", await resp.text()): + self._id_token = id_token[0] + return True + return False + + async def authorize(self, email, password, mobile_id): + async with aiohttp.ClientSession() as session: + if not await self._prepare_login(session, email, password): + return False + if not await self._login(session): + return False + + post_headers = {"Content-Type": "application/json", "id-token": self._id_token} + data = {"appVersion": const.APP_VERSION, "mobileId": mobile_id, "osVersion": const.OS_VERSION, + "os": const.OS, "deviceModel": const.DEVICE_MODEL} + async with session.post(f"{const.API_URL}/auth/v1/login", headers=post_headers, json=data) as resp: + try: + json_data = await resp.json() + except json.JSONDecodeError: + _LOGGER.error("No JSON Data after POST: %s", await resp.text()) + return False + self._cognito_token = json_data["cognitoUser"]["Token"] + return True diff --git a/pyhon/commands.py b/pyhon/commands.py new file mode 100644 index 0000000..ed7c515 --- /dev/null +++ b/pyhon/commands.py @@ -0,0 +1,42 @@ +from parameter import HonParameterFixed, HonParameterEnum, HonParameterRange + + +class HonCommand: + def __init__(self, name, attributes, connector, device, multi=None): + self._connector = connector + self._device = device + self._name = name + self._description = attributes.get("description", "") + self._parameters = self._create_parameters(attributes.get("parameters", {})) + self._ancillary_parameters = self._create_parameters(attributes.get("ancillaryParameters", {})) + self._multi = multi + + def _create_parameters(self, parameters): + result = {} + for parameter, attributes in parameters.items(): + match attributes.get("typology"): + case "range": + result[parameter] = HonParameterRange(parameter, attributes) + case "enum": + result[parameter] = HonParameterEnum(parameter, attributes) + case "fixed": + result[parameter] = HonParameterFixed(parameter, attributes) + return result + + @property + def parameters(self): + return {key: parameter.value for key, parameter in self._parameters.items()} + + @property + def ancillary_parameters(self): + return {key: parameter.value for key, parameter in self._ancillary_parameters.items()} + + async def send(self): + return await self._connector.send_command(self._device, self._name, self.parameters, + self.ancillary_parameters) + + async def get_programs(self): + return self._multi + + async def set_program(self, program): + self._device.commands[self._name] = self._multi[program] diff --git a/pyhon/const.py b/pyhon/const.py new file mode 100644 index 0000000..f984c50 --- /dev/null +++ b/pyhon/const.py @@ -0,0 +1,10 @@ +AUTH_API = "https://he-accounts.force.com/SmartHome" +API_URL = "https://api-iot.he.services" +APP = "hon" +# All seen id's (different accounts, different devices) are the same, so I guess this hash is static +CLIENT_ID = "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9.HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6" +APP_VERSION = "1.51.9" +OS_VERSION = 31 +OS = "android" +DEVICE_MODEL = "exynos9820" +LANGUAGE = "en" diff --git a/pyhon/device.py b/pyhon/device.py new file mode 100644 index 0000000..0b59ea2 --- /dev/null +++ b/pyhon/device.py @@ -0,0 +1,149 @@ +from commands import HonCommand + + +class HonDevice: + def __init__(self, connector, appliance): + self._appliance = appliance + self._connector = connector + self._appliance_model = {} + + self._commands = {} + self._statistics = {} + self._attributs = {} + + @property + def appliance_id(self): + return self._appliance.get("applianceId") + + @property + def appliance_model_id(self): + return self._appliance.get("applianceModelId") + + @property + def appliance_status(self): + return self._appliance.get("applianceStatus") + + @property + def appliance_type_id(self): + return self._appliance.get("applianceTypeId") + + @property + def appliance_type_name(self): + return self._appliance.get("applianceTypeName") + + @property + def brand(self): + return self._appliance.get("brand") + + @property + def code(self): + return self._appliance.get("code") + + @property + def connectivity(self): + return self._appliance.get("connectivity") + + @property + def coords(self): + return self._appliance.get("coords") + + @property + def eeprom_id(self): + return self._appliance.get("eepromId") + + @property + def eeprom_name(self): + return self._appliance.get("eepromName") + + @property + def enrollment_date(self): + return self._appliance.get("enrollmentDate") + + @property + def first_enrollment(self): + return self._appliance.get("firstEnrollment") + + @property + def first_enrollment_tbc(self): + return self._appliance.get("firstEnrollmentTBC") + + @property + def fw_version(self): + return self._appliance.get("fwVersion") + + @property + def id(self): + return self._appliance.get("id") + + @property + def last_update(self): + return self._appliance.get("lastUpdate") + + @property + def mac_address(self): + return self._appliance.get("macAddress") + + @property + def model_name(self): + return self._appliance.get("modelName") + + @property + def nick_name(self): + return self._appliance.get("nickName") + + @property + def purchase_date(self): + return self._appliance.get("purchaseDate") + + @property + def serial_number(self): + return self._appliance.get("serialNumber") + + @property + def series(self): + return self._appliance.get("series") + + @property + def water_hard(self): + return self._appliance.get("waterHard") + + @property + def commands_options(self): + return self._appliance_model.get("options") + + @property + def commands(self): + return self._commands + + @property + def attributes(self): + return self._attributs + + @property + def statistics(self): + return self._statistics + + async def load_commands(self): + raw = await self._connector.load_commands(self) + self._appliance_model = raw.pop("applianceModel") + for item in ["settings", "options", "dictionaryId"]: + raw.pop(item) + commands = {} + for command, attr in raw.items(): + if "parameters" in attr: + commands[command] = HonCommand(command, attr, self._connector, self) + elif "parameters" in attr[list(attr)[0]]: + multi = {} + for category, attr2 in attr.items(): + cmd = HonCommand(command, attr2, self._connector, self, multi=multi) + multi[category] = cmd + commands[command] = cmd + self._commands = commands + + async def load_attributes(self): + data = await self._connector.load_attributes(self) + for name, values in data.get("shadow").get("parameters").items(): + self._attributs[name] = values["parNewVal"] + + async def load_statistics(self): + self._statistics = await self._connector.load_statistics(self) diff --git a/pyhon/parameter.py b/pyhon/parameter.py new file mode 100644 index 0000000..589aca9 --- /dev/null +++ b/pyhon/parameter.py @@ -0,0 +1,35 @@ +class HonParameter: + def __init__(self, key, attributes): + self._key = key + self._category = attributes.get("category") + self._typology = attributes.get("typology") + self._mandatory = attributes.get("mandatory") + self._value = "" + + @property + def value(self): + return self._value if self._value is not None else "0" + + +class HonParameterFixed(HonParameter): + def __init__(self, key, attributes): + super().__init__(key, attributes) + self._value = attributes["fixedValue"] + + +class HonParameterRange(HonParameter): + def __init__(self, key, attributes): + super().__init__(key, attributes) + self._value = attributes.get("defaultValue") + self._default = attributes.get("defaultValue") + self._min = attributes["minimumValue"] + self._max = attributes["maximumValue"] + self._step = attributes["incrementValue"] + + +class HonParameterEnum(HonParameter): + def __init__(self, key, attributes): + super().__init__(key, attributes) + self._value = attributes.get("defaultValue", "0") + self._default = attributes["defaultValue"] + self._values = attributes["enumValues"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ee4ba4f --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +aiohttp diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..c49de9f --- /dev/null +++ b/setup.py @@ -0,0 +1,23 @@ +#!/usr/bin/env python3 + +from setuptools import setup + +with open("README.md", "r") as f: + long_description = f.read() + +setup( + name="pyhon", + version="0.0.1", + author="Andre Basche", + description="Control Haier devices with pyhon", + long_description=long_description, + long_description_content_type='text/markdown', + url="https://github.com/Andre0512/pyhon", + license="MIT", + platforms="any", + package_dir={"": "pyhon"}, + packages=[""], + include_package_data=True, + python_requires=">=3.10", + install_requires=["aiohttp"] +)