Compare commits

...

14 Commits

Author SHA1 Message Date
44f40c531e Fix some minor issues 2023-06-29 22:08:51 +02:00
4e88bc7a9f Fix error in archive again 2023-06-29 18:49:33 +02:00
b5d8a90d79 Fix messed up parameters in request 2023-06-28 20:25:52 +02:00
52837f16e3 Improve appliance import 2023-06-28 19:45:37 +02:00
2a6b040193 Add import parameter 2023-06-28 19:17:17 +02:00
9eb99f283b Add more type hints 2023-06-28 19:02:11 +02:00
ad0d065b03 Fix creating wrong zip archive 2023-06-26 02:19:07 +02:00
924e2c240d Fix wrong archive link 2023-06-25 18:30:29 +02:00
76bd189e7b Bump version 2023-06-25 17:49:03 +02:00
ef67188b93 Create data archive and use it to test 2023-06-25 17:30:15 +02:00
66cb7bcc24 Merge branch 'refactor2' 2023-06-22 00:03:07 +02:00
8c65a37f29 Add command loader class 2023-06-15 02:16:03 +02:00
1ca89995a2 Lock attributes 2023-06-13 00:39:18 +02:00
f6139db0b5 Use class for attributes 2023-06-13 00:12:29 +02:00
32 changed files with 918 additions and 424 deletions

View File

@ -107,3 +107,6 @@ This library is used for the custom [HomeAssistant Integration "Haier hOn"](http
## Contribution ## Contribution
Any kind of contribution is welcome! Any kind of contribution is welcome!
| Please add your appliances data to our [hon-test-data collection](https://github.com/Andre0512/hon-test-data). <br/>This helps us to develop new features and not to break compatibility in newer versions. |
|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|

4
mypy.ini Normal file
View File

@ -0,0 +1,4 @@
[mypy]
check_untyped_defs = True
disallow_any_generics = True
disallow_untyped_defs = True

View File

@ -6,16 +6,17 @@ import logging
import sys import sys
from getpass import getpass from getpass import getpass
from pathlib import Path from pathlib import Path
from typing import Tuple, Dict, Any
if __name__ == "__main__": if __name__ == "__main__":
sys.path.insert(0, str(Path(__file__).parent.parent)) sys.path.insert(0, str(Path(__file__).parent.parent))
from pyhon import Hon, HonAPI, helper from pyhon import Hon, HonAPI, diagnose, printer
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
def get_arguments(): def get_arguments() -> Dict[str, Any]:
"""Get parsed arguments.""" """Get parsed arguments."""
parser = argparse.ArgumentParser(description="pyhOn: Command Line Utility") parser = argparse.ArgumentParser(description="pyhOn: Command Line Utility")
parser.add_argument("-u", "--user", help="user for haier hOn account") parser.add_argument("-u", "--user", help="user for haier hOn account")
@ -24,6 +25,11 @@ def get_arguments():
keys = subparser.add_parser("keys", help="print as key format") 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("keys", help="print as key format", action="store_true")
keys.add_argument("--all", help="print also full keys", 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 = subparser.add_parser(
"translate", help="print available translation keys" "translate", help="print available translation keys"
) )
@ -31,10 +37,13 @@ def get_arguments():
"translate", help="language (de, en, fr...)", metavar="LANGUAGE" "translate", help="language (de, en, fr...)", metavar="LANGUAGE"
) )
translate.add_argument("--json", help="print as json", action="store_true") translate.add_argument("--json", help="print as json", action="store_true")
parser.add_argument(
"-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd()
)
return vars(parser.parse_args()) return vars(parser.parse_args())
async def translate(language, json_output=False): async def translate(language: str, json_output: bool = False) -> None:
async with HonAPI(anonymous=True) as hon: async with HonAPI(anonymous=True) as hon:
keys = await hon.translation_keys(language) keys = await hon.translation_keys(language)
if json_output: if json_output:
@ -47,41 +56,57 @@ async def translate(language, json_output=False):
.replace("\\r", "") .replace("\\r", "")
) )
keys = json.loads(clean_keys) keys = json.loads(clean_keys)
print(helper.pretty_print(keys)) print(printer.pretty_print(keys))
async def main(): def get_login_data(args: Dict[str, str]) -> Tuple[str, str]:
args = get_arguments()
if language := args.get("translate"):
await translate(language, json_output=args.get("json"))
return
if not (user := args["user"]): if not (user := args["user"]):
user = input("User for hOn account: ") user = input("User for hOn account: ")
if not (password := args["password"]): if not (password := args["password"]):
password = getpass("Password for hOn account: ") password = getpass("Password for hOn account: ")
async with Hon(user, password) as hon: return user, password
async def main() -> None:
args = get_arguments()
if language := args.get("translate"):
await translate(language, json_output=args.get("json", ""))
return
async with Hon(
*get_login_data(args), test_data_path=Path(args.get("import", ""))
) as hon:
for device in hon.appliances: 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:
archive = await diagnose.zip_archive(device, path, anonymous)
print(f"Created {archive}")
continue
print("=" * 10, device.appliance_type, "-", device.nick_name, "=" * 10) print("=" * 10, device.appliance_type, "-", device.nick_name, "=" * 10)
if args.get("keys"): if args.get("keys"):
data = device.data.copy() data = device.data.copy()
attr = "get" if args.get("all") else "pop" attr = "get" if args.get("all") else "pop"
print( print(
helper.key_print( printer.key_print(
data["attributes"].__getattribute__(attr)("parameters") data["attributes"].__getattribute__(attr)("parameters")
) )
) )
print(helper.key_print(data.__getattribute__(attr)("appliance"))) print(printer.key_print(data.__getattribute__(attr)("appliance")))
print(helper.key_print(data)) print(printer.key_print(data))
print( print(
helper.pretty_print( printer.pretty_print(
helper.create_command(device.commands, concat=True) printer.create_command(device.commands, concat=True)
) )
) )
else: else:
print(device.diagnose(" ")) print(diagnose.yaml_export(device))
def start(): def start() -> None:
try: try:
asyncio.run(main()) asyncio.run(main())
except KeyboardInterrupt: except KeyboardInterrupt:

View File

