diff --git a/pyhon/__main__.py b/pyhon/__main__.py index 41ffdf6..f8d810f 100755 --- a/pyhon/__main__.py +++ b/pyhon/__main__.py @@ -10,7 +10,7 @@ from pathlib import Path if __name__ == "__main__": sys.path.insert(0, str(Path(__file__).parent.parent)) -from pyhon import Hon, HonAPI, helper +from pyhon import Hon, HonAPI, helper, diagnose _LOGGER = logging.getLogger(__name__) @@ -24,6 +24,11 @@ def get_arguments(): keys = subparser.add_parser("keys", help="print as key format") keys.add_argument("keys", help="print as key format", action="store_true") keys.add_argument("--all", help="print also full keys", action="store_true") + export = subparser.add_parser("export") + export.add_argument("export", help="export pyhon data", action="store_true") + export.add_argument("--zip", help="create zip archive", action="store_true") + export.add_argument("--anonymous", help="anonymize data", action="store_true") + export.add_argument("directory", nargs="?", default=Path().cwd()) translate = subparser.add_parser( "translate", help="print available translation keys" ) @@ -50,17 +55,31 @@ async def translate(language, json_output=False): print(helper.pretty_print(keys)) +def get_login_data(args): + if not (user := args["user"]): + user = input("User for hOn account: ") + if not (password := args["password"]): + password = getpass("Password for hOn account: ") + return user, password + + async def main(): args = get_arguments() if language := args.get("translate"): await translate(language, json_output=args.get("json")) return - if not (user := args["user"]): - user = input("User for hOn account: ") - if not (password := args["password"]): - password = getpass("Password for hOn account: ") - async with Hon(user, password) as hon: + async with Hon(*get_login_data(args)) as hon: for device in hon.appliances: + if args.get("export"): + anonymous = args.get("anonymous", False) + path = Path(args.get("directory")) + if not args.get("zip"): + for file in await diagnose.appliance_data(device, path, anonymous): + print(f"Created {file}") + else: + file = await diagnose.zip_archive(device, path, anonymous) + print(f"Created {file}") + continue print("=" * 10, device.appliance_type, "-", device.nick_name, "=" * 10) if args.get("keys"): data = device.data.copy() @@ -78,7 +97,7 @@ async def main(): ) ) else: - print(device.diagnose(" ")) + print(diagnose.yaml_export(device)) def start(): diff --git a/pyhon/appliance.py b/pyhon/appliance.py index 23f58c9..3eaa525 100644 --- a/pyhon/appliance.py +++ b/pyhon/appliance.py @@ -1,11 +1,10 @@ import importlib -import json import logging from datetime import datetime, timedelta from pathlib import Path from typing import Optional, Dict, Any, TYPE_CHECKING -from pyhon import helper +from pyhon import diagnose from pyhon.attributes import HonAttribute from pyhon.command_loader import HonCommandLoader from pyhon.commands import HonCommand @@ -94,6 +93,10 @@ class HonAppliance: def model_name(self) -> str: return self._check_name_zone("modelName") + @property + def brand(self) -> str: + return self._check_name_zone("brand") + @property def nick_name(self) -> str: return self._check_name_zone("nickName") @@ -105,6 +108,10 @@ class HonAppliance: serial_number = self.info.get("serialNumber", "") return serial_number[:8] if len(serial_number) < 18 else serial_number[:11] + @property + def model_id(self) -> int: + return self._info.get("applianceModelId", 0) + @property def options(self): return self._appliance_model.get("options", {}) @@ -137,9 +144,9 @@ class HonAppliance: def api(self) -> Optional["HonAPI"]: return self._api - async def load_commands(self, data=None): + async def load_commands(self): command_loader = HonCommandLoader(self.api, self) - await command_loader.load_commands(data) + await command_loader.load_commands() self._commands = command_loader.commands self._additional_data = command_loader.additional_data self._appliance_model = command_loader.appliance_data @@ -206,32 +213,12 @@ class HonAppliance: } return result - def diagnose(self, whitespace=" ", command_only=False): - data = { - "attributes": self.attributes.copy(), - "appliance": self.info, - "statistics": self.statistics, - "additional_data": self._additional_data, - } - if command_only: - data.pop("attributes") - data.pop("appliance") - data.pop("statistics") - data |= {n: c.parameter_groups for n, c in self._commands.items()} - extra = {n: c.data for n, c in self._commands.items() if c.data} - if extra: - data |= {"extra_command_data": extra} - for sensible in ["PK", "SK", "serialNumber", "coords", "device"]: - data.get("appliance", {}).pop(sensible, None) - result = helper.pretty_print({"data": data}, whitespace=whitespace) - result += helper.pretty_print( - { - "commands": helper.create_command(self.commands), - "rules": helper.create_rules(self.commands), - }, - whitespace=whitespace, - ) - return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx") + @property + def diagnose(self) -> str: + return diagnose.yaml_export(self, anonymous=True) + + async def data_archive(self, path: Path) -> str: + return await diagnose.zip_archive(self, path, anonymous=True) def sync_to_params(self, command_name): command: HonCommand = self.commands.get(command_name) @@ -261,35 +248,3 @@ class HonAppliance: parameter.min = int(base_value.value) parameter.step = 1 parameter.value = base_value.value - - -class HonApplianceTest(HonAppliance): - def __init__(self, name): - super().__init__(None, {}) - self._name = name - self.load_commands() - self.load_attributes() - self._info = self._appliance_model - - def load_commands(self): - device = Path(__file__).parent / "test_data" / f"{self._name}.json" - with open(str(device)) as f: - raw = json.loads(f.read()) - self._appliance_model = raw.pop("applianceModel") - raw.pop("dictionaryId", None) - self._commands = self._get_commands(raw) - - async def update(self): - return - - @property - def nick_name(self) -> str: - return self._name - - @property - def unique_id(self) -> str: - return self._name - - @property - def mac_address(self) -> str: - return "xx-xx-xx-xx-xx-xx" diff --git a/pyhon/command_loader.py b/pyhon/command_loader.py index 7ccb3c5..e6ab721 100644 --- a/pyhon/command_loader.py +++ b/pyhon/command_loader.py @@ -1,5 +1,4 @@ import asyncio -import json from contextlib import suppress from copy import copy from typing import Dict, Any, Optional, TYPE_CHECKING, List @@ -53,12 +52,9 @@ class HonCommandLoader: """Get command additional data""" return self._additional_data - async def load_commands(self, data=None): + async def load_commands(self): """Trigger loading of command data""" - if data: - self._api_commands = data - else: - await self._load_data() + await self._load_data() self._appliance_data = self._api_commands.pop("applianceModel") self._get_commands() self._add_favourites() @@ -68,10 +64,10 @@ class HonCommandLoader: self._api_commands = await self._api.load_commands(self._appliance) async def _load_favourites(self): - self._favourites = await self._api.command_favourites(self._appliance) + self._favourites = await self._api.load_favourites(self._appliance) async def _load_command_history(self): - self._command_history = await self._api.command_history(self._appliance) + self._command_history = await self._api.load_command_history(self._appliance) async def _load_data(self): """Request parallel all relevant data""" diff --git a/pyhon/connection/api.py b/pyhon/connection/api.py index 6eebc7b..5819776 100644 --- a/pyhon/connection/api.py +++ b/pyhon/connection/api.py @@ -1,8 +1,9 @@ import json import logging from datetime import datetime +from pathlib import Path from pprint import pformat -from typing import Dict, Optional +from typing import Dict, Optional, Any, List, no_type_check from aiohttp import ClientSession from typing_extensions import Self @@ -66,11 +67,13 @@ class HonAPI: ).create() return self - async def load_appliances(self) -> Dict: + async def load_appliances(self) -> List[Dict[str, Any]]: async with self._hon.get(f"{const.API_URL}/commands/v1/appliance") as resp: - return await resp.json() + if result := await resp.json(): + return result.get("payload", {}).get("appliances", {}) + return [] - async def load_commands(self, appliance: HonAppliance) -> Dict: + async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]: params: Dict = { "applianceType": appliance.appliance_type, "applianceModelId": appliance.appliance_model_id, @@ -93,27 +96,29 @@ class HonAPI: return {} return result - async def command_history(self, appliance: HonAppliance) -> Dict: + async def load_command_history( + self, appliance: HonAppliance + ) -> List[Dict[str, Any]]: url: str = ( f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/history" ) async with self._hon.get(url) as response: result: Dict = await response.json() if not result or not result.get("payload"): - return {} + return [] return result["payload"]["history"] - async def command_favourites(self, appliance: HonAppliance) -> Dict: + async def load_favourites(self, appliance: HonAppliance) -> List[Dict[str, Any]]: url: str = ( f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/favourite" ) async with self._hon.get(url) as response: result: Dict = await response.json() if not result or not result.get("payload"): - return {} + return [] return result["payload"]["favourites"] - async def last_activity(self, appliance: HonAppliance) -> Dict: + async def load_last_activity(self, appliance: HonAppliance) -> Dict[str, Any]: url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity" params: Dict = {"macAddress": appliance.mac_address} async with self._hon.get(url, params=params) as response: @@ -122,19 +127,19 @@ class HonAPI: return activity return {} - async def appliance_model(self, appliance: HonAppliance) -> Dict: + async def load_appliance_data(self, appliance: HonAppliance) -> Dict[str, Any]: url: str = f"{const.API_URL}/commands/v1/appliance-model" params: Dict = { - "code": appliance.info["code"], + "code": appliance.code, "macAddress": appliance.mac_address, } async with self._hon.get(url, params=params) as response: result: Dict = await response.json() - if result and (activity := result.get("attributes")): - return activity + if result: + return result.get("payload", {}).get("applianceModel", {}) return {} - async def load_attributes(self, appliance: HonAppliance) -> Dict: + async def load_attributes(self, appliance: HonAppliance) -> Dict[str, Any]: params: Dict = { "macAddress": appliance.mac_address, "applianceType": appliance.appliance_type, @@ -144,7 +149,7 @@ class HonAPI: async with self._hon.get(url, params=params) as response: return (await response.json()).get("payload", {}) - async def load_statistics(self, appliance: HonAppliance) -> Dict: + async def load_statistics(self, appliance: HonAppliance) -> Dict[str, Any]: params: Dict = { "macAddress": appliance.mac_address, "applianceType": appliance.appliance_type, @@ -153,7 +158,7 @@ class HonAPI: async with self._hon.get(url, params=params) as response: return (await response.json()).get("payload", {}) - async def load_maintenance(self, appliance: HonAppliance): + async def load_maintenance(self, appliance: HonAppliance) -> Dict[str, Any]: url = f"{const.API_URL}/commands/v1/maintenance-cycle" params = {"macAddress": appliance.mac_address} async with self._hon.get(url, params=params) as response: @@ -192,7 +197,7 @@ class HonAPI: _LOGGER.error("%s - Payload:\n%s", url, pformat(data)) return False - async def appliance_configuration(self) -> Dict: + async def appliance_configuration(self) -> Dict[str, Any]: url: str = f"{const.API_URL}/config/v1/program-list-rules" async with self._hon_anonymous.get(url) as response: result: Dict = await response.json() @@ -200,7 +205,9 @@ class HonAPI: return data return {} - async def app_config(self, language: str = "en", beta: bool = True) -> Dict: + async def app_config( + self, language: str = "en", beta: bool = True + ) -> Dict[str, Any]: url: str = f"{const.API_URL}/app-config" payload_data: Dict = { "languageCode": language, @@ -214,7 +221,7 @@ class HonAPI: return data return {} - async def translation_keys(self, language: str = "en") -> Dict: + async def translation_keys(self, language: str = "en") -> Dict[str, Any]: config = await self.app_config(language=language) if url := config.get("language", {}).get("jsonPath"): async with self._hon_anonymous.get(url) as response: @@ -227,3 +234,61 @@ class HonAPI: await self._hon_handler.close() if self._hon_anonymous_handler is not None: await self._hon_anonymous_handler.close() + + +class TestAPI(HonAPI): + def __init__(self, path): + super().__init__() + self._anonymous = True + self._path: Path = path + + def _load_json(self, appliance: HonAppliance, file) -> Dict[str, Any]: + directory = f"{appliance.appliance_type}_{appliance.appliance_model_id}".lower() + path = f"{self._path}/{directory}/{file}.json" + with open(path, "r", encoding="utf-8") as json_file: + return json.loads(json_file.read()) + + async def load_appliances(self) -> List[Dict[str, Any]]: + result = [] + for appliance in self._path.glob("*/"): + with open( + appliance / "appliance_data.json", "r", encoding="utf-8" + ) as json_file: + result.append(json.loads(json_file.read())) + return result + + async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]: + return self._load_json(appliance, "commands") + + @no_type_check + async def load_command_history( + self, appliance: HonAppliance + ) -> List[Dict[str, Any]]: + return self._load_json(appliance, "command_history") + + async def load_favourites(self, appliance: HonAppliance) -> List[Dict[str, Any]]: + return [] + + async def load_last_activity(self, appliance: HonAppliance) -> Dict[str, Any]: + return {} + + async def load_appliance_data(self, appliance: HonAppliance) -> Dict[str, Any]: + return self._load_json(appliance, "appliance_data") + + async def load_attributes(self, appliance: HonAppliance) -> Dict[str, Any]: + return self._load_json(appliance, "attributes") + + async def load_statistics(self, appliance: HonAppliance) -> Dict[str, Any]: + return self._load_json(appliance, "statistics") + + async def load_maintenance(self, appliance: HonAppliance) -> Dict[str, Any]: + return self._load_json(appliance, "maintenance") + + async def send_command( + self, + appliance: HonAppliance, + command: str, + parameters: Dict, + ancillary_parameters: Dict, + ) -> bool: + return True diff --git a/pyhon/diagnose.py b/pyhon/diagnose.py new file mode 100644 index 0000000..90ab9b7 --- /dev/null +++ b/pyhon/diagnose.py @@ -0,0 +1,97 @@ +import asyncio +import json +import re +import shutil +from pathlib import Path +from typing import TYPE_CHECKING, List, Tuple + +from pyhon import helper + +if TYPE_CHECKING: + from pyhon.appliance import HonAppliance + + +def anonymize_data(data: str) -> str: + default_date = "1970-01-01T00:00:00.0Z" + default_mac = "xx-xx-xx-xx-xx-xx" + data = re.sub("[0-9A-Fa-f]{2}(-[0-9A-Fa-f]{2}){5}", default_mac, data) + data = re.sub("[\\d-]{10}T[\\d:]{8}(.\\d+)?Z", default_date, data) + for sensible in [ + "serialNumber", + "code", + "nickName", + "mobileId", + "PK", + "SK", + "lat", + "lng", + ]: + for match in re.findall(f'"{sensible}.*?":\\s"?(.+?)"?,?\\n', data): + replace = re.sub("[a-z]", "x", match) + replace = re.sub("[A-Z]", "X", replace) + replace = re.sub("\\d", "0", replace) + data = data.replace(match, replace) + return data + + +async def load_data(appliance: "HonAppliance", topic: str) -> Tuple[str, str]: + return topic, await getattr(appliance.api, f"load_{topic}")(appliance) + + +def write_to_json(data: str, topic: str, path: Path, anonymous: bool = False): + json_data = json.dumps(data, indent=4) + if anonymous: + json_data = anonymize_data(json_data) + file = path / f"{topic}.json" + with open(file, "w", encoding="utf-8") as json_file: + json_file.write(json_data) + return file + + +async def appliance_data( + appliance: "HonAppliance", path: Path, anonymous: bool = False +) -> List[Path]: + requests = [ + "commands", + "attributes", + "command_history", + "statistics", + "maintenance", + "appliance_data", + ] + path /= f"{appliance.appliance_type}_{appliance.model_id}".lower() + path.mkdir(parents=True, exist_ok=True) + api_data = await asyncio.gather(*[load_data(appliance, name) for name in requests]) + return [write_to_json(data, topic, path, anonymous) for topic, data in api_data] + + +async def zip_archive(appliance: "HonAppliance", path: Path, anonymous: bool = False): + data = await appliance_data(appliance, path, anonymous) + shutil.make_archive(str(path), "zip", path) + shutil.rmtree(path) + return f"{data[0].parent.stem}.zip" + + +def yaml_export(appliance: "HonAppliance", anonymous=False) -> str: + data = { + "attributes": appliance.attributes.copy(), + "appliance": appliance.info, + "statistics": appliance.statistics, + "additional_data": appliance.additional_data, + } + data |= {n: c.parameter_groups for n, c in appliance.commands.items()} + extra = {n: c.data for n, c in appliance.commands.items() if c.data} + if extra: + data |= {"extra_command_data": extra} + if anonymous: + for sensible in ["serialNumber", "coords"]: + data.get("appliance", {}).pop(sensible, None) + data = { + "data": data, + "commands": helper.create_command(appliance.commands), + "rules": helper.create_rules(appliance.commands), + } + result = helper.pretty_print(data) + if anonymous: + result = anonymize_data(result) + return result diff --git a/pyhon/hon.py b/pyhon/hon.py index 2f22dc1..e385337 100644 --- a/pyhon/hon.py +++ b/pyhon/hon.py @@ -1,5 +1,6 @@ import asyncio import logging +from pathlib import Path from types import TracebackType from typing import List, Optional, Dict, Any, Type @@ -8,6 +9,7 @@ from typing_extensions import Self from pyhon import HonAPI, exceptions from pyhon.appliance import HonAppliance +from pyhon.connection.api import TestAPI _LOGGER = logging.getLogger(__name__) @@ -18,12 +20,14 @@ class Hon: email: Optional[str] = "", password: Optional[str] = "", session: Optional[ClientSession] = None, + test_data_path: Optional[Path] = None, ): self._email: Optional[str] = email self._password: Optional[str] = password self._session: ClientSession | None = session self._appliances: List[HonAppliance] = [] self._api: Optional[HonAPI] = None + self._test_data_path: Path = test_data_path or Path().cwd() async def __aenter__(self) -> Self: return await self.create() @@ -69,8 +73,10 @@ class Hon: def appliances(self, appliances) -> None: self._appliances = appliances - async def _create_appliance(self, appliance_data: Dict[str, Any], zone=0) -> None: - appliance = HonAppliance(self._api, appliance_data, zone=zone) + async def _create_appliance( + self, appliance_data: Dict[str, Any], api: HonAPI, zone=0 + ) -> None: + appliance = HonAppliance(api, appliance_data, zone=zone) if appliance.mac_address == "": return try: @@ -87,12 +93,20 @@ class Hon: self._appliances.append(appliance) async def setup(self) -> None: - appliance: Dict - for appliance in (await self.api.load_appliances())["payload"]["appliances"]: + appliances = await self.api.load_appliances() + for appliance in appliances: if (zones := int(appliance.get("zone", "0"))) > 1: for zone in range(zones): - await self._create_appliance(appliance.copy(), zone=zone + 1) - await self._create_appliance(appliance) + await self._create_appliance( + appliance.copy(), self.api, zone=zone + 1 + ) + await self._create_appliance(appliance, self.api) + if ( + test_data := self._test_data_path / "hon-test-data" / "test_data" + ).exists() or (test_data := test_data / "test_data").exists(): + api = TestAPI(test_data) + for appliance in await api.load_appliances(): + await self._create_appliance(appliance, api) async def close(self) -> None: await self.api.close()