Compare commits
	
		
			3 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| ccff32e6c1 | |||
| 22cbd7474a | |||
| dd61b24eed | 
| @ -78,12 +78,7 @@ async def main(): | ||||
|                     ) | ||||
|                 ) | ||||
|             else: | ||||
|                 print(helper.pretty_print({"data": device.data})) | ||||
|                 print( | ||||
|                     helper.pretty_print( | ||||
|                         {"settings": helper.create_command(device.commands)} | ||||
|                     ) | ||||
|                 ) | ||||
|                 print(device.diagnose("  ")) | ||||
|  | ||||
|  | ||||
| def start(): | ||||
|  | ||||
| @ -1,12 +1,12 @@ | ||||
| import importlib | ||||
| import logging | ||||
| from contextlib import suppress | ||||
| from datetime import datetime, timedelta | ||||
| from typing import Optional, Dict, Any | ||||
| from typing import TYPE_CHECKING | ||||
|  | ||||
| from pyhon import helper | ||||
| from pyhon import helper, exceptions | ||||
| from pyhon.commands import HonCommand | ||||
| from pyhon.parameter.base import HonParameter | ||||
| from pyhon.parameter.fixed import HonParameterFixed | ||||
|  | ||||
| if TYPE_CHECKING: | ||||
| @ -16,6 +16,8 @@ _LOGGER = logging.getLogger(__name__) | ||||
|  | ||||
|  | ||||
| class HonAppliance: | ||||
|     _MINIMAL_UPDATE_INTERVAL = 5  # seconds | ||||
|  | ||||
|     def __init__( | ||||
|         self, api: Optional["HonAPI"], info: Dict[str, Any], zone: int = 0 | ||||
|     ) -> None: | ||||
| @ -29,7 +31,8 @@ class HonAppliance: | ||||
|         self._statistics: Dict = {} | ||||
|         self._attributes: Dict = {} | ||||
|         self._zone: int = zone | ||||
|         self._additional_data = {} | ||||
|         self._additional_data: Dict[str, Any] = {} | ||||
|         self._last_update = None | ||||
|  | ||||
|         try: | ||||
|             self._extra = importlib.import_module( | ||||
| @ -119,8 +122,14 @@ class HonAppliance: | ||||
|     def zone(self) -> int: | ||||
|         return self._zone | ||||
|  | ||||
|     @property | ||||
|     def api(self) -> "HonAPI": | ||||
|         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) | ||||
|         command_history = await self.api.command_history(self) | ||||
|         for name, command in self._commands.items(): | ||||
|             last = next( | ||||
|                 ( | ||||
| @ -133,8 +142,11 @@ class HonAppliance: | ||||
|             if last is None: | ||||
|                 continue | ||||
|             parameters = command_history[last].get("command", {}).get("parameters", {}) | ||||
|             if command.categories and parameters.get("category"): | ||||
|                 command.category = parameters.pop("category").split(".")[-1].lower() | ||||
|             if command.categories: | ||||
|                 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 ( | ||||
| @ -170,7 +182,6 @@ class HonAppliance: | ||||
|                     HonCommand( | ||||
|                         command, | ||||
|                         data, | ||||
|                         self._api, | ||||
|                         self, | ||||
|                         category_name=category, | ||||
|                         categories=categories, | ||||
| @ -178,47 +189,58 @@ class HonAppliance: | ||||
|                 ] | ||||
|             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) | ||||
|         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) | ||||
|         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) | ||||
|         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 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 | ||||
|     def command_parameters(self): | ||||
|         return {n: c.parameter_value for n, c in self._commands.items()} | ||||
|  | ||||
|     @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)) | ||||
|                 setting = command.settings.get(key) | ||||
|                 result[f"{name}.{key}"] = setting | ||||
|         if self._extra: | ||||
|             return self._extra.settings(result) | ||||
|         return result | ||||
|  | ||||
|     @property | ||||
|     def available_settings(self): | ||||
|         result = [] | ||||
|         for name, command in self._commands.items(): | ||||
|             for key in command.setting_keys: | ||||
|                 result.append(f"{name}.{key}") | ||||
|         return result | ||||
|  | ||||
|     @property | ||||
|     def data(self): | ||||
|         result = { | ||||
| @ -226,20 +248,27 @@ class HonAppliance: | ||||
|             "appliance": self.info, | ||||
|             "statistics": self.statistics, | ||||
|             "additional_data": self._additional_data, | ||||
|             **self.parameters, | ||||
|             **self.command_parameters, | ||||
|         } | ||||
|         if self._extra: | ||||
|             return self._extra.data(result) | ||||
|         return result | ||||
|  | ||||
|     @property | ||||
|     def diagnose(self): | ||||
|         data = self.data.copy() | ||||
|     def diagnose(self, whitespace="\u200B \u200B "): | ||||
|         data = { | ||||
|             "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"]: | ||||
|             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( | ||||
|             {"commands": helper.create_command(self.commands)}, | ||||
|             whitespace="\u200B \u200B ", | ||||
|             whitespace=whitespace, | ||||
|         ) | ||||
|         return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx") | ||||
|  | ||||
| @ -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.enum import HonParameterEnum | ||||
| @ -16,12 +16,11 @@ class HonCommand: | ||||
|         self, | ||||
|         name: str, | ||||
|         attributes: Dict[str, Any], | ||||
|         api: "HonAPI", | ||||
|         appliance: "HonAppliance", | ||||
|         categories: Optional[Dict[str, "HonCommand"]] = None, | ||||
|         category_name: str = "", | ||||
|     ): | ||||
|         self._api: HonAPI = api | ||||
|         self._api: HonAPI = appliance.api | ||||
|         self._appliance: "HonAppliance" = appliance | ||||
|         self._name: str = name | ||||
|         self._categories: Optional[Dict[str, "HonCommand"]] = categories | ||||
| @ -29,7 +28,8 @@ class HonCommand: | ||||
|         self._description: str = attributes.pop("description", "") | ||||
|         self._protocol_type: str = attributes.pop("protocolType", "") | ||||
|         self._parameters: Dict[str, HonParameter] = {} | ||||
|         self._data = {} | ||||
|         self._data: Dict[str, Any] = {} | ||||
|         self._available_settings: Dict[str, HonParameter] = {} | ||||
|         self._load_parameters(attributes) | ||||
|  | ||||
|     def __repr__(self) -> str: | ||||
| @ -47,6 +47,21 @@ class HonCommand: | ||||
|     def parameters(self) -> Dict[str, HonParameter]: | ||||
|         return self._parameters | ||||
|  | ||||
|     @property | ||||
|     def settings(self) -> Dict[str, HonParameter]: | ||||
|         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(): | ||||
| @ -66,17 +81,12 @@ class HonCommand: | ||||
|                 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 | ||||
|         } | ||||
|             name = "program" if "PROGRAM" in self._category_name else "category" | ||||
|             self._parameters[name] = HonParameterProgram(name, self, "custom") | ||||
|  | ||||
|     async def send(self) -> bool: | ||||
|         params = self._parameters_by_group("parameters") | ||||
|         ancillary_params = self._parameters_by_group("ancillary_parameters") | ||||
|         params = self.parameter_groups["parameters"] | ||||
|         ancillary_params = self.parameter_groups["ancillary_parameters"] | ||||
|         return await self._api.send_command( | ||||
|             self._appliance, self._name, params, ancillary_params | ||||
|         ) | ||||
| @ -101,12 +111,22 @@ class HonCommand: | ||||
|             {param for cmd in self.categories.values() for param in cmd.parameters} | ||||
|         ) | ||||
|  | ||||
|     @staticmethod | ||||
|     def _more_options(first: HonParameter, second: HonParameter): | ||||
|         if isinstance(first, HonParameterFixed) and not isinstance( | ||||
|             second, HonParameterFixed | ||||
|         ): | ||||
|             return second | ||||
|         if len(second.values) > len(first.values): | ||||
|             return second | ||||
|         return first | ||||
|  | ||||
|     @property | ||||
|     def settings(self) -> Dict[str, HonParameter]: | ||||
|         result = {} | ||||
|     def available_settings(self) -> Dict[str, HonParameter]: | ||||
|         result: Dict[str, HonParameter] = {} | ||||
|         for command in self.categories.values(): | ||||
|             for name, parameter in command.parameters.items(): | ||||
|                 if name in result: | ||||
|                     continue | ||||
|                     result[name] = self._more_options(result[name], parameter) | ||||
|                 result[name] = parameter | ||||
|         return result | ||||
|  | ||||
| @ -1,6 +1,3 @@ | ||||
| from pyhon.parameter.base import HonParameter | ||||
|  | ||||
|  | ||||
| def key_print(data, key="", start=True): | ||||
|     result = "" | ||||
|     if isinstance(data, list): | ||||
| @ -47,21 +44,12 @@ def pretty_print(data, key="", intend=0, is_list=False, whitespace="  "): | ||||
|     return result | ||||
|  | ||||
|  | ||||
| def get_parameter(command, parameter): | ||||
|     if programs := command.categories: | ||||
|         for program in programs.values(): | ||||
|             if data := program.settings.get(parameter): | ||||
|                 return data | ||||
|     return command.settings.get(parameter) | ||||
|  | ||||
|  | ||||
| def create_command(commands, concat=False): | ||||
|     result = {} | ||||
|     for name, command in commands.items(): | ||||
|         if not concat: | ||||
|             result[name] = {} | ||||
|         for parameter in command.setting_keys: | ||||
|             data = get_parameter(command, parameter) | ||||
|         for parameter, data in command.available_settings.items(): | ||||
|             if data.typology == "enum": | ||||
|                 value = data.values | ||||
|             elif data.typology == "range": | ||||
|  | ||||
| @ -12,6 +12,9 @@ class HonParameterProgram(HonParameterEnum): | ||||
|     def __init__(self, key: str, command: "HonCommand", group: str) -> None: | ||||
|         super().__init__(key, {}, group) | ||||
|         self._command = command | ||||
|         if "PROGRAM" in command.category: | ||||
|             self._value: str = command.category.split(".")[-1].lower() | ||||
|         else: | ||||
|             self._value: str = command.category | ||||
|         self._programs: Dict[str, "HonCommand"] = command.categories | ||||
|         self._typology: str = "enum" | ||||
| @ -23,7 +26,7 @@ class HonParameterProgram(HonParameterEnum): | ||||
|     @value.setter | ||||
|     def value(self, value: str) -> None: | ||||
|         if value in self.values: | ||||
|             self._command.program = value | ||||
|             self._command.category = value | ||||
|         else: | ||||
|             raise ValueError(f"Allowed values {self.values}") | ||||
|  | ||||
|  | ||||
		Reference in New Issue
	
	Block a user
	