@ -1,18 +1,17 @@
import importlib import importlib
import json
import logging import logging
from contextlib import suppress import re
from copy import copy
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pathlib import Path from pathlib import Path
from typing import Optional, Dict, Any from typing import Optional, Dict, Any, TYPE_CHECKING, List
from typing import TYPE_CHECKING
from pyhon import helper from pyhon import diagnose, exceptions
from pyhon.attributes import HonAttribute
from pyhon.command_loader import HonCommandLoader
from pyhon.commands import HonCommand from pyhon.commands import HonCommand
from pyhon.parameter.base import HonParameter from pyhon.parameter.base import HonParameter
from pyhon.parameter.fixed import HonParameterFixed
from pyhon.parameter.range import HonParameterRange from pyhon.parameter.range import HonParameterRange
from pyhon.typedefs import Parameter
if TYPE_CHECKING: if TYPE_CHECKING:
from pyhon import HonAPI from pyhon import HonAPI
@ -28,16 +27,16 @@ class HonAppliance:
) -> None: ) -> None:
if attributes := info.get("attributes"): if attributes := info.get("attributes"):
info["attributes"] = {v["parName"]: v["parValue"] for v in attributes} info["attributes"] = {v["parName"]: v["parValue"] for v in attributes}
self._info: Dict = info self._info: Dict[str, Any] = info
self._api: Optional[HonAPI] = api self._api: Optional[HonAPI] = api
self._appliance_model: Dict = {} self._appliance_model: Dict[str, Any] = {}
self._commands: Dict[str, HonCommand] = {} self._commands: Dict[str, HonCommand] = {}
self._statistics: Dict = {} self._statistics: Dict[str, Any] = {}
self._attributes: Dict = {} self._attributes: Dict[str, Any] = {}
self._zone: int = zone self._zone: int = zone
self._additional_data: Dict[str, Any] = {} self._additional_data: Dict[str, Any] = {}
self._last_update = None self._last_update: Optional[datetime] = None
self._default_setting = HonParameter("", {}, "") self._default_setting = HonParameter("", {}, "")
try: try:
@ -47,7 +46,7 @@ class HonAppliance:
except ModuleNotFoundError: except ModuleNotFoundError:
self._extra = None self._extra = None
def __getitem__(self, item): def __getitem__(self, item: str) -> Any:
if self._zone: if self._zone:
item += f"Z{self._zone}" item += f"Z{self._zone}"
if "." in item: if "." in item:
@ -61,19 +60,19 @@ class HonAppliance:
if item in self.data: if item in self.data:
return self.data[item] return self.data[item]
if item in self.attributes["parameters"]: if item in self.attributes["parameters"]:
return self.attributes["parameters"].get(item) return self.attributes["parameters"][item].value
return self.info[item] return self.info[item]
def get(self, item, default=None): def get(self, item: str, default: Any = None) -> Any:
try: try:
return self[item] return self[item]
except (KeyError, IndexError): except (KeyError, IndexError):
return default return default
def _check_name_zone(self, name: str, frontend: bool = True) -> str: def _check_name_zone(self, name: str, frontend: bool = True) -> str:
middle = " Z" if frontend else "_z" zone = " Z" if frontend else "_z"
if (attribute := self._info.get(name, "")) and self._zone: if (attribute := self._info.get(name, "")) and self._zone:
return f"{attribute}{middle}{self._zone}" return f"{attribute}{zone}{self._zone}"
return attribute return attribute
@property @property
@ -90,15 +89,26 @@ class HonAppliance:
@property @property
def unique_id(self) -> str: def unique_id(self) -> str:
return self._check_name_zone("macAddress", frontend=False) default_mac = "xx-xx-xx-xx-xx-xx"
import_name = f"{self.appliance_type.lower()}_{self.appliance_model_id}"
result = self._check_name_zone("macAddress", frontend=False)
result = result.replace(default_mac, import_name)
return result
@property @property
def model_name(self) -> str: def model_name(self) -> str:
return self._check_name_zone("modelName") return self._check_name_zone("modelName")
@property
def brand(self) -> str:
return self._check_name_zone("brand")
@property @property
def nick_name(self) -> str: def nick_name(self) -> str:
return self._check_name_zone("nickName") result = self._check_name_zone("nickName")
if not result or re.findall("^[xX\s]+$", result):
return self.model_name
return result
@property @property
def code(self) -> str: def code(self) -> str:
@ -108,7 +118,11 @@ class HonAppliance:
return serial_number[:8] if len(serial_number) < 18 else serial_number[:11] return serial_number[:8] if len(serial_number) < 18 else serial_number[:11]
@property @property
def options(self): def model_id(self) -> int:
return self._info.get("applianceModelId", 0)
@property
def options(self) -> Dict[str, Any]:
return self._appliance_model.get("options", {}) return self._appliance_model.get("options", {})
@property @property
@ -116,19 +130,19 @@ class HonAppliance:
return self._commands return self._commands
@property @property
def attributes(self): def attributes(self) -> Dict[str, Any]:
return self._attributes return self._attributes
@property @property
def statistics(self): def statistics(self) -> Dict[str, Any]:
return self._statistics return self._statistics
@property @property
def info(self): def info(self) -> Dict[str, Any]:
return self._info return self._info
@property @property
def additional_data(self): def additional_data(self) -> Dict[str, Any]:
return self._additional_data return self._additional_data
@property @property
@ -136,120 +150,38 @@ class HonAppliance:
return self._zone return self._zone
@property @property
def api(self) -> Optional["HonAPI"]: def api(self) -> "HonAPI":
"""api connection object"""
if self._api is None:
raise exceptions.NoAuthenticationException("Missing hOn login")
return self._api return self._api
async def _recover_last_command_states(self): async def load_commands(self) -> None:
command_history = await self.api.command_history(self) command_loader = HonCommandLoader(self.api, self)
for name, command in self._commands.items(): await command_loader.load_commands()
last = next( self._commands = command_loader.commands
( self._additional_data = command_loader.additional_data
index self._appliance_model = command_loader.appliance_data
for (index, d) in enumerate(command_history)
if d.get("command", {}).get("commandName") == name
),
None,
)
if last is None:
continue
parameters = command_history[last].get("command", {}).get("parameters", {})
if command.categories and (
parameters.get("program") or parameters.get("category")
):
if parameters.get("program"):
command.category = parameters.pop("program").split(".")[-1].lower()
else:
command.category = parameters.pop("category")
command = self.commands[name]
for key, data in command.settings.items():
if (
not isinstance(data, HonParameterFixed)
and parameters.get(key) is not None
):
with suppress(ValueError):
data.value = parameters.get(key)
def _get_categories(self, command, data): async def load_attributes(self) -> None:
categories = {}
for category, value in data.items():
result = self._get_command(value, command, category, categories)
if result:
if "PROGRAM" in category:
category = category.split(".")[-1].lower()
categories[category] = result[0]
if categories:
if "setParameters" in categories:
return [categories["setParameters"]]
return [list(categories.values())[0]]
return []
def _get_commands(self, data):
commands = []
for command, value in data.items():
commands += self._get_command(value, command, "")
return {c.name: c for c in commands}
def _get_command(self, data, command="", category="", categories=None):
commands = []
if isinstance(data, dict):
if data.get("description") and data.get("protocolType", None):
commands += [
HonCommand(
command,
data,
self,
category_name=category,
categories=categories,
)
]
else:
commands += self._get_categories(command, data)
elif category:
self._additional_data.setdefault(command, {})[category] = data
else:
self._additional_data[command] = data
return commands
async def load_commands(self):
raw = await self.api.load_commands(self)
self._appliance_model = raw.pop("applianceModel")
raw.pop("dictionaryId", None)
self._commands = self._get_commands(raw)
await self._add_favourites()
await self._recover_last_command_states()
async def _add_favourites(self):
favourites = await self._api.command_favourites(self)
for favourite in favourites:
name = favourite.get("favouriteName")
command = favourite.get("command")
command_name = command.get("commandName")
program_name = command.get("programName", "").split(".")[-1].lower()
base = copy(self._commands[command_name].categories[program_name])
for data in command.values():
if isinstance(data, str):
continue
for key, value in data.items():
if parameter := base.parameters.get(key):
with suppress(ValueError):
parameter.value = value
extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom")
base.parameters.update(favourite=extra_param)
base.parameters["program"].set_value(name)
self._commands[command_name].categories[name] = base
async def load_attributes(self):
self._attributes = await self.api.load_attributes(self) self._attributes = await self.api.load_attributes(self)
for name, values in self._attributes.pop("shadow").get("parameters").items(): for name, values in (
self._attributes.setdefault("parameters", {})[name] = values["parNewVal"] self._attributes.pop("shadow", {}).get("parameters", {}).items()
):
if name in self._attributes.get("parameters", {}):
self._attributes["parameters"][name].update(values)
else:
self._attributes.setdefault("parameters", {})[name] = HonAttribute(
values
)
if self._extra: if self._extra:
self._attributes = self._extra.attributes(self._attributes) self._attributes = self._extra.attributes(self._attributes)
async def load_statistics(self): async def load_statistics(self) -> None:
self._statistics = await self.api.load_statistics(self) self._statistics = await self.api.load_statistics(self)
self._statistics |= await self.api.load_maintenance(self) self._statistics |= await self.api.load_maintenance(self)
async def update(self, force=False): async def update(self, force: bool = False) -> None:
now = datetime.now() now = datetime.now()
if ( if (
force force
@ -261,11 +193,11 @@ class HonAppliance:
await self.load_attributes() await self.load_attributes()
@property @property
def command_parameters(self): def command_parameters(self) -> Dict[str, Dict[str, str | float]]:
return {n: c.parameter_value for n, c in self._commands.items()} return {n: c.parameter_value for n, c in self._commands.items()}
@property @property
def settings(self): def settings(self) -> Dict[str, Parameter]:
result = {} result = {}
for name, command in self._commands.items(): for name, command in self._commands.items():
for key in command.setting_keys: for key in command.setting_keys:
@ -276,7 +208,7 @@ class HonAppliance:
return result return result
@property @property
def available_settings(self): def available_settings(self) -> List[str]:
result = [] result = []
for name, command in self._commands.items(): for name, command in self._commands.items():
for key in command.setting_keys: for key in command.setting_keys:
@ -284,7 +216,7 @@ class HonAppliance:
return result return result
@property @property
def data(self): def data(self) -> Dict[str, Any]:
result = { result = {
"attributes": self.attributes, "attributes": self.attributes,
"appliance": self.info, "appliance": self.info,
@ -295,40 +227,23 @@ class HonAppliance:
} }
return result return result
def diagnose(self, whitespace=" ", command_only=False): @property
data = { def diagnose(self) -> str:
"attributes": self.attributes.copy(), return diagnose.yaml_export(self, anonymous=True)
"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")
def sync_to_params(self, command_name): async def data_archive(self, path: Path) -> str:
command: HonCommand = self.commands.get(command_name) return await diagnose.zip_archive(self, path, anonymous=True)
def sync_to_params(self, command_name: str) -> None:
if not (command := self.commands.get(command_name)):
return
for key, value in self.attributes.get("parameters", {}).items(): for key, value in self.attributes.get("parameters", {}).items():
if isinstance(value, str) and (new := command.parameters.get(key)): if isinstance(value, str) and (new := command.parameters.get(key)):
self.attributes["parameters"][key] = str(new.intern_value) self.attributes["parameters"][key].update(
str(new.intern_value), shield=True
)
def sync_command(self, main, target=None) -> None: def sync_command(self, main: str, target: Optional[List[str]] = None) -> None:
base: Optional[HonCommand] = self.commands.get(main) base: Optional[HonCommand] = self.commands.get(main)
if not base: if not base:
return return
@ -348,35 +263,3 @@ class HonAppliance:
parameter.min = int(base_value.value) parameter.min = int(base_value.value)
parameter.step = 1 parameter.step = 1
parameter.value = base_value.value 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"

View File

@ -1,15 +1,25 @@
from typing import Dict, Any, TYPE_CHECKING
from pyhon.parameter.program import HonParameterProgram
if TYPE_CHECKING:
from pyhon.appliance import HonAppliance
class ApplianceBase: class ApplianceBase:
def __init__(self, appliance): def __init__(self, appliance: "HonAppliance"):
self.parent = appliance self.parent = appliance
def attributes(self, data): def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
program_name = "No Program" program_name = "No Program"
if program := int(data["parameters"].get("prCode", "0")): if program := int(str(data.get("parameters", {}).get("prCode", "0"))):
if start_cmd := self.parent.settings.get("startProgram.program"): if start_cmd := self.parent.settings.get("startProgram.program"):
if ids := start_cmd.ids: if isinstance(start_cmd, HonParameterProgram) and (
ids := start_cmd.ids
):
program_name = ids.get(program, program_name) program_name = ids.get(program, program_name)
data["programName"] = program_name data["programName"] = program_name
return data return data
def settings(self, settings): def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]:
return settings return settings

View File

@ -1,10 +1,12 @@
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase from pyhon.appliances.base import ApplianceBase
class Appliance(ApplianceBase): class Appliance(ApplianceBase):
def attributes(self, data): def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["machMode"] = "0" data["parameters"]["machMode"].value = "0"
data["active"] = bool(data.get("activity")) data["active"] = bool(data.get("activity"))
return data return data

View File

@ -1,19 +1,24 @@
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase from pyhon.appliances.base import ApplianceBase
from pyhon.parameter.program import HonParameterProgram
class Appliance(ApplianceBase): class Appliance(ApplianceBase):
def attributes(self, data): def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["temp"] = "0" data["parameters"]["temp"].value = "0"
data["parameters"]["onOffStatus"] = "0" data["parameters"]["onOffStatus"].value = "0"
data["parameters"]["remoteCtrValid"] = "0" data["parameters"]["remoteCtrValid"].value = "0"
data["parameters"]["remainingTimeMM"] = "0" data["parameters"]["remainingTimeMM"].value = "0"
data["active"] = data["parameters"]["onOffStatus"] == "1" data["active"] = data["parameters"]["onOffStatus"] == "1"
if program := int(data["parameters"]["prCode"]): if program := int(data["parameters"]["prCode"].value):
ids = self.parent.settings["startProgram.program"].ids if (setting := self.parent.settings["startProgram.program"]) and isinstance(
data["programName"] = ids.get(program, "") setting, HonParameterProgram
):
data["programName"] = setting.ids.get(program, "")
return data return data

View File

@ -1,8 +1,10 @@
from typing import Dict, Any
from pyhon.appliances.base import ApplianceBase from pyhon.appliances.base import ApplianceBase
class Appliance(ApplianceBase): class Appliance(ApplianceBase):
def attributes(self, data): def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
if data["parameters"]["holidayMode"] == "1": if data["parameters"]["holidayMode"] == "1":
data["modeZ1"] = "holiday" data["modeZ1"] = "holiday"

View File

@ -1,17 +1,19 @@
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase from pyhon.appliances.base import ApplianceBase
from pyhon.parameter.fixed import HonParameterFixed from pyhon.parameter.fixed import HonParameterFixed
class Appliance(ApplianceBase): class Appliance(ApplianceBase):
def attributes(self, data): def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["machMode"] = "0" data["parameters"]["machMode"].value = "0"
data["active"] = bool(data.get("activity")) data["active"] = bool(data.get("activity"))
data["pause"] = data["parameters"]["machMode"] == "3" data["pause"] = data["parameters"]["machMode"] == "3"
return data return data
def settings(self, settings): def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]:
dry_level = settings.get("startProgram.dryLevel") dry_level = settings.get("startProgram.dryLevel")
if isinstance(dry_level, HonParameterFixed) and dry_level.value == "11": if isinstance(dry_level, HonParameterFixed) and dry_level.value == "11":
settings.pop("startProgram.dryLevel", None) settings.pop("startProgram.dryLevel", None)

