From ea8f481b0126f38493d59c706ec5694dfd1bd5a9 Mon Sep 17 00:00:00 2001 From: Andre Basche Date: Sat, 6 May 2023 16:07:28 +0200 Subject: [PATCH] More general parsing --- pyhon/appliance.py | 118 ++++++++++++++++++++-------------- pyhon/commands.py | 128 ++++++++++++++++++------------------- pyhon/connection/api.py | 14 +++- pyhon/helper.py | 2 +- pyhon/parameter/base.py | 9 ++- pyhon/parameter/enum.py | 4 +- pyhon/parameter/fixed.py | 4 +- pyhon/parameter/program.py | 8 +-- pyhon/parameter/range.py | 4 +- setup.py | 2 +- 10 files changed, 165 insertions(+), 128 deletions(-) diff --git a/pyhon/appliance.py b/pyhon/appliance.py index e733646..22b95a2 100644 --- a/pyhon/appliance.py +++ b/pyhon/appliance.py @@ -12,7 +12,6 @@ from pyhon.parameter.fixed import HonParameterFixed if TYPE_CHECKING: from pyhon import HonAPI - _LOGGER = logging.getLogger(__name__) @@ -30,6 +29,7 @@ class HonAppliance: self._statistics: Dict = {} self._attributes: Dict = {} self._zone: int = zone + self._additional_data = {} try: self._extra = importlib.import_module( @@ -111,13 +111,17 @@ class HonAppliance: def info(self): return self._info + @property + def additional_data(self): + return self._additional_data + @property def zone(self) -> int: return self._zone - async def _recover_last_command_states(self, commands): + async def _recover_last_command_states(self): command_history = await self._api.command_history(self) - for name, command in commands.items(): + for name, command in self._commands.items(): last = next( ( index @@ -129,8 +133,8 @@ class HonAppliance: if last is None: continue parameters = command_history[last].get("command", {}).get("parameters", {}) - if command.programs and parameters.get("program"): - command.program = parameters.pop("program").split(".")[-1].lower() + if command.categories and parameters.get("category"): + command.category = parameters.pop("category").split(".")[-1].lower() command = self.commands[name] for key, data in command.settings.items(): if ( @@ -140,52 +144,50 @@ class HonAppliance: with suppress(ValueError): data.value = parameters.get(key) + def _get_categories(self, command, data): + 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: + 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._api, + self, + category_name=category, + categories=categories, + ) + ] + else: + commands += self._get_categories(command, 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") - for item in ["settings", "options", "dictionaryId"]: - raw.pop(item) - commands = {} - for command, attr in raw.items(): - if "parameters" in attr: - commands[command] = HonCommand(command, attr, self._api, self) - elif "parameters" in attr[list(attr)[0]]: - multi = {} - for program, attr2 in attr.items(): - program = program.split(".")[-1].lower() - cmd = HonCommand( - command, - attr2, - self._api, - self, - programs=multi, - program_name=program, - ) - multi[program] = cmd - commands[command] = cmd - self._commands = commands - await self._recover_last_command_states(commands) - - @property - def settings(self): - result = {} - for name, command in self._commands.items(): - for key in command.setting_keys: - setting = command.settings.get(key, HonParameter(key, {})) - result[f"{name}.{key}"] = setting - if self._extra: - return self._extra.settings(result) - return result - - @property - def parameters(self): - result = {} - for name, command in self._commands.items(): - for key, parameter in ( - command.parameters | command.ancillary_parameters - ).items(): - result.setdefault(name, {})[key] = parameter.value - return result + raw.pop("dictionaryId") + self._commands = self._get_commands(raw) + await self._recover_last_command_states() async def load_attributes(self): self._attributes = await self._api.load_attributes(self) @@ -198,12 +200,32 @@ class HonAppliance: async def update(self): await self.load_attributes() + @property + def parameters(self): + result = {} + for name, command in self._commands.items(): + for key, parameter in command.parameters.items(): + result.setdefault(name, {})[key] = parameter.value + return result + + @property + def settings(self): + result = {} + for name, command in self._commands.items(): + for key in command.setting_keys: + setting = command.settings.get(key, HonParameter(key, {}, name)) + result[f"{name}.{key}"] = setting + if self._extra: + return self._extra.settings(result) + return result + @property def data(self): result = { "attributes": self.attributes, "appliance": self.info, "statistics": self.statistics, + "additional_data": self._additional_data, **self.parameters, } if self._extra: diff --git a/pyhon/commands.py b/pyhon/commands.py index f0075f3..180a4bf 100644 --- a/pyhon/commands.py +++ b/pyhon/commands.py @@ -18,97 +18,95 @@ class HonCommand: attributes: Dict[str, Any], api: "HonAPI", appliance: "HonAppliance", - programs: Optional[Dict[str, "HonCommand"]] = None, - program_name: str = "", + categories: Optional[Dict[str, "HonCommand"]] = None, + category_name: str = "", ): self._api: HonAPI = api self._appliance: "HonAppliance" = appliance self._name: str = name - self._programs: Optional[Dict[str, "HonCommand"]] = programs - self._program_name: str = program_name - self._description: str = attributes.get("description", "") - self._parameters: Dict[str, HonParameter] = self._create_parameters( - attributes.get("parameters", {}) - ) - self._ancillary_parameters: Dict[str, HonParameter] = self._create_parameters( - attributes.get("ancillaryParameters", {}) - ) + self._categories: Optional[Dict[str, "HonCommand"]] = categories + self._category_name: str = category_name + self._description: str = attributes.pop("description", "") + self._protocol_type: str = attributes.pop("protocolType", "") + self._parameters: Dict[str, HonParameter] = {} + self._data = {} + self._load_parameters(attributes) def __repr__(self) -> str: return f"{self._name} command" - def _create_parameters(self, parameters: Dict) -> Dict[str, HonParameter]: - result: Dict[str, HonParameter] = {} - for parameter, attributes in parameters.items(): - if parameter == "zoneMap" and self._appliance.zone: - attributes["default"] = self._appliance.zone - match attributes.get("typology"): - case "range": - result[parameter] = HonParameterRange(parameter, attributes) - case "enum": - result[parameter] = HonParameterEnum(parameter, attributes) - case "fixed": - result[parameter] = HonParameterFixed(parameter, attributes) - if self._programs: - result["program"] = HonParameterProgram("program", self) - return result + @property + def name(self): + return self._name + + @property + def data(self): + return self._data @property def parameters(self) -> Dict[str, HonParameter]: return self._parameters - @property - def ancillary_parameters(self) -> Dict[str, HonParameter]: - return self._ancillary_parameters + def _load_parameters(self, attributes): + for key, items in attributes.items(): + for name, data in items.items(): + self._create_parameters(data, name, key) + + def _create_parameters(self, data: Dict, name: str, parameter: str) -> None: + if name == "zoneMap" and self._appliance.zone: + data["default"] = self._appliance.zone + match data.get("typology"): + case "range": + self._parameters[name] = HonParameterRange(name, data, parameter) + case "enum": + self._parameters[name] = HonParameterEnum(name, data, parameter) + case "fixed": + self._parameters[name] = HonParameterFixed(name, data, parameter) + case _: + self._data[name] = data + return + if self._category_name: + if not self._categories: + self._parameters["program"] = HonParameterProgram("program", self, name) + + def _parameters_by_group(self, group): + return { + name: v.value for name, v in self._parameters.items() if v.group == group + } async def send(self) -> bool: - params = {k: v.value for k, v in self._parameters.items()} - ancillary_params = {k: v.value for k, v in self._ancillary_parameters.items()} + params = self._parameters_by_group("parameters") + ancillary_params = self._parameters_by_group("ancillary_parameters") return await self._api.send_command( self._appliance, self._name, params, ancillary_params ) @property - def programs(self) -> Dict[str, "HonCommand"]: - if self._programs is None: - return {} - return self._programs + def categories(self) -> Dict[str, "HonCommand"]: + if self._categories is None: + return {"_": self} + return self._categories @property - def program(self) -> str: - return self._program_name + def category(self) -> str: + return self._category_name - @program.setter - def program(self, program: str) -> None: - self._appliance.commands[self._name] = self.programs[program] - - def _get_settings_keys(self, command: Optional["HonCommand"] = None) -> List[str]: - if command is None: - command = self - keys = [] - for key, parameter in ( - command._parameters | command._ancillary_parameters - ).items(): - if key not in keys: - keys.append(key) - return keys + @category.setter + def category(self, category: str) -> None: + self._appliance.commands[self._name] = self.categories[category] @property def setting_keys(self) -> List[str]: - if not self._programs: - return self._get_settings_keys() - result = [ - key - for cmd in self._programs.values() - for key in self._get_settings_keys(cmd) - ] - return list(set(result + ["program"])) + return list( + {param for cmd in self.categories.values() for param in cmd.parameters} + ) @property def settings(self) -> Dict[str, HonParameter]: - return { - s: param - for s in self.setting_keys - if (param := self._parameters.get(s)) is not None - or (param := self._ancillary_parameters.get(s)) is not None - } + result = {} + for command in self.categories.values(): + for name, parameter in command.parameters.items(): + if name in result: + continue + result[name] = parameter + return result diff --git a/pyhon/connection/api.py b/pyhon/connection/api.py index f40f5b8..87cc757 100644 --- a/pyhon/connection/api.py +++ b/pyhon/connection/api.py @@ -109,6 +109,18 @@ class HonAPI: return activity return {} + async def appliance_model(self, appliance: HonAppliance) -> Dict: + url: str = f"{const.API_URL}/commands/v1/appliance-model" + params: Dict = { + "code": appliance.info["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 + return {} + async def load_attributes(self, appliance: HonAppliance) -> Dict: params: Dict = { "macAddress": appliance.mac_address, @@ -161,7 +173,7 @@ class HonAPI: return False async def appliance_configuration(self) -> Dict: - url: str = f"{const.API_URL}/config/v1/appliance-configuration" + 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() if result and (data := result.get("payload")): diff --git a/pyhon/helper.py b/pyhon/helper.py index bb1c00f..9ee8c75 100644 --- a/pyhon/helper.py +++ b/pyhon/helper.py @@ -48,7 +48,7 @@ def pretty_print(data, key="", intend=0, is_list=False, whitespace=" "): def get_parameter(command, parameter): - if programs := command.programs: + if programs := command.categories: for program in programs.values(): if data := program.settings.get(parameter): return data diff --git a/pyhon/parameter/base.py b/pyhon/parameter/base.py index 3672d29..ac14f3b 100644 --- a/pyhon/parameter/base.py +++ b/pyhon/parameter/base.py @@ -2,12 +2,13 @@ from typing import Dict, Any, List class HonParameter: - def __init__(self, key: str, attributes: Dict[str, Any]) -> None: + def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: self._key = key self._category: str = attributes.get("category", "") self._typology: str = attributes.get("typology", "") self._mandatory: int = attributes.get("mandatory", 0) self._value: str | float = "" + self._group: str = group @property def key(self) -> str: @@ -19,7 +20,7 @@ class HonParameter: @property def values(self) -> List[str]: - return list(str(self.value)) + return [str(self.value)] @property def category(self) -> str: @@ -32,3 +33,7 @@ class HonParameter: @property def mandatory(self) -> int: return self._mandatory + + @property + def group(self) -> str: + return self._group diff --git a/pyhon/parameter/enum.py b/pyhon/parameter/enum.py index c5afcda..02d01b4 100644 --- a/pyhon/parameter/enum.py +++ b/pyhon/parameter/enum.py @@ -4,8 +4,8 @@ from pyhon.parameter.base import HonParameter class HonParameterEnum(HonParameter): - def __init__(self, key: str, attributes: Dict[str, Any]) -> None: - super().__init__(key, attributes) + def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: + super().__init__(key, attributes, group) self._default = attributes.get("defaultValue") self._value = self._default or "0" self._values: List[str] = attributes.get("enumValues", []) diff --git a/pyhon/parameter/fixed.py b/pyhon/parameter/fixed.py index 52595ed..eb925e0 100644 --- a/pyhon/parameter/fixed.py +++ b/pyhon/parameter/fixed.py @@ -4,8 +4,8 @@ from pyhon.parameter.base import HonParameter class HonParameterFixed(HonParameter): - def __init__(self, key: str, attributes: Dict[str, Any]) -> None: - super().__init__(key, attributes) + def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: + super().__init__(key, attributes, group) self._value = attributes.get("fixedValue", None) def __repr__(self) -> str: diff --git a/pyhon/parameter/program.py b/pyhon/parameter/program.py index 80ff38a..37b807d 100644 --- a/pyhon/parameter/program.py +++ b/pyhon/parameter/program.py @@ -9,11 +9,11 @@ if TYPE_CHECKING: class HonParameterProgram(HonParameterEnum): _FILTER = ["iot_recipe", "iot_guided"] - def __init__(self, key: str, command: "HonCommand") -> None: - super().__init__(key, {}) + def __init__(self, key: str, command: "HonCommand", group: str) -> None: + super().__init__(key, {}, group) self._command = command - self._value: str = command.program - self._programs: Dict[str, "HonCommand"] = command.programs + self._value: str = command.category + self._programs: Dict[str, "HonCommand"] = command.categories self._typology: str = "enum" @property diff --git a/pyhon/parameter/range.py b/pyhon/parameter/range.py index e58accb..ca6fb1b 100644 --- a/pyhon/parameter/range.py +++ b/pyhon/parameter/range.py @@ -11,8 +11,8 @@ def str_to_float(string: str | float) -> float: class HonParameterRange(HonParameter): - def __init__(self, key: str, attributes: Dict[str, Any]) -> None: - super().__init__(key, attributes) + def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: + super().__init__(key, attributes, group) self._min: float = str_to_float(attributes["minimumValue"]) self._max: float = str_to_float(attributes["maximumValue"]) self._step: float = str_to_float(attributes["incrementValue"]) diff --git a/setup.py b/setup.py index ca6fad6..f665f08 100644 --- a/setup.py +++ b/setup.py @@ -7,7 +7,7 @@ with open("README.md", "r") as f: setup( name="pyhOn", - version="0.9.1", + version="0.10.0b0", author="Andre Basche", description="Control hOn devices with python", long_description=long_description,