xiconfjs/pyhOn
1
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases 91 Wiki Activity

Create data archive and use it to test

Browse Source
This commit is contained in:
Andre Basche 2023-06-25 17:29:04 +02:00
parent 66cb7bcc24
commit ef67188b93
6 changed files with 248 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
pyhon/__main__.py
View File

@ -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():

79
pyhon/appliance.py
View File

@ -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"

10
pyhon/command_loader.py
View File

@ -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,11 +52,8 @@ 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()
self._appliance_data = self._api_commands.pop("applianceModel")
self._get_commands()
@ -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"""

103
pyhon/connection/api.py
View File

@ -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

97
pyhon/diagnose.py Normal file
View File

@ -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

26
pyhon/hon.py
View File

@ -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()