View File

@ -1,14 +1,16 @@
from typing import Dict, Any
from pyhon.appliances.base import ApplianceBase from pyhon.appliances.base import ApplianceBase
class Appliance(ApplianceBase): class Appliance(ApplianceBase):
def attributes(self, data): def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["machMode"] = "0" data["parameters"]["machMode"].value = "0"
data["active"] = bool(data.get("activity")) data["active"] = bool(data.get("activity"))
data["pause"] = data["parameters"]["machMode"] == "3" data["pause"] = data["parameters"]["machMode"] == "3"
return data return data
def settings(self, settings): def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]:
return settings return settings

View File

@ -1,14 +1,16 @@
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase from pyhon.appliances.base import ApplianceBase
class Appliance(ApplianceBase): class Appliance(ApplianceBase):
def attributes(self, data): def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["machMode"] = "0" data["parameters"]["machMode"].value = "0"
data["active"] = bool(data.get("activity")) data["active"] = bool(data.get("activity"))
data["pause"] = data["parameters"]["machMode"] == "3" data["pause"] = data["parameters"]["machMode"] == "3"
return data return data
def settings(self, settings): def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]:
return settings return settings

58
pyhon/attributes.py Normal file
View File

@ -0,0 +1,58 @@
from datetime import datetime, timedelta
from typing import Optional, Final, Dict
from pyhon.helper import str_to_float
class HonAttribute:
_LOCK_TIMEOUT: Final = 10
def __init__(self, data: Dict[str, str] | str):
self._value: str = ""
self._last_update: Optional[datetime] = None
self._lock_timestamp: Optional[datetime] = None
self.update(data)
@property
def value(self) -> float | str:
"""Attribute value"""
try:
return str_to_float(self._value)
except ValueError:
return self._value
@value.setter
def value(self, value: str) -> None:
self._value = value
@property
def last_update(self) -> Optional[datetime]:
"""Timestamp of last api update"""
return self._last_update
@property
def lock(self) -> bool:
"""Shows if value changes are forbidden"""
if not self._lock_timestamp:
return False
lock_until = self._lock_timestamp + timedelta(seconds=self._LOCK_TIMEOUT)
return lock_until >= datetime.utcnow()
def update(self, data: Dict[str, str] | str, shield: bool = False) -> bool:
if self.lock and not shield:
return False
if shield:
self._lock_timestamp = datetime.utcnow()
if isinstance(data, str):
self.value = data
return True
self.value = data.get("parNewVal", "")
if last_update := data.get("lastUpdate"):
try:
self._last_update = datetime.fromisoformat(last_update)
except ValueError:
self._last_update = None
return True
def __str__(self) -> str:
return self._value

204
pyhon/command_loader.py Normal file
View File

