Compare commits
	
		
			27 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| 7bd3aac7c5 | |||
| 365a37b42d | |||
| 2bde6bb61c | |||
| ccff32e6c1 | |||
| 22cbd7474a | |||
| dd61b24eed | |||
| ea8f481b01 | |||
| 7dcb34559b | |||
| 5db13a90e7 | |||
| 9ee5dbc956 | |||
| d4c6ccdce3 | |||
| 9594b9ebd8 | |||
| b011d98e07 | |||
| ad864286fc | |||
| 13cff8caa0 | |||
| 5fc6245806 | |||
| 1dad0e14b8 | |||
| b04c601ad6 | |||
| 3ec0f5a006 | |||
| 78bc85132f | |||
| 191928287f | |||
| c0aab8b99d | |||
| a68dcac379 | |||
| 40cc0013a5 | |||
| f2aa3dc8fd | |||
| e887371bec | |||
| 18b0ecdd68 | 
| @ -78,12 +78,7 @@ async def main(): | |||||||
|                     ) |                     ) | ||||||
|                 ) |                 ) | ||||||
|             else: |             else: | ||||||
|                 print(helper.pretty_print({"data": device.data})) |                 print(device.diagnose("  ")) | ||||||
|                 print( |  | ||||||
|                     helper.pretty_print( |  | ||||||
|                         {"settings": helper.create_command(device.commands)} |  | ||||||
|                     ) |  | ||||||
|                 ) |  | ||||||
|  |  | ||||||
|  |  | ||||||
| def start(): | def start(): | ||||||
|  | |||||||
| @ -1,17 +1,23 @@ | |||||||
| import importlib | import importlib | ||||||
|  | import logging | ||||||
| from contextlib import suppress | from contextlib import suppress | ||||||
|  | from datetime import datetime, timedelta | ||||||
| from typing import Optional, Dict, Any | from typing import Optional, Dict, Any | ||||||
| from typing import TYPE_CHECKING | from typing import TYPE_CHECKING | ||||||
|  |  | ||||||
| from pyhon import helper | from pyhon import helper, exceptions | ||||||
| from pyhon.commands import HonCommand | from pyhon.commands import HonCommand | ||||||
| from pyhon.parameter.fixed import HonParameterFixed | from pyhon.parameter.fixed import HonParameterFixed | ||||||
|  |  | ||||||
| if TYPE_CHECKING: | if TYPE_CHECKING: | ||||||
|     from pyhon import HonAPI |     from pyhon import HonAPI | ||||||
|  |  | ||||||
|  | _LOGGER = logging.getLogger(__name__) | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonAppliance: | class HonAppliance: | ||||||
|  |     _MINIMAL_UPDATE_INTERVAL = 5  # seconds | ||||||
|  |  | ||||||
|     def __init__( |     def __init__( | ||||||
|         self, api: Optional["HonAPI"], info: Dict[str, Any], zone: int = 0 |         self, api: Optional["HonAPI"], info: Dict[str, Any], zone: int = 0 | ||||||
|     ) -> None: |     ) -> None: | ||||||
| @ -25,11 +31,13 @@ class HonAppliance: | |||||||
|         self._statistics: Dict = {} |         self._statistics: Dict = {} | ||||||
|         self._attributes: Dict = {} |         self._attributes: Dict = {} | ||||||
|         self._zone: int = zone |         self._zone: int = zone | ||||||
|  |         self._additional_data: Dict[str, Any] = {} | ||||||
|  |         self._last_update = None | ||||||
|  |  | ||||||
|         try: |         try: | ||||||
|             self._extra = importlib.import_module( |             self._extra = importlib.import_module( | ||||||
|                 f"pyhon.appliances.{self.appliance_type.lower()}" |                 f"pyhon.appliances.{self.appliance_type.lower()}" | ||||||
|             ).Appliance() |             ).Appliance(self) | ||||||
|         except ModuleNotFoundError: |         except ModuleNotFoundError: | ||||||
|             self._extra = None |             self._extra = None | ||||||
|  |  | ||||||
| @ -106,13 +114,23 @@ class HonAppliance: | |||||||
|     def info(self): |     def info(self): | ||||||
|         return self._info |         return self._info | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def additional_data(self): | ||||||
|  |         return self._additional_data | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def zone(self) -> int: |     def zone(self) -> int: | ||||||
|         return self._zone |         return self._zone | ||||||
|  |  | ||||||
|     async def _recover_last_command_states(self, commands): |     @property | ||||||
|         command_history = await self._api.command_history(self) |     def api(self) -> "HonAPI": | ||||||
|         for name, command in commands.items(): |         if self._api is None: | ||||||
|  |             raise exceptions.NoAuthenticationException | ||||||
|  |         return self._api | ||||||
|  |  | ||||||
|  |     async def _recover_last_command_states(self): | ||||||
|  |         command_history = await self.api.command_history(self) | ||||||
|  |         for name, command in self._commands.items(): | ||||||
|             last = next( |             last = next( | ||||||
|                 ( |                 ( | ||||||
|                     index |                     index | ||||||
| @ -124,8 +142,13 @@ class HonAppliance: | |||||||
|             if last is None: |             if last is None: | ||||||
|                 continue |                 continue | ||||||
|             parameters = command_history[last].get("command", {}).get("parameters", {}) |             parameters = command_history[last].get("command", {}).get("parameters", {}) | ||||||
|             if command.programs and parameters.get("program"): |             if command.categories and ( | ||||||
|                 command.program = parameters.pop("program").split(".")[-1].lower() |                 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] |                 command = self.commands[name] | ||||||
|             for key, data in command.settings.items(): |             for key, data in command.settings.items(): | ||||||
|                 if ( |                 if ( | ||||||
| @ -135,83 +158,119 @@ class HonAppliance: | |||||||
|                     with suppress(ValueError): |                     with suppress(ValueError): | ||||||
|                         data.value = parameters.get(key) |                         data.value = parameters.get(key) | ||||||
|  |  | ||||||
|     async def load_commands(self): |     def _get_categories(self, command, data): | ||||||
|         raw = await self._api.load_commands(self) |         categories = {} | ||||||
|         self._appliance_model = raw.pop("applianceModel") |         for category, value in data.items(): | ||||||
|         for item in ["settings", "options", "dictionaryId"]: |             result = self._get_command(value, command, category, categories) | ||||||
|             raw.pop(item) |             if result: | ||||||
|         commands = {} |                 if "PROGRAM" in category: | ||||||
|         for command, attr in raw.items(): |                     category = category.split(".")[-1].lower() | ||||||
|             if "parameters" in attr: |                 categories[category] = result[0] | ||||||
|                 commands[command] = HonCommand(command, attr, self._api, self) |         if categories: | ||||||
|             elif "parameters" in attr[list(attr)[0]]: |             return [list(categories.values())[0]] | ||||||
|                 multi = {} |         return [] | ||||||
|                 for program, attr2 in attr.items(): |  | ||||||
|                     program = program.split(".")[-1].lower() |     def _get_commands(self, data): | ||||||
|                     cmd = HonCommand( |         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, |                         command, | ||||||
|                         attr2, |                         data, | ||||||
|                         self._api, |  | ||||||
|                         self, |                         self, | ||||||
|                         programs=multi, |                         category_name=category, | ||||||
|                         program_name=program, |                         categories=categories, | ||||||
|                     ) |                     ) | ||||||
|                     multi[program] = cmd |                 ] | ||||||
|                     commands[command] = cmd |             else: | ||||||
|         self._commands = commands |                 commands += self._get_categories(command, data) | ||||||
|         await self._recover_last_command_states(commands) |         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") | ||||||
|  |         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) | ||||||
|  |         for name, values in self._attributes.pop("shadow").get("parameters").items(): | ||||||
|  |             self._attributes.setdefault("parameters", {})[name] = values["parNewVal"] | ||||||
|  |  | ||||||
|  |     async def load_statistics(self): | ||||||
|  |         self._statistics = await self.api.load_statistics(self) | ||||||
|  |  | ||||||
|  |     async def update(self): | ||||||
|  |         now = datetime.now() | ||||||
|  |         if not self._last_update or self._last_update < now - timedelta( | ||||||
|  |             seconds=self._MINIMAL_UPDATE_INTERVAL | ||||||
|  |         ): | ||||||
|  |             self._last_update = now | ||||||
|  |             await self.load_attributes() | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def command_parameters(self): | ||||||
|  |         return {n: c.parameter_value for n, c in self._commands.items()} | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def settings(self): |     def settings(self): | ||||||
|         result = {} |         result = {} | ||||||
|         for name, command in self._commands.items(): |         for name, command in self._commands.items(): | ||||||
|             for key, setting in command.settings.items(): |             for key in command.setting_keys: | ||||||
|  |                 setting = command.settings.get(key) | ||||||
|                 result[f"{name}.{key}"] = setting |                 result[f"{name}.{key}"] = setting | ||||||
|         if self._extra: |         if self._extra: | ||||||
|             return self._extra.settings(result) |             return self._extra.settings(result) | ||||||
|         return result |         return result | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def parameters(self): |     def available_settings(self): | ||||||
|         result = {} |         result = [] | ||||||
|         for name, command in self._commands.items(): |         for name, command in self._commands.items(): | ||||||
|             for key, parameter in ( |             for key in command.setting_keys: | ||||||
|                 command.parameters | command.ancillary_parameters |                 result.append(f"{name}.{key}") | ||||||
|             ).items(): |  | ||||||
|                 result.setdefault(name, {})[key] = parameter.value |  | ||||||
|         return result |         return result | ||||||
|  |  | ||||||
|     async def load_attributes(self): |  | ||||||
|         self._attributes = await self._api.load_attributes(self) |  | ||||||
|         for name, values in self._attributes.pop("shadow").get("parameters").items(): |  | ||||||
|             self._attributes.setdefault("parameters", {})[name] = values["parNewVal"] |  | ||||||
|  |  | ||||||
|     async def load_statistics(self): |  | ||||||
|         self._statistics = await self._api.load_statistics(self) |  | ||||||
|  |  | ||||||
|     async def update(self): |  | ||||||
|         await self.load_attributes() |  | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def data(self): |     def data(self): | ||||||
|         result = { |         result = { | ||||||
|             "attributes": self.attributes, |             "attributes": self.attributes, | ||||||
|             "appliance": self.info, |             "appliance": self.info, | ||||||
|             "statistics": self.statistics, |             "statistics": self.statistics, | ||||||
|             **self.parameters, |             "additional_data": self._additional_data, | ||||||
|  |             **self.command_parameters, | ||||||
|         } |         } | ||||||
|         if self._extra: |         if self._extra: | ||||||
|             return self._extra.data(result) |             return self._extra.data(result) | ||||||
|         return result |         return result | ||||||
|  |  | ||||||
|     @property |     def diagnose(self, whitespace="\u200B \u200B "): | ||||||
|     def diagnose(self): |         data = { | ||||||
|         data = self.data.copy() |             "attributes": self.attributes.copy(), | ||||||
|  |             "appliance": self.info, | ||||||
|  |             "additional_data": self._additional_data, | ||||||
|  |         } | ||||||
|  |         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", "code", "coords"]: |         for sensible in ["PK", "SK", "serialNumber", "code", "coords"]: | ||||||
|             data["appliance"].pop(sensible, None) |             data["appliance"].pop(sensible, None) | ||||||
|         result = helper.pretty_print({"data": self.data}, whitespace="\u200B \u200B ") |         result = helper.pretty_print({"data": data}, whitespace=whitespace) | ||||||
|         result += helper.pretty_print( |         result += helper.pretty_print( | ||||||
|             {"commands": helper.create_command(self.commands)}, |             {"commands": helper.create_command(self.commands)}, | ||||||
|             whitespace="\u200B \u200B ", |             whitespace=whitespace, | ||||||
|         ) |         ) | ||||||
|         return result.replace(self.mac_address, "12-34-56-78-90-ab") |         return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx") | ||||||
|  | |||||||
| @ -1,4 +1,7 @@ | |||||||
| class Appliance: | class Appliance: | ||||||
|  |     def __init__(self, appliance): | ||||||
|  |         self.parent = appliance | ||||||
|  |  | ||||||
|     def data(self, data): |     def data(self, data): | ||||||
|         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": |         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||||
|             data["attributes"]["parameters"]["machMode"] = "0" |             data["attributes"]["parameters"]["machMode"] = "0" | ||||||
|  | |||||||
							
								
								
									
										17
									
								
								pyhon/appliances/ov.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										17
									
								
								pyhon/appliances/ov.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,17 @@ | |||||||
|  | class Appliance: | ||||||
|  |     def __init__(self, appliance): | ||||||
|  |         self.parent = appliance | ||||||
|  |  | ||||||
|  |     def data(self, data): | ||||||
|  |         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||||
|  |             data["attributes"]["parameters"]["temp"] = "0" | ||||||
|  |             data["attributes"]["parameters"]["onOffStatus"] = "0" | ||||||
|  |             data["attributes"]["parameters"]["remoteCtrValid"] = "0" | ||||||
|  |             data["attributes"]["parameters"]["remainingTimeMM"] = "0" | ||||||
|  |  | ||||||
|  |         data["active"] = data["attributes"]["parameters"]["onOffStatus"] == "1" | ||||||
|  |  | ||||||
|  |         return data | ||||||
|  |  | ||||||
|  |     def settings(self, settings): | ||||||
|  |         return settings | ||||||
| @ -1,10 +1,22 @@ | |||||||
|  | from pyhon.parameter.fixed import HonParameterFixed | ||||||
|  |  | ||||||
|  |  | ||||||
| class Appliance: | class Appliance: | ||||||
|  |     def __init__(self, appliance): | ||||||
|  |         self.parent = appliance | ||||||
|  |  | ||||||
|     def data(self, data): |     def data(self, data): | ||||||
|         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": |         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||||
|             data["attributes"]["parameters"]["machMode"] = "0" |             data["attributes"]["parameters"]["machMode"] = "0" | ||||||
|         data["active"] = bool(data.get("attributes", {}).get("activity")) |         data["active"] = bool(data.get("attributes", {}).get("activity")) | ||||||
|         data["pause"] = data["attributes"]["parameters"]["machMode"] == "3" |         data["pause"] = data["attributes"]["parameters"]["machMode"] == "3" | ||||||
|  |         if program := int(data["attributes"]["parameters"]["prCode"]): | ||||||
|  |             ids = self.parent.settings["startProgram.program"].ids | ||||||
|  |             data["programName"] = ids.get(program, "") | ||||||
|         return data |         return data | ||||||
|  |  | ||||||
|     def settings(self, settings): |     def settings(self, settings): | ||||||
|  |         dry_level = settings.get("startProgram.dryLevel") | ||||||
|  |         if isinstance(dry_level, HonParameterFixed) and dry_level.value == "11": | ||||||
|  |             settings.pop("startProgram.dryLevel", None) | ||||||
|         return settings |         return settings | ||||||
|  | |||||||
| @ -1,4 +1,7 @@ | |||||||
| class Appliance: | class Appliance: | ||||||
|  |     def __init__(self, appliance): | ||||||
|  |         self.parent = appliance | ||||||
|  |  | ||||||
|     def data(self, data): |     def data(self, data): | ||||||
|         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": |         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||||
|             data["attributes"]["parameters"]["machMode"] = "0" |             data["attributes"]["parameters"]["machMode"] = "0" | ||||||
|  | |||||||
| @ -1,4 +1,7 @@ | |||||||
| class Appliance: | class Appliance: | ||||||
|  |     def __init__(self, appliance): | ||||||
|  |         self.parent = appliance | ||||||
|  |  | ||||||
|     def data(self, data): |     def data(self, data): | ||||||
|         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": |         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||||
|             data["attributes"]["parameters"]["machMode"] = "0" |             data["attributes"]["parameters"]["machMode"] = "0" | ||||||
|  | |||||||
| @ -1,4 +1,4 @@ | |||||||
| from typing import Optional, Dict, Any, List, TYPE_CHECKING | from typing import Optional, Dict, Any, List, TYPE_CHECKING, Union | ||||||
|  |  | ||||||
| from pyhon.parameter.base import HonParameter | from pyhon.parameter.base import HonParameter | ||||||
| from pyhon.parameter.enum import HonParameterEnum | from pyhon.parameter.enum import HonParameterEnum | ||||||
| @ -16,102 +16,117 @@ class HonCommand: | |||||||
|         self, |         self, | ||||||
|         name: str, |         name: str, | ||||||
|         attributes: Dict[str, Any], |         attributes: Dict[str, Any], | ||||||
|         api: "HonAPI", |  | ||||||
|         appliance: "HonAppliance", |         appliance: "HonAppliance", | ||||||
|         programs: Optional[Dict[str, "HonCommand"]] = None, |         categories: Optional[Dict[str, "HonCommand"]] = None, | ||||||
|         program_name: str = "", |         category_name: str = "", | ||||||
|     ): |     ): | ||||||
|         self._api: HonAPI = api |         self._api: HonAPI = appliance.api | ||||||
|         self._appliance: "HonAppliance" = appliance |         self._appliance: "HonAppliance" = appliance | ||||||
|         self._name: str = name |         self._name: str = name | ||||||
|         self._programs: Optional[Dict[str, "HonCommand"]] = programs or {} |         self._categories: Optional[Dict[str, "HonCommand"]] = categories | ||||||
|         self._program_name: str = program_name |         self._category_name: str = category_name | ||||||
|         self._description: str = attributes.get("description", "") |         self._description: str = attributes.pop("description", "") | ||||||
|         self._parameters: Dict[str, HonParameter] = self._create_parameters( |         self._protocol_type: str = attributes.pop("protocolType", "") | ||||||
|             attributes.get("parameters", {}) |         self._parameters: Dict[str, HonParameter] = {} | ||||||
|         ) |         self._data: Dict[str, Any] = {} | ||||||
|         self._ancillary_parameters: Dict[str, HonParameter] = self._create_parameters( |         self._available_settings: Dict[str, HonParameter] = {} | ||||||
|             attributes.get("ancillaryParameters", {}) |         self._load_parameters(attributes) | ||||||
|         ) |  | ||||||
|  |  | ||||||
|     def __repr__(self) -> str: |     def __repr__(self) -> str: | ||||||
|         return f"{self._name} command" |         return f"{self._name} command" | ||||||
|  |  | ||||||
|     def _create_parameters(self, parameters: Dict) -> Dict[str, HonParameter]: |     @property | ||||||
|         result: Dict[str, HonParameter] = {} |     def name(self): | ||||||
|         for parameter, attributes in parameters.items(): |         return self._name | ||||||
|             if parameter == "zoneMap" and self._appliance.zone: |  | ||||||
|                 attributes["default"] = self._appliance.zone |     @property | ||||||
|             match attributes.get("typology"): |     def data(self): | ||||||
|                 case "range": |         return self._data | ||||||
|                     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 |     @property | ||||||
|     def parameters(self) -> Dict[str, HonParameter]: |     def parameters(self) -> Dict[str, HonParameter]: | ||||||
|         return self._parameters |         return self._parameters | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def ancillary_parameters(self) -> Dict[str, HonParameter]: |     def settings(self) -> Dict[str, HonParameter]: | ||||||
|         return self._ancillary_parameters |         return self._parameters | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]: | ||||||
|  |         result: Dict[str, Dict[str, Union[str, float]]] = {} | ||||||
|  |         for name, parameter in self._parameters.items(): | ||||||
|  |             result.setdefault(parameter.group, {})[name] = parameter.value | ||||||
|  |         return result | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def parameter_value(self) -> Dict[str, Union[str, float]]: | ||||||
|  |         return {n: p.value for n, p in self._parameters.items()} | ||||||
|  |  | ||||||
|  |     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: | ||||||
|  |             name = "program" if "PROGRAM" in self._category_name else "category" | ||||||
|  |             self._parameters[name] = HonParameterProgram(name, self, "custom") | ||||||
|  |  | ||||||
|     async def send(self) -> bool: |     async def send(self) -> bool: | ||||||
|         params = {k: v.value for k, v in self._parameters.items()} |         params = self.parameter_groups.get("parameters", {}) | ||||||
|         ancillary_params = {k: v.value for k, v in self._ancillary_parameters.items()} |         ancillary_params = self.parameter_groups.get("ancillaryParameters", {}) | ||||||
|         return await self._api.send_command( |         return await self._api.send_command( | ||||||
|             self._appliance, self._name, params, ancillary_params |             self._appliance, self._name, params, ancillary_params | ||||||
|         ) |         ) | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def programs(self) -> Dict[str, "HonCommand"]: |     def categories(self) -> Dict[str, "HonCommand"]: | ||||||
|         if self._programs is None: |         if self._categories is None: | ||||||
|             return {} |             return {"_": self} | ||||||
|         return self._programs |         return self._categories | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def program(self) -> str: |     def category(self) -> str: | ||||||
|         return self._program_name |         return self._category_name | ||||||
|  |  | ||||||
|     @program.setter |     @category.setter | ||||||
|     def program(self, program: str) -> None: |     def category(self, category: str) -> None: | ||||||
|         self._appliance.commands[self._name] = self.programs[program] |         self._appliance.commands[self._name] = self.categories[category] | ||||||
|  |  | ||||||
|     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 isinstance(parameter, HonParameterFixed): |  | ||||||
|                 continue |  | ||||||
|             if key not in keys: |  | ||||||
|                 keys.append(key) |  | ||||||
|         return keys |  | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def setting_keys(self) -> List[str]: |     def setting_keys(self) -> List[str]: | ||||||
|         if not self._programs: |         return list( | ||||||
|             return self._get_settings_keys() |             {param for cmd in self.categories.values() for param in cmd.parameters} | ||||||
|         result = [ |         ) | ||||||
|             key |  | ||||||
|             for cmd in self._programs.values() |     @staticmethod | ||||||
|             for key in self._get_settings_keys(cmd) |     def _more_options(first: HonParameter, second: HonParameter): | ||||||
|         ] |         if isinstance(first, HonParameterFixed) and not isinstance( | ||||||
|         return list(set(result + ["program"])) |             second, HonParameterFixed | ||||||
|  |         ): | ||||||
|  |             return second | ||||||
|  |         if len(second.values) > len(first.values): | ||||||
|  |             return second | ||||||
|  |         return first | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def settings(self) -> Dict[str, HonParameter]: |     def available_settings(self) -> Dict[str, HonParameter]: | ||||||
|         """Parameters with typology enum and range""" |         result: Dict[str, HonParameter] = {} | ||||||
|         return { |         for command in self.categories.values(): | ||||||
|             s: param |             for name, parameter in command.parameters.items(): | ||||||
|             for s in self.setting_keys |                 if name in result: | ||||||
|             if (param := self._parameters.get(s)) is not None |                     result[name] = self._more_options(result[name], parameter) | ||||||
|             or (param := self._ancillary_parameters.get(s)) is not None |                 result[name] = parameter | ||||||
|         } |         return result | ||||||
|  | |||||||
| @ -12,7 +12,7 @@ from pyhon.connection.auth import HonAuth | |||||||
| from pyhon.connection.handler.anonym import HonAnonymousConnectionHandler | from pyhon.connection.handler.anonym import HonAnonymousConnectionHandler | ||||||
| from pyhon.connection.handler.hon import HonConnectionHandler | from pyhon.connection.handler.hon import HonConnectionHandler | ||||||
|  |  | ||||||
| _LOGGER = logging.getLogger() | _LOGGER = logging.getLogger(__name__) | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonAPI: | class HonAPI: | ||||||
| @ -74,13 +74,15 @@ class HonAPI: | |||||||
|             "applianceType": appliance.appliance_type, |             "applianceType": appliance.appliance_type, | ||||||
|             "code": appliance.info["code"], |             "code": appliance.info["code"], | ||||||
|             "applianceModelId": appliance.appliance_model_id, |             "applianceModelId": appliance.appliance_model_id, | ||||||
|             "firmwareId": appliance.info["eepromId"], |  | ||||||
|             "macAddress": appliance.mac_address, |             "macAddress": appliance.mac_address, | ||||||
|             "fwVersion": appliance.info["fwVersion"], |  | ||||||
|             "os": const.OS, |             "os": const.OS, | ||||||
|             "appVersion": const.APP_VERSION, |             "appVersion": const.APP_VERSION, | ||||||
|             "series": appliance.info["series"], |             "series": appliance.info["series"], | ||||||
|         } |         } | ||||||
|  |         if firmware_id := appliance.info.get("eepromId"): | ||||||
|  |             params["firmwareId"] = firmware_id | ||||||
|  |         if firmware_version := appliance.info.get("fwVersion"): | ||||||
|  |             params["fwVersion"] = firmware_version | ||||||
|         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 = (await response.json()).get("payload", {}) | ||||||
| @ -107,6 +109,18 @@ class HonAPI: | |||||||
|                 return activity |                 return activity | ||||||
|         return {} |         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: |     async def load_attributes(self, appliance: HonAppliance) -> Dict: | ||||||
|         params: Dict = { |         params: Dict = { | ||||||
|             "macAddress": appliance.mac_address, |             "macAddress": appliance.mac_address, | ||||||
| @ -159,7 +173,7 @@ class HonAPI: | |||||||
|         return False |         return False | ||||||
|  |  | ||||||
|     async def appliance_configuration(self) -> Dict: |     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: |         async with self._hon_anonymous.get(url) as response: | ||||||
|             result: Dict = await response.json() |             result: Dict = await response.json() | ||||||
|             if result and (data := result.get("payload")): |             if result and (data := result.get("payload")): | ||||||
|  | |||||||
| @ -182,13 +182,14 @@ class HonAuth: | |||||||
|             await self._error_logger(response) |             await self._error_logger(response) | ||||||
|             return "" |             return "" | ||||||
|  |  | ||||||
|     def _parse_token_data(self, text: str) -> None: |     def _parse_token_data(self, text: str) -> bool: | ||||||
|         if access_token := re.findall("access_token=(.*?)&", text): |         if access_token := re.findall("access_token=(.*?)&", text): | ||||||
|             self._access_token = access_token[0] |             self._access_token = access_token[0] | ||||||
|         if refresh_token := re.findall("refresh_token=(.*?)&", text): |         if refresh_token := re.findall("refresh_token=(.*?)&", text): | ||||||
|             self._refresh_token = refresh_token[0] |             self._refresh_token = refresh_token[0] | ||||||
|         if id_token := re.findall("id_token=(.*?)&", text): |         if id_token := re.findall("id_token=(.*?)&", text): | ||||||
|             self._id_token = id_token[0] |             self._id_token = id_token[0] | ||||||
|  |         return True if access_token and refresh_token and id_token else False | ||||||
|  |  | ||||||
|     async def _get_token(self, url: str) -> bool: |     async def _get_token(self, url: str) -> bool: | ||||||
|         async with self._request.get(url) as response: |         async with self._request.get(url) as response: | ||||||
| @ -214,7 +215,9 @@ class HonAuth: | |||||||
|             if response.status != 200: |             if response.status != 200: | ||||||
|                 await self._error_logger(response) |                 await self._error_logger(response) | ||||||
|                 return False |                 return False | ||||||
|             self._parse_token_data(await response.text()) |             if not self._parse_token_data(await response.text()): | ||||||
|  |                 await self._error_logger(response) | ||||||
|  |                 return False | ||||||
|         return True |         return True | ||||||
|  |  | ||||||
|     async def _api_auth(self) -> bool: |     async def _api_auth(self) -> bool: | ||||||
| @ -228,7 +231,10 @@ class HonAuth: | |||||||
|             except json.JSONDecodeError: |             except json.JSONDecodeError: | ||||||
|                 await self._error_logger(response) |                 await self._error_logger(response) | ||||||
|                 return False |                 return False | ||||||
|             self._cognito_token = json_data["cognitoUser"]["Token"] |             self._cognito_token = json_data.get("cognitoUser", {}).get("Token", "") | ||||||
|  |             if not self._cognito_token: | ||||||
|  |                 _LOGGER.error(json_data) | ||||||
|  |                 raise exceptions.HonAuthenticationError() | ||||||
|         return True |         return True | ||||||
|  |  | ||||||
|     async def authenticate(self) -> None: |     async def authenticate(self) -> None: | ||||||
|  | |||||||
| @ -49,7 +49,7 @@ def create_command(commands, concat=False): | |||||||
|     for name, command in commands.items(): |     for name, command in commands.items(): | ||||||
|         if not concat: |         if not concat: | ||||||
|             result[name] = {} |             result[name] = {} | ||||||
|         for parameter, data in command.settings.items(): |         for parameter, data in command.available_settings.items(): | ||||||
|             if data.typology == "enum": |             if data.typology == "enum": | ||||||
|                 value = data.values |                 value = data.values | ||||||
|             elif data.typology == "range": |             elif data.typology == "range": | ||||||
|  | |||||||
| @ -47,7 +47,7 @@ class Hon: | |||||||
|  |  | ||||||
|     async def _create_appliance(self, appliance_data: Dict[str, Any], zone=0) -> None: |     async def _create_appliance(self, appliance_data: Dict[str, Any], zone=0) -> None: | ||||||
|         appliance = HonAppliance(self._api, appliance_data, zone=zone) |         appliance = HonAppliance(self._api, appliance_data, zone=zone) | ||||||
|         if appliance.mac_address is None: |         if appliance.mac_address == "": | ||||||
|             return |             return | ||||||
|         await asyncio.gather( |         await asyncio.gather( | ||||||
|             *[ |             *[ | ||||||
|  | |||||||
| @ -1,13 +1,14 @@ | |||||||
| from typing import Dict, Any | from typing import Dict, Any, List | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonParameter: | 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._key = key | ||||||
|         self._category: str = attributes.get("category", "") |         self._category: str = attributes.get("category", "") | ||||||
|         self._typology: str = attributes.get("typology", "") |         self._typology: str = attributes.get("typology", "") | ||||||
|         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 | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def key(self) -> str: |     def key(self) -> str: | ||||||
| @ -17,6 +18,10 @@ class HonParameter: | |||||||
|     def value(self) -> str | float: |     def value(self) -> str | float: | ||||||
|         return self._value if self._value is not None else "0" |         return self._value if self._value is not None else "0" | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def values(self) -> List[str]: | ||||||
|  |         return [str(self.value)] | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def category(self) -> str: |     def category(self) -> str: | ||||||
|         return self._category |         return self._category | ||||||
| @ -28,3 +33,7 @@ class HonParameter: | |||||||
|     @property |     @property | ||||||
|     def mandatory(self) -> int: |     def mandatory(self) -> int: | ||||||
|         return self._mandatory |         return self._mandatory | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def group(self) -> str: | ||||||
|  |         return self._group | ||||||
|  | |||||||
| @ -4,8 +4,8 @@ from pyhon.parameter.base import HonParameter | |||||||
|  |  | ||||||
|  |  | ||||||
| class HonParameterEnum(HonParameter): | class HonParameterEnum(HonParameter): | ||||||
|     def __init__(self, key: str, attributes: Dict[str, Any]) -> None: |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|         super().__init__(key, attributes) |         super().__init__(key, attributes, group) | ||||||
|         self._default = attributes.get("defaultValue") |         self._default = attributes.get("defaultValue") | ||||||
|         self._value = self._default or "0" |         self._value = self._default or "0" | ||||||
|         self._values: List[str] = attributes.get("enumValues", []) |         self._values: List[str] = attributes.get("enumValues", []) | ||||||
|  | |||||||
| @ -4,8 +4,8 @@ from pyhon.parameter.base import HonParameter | |||||||
|  |  | ||||||
|  |  | ||||||
| class HonParameterFixed(HonParameter): | class HonParameterFixed(HonParameter): | ||||||
|     def __init__(self, key: str, attributes: Dict[str, Any]) -> None: |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|         super().__init__(key, attributes) |         super().__init__(key, attributes, group) | ||||||
|         self._value = attributes.get("fixedValue", None) |         self._value = attributes.get("fixedValue", None) | ||||||
|  |  | ||||||
|     def __repr__(self) -> str: |     def __repr__(self) -> str: | ||||||
|  | |||||||
| @ -9,11 +9,14 @@ if TYPE_CHECKING: | |||||||
| class HonParameterProgram(HonParameterEnum): | class HonParameterProgram(HonParameterEnum): | ||||||
|     _FILTER = ["iot_recipe", "iot_guided"] |     _FILTER = ["iot_recipe", "iot_guided"] | ||||||
|  |  | ||||||
|     def __init__(self, key: str, command: "HonCommand") -> None: |     def __init__(self, key: str, command: "HonCommand", group: str) -> None: | ||||||
|         super().__init__(key, {}) |         super().__init__(key, {}, group) | ||||||
|         self._command = command |         self._command = command | ||||||
|         self._value: str = command.program |         if "PROGRAM" in command.category: | ||||||
|         self._programs: Dict[str, "HonCommand"] = command.programs |             self._value = command.category.split(".")[-1].lower() | ||||||
|  |         else: | ||||||
|  |             self._value = command.category | ||||||
|  |         self._programs: Dict[str, "HonCommand"] = command.categories | ||||||
|         self._typology: str = "enum" |         self._typology: str = "enum" | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
| @ -23,7 +26,7 @@ class HonParameterProgram(HonParameterEnum): | |||||||
|     @value.setter |     @value.setter | ||||||
|     def value(self, value: str) -> None: |     def value(self, value: str) -> None: | ||||||
|         if value in self.values: |         if value in self.values: | ||||||
|             self._command.program = value |             self._command.category = value | ||||||
|         else: |         else: | ||||||
|             raise ValueError(f"Allowed values {self.values}") |             raise ValueError(f"Allowed values {self.values}") | ||||||
|  |  | ||||||
| @ -31,3 +34,12 @@ class HonParameterProgram(HonParameterEnum): | |||||||
|     def values(self) -> List[str]: |     def values(self) -> List[str]: | ||||||
|         values = [v for v in self._programs if all(f not in v for f in self._FILTER)] |         values = [v for v in self._programs if all(f not in v for f in self._FILTER)] | ||||||
|         return sorted(values) |         return sorted(values) | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def ids(self): | ||||||
|  |         values = { | ||||||
|  |             int(p.parameters["prCode"].value): n | ||||||
|  |             for i, (n, p) in enumerate(self._programs.items()) | ||||||
|  |             if "iot_" not in n and p.parameters.get("prCode") | ||||||
|  |         } | ||||||
|  |         return dict(sorted(values.items())) | ||||||
|  | |||||||
| @ -1,4 +1,4 @@ | |||||||
| from typing import Dict, Any | from typing import Dict, Any, List | ||||||
|  |  | ||||||
| from pyhon.parameter.base import HonParameter | from pyhon.parameter.base import HonParameter | ||||||
|  |  | ||||||
| @ -11,8 +11,8 @@ def str_to_float(string: str | float) -> float: | |||||||
|  |  | ||||||
|  |  | ||||||
| class HonParameterRange(HonParameter): | class HonParameterRange(HonParameter): | ||||||
|     def __init__(self, key: str, attributes: Dict[str, Any]) -> None: |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|         super().__init__(key, attributes) |         super().__init__(key, attributes, group) | ||||||
|         self._min: float = str_to_float(attributes["minimumValue"]) |         self._min: float = str_to_float(attributes["minimumValue"]) | ||||||
|         self._max: float = str_to_float(attributes["maximumValue"]) |         self._max: float = str_to_float(attributes["maximumValue"]) | ||||||
|         self._step: float = str_to_float(attributes["incrementValue"]) |         self._step: float = str_to_float(attributes["incrementValue"]) | ||||||
| @ -47,3 +47,7 @@ class HonParameterRange(HonParameter): | |||||||
|             raise ValueError( |             raise ValueError( | ||||||
|                 f"Allowed: min {self._min} max {self._max} step {self._step}" |                 f"Allowed: min {self._min} max {self._max} step {self._step}" | ||||||
|             ) |             ) | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def values(self) -> List[str]: | ||||||
|  |         return [str(i) for i in range(int(self.min), int(self.max) + 1, int(self.step))] | ||||||
|  | |||||||
							
								
								
									
										2
									
								
								setup.py
									
									
									
									
									
								
							
							
						
						
									
										2
									
								
								setup.py
									
									
									
									
									
								
							| @ -7,7 +7,7 @@ with open("README.md", "r") as f: | |||||||
|  |  | ||||||
| setup( | setup( | ||||||
|     name="pyhOn", |     name="pyhOn", | ||||||
|     version="0.8.0", |     version="0.10.4", | ||||||
|     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, | ||||||
|  | |||||||
		Reference in New Issue
	
	Block a user
	