@ -0,0 +1,204 @@
import asyncio
from contextlib import suppress
from copy import copy
from typing import Dict, Any, Optional, TYPE_CHECKING, List, Collection
from pyhon.commands import HonCommand
from pyhon.parameter.fixed import HonParameterFixed
from pyhon.parameter.program import HonParameterProgram
if TYPE_CHECKING:
from pyhon import HonAPI, exceptions
from pyhon.appliance import HonAppliance
class HonCommandLoader:
"""Loads and parses hOn command data"""
def __init__(self, api: "HonAPI", appliance: "HonAppliance") -> None:
self._api_commands: Dict[str, Any] = {}
self._favourites: List[Dict[str, Any]] = []
self._command_history: List[Dict[str, Any]] = []
self._commands: Dict[str, HonCommand] = {}
self._api: "HonAPI" = api
self._appliance: "HonAppliance" = appliance
self._appliance_data: Dict[str, Any] = {}
self._additional_data: Dict[str, Any] = {}
@property
def api(self) -> "HonAPI":
"""api connection object"""
if self._api is None:
raise exceptions.NoAuthenticationException("Missing hOn login")
return self._api
@property
def appliance(self) -> "HonAppliance":
"""appliance object"""
return self._appliance
@property
def commands(self) -> Dict[str, HonCommand]:
"""Get list of hon commands"""
return self._commands
@property
def appliance_data(self) -> Dict[str, Any]:
"""Get command appliance data"""
return self._appliance_data
@property
def additional_data(self) -> Dict[str, Any]:
"""Get command additional data"""
return self._additional_data
async def load_commands(self) -> None:
"""Trigger loading of command data"""
await self._load_data()
self._appliance_data = self._api_commands.pop("applianceModel", {})
self._get_commands()
self._add_favourites()
self._recover_last_command_states()
async def _load_commands(self) -> None:
self._api_commands = await self._api.load_commands(self._appliance)
async def _load_favourites(self) -> None:
self._favourites = await self._api.load_favourites(self._appliance)
async def _load_command_history(self) -> None:
self._command_history = await self._api.load_command_history(self._appliance)
async def _load_data(self) -> None:
"""Callback parallel all relevant data"""
await asyncio.gather(
*[
self._load_commands(),
self._load_favourites(),
self._load_command_history(),
]
)
@staticmethod
def _is_command(data: Dict[str, Any]) -> bool:
"""Check if dict can be parsed as command"""
return (
data.get("description") is not None and data.get("protocolType") is not None
)
@staticmethod
def _clean_name(category: str) -> str:
"""Clean up category name"""
if "PROGRAM" in category:
return category.split(".")[-1].lower()
return category
def _get_commands(self) -> None:
"""Generates HonCommand dict from api data"""
commands = []
for name, data in self._api_commands.items():
if command := self._parse_command(data, name):
commands.append(command)
self._commands = {c.name: c for c in commands}
def _parse_command(
self,
data: Dict[str, Any] | str,
command_name: str,
categories: Optional[Dict[str, "HonCommand"]] = None,
category_name: str = "",
) -> Optional[HonCommand]:
"""Try to crate HonCommand object"""
if not isinstance(data, dict):
self._additional_data[command_name] = data
return None
if self._is_command(data):
return HonCommand(
command_name,
data,
self._appliance,
category_name=category_name,
categories=categories,
)
if category := self._parse_categories(data, command_name):
return category
return None
def _parse_categories(
self, data: Dict[str, Any], command_name: str
) -> Optional[HonCommand]:
"""Parse categories and create reference to other"""
categories: Dict[str, HonCommand] = {}
for category, value in data.items():
if command := self._parse_command(
value, command_name, category_name=category, categories=categories
):
categories[self._clean_name(category)] = command
if categories:
# setParameters should be at first place
if "setParameters" in categories:
return categories["setParameters"]
return list(categories.values())[0]
return None
def _get_last_command_index(self, name: str) -> Optional[int]:
"""Get index of last command execution"""
return next(
(
index
for (index, d) in enumerate(self._command_history)
if d.get("command", {}).get("commandName") == name
),
None,
)
def _set_last_category(
self, command: HonCommand, name: str, parameters: Dict[str, Any]
) -> HonCommand:
"""Set category to last state"""
if command.categories:
if program := parameters.pop("program", None):
command.category = self._clean_name(program)
elif category := parameters.pop("category", None):
command.category = category
else:
return command
return self.commands[name]
return command
def _recover_last_command_states(self) -> None:
"""Set commands to last state"""
for name, command in self.commands.items():
if (last_index := self._get_last_command_index(name)) is None:
continue
last_command = self._command_history[last_index]
parameters = last_command.get("command", {}).get("parameters", {})
command = self._set_last_category(command, name, parameters)
for key, data in command.settings.items():
if parameters.get(key) is None:
continue
with suppress(ValueError):
data.value = parameters.get(key)
def _add_favourites(self) -> None:
"""Patch program categories with favourites"""
for favourite in self._favourites:
name = favourite.get("favouriteName", {})
command = favourite.get("command", {})
command_name = command.get("commandName", "")
program_name = self._clean_name(command.get("programName", ""))
base: HonCommand = copy(
self.commands[command_name].categories[program_name]
)
for data in command.values():
if isinstance(data, str):
continue
for key, value in data.items():
if parameter := base.parameters.get(key):
with suppress(ValueError):
parameter.value = value
extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom")
base.parameters.update(favourite=extra_param)
if isinstance(program := base.parameters["program"], HonParameterProgram):
program.set_value(name)
self.commands[command_name].categories[name] = base

View File

@ -9,6 +9,7 @@ from pyhon.parameter.fixed import HonParameterFixed
from pyhon.parameter.program import HonParameterProgram from pyhon.parameter.program import HonParameterProgram
from pyhon.parameter.range import HonParameterRange from pyhon.parameter.range import HonParameterRange
from pyhon.rules import HonRuleSet from pyhon.rules import HonRuleSet
from pyhon.typedefs import Parameter
if TYPE_CHECKING: if TYPE_CHECKING:
from pyhon import HonAPI from pyhon import HonAPI
@ -43,7 +44,7 @@ class HonCommand:
return f"{self._name} command" return f"{self._name} command"
@property @property
def name(self): def name(self) -> str:
return self._name return self._name
@property @property
@ -57,7 +58,7 @@ class HonCommand:
return self._appliance return self._appliance
@property @property
def data(self): def data(self) -> Dict[str, Any]:
return self._data return self._data
@property @property
@ -79,14 +80,16 @@ class HonCommand:
def parameter_value(self) -> Dict[str, Union[str, float]]: def parameter_value(self) -> Dict[str, Union[str, float]]:
return {n: p.value for n, p in self._parameters.items()} return {n: p.value for n, p in self._parameters.items()}
def _load_parameters(self, attributes): def _load_parameters(self, attributes: Dict[str, Dict[str, Any]]) -> None:
for key, items in attributes.items(): for key, items in attributes.items():
for name, data in items.items(): for name, data in items.items():
self._create_parameters(data, name, key) self._create_parameters(data, name, key)
for rule in self._rules: for rule in self._rules:
rule.patch() rule.patch()
def _create_parameters(self, data: Dict, name: str, parameter: str) -> None: def _create_parameters(
self, data: Dict[str, Any], name: str, parameter: str
) -> None:
if name == "zoneMap" and self._appliance.zone: if name == "zoneMap" and self._appliance.zone:
data["default"] = self._appliance.zone data["default"] = self._appliance.zone
if data.get("category") == "rule": if data.get("category") == "rule":
@ -147,7 +150,7 @@ class HonCommand:
) )
@staticmethod @staticmethod
def _more_options(first: HonParameter, second: HonParameter): def _more_options(first: Parameter, second: Parameter) -> Parameter:
if isinstance(first, HonParameterFixed) and not isinstance( if isinstance(first, HonParameterFixed) and not isinstance(
second, HonParameterFixed second, HonParameterFixed
): ):
@ -157,8 +160,8 @@ class HonCommand:
return first return first
@property @property
def available_settings(self) -> Dict[str, HonParameter]: def available_settings(self) -> Dict[str, Parameter]:
result: Dict[str, HonParameter] = {} result: Dict[str, Parameter] = {}
for command in self.categories.values(): for command in self.categories.values():
for name, parameter in command.parameters.items(): for name, parameter in command.parameters.items():
if name in result: if name in result:

View File

@ -1,8 +1,10 @@
import json import json
import logging import logging
from datetime import datetime from datetime import datetime
from pathlib import Path
from pprint import pformat from pprint import pformat
from typing import Dict, Optional from types import TracebackType
from typing import Dict, Optional, Any, List, no_type_check, Type
from aiohttp import ClientSession from aiohttp import ClientSession
from typing_extensions import Self from typing_extensions import Self
@ -35,7 +37,12 @@ class HonAPI:
async def __aenter__(self) -> Self: async def __aenter__(self) -> Self:
return await self.create() return await self.create()
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
await self.close() await self.close()
@property @property
@ -45,13 +52,13 @@ class HonAPI:
return self._hon.auth return self._hon.auth
@property @property
def _hon(self): def _hon(self) -> HonConnectionHandler:
if self._hon_handler is None: if self._hon_handler is None:
raise exceptions.NoAuthenticationException raise exceptions.NoAuthenticationException
return self._hon_handler return self._hon_handler
@property @property
def _hon_anonymous(self): def _hon_anonymous(self) -> HonAnonymousConnectionHandler:
if self._hon_anonymous_handler is None: if self._hon_anonymous_handler is None:
raise exceptions.NoAuthenticationException raise exceptions.NoAuthenticationException
return self._hon_anonymous_handler return self._hon_anonymous_handler
@ -66,12 +73,14 @@ class HonAPI:
).create() ).create()
return self 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: 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 = { params: Dict[str, str | int] = {
"applianceType": appliance.appliance_type, "applianceType": appliance.appliance_type,
"applianceModelId": appliance.appliance_model_id, "applianceModelId": appliance.appliance_model_id,
"macAddress": appliance.mac_address, "macAddress": appliance.mac_address,
@ -87,55 +96,57 @@ class HonAPI:
params["series"] = series params["series"] = series
url: str = f"{const.API_URL}/commands/v1/retrieve" url: str = f"{const.API_URL}/commands/v1/retrieve"
async with self._hon.get(url, params=params) as response: async with self._hon.get(url, params=params) as response:
result: Dict = (await response.json()).get("payload", {}) result: Dict[str, Any] = (await response.json()).get("payload", {})
if not result or result.pop("resultCode") != "0": if not result or result.pop("resultCode") != "0":
_LOGGER.error(await response.json()) _LOGGER.error(await response.json())
return {} return {}
return result return result
async def command_history(self, appliance: HonAppliance) -> Dict: async def load_command_history(
self, appliance: HonAppliance
) -> List[Dict[str, Any]]:
url: str = ( url: str = (
f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/history" f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/history"
) )
async with self._hon.get(url) as response: async with self._hon.get(url) as response:
result: Dict = await response.json() result: Dict[str, Any] = await response.json()
if not result or not result.get("payload"): if not result or not result.get("payload"):
return {} return []
return result["payload"]["history"] 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 = ( url: str = (
f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/favourite" f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/favourite"
) )
async with self._hon.get(url) as response: async with self._hon.get(url) as response:
result: Dict = await response.json() result: Dict[str, Any] = await response.json()
if not result or not result.get("payload"): if not result or not result.get("payload"):
return {} return []
return result["payload"]["favourites"] 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" url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity"
params: Dict = {"macAddress": appliance.mac_address} params: Dict[str, str] = {"macAddress": appliance.mac_address}
async with self._hon.get(url, params=params) as response: async with self._hon.get(url, params=params) as response:
result: Dict = await response.json() result: Dict[str, Any] = await response.json()
if result and (activity := result.get("attributes")): if result and (activity := result.get("attributes")):
return activity return activity
return {} 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" url: str = f"{const.API_URL}/commands/v1/appliance-model"
params: Dict = { params: Dict[str, str] = {
"code": appliance.info["code"], "code": appliance.code,
"macAddress": appliance.mac_address, "macAddress": appliance.mac_address,
} }
async with self._hon.get(url, params=params) as response: async with self._hon.get(url, params=params) as response:
result: Dict = await response.json() result: Dict[str, Any] = await response.json()
if result and (activity := result.get("attributes")): if result:
return activity return result.get("payload", {}).get("applianceModel", {})
return {} return {}
async def load_attributes(self, appliance: HonAppliance) -> Dict: async def load_attributes(self, appliance: HonAppliance) -> Dict[str, Any]:
params: Dict = { params: Dict[str, str] = {
"macAddress": appliance.mac_address, "macAddress": appliance.mac_address,
"applianceType": appliance.appliance_type, "applianceType": appliance.appliance_type,
"category": "CYCLE", "category": "CYCLE",
@ -144,8 +155,8 @@ class HonAPI:
async with self._hon.get(url, params=params) as response: async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {}) 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 = { params: Dict[str, str] = {
"macAddress": appliance.mac_address, "macAddress": appliance.mac_address,
"applianceType": appliance.appliance_type, "applianceType": appliance.appliance_type,
} }
@ -153,7 +164,7 @@ class HonAPI:
async with self._hon.get(url, params=params) as response: async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {}) 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" url = f"{const.API_URL}/commands/v1/maintenance-cycle"
params = {"macAddress": appliance.mac_address} params = {"macAddress": appliance.mac_address}
async with self._hon.get(url, params=params) as response: async with self._hon.get(url, params=params) as response:
@ -163,11 +174,11 @@ class HonAPI:
self, self,
appliance: HonAppliance, appliance: HonAppliance,
command: str, command: str,
parameters: Dict, parameters: Dict[str, Any],
ancillary_parameters: Dict, ancillary_parameters: Dict[str, Any],
) -> bool: ) -> bool:
now: str = datetime.utcnow().isoformat() now: str = datetime.utcnow().isoformat()
data: Dict = { data: Dict[str, Any] = {
"macAddress": appliance.mac_address, "macAddress": appliance.mac_address,
"timestamp": f"{now[:-3]}Z", "timestamp": f"{now[:-3]}Z",
"commandName": command, "commandName": command,
@ -185,24 +196,26 @@ class HonAPI:
} }
url: str = f"{const.API_URL}/commands/v1/send" url: str = f"{const.API_URL}/commands/v1/send"
async with self._hon.post(url, json=data) as response: async with self._hon.post(url, json=data) as response:
json_data: Dict = await response.json() json_data: Dict[str, Any] = await response.json()
if json_data.get("payload", {}).get("resultCode") == "0": if json_data.get("payload", {}).get("resultCode") == "0":
return True return True
_LOGGER.error(await response.text()) _LOGGER.error(await response.text())
_LOGGER.error("%s - Payload:\n%s", url, pformat(data)) _LOGGER.error("%s - Payload:\n%s", url, pformat(data))
return False 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" url: str = f"{const.API_URL}/config/v1/program-list-rules"
async with self._hon_anonymous.get(url) as response: async with self._hon_anonymous.get(url) as response:
result: Dict = await response.json() result: Dict[str, Any] = await response.json()
if result and (data := result.get("payload")): if result and (data := result.get("payload")):
return data return data
return {} 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" url: str = f"{const.API_URL}/app-config"
payload_data: Dict = { payload_data: Dict[str, str | int] = {
"languageCode": language, "languageCode": language,
"beta": beta, "beta": beta,
"appVersion": const.APP_VERSION, "appVersion": const.APP_VERSION,
@ -214,7 +227,7 @@ class HonAPI:
return data return data
return {} 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) config = await self.app_config(language=language)
if url := config.get("language", {}).get("jsonPath"): if url := config.get("language", {}).get("jsonPath"):
async with self._hon_anonymous.get(url) as response: async with self._hon_anonymous.get(url) as response:
@ -227,3 +240,63 @@ class HonAPI:
await self._hon_handler.close() await self._hon_handler.close()
if self._hon_anonymous_handler is not None: if self._hon_anonymous_handler is not None:
await self._hon_anonymous_handler.close() await self._hon_anonymous_handler.close()
class TestAPI(HonAPI):
def __init__(self, path: Path):
super().__init__()
self._anonymous = True
self._path: Path = path
def _load_json(self, appliance: HonAppliance, file: str) -> Dict[str, Any]:
directory = f"{appliance.appliance_type}_{appliance.appliance_model_id}".lower()
if (path := self._path / directory / f"{file}.json").exists():
with open(path, "r", encoding="utf-8") as json_file:
return json.loads(json_file.read())
_LOGGER.warning(f"Can't open {str(path)}")
return {}
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[str, Any],
ancillary_parameters: Dict[str, Any],
) -> bool:
return True

View File

@ -6,14 +6,16 @@ import urllib
from contextlib import suppress from contextlib import suppress
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Dict, Optional from typing import Dict, Optional, Any
from urllib import parse from urllib import parse
from urllib.parse import quote from urllib.parse import quote
import aiohttp
from aiohttp import ClientResponse from aiohttp import ClientResponse
from yarl import URL from yarl import URL
from pyhon import const, exceptions from pyhon import const, exceptions
from pyhon.connection.device import HonDevice
from pyhon.connection.handler.auth import HonAuthConnectionHandler from pyhon.connection.handler.auth import HonAuthConnectionHandler
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -25,14 +27,20 @@ class HonLoginData:
email: str = "" email: str = ""
password: str = "" password: str = ""
fw_uid: str = "" fw_uid: str = ""
loaded: Optional[Dict] = None loaded: Optional[Dict[str, Any]] = None
class HonAuth: class HonAuth:
_TOKEN_EXPIRES_AFTER_HOURS = 8 _TOKEN_EXPIRES_AFTER_HOURS = 8
_TOKEN_EXPIRE_WARNING_HOURS = 7 _TOKEN_EXPIRE_WARNING_HOURS = 7
def __init__(self, session, email, password, device) -> None: def __init__(
self,
session: aiohttp.ClientSession,
email: str,
password: str,
device: HonDevice,
) -> None:
self._session = session self._session = session
self._request = HonAuthConnectionHandler(session) self._request = HonAuthConnectionHandler(session)
self._login_data = HonLoginData() self._login_data = HonLoginData()
@ -120,7 +128,7 @@ class HonAuth:
await self._error_logger(response) await self._error_logger(response)
return new_location return new_location
async def _handle_redirects(self, login_url) -> str: async def _handle_redirects(self, login_url: str) -> str:
redirect1 = await self._manual_redirect(login_url) redirect1 = await self._manual_redirect(login_url)
redirect2 = await self._manual_redirect(redirect1) redirect2 = await self._manual_redirect(redirect1)
return f"{redirect2}&System=IoT_Mobile_App&RegistrationSubChannel=hOn" return f"{redirect2}&System=IoT_Mobile_App&RegistrationSubChannel=hOn"

View File

@ -32,12 +32,14 @@ class HonDevice:
def mobile_id(self) -> str: def mobile_id(self) -> str:
return self._mobile_id return self._mobile_id
def get(self, mobile: bool = False) -> Dict: def get(self, mobile: bool = False) -> Dict[str, str | int]:
result = { result: Dict[str, str | int] = {
"appVersion": self.app_version, "appVersion": self.app_version,
"mobileId": self.mobile_id, "mobileId": self.mobile_id,
"os": self.os, "os": self.os,
"osVersion": self.os_version, "osVersion": self.os_version,
"deviceModel": self.device_model, "deviceModel": self.device_model,
} }
return (result | {"mobileOs": result.pop("os")}) if mobile else result if mobile:
result |= {"mobileOs": result.pop("os", "")}
return result

View File

@ -1,21 +1,27 @@
import logging import logging
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Callable, Dict from typing import Dict, Any
import aiohttp
from yarl import URL
from pyhon import const from pyhon import const
from pyhon.connection.handler.base import ConnectionHandler from pyhon.connection.handler.base import ConnectionHandler
from pyhon.typedefs import Callback
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
class HonAnonymousConnectionHandler(ConnectionHandler): class HonAnonymousConnectionHandler(ConnectionHandler):
_HEADERS: Dict = ConnectionHandler._HEADERS | {"x-api-key": const.API_KEY} _HEADERS: Dict[str, str] = ConnectionHandler._HEADERS | {"x-api-key": const.API_KEY}
@asynccontextmanager @asynccontextmanager
async def _intercept(self, method: Callable, *args, **kwargs) -> AsyncIterator: async def _intercept(
self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any]
) -> AsyncIterator[aiohttp.ClientResponse]:
kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS
async with method(*args, **kwargs) as response: async with method(url, *args, **kwargs) as response:
if response.status == 403: if response.status == 403:
_LOGGER.error("Can't authenticate anymore") _LOGGER.error("Can't authenticate anymore")
yield response yield response

View File

@ -1,12 +1,14 @@
import logging import logging
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Optional, Callable, List, Tuple from typing import Optional, List, Tuple, Any, Dict
import aiohttp import aiohttp
from yarl import URL
from pyhon import const from pyhon import const
from pyhon.connection.handler.base import ConnectionHandler from pyhon.connection.handler.base import ConnectionHandler
from pyhon.typedefs import Callback
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -28,9 +30,9 @@ class HonAuthConnectionHandler(ConnectionHandler):
@asynccontextmanager @asynccontextmanager
async def _intercept( async def _intercept(
self, method: Callable, *args, loop: int = 0, **kwargs self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any]
) -> AsyncIterator: ) -> AsyncIterator[aiohttp.ClientResponse]:
kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS
async with method(*args, **kwargs) as response: async with method(url, *args, **kwargs) as response:
self._called_urls.append((response.status, response.request_info.url)) self._called_urls.append((response.status, str(response.request_info.url)))
yield response yield response

View File

@ -1,18 +1,21 @@
import logging import logging
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Optional, Callable, Dict from types import TracebackType
from typing import Optional, Dict, Type, Any, Protocol
import aiohttp import aiohttp
from typing_extensions import Self from typing_extensions import Self
from yarl import URL
from pyhon import const, exceptions from pyhon import const, exceptions
from pyhon.typedefs import Callback
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
class ConnectionHandler: class ConnectionHandler:
_HEADERS: Dict = { _HEADERS: Dict[str, str] = {
"user-agent": const.USER_AGENT, "user-agent": const.USER_AGENT,
"Content-Type": "application/json", "Content-Type": "application/json",
} }
@ -24,32 +27,49 @@ class ConnectionHandler:
async def __aenter__(self) -> Self: async def __aenter__(self) -> Self:
return await self.create() return await self.create()
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc: Optional[BaseException],
traceback: Optional[TracebackType],
) -> None:
await self.close() await self.close()
@property
def session(self) -> aiohttp.ClientSession:
if self._session is None:
raise exceptions.NoSessionException
return self._session
async def create(self) -> Self: async def create(self) -> Self:
if self._create_session: if self._create_session:
self._session = aiohttp.ClientSession() self._session = aiohttp.ClientSession()
return self return self
@asynccontextmanager @asynccontextmanager
def _intercept(self, method: Callable, *args, loop: int = 0, **kwargs): def _intercept(
self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any]
) -> AsyncIterator[aiohttp.ClientResponse]:
raise NotImplementedError raise NotImplementedError
@asynccontextmanager @asynccontextmanager
async def get(self, *args, **kwargs) -> AsyncIterator[aiohttp.ClientResponse]: async def get(
self, *args: Any, **kwargs: Any
) -> AsyncIterator[aiohttp.ClientResponse]:
if self._session is None: if self._session is None:
raise exceptions.NoSessionException() raise exceptions.NoSessionException()
response: aiohttp.ClientResponse response: aiohttp.ClientResponse
async with self._intercept(self._session.get, *args, **kwargs) as response: async with self._intercept(self._session.get, *args, **kwargs) as response: # type: ignore[arg-type]
yield response yield response
@asynccontextmanager @asynccontextmanager
async def post(self, *args, **kwargs) -> AsyncIterator[aiohttp.ClientResponse]: async def post(
self, *args: Any, **kwargs: Any
) -> AsyncIterator[aiohttp.ClientResponse]:
if self._session is None: if self._session is None:
raise exceptions.NoSessionException() raise exceptions.NoSessionException()
response: aiohttp.ClientResponse response: aiohttp.ClientResponse
async with self._intercept(self._session.post, *args, **kwargs) as response: async with self._intercept(self._session.post, *args, **kwargs) as response: # type: ignore[arg-type]
yield response yield response
async def close(self) -> None: async def close(self) -> None:

View File

@ -2,15 +2,17 @@ import json
import logging import logging
from collections.abc import AsyncIterator from collections.abc import AsyncIterator
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Optional, Callable, Dict from typing import Optional, Dict, Any
import aiohttp import aiohttp
from typing_extensions import Self from typing_extensions import Self
from yarl import URL
from pyhon.connection.auth import HonAuth from pyhon.connection.auth import HonAuth
from pyhon.connection.device import HonDevice from pyhon.connection.device import HonDevice
from pyhon.connection.handler.base import ConnectionHandler from pyhon.connection.handler.base import ConnectionHandler
from pyhon.exceptions import HonAuthenticationError, NoAuthenticationException from pyhon.exceptions import HonAuthenticationError, NoAuthenticationException
from pyhon.typedefs import Callback
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -41,10 +43,10 @@ class HonConnectionHandler(ConnectionHandler):
async def create(self) -> Self: async def create(self) -> Self:
await super().create() await super().create()
self._auth = HonAuth(self._session, self._email, self._password, self._device) self._auth = HonAuth(self.session, self._email, self._password, self._device)
return self return self
async def _check_headers(self, headers: Dict) -> Dict: async def _check_headers(self, headers: Dict[str, str]) -> Dict[str, str]:
if not (self.auth.cognito_token and self.auth.id_token): if not (self.auth.cognito_token and self.auth.id_token):
await self.auth.authenticate() await self.auth.authenticate()
headers["cognito-token"] = self.auth.cognito_token headers["cognito-token"] = self.auth.cognito_token
@ -53,17 +55,18 @@ class HonConnectionHandler(ConnectionHandler):
@asynccontextmanager @asynccontextmanager
async def _intercept( async def _intercept(
self, method: Callable, *args, loop: int = 0, **kwargs self, method: Callback, url: str | URL, *args: Any, **kwargs: Any
) -> AsyncIterator: ) -> AsyncIterator[aiohttp.ClientResponse]:
loop: int = kwargs.pop("loop", 0)
kwargs["headers"] = await self._check_headers(kwargs.get("headers", {})) kwargs["headers"] = await self._check_headers(kwargs.get("headers", {}))
async with method(*args, **kwargs) as response: async with method(url, *args, **kwargs) as response:
if ( if (
self.auth.token_expires_soon or response.status in [401, 403] self.auth.token_expires_soon or response.status in [401, 403]
) and loop == 0: ) and loop == 0:
_LOGGER.info("Try refreshing token...") _LOGGER.info("Try refreshing token...")
await self.auth.refresh() await self.auth.refresh()
async with self._intercept( async with self._intercept(
method, *args, loop=loop + 1, **kwargs method, url, *args, loop=loop + 1, **kwargs
) as result: ) as result:
yield result yield result
elif ( elif (
@ -77,7 +80,7 @@ class HonConnectionHandler(ConnectionHandler):
) )
await self.create() await self.create()
async with self._intercept( async with self._intercept(
method, *args, loop=loop + 1, **kwargs method, url, *args, loop=loop + 1, **kwargs
) as result: ) as result:
yield result yield result
elif loop >= 2: elif loop >= 2:

100
pyhon/diagnose.py Normal file
View File

@ -0,0 +1,100 @@
import asyncio
import json
import re
import shutil
from pathlib import Path
from typing import TYPE_CHECKING, List, Tuple
from pyhon import printer
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", "1", 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) -> Path:
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
) -> str:
data = await appliance_data(appliance, path, anonymous)
archive = data[0].parent
shutil.make_archive(str(archive), "zip", archive)
shutil.rmtree(archive)
return f"{archive.stem}.zip"
def yaml_export(appliance: "HonAppliance", anonymous: bool = 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": printer.create_command(appliance.commands),
"rules": printer.create_rules(appliance.commands),
}
result = printer.pretty_print(data)
if anonymous:
result = anonymize_data(result)
return result

View File

@ -1,75 +1,5 @@
def key_print(data, key="", start=True): def str_to_float(string: str | float) -> float:
result = "" try:
if isinstance(data, list): return int(string)
for i, value in enumerate(data): except ValueError:
result += key_print(value, key=f"{key}.{i}", start=False) return float(str(string).replace(",", "."))
elif isinstance(data, dict):
for k, value in sorted(data.items()):
result += key_print(value, key=k if start else f"{key}.{k}", start=False)
else:
result += f"{key}: {data}\n"
return result
# yaml.dump() would be done the same, but needs an additional dependency...
def pretty_print(data, key="", intend=0, is_list=False, whitespace=" "):
result = ""
if isinstance(data, list):
if key:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
intend += 1
for i, value in enumerate(data):
result += pretty_print(
value, intend=intend, is_list=True, whitespace=whitespace
)
elif isinstance(data, dict):
if key:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
intend += 1
for i, (key, value) in enumerate(sorted(data.items())):
if is_list and not i:
result += pretty_print(
value, key=key, intend=intend, is_list=True, whitespace=whitespace
)
elif is_list:
result += pretty_print(
value, key=key, intend=intend + 1, whitespace=whitespace
)
else:
result += pretty_print(
value, key=key, intend=intend, whitespace=whitespace
)
else:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n"
return result
def create_command(commands, concat=False):
result = {}
for name, command in commands.items():
for parameter, data in command.available_settings.items():
if data.typology == "enum":
value = data.values
elif data.typology == "range":
value = {"min": data.min, "max": data.max, "step": data.step}
else:
continue
if not concat:
result.setdefault(name, {})[parameter] = value
else:
result[f"{name}.{parameter}"] = value
return result
def create_rules(commands, concat=False):
result = {}
for name, command in commands.items():
for parameter, data in command.available_settings.items():
value = data.triggers
if not value:
continue
if not concat:
result.setdefault(name, {})[parameter] = value
else:
result[f"{name}.{parameter}"] = value
return result

View File

@ -1,5 +1,6 @@
import asyncio import asyncio
import logging import logging
from pathlib import Path
from types import TracebackType from types import TracebackType
from typing import List, Optional, Dict, Any, Type from typing import List, Optional, Dict, Any, Type
@ -8,6 +9,7 @@ from typing_extensions import Self
from pyhon import HonAPI, exceptions from pyhon import HonAPI, exceptions
from pyhon.appliance import HonAppliance from pyhon.appliance import HonAppliance
from pyhon.connection.api import TestAPI
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -18,12 +20,14 @@ class Hon:
email: Optional[str] = "", email: Optional[str] = "",
password: Optional[str] = "", password: Optional[str] = "",
session: Optional[ClientSession] = None, session: Optional[ClientSession] = None,
test_data_path: Optional[Path] = None,
): ):
self._email: Optional[str] = email self._email: Optional[str] = email
self._password: Optional[str] = password self._password: Optional[str] = password
self._session: ClientSession | None = session self._session: ClientSession | None = session
self._appliances: List[HonAppliance] = [] self._appliances: List[HonAppliance] = []
self._api: Optional[HonAPI] = None self._api: Optional[HonAPI] = None
self._test_data_path: Path = test_data_path or Path().cwd()
async def __aenter__(self) -> Self: async def __aenter__(self) -> Self:
return await self.create() return await self.create()
@ -66,11 +70,13 @@ class Hon:
return self._appliances return self._appliances
@appliances.setter @appliances.setter
def appliances(self, appliances) -> None: def appliances(self, appliances: List[HonAppliance]) -> None:
self._appliances = appliances self._appliances = appliances
async def _create_appliance(self, appliance_data: Dict[str, Any], zone=0) -> None: async def _create_appliance(
appliance = HonAppliance(self._api, appliance_data, zone=zone) self, appliance_data: Dict[str, Any], api: HonAPI, zone: int = 0
) -> None:
appliance = HonAppliance(api, appliance_data, zone=zone)
if appliance.mac_address == "": if appliance.mac_address == "":
return return
try: try:
@ -87,12 +93,20 @@ class Hon:
self._appliances.append(appliance) self._appliances.append(appliance)
async def setup(self) -> None: async def setup(self) -> None:
appliance: Dict appliances = await self.api.load_appliances()
for appliance in (await self.api.load_appliances())["payload"]["appliances"]: for appliance in appliances:
if (zones := int(appliance.get("zone", "0"))) > 1: if (zones := int(appliance.get("zone", "0"))) > 1:
for zone in range(zones): for zone in range(zones):
await self._create_appliance(appliance.copy(), zone=zone + 1) await self._create_appliance(
await self._create_appliance(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: async def close(self) -> None:
await self.api.close() await self.api.close()

View File

@ -12,7 +12,9 @@ class HonParameter:
self._mandatory: int = attributes.get("mandatory", 0) self._mandatory: int = attributes.get("mandatory", 0)
self._value: str | float = "" self._value: str | float = ""
self._group: str = group self._group: str = group
self._triggers: Dict[str, List[Tuple[Callable, "HonRule"]]] = {} self._triggers: Dict[
str, List[Tuple[Callable[["HonRule"], None], "HonRule"]]
] = {}
@property @property
def key(self) -> str: def key(self) -> str:
@ -51,20 +53,22 @@ class HonParameter:
def group(self) -> str: def group(self) -> str:
return self._group return self._group
def add_trigger(self, value, func, data): def add_trigger(
self, value: str, func: Callable[["HonRule"], None], data: "HonRule"
) -> None:
if self._value == value: if self._value == value:
func(data) func(data)
self._triggers.setdefault(value, []).append((func, data)) self._triggers.setdefault(value, []).append((func, data))
def check_trigger(self, value) -> None: def check_trigger(self, value: str | float) -> None:
if str(value) in self._triggers: if str(value) in self._triggers:
for trigger in self._triggers[str(value)]: for trigger in self._triggers[str(value)]:
func, args = trigger func, args = trigger
func(args) func(args)
@property @property
def triggers(self): def triggers(self) -> Dict[str, Any]:
result = {} result: Dict[str, Any] = {}
for value, rules in self._triggers.items(): for value, rules in self._triggers.items():
for _, rule in rules: for _, rule in rules:
if rule.extras: if rule.extras:

View File

@ -3,7 +3,7 @@ from typing import Dict, Any, List
from pyhon.parameter.base import HonParameter from pyhon.parameter.base import HonParameter
def clean_value(value): def clean_value(value: str | float) -> str:
return str(value).strip("[]").replace("|", "_").lower() return str(value).strip("[]").replace("|", "_").lower()
@ -24,7 +24,7 @@ class HonParameterEnum(HonParameter):
return [clean_value(value) for value in self._values] return [clean_value(value) for value in self._values]
@values.setter @values.setter
def values(self, values) -> None: def values(self, values: List[str]) -> None:
self._values = values self._values = values
@property @property

View File

@ -36,7 +36,7 @@ class HonParameterProgram(HonParameterEnum):
return sorted(values) return sorted(values)
@values.setter @values.setter
def values(self, values) -> None: def values(self, values: List[str]) -> None:
return return
@property @property
@ -50,5 +50,5 @@ class HonParameterProgram(HonParameterEnum):
} }
return dict(sorted(values.items())) return dict(sorted(values.items()))
def set_value(self, value: str): def set_value(self, value: str) -> None:
self._value = value self._value = value

View File

@ -1,15 +1,9 @@
from typing import Dict, Any, List from typing import Dict, Any, List
from pyhon.helper import str_to_float
from pyhon.parameter.base import HonParameter from pyhon.parameter.base import HonParameter
def str_to_float(string: str | float) -> float:
try:
return int(string)
except ValueError:
return float(str(string).replace(",", "."))
class HonParameterRange(HonParameter): class HonParameterRange(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
super().__init__(key, attributes, group) super().__init__(key, attributes, group)
@ -19,7 +13,7 @@ class HonParameterRange(HonParameter):
self._default: float = str_to_float(attributes.get("defaultValue", self.min)) self._default: float = str_to_float(attributes.get("defaultValue", self.min))
self._value: float = self._default self._value: float = self._default
def __repr__(self): def __repr__(self) -> str:
return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])" return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])"
@property @property

94
pyhon/printer.py Normal file
View File

@ -0,0 +1,94 @@
from typing import Dict, Any, TYPE_CHECKING, List
from pyhon.parameter.enum import HonParameterEnum
from pyhon.parameter.range import HonParameterRange
if TYPE_CHECKING:
from pyhon.commands import HonCommand
def key_print(data: Any, key: str = "", start: bool = True) -> str:
result = ""
if isinstance(data, list):
for i, value in enumerate(data):
result += key_print(value, key=f"{key}.{i}", start=False)
elif isinstance(data, dict):
for k, value in sorted(data.items()):
result += key_print(value, key=k if start else f"{key}.{k}", start=False)
else:
result += f"{key}: {data}\n"
return result
# yaml.dump() would be done the same, but needs an additional dependency...
def pretty_print(
data: Any,
key: str = "",
intend: int = 0,
is_list: bool = False,
whitespace: str = " ",
) -> str:
result = ""
if isinstance(data, list):
if key:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
intend += 1
for i, value in enumerate(data):
result += pretty_print(
value, intend=intend, is_list=True, whitespace=whitespace
)
elif isinstance(data, dict):
if key:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
intend += 1
for i, (key, value) in enumerate(sorted(data.items())):
if is_list and not i:
result += pretty_print(
value, key=key, intend=intend, is_list=True, whitespace=whitespace
)
elif is_list:
result += pretty_print(
value, key=key, intend=intend + 1, whitespace=whitespace
)
else:
result += pretty_print(
value, key=key, intend=intend, whitespace=whitespace
)
else:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n"
return result
def create_command(
commands: Dict[str, "HonCommand"], concat: bool = False
) -> Dict[str, Any]:
result: Dict[str, Any] = {}
for name, command in commands.items():
for parameter, data in command.available_settings.items():
if isinstance(data, HonParameterEnum):
value: List[str] | Dict[str, str | float] = data.values
elif isinstance(data, HonParameterRange):
value = {"min": data.min, "max": data.max, "step": data.step}
else:
continue
if not concat:
result.setdefault(name, {})[parameter] = value
else:
result[f"{name}.{parameter}"] = value
return result
def create_rules(
commands: Dict[str, "HonCommand"], concat: bool = False
) -> Dict[str, Any]:
result: Dict[str, Any] = {}
for name, command in commands.items():
for parameter, data in command.available_settings.items():
value = data.triggers
if not value:
continue
if not concat:
result.setdefault(name, {})[parameter] = value
else:
result[f"{name}.{parameter}"] = value
return result

View File

@ -6,6 +6,7 @@ from pyhon.parameter.range import HonParameterRange
if TYPE_CHECKING: if TYPE_CHECKING:
from pyhon.commands import HonCommand from pyhon.commands import HonCommand
from pyhon.parameter.base import HonParameter
@dataclass @dataclass
@ -18,18 +19,24 @@ class HonRule:
class HonRuleSet: class HonRuleSet:
def __init__(self, command: "HonCommand", rule): def __init__(self, command: "HonCommand", rule: Dict[str, Any]):
self._command: "HonCommand" = command self._command: "HonCommand" = command
self._rules: Dict[str, List[HonRule]] = {} self._rules: Dict[str, List[HonRule]] = {}
self._parse_rule(rule) self._parse_rule(rule)
def _parse_rule(self, rule): def _parse_rule(self, rule: Dict[str, Any]) -> None:
for param_key, params in rule.items(): for param_key, params in rule.items():
param_key = self._command.appliance.options.get(param_key, param_key) param_key = self._command.appliance.options.get(param_key, param_key)
for trigger_key, trigger_data in params.items(): for trigger_key, trigger_data in params.items():
self._parse_conditions(param_key, trigger_key, trigger_data) self._parse_conditions(param_key, trigger_key, trigger_data)
def _parse_conditions(self, param_key, trigger_key, trigger_data, extra=None): def _parse_conditions(
self,
param_key: str,
trigger_key: str,
trigger_data: Dict[str, Any],
extra: Optional[Dict[str, str]] = None,
) -> None:
trigger_key = trigger_key.replace("@", "") trigger_key = trigger_key.replace("@", "")
trigger_key = self._command.appliance.options.get(trigger_key, trigger_key) trigger_key = self._command.appliance.options.get(trigger_key, trigger_key)
for multi_trigger_value, param_data in trigger_data.items(): for multi_trigger_value, param_data in trigger_data.items():
@ -46,16 +53,21 @@ class HonRuleSet:
self._parse_conditions(param_key, extra_key, extra_data, extra) self._parse_conditions(param_key, extra_key, extra_data, extra)
def _create_rule( def _create_rule(
self, param_key, trigger_key, trigger_value, param_data, extras=None self,
): param_key: str,
trigger_key: str,
trigger_value: str,
param_data: Dict[str, Any],
extras: Optional[Dict[str, str]] = None,
) -> None:
if param_data.get("fixedValue") == f"@{param_key}": if param_data.get("fixedValue") == f"@{param_key}":
return return
self._rules.setdefault(trigger_key, []).append( self._rules.setdefault(trigger_key, []).append(
HonRule(trigger_key, trigger_value, param_key, param_data, extras) HonRule(trigger_key, trigger_value, param_key, param_data, extras)
) )
def _duplicate_for_extra_conditions(self): def _duplicate_for_extra_conditions(self) -> None:
new = {} new: Dict[str, List[HonRule]] = {}
for rules in self._rules.values(): for rules in self._rules.values():
for rule in rules: for rule in rules:
if rule.extras is None: if rule.extras is None:
@ -71,8 +83,8 @@ class HonRuleSet:
for rule in rules: for rule in rules:
self._rules.setdefault(key, []).append(rule) self._rules.setdefault(key, []).append(rule)
def _add_trigger(self, parameter, data): def _add_trigger(self, parameter: "HonParameter", data: HonRule) -> None:
def apply(rule: HonRule): def apply(rule: HonRule) -> None:
if rule.extras is not None: if rule.extras is not None:
for key, value in rule.extras.items(): for key, value in rule.extras.items():
if str(self._command.parameters.get(key)) != str(value): if str(self._command.parameters.get(key)) != str(value):
@ -96,10 +108,10 @@ class HonRuleSet:
parameter.add_trigger(data.trigger_value, apply, data) parameter.add_trigger(data.trigger_value, apply, data)
def patch(self): def patch(self) -> None:
self._duplicate_for_extra_conditions() self._duplicate_for_extra_conditions()
for name, parameter in self._command.parameters.items(): for name, parameter in self._command.parameters.items():
if name not in self._rules: if name not in self._rules:
continue continue
for data in self._rules.get(name): for data in self._rules.get(name, []):
self._add_trigger(parameter, data) self._add_trigger(parameter, data)

27
pyhon/typedefs.py Normal file
View File

@ -0,0 +1,27 @@
from typing import Union, Any, TYPE_CHECKING, Protocol
import aiohttp
from yarl import URL
if TYPE_CHECKING:
from pyhon.parameter.base import HonParameter
from pyhon.parameter.enum import HonParameterEnum
from pyhon.parameter.fixed import HonParameterFixed
from pyhon.parameter.program import HonParameterProgram
from pyhon.parameter.range import HonParameterRange
class Callback(Protocol):
def __call__(
self, url: str | URL, *args: Any, **kwargs: Any
) -> aiohttp.client._RequestContextManager:
...
Parameter = Union[
"HonParameter",
"HonParameterRange",
"HonParameterEnum",
"HonParameterFixed",
"HonParameterProgram",
]

View File

@ -7,7 +7,7 @@ with open("README.md", "r") as f:
setup( setup(
name="pyhOn", name="pyhOn",
version="0.13.1", version="0.14.6",
author="Andre Basche", author="Andre Basche",
description="Control hOn devices with python", description="Control hOn devices with python",
long_description=long_description, long_description=long_description,