Compare commits
	
		
			6 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| 7bd3aac7c5 | |||
| 365a37b42d | |||
| 2bde6bb61c | |||
| ccff32e6c1 | |||
| 22cbd7474a | |||
| dd61b24eed | 
| @ -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,12 +1,12 @@ | |||||||
| import importlib | import importlib | ||||||
| import logging | 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.base import HonParameter |  | ||||||
| from pyhon.parameter.fixed import HonParameterFixed | from pyhon.parameter.fixed import HonParameterFixed | ||||||
|  |  | ||||||
| if TYPE_CHECKING: | if TYPE_CHECKING: | ||||||
| @ -16,6 +16,8 @@ _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: | ||||||
| @ -29,7 +31,8 @@ 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 = {} |         self._additional_data: Dict[str, Any] = {} | ||||||
|  |         self._last_update = None | ||||||
|  |  | ||||||
|         try: |         try: | ||||||
|             self._extra = importlib.import_module( |             self._extra = importlib.import_module( | ||||||
| @ -119,8 +122,14 @@ class HonAppliance: | |||||||
|     def zone(self) -> int: |     def zone(self) -> int: | ||||||
|         return self._zone |         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): |     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(): |         for name, command in self._commands.items(): | ||||||
|             last = next( |             last = next( | ||||||
|                 ( |                 ( | ||||||
| @ -133,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.categories and parameters.get("category"): |             if command.categories and ( | ||||||
|                 command.category = parameters.pop("category").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 ( | ||||||
| @ -170,7 +184,6 @@ class HonAppliance: | |||||||
|                     HonCommand( |                     HonCommand( | ||||||
|                         command, |                         command, | ||||||
|                         data, |                         data, | ||||||
|                         self._api, |  | ||||||
|                         self, |                         self, | ||||||
|                         category_name=category, |                         category_name=category, | ||||||
|                         categories=categories, |                         categories=categories, | ||||||
| @ -178,47 +191,58 @@ class HonAppliance: | |||||||
|                 ] |                 ] | ||||||
|             else: |             else: | ||||||
|                 commands += self._get_categories(command, data) |                 commands += self._get_categories(command, data) | ||||||
|  |         elif category: | ||||||
|  |             self._additional_data.setdefault(command, {})[category] = data | ||||||
|         else: |         else: | ||||||
|             self._additional_data[command] = data |             self._additional_data[command] = data | ||||||
|         return commands |         return commands | ||||||
|  |  | ||||||
|     async def load_commands(self): |     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") |         self._appliance_model = raw.pop("applianceModel") | ||||||
|         raw.pop("dictionaryId") |         raw.pop("dictionaryId") | ||||||
|         self._commands = self._get_commands(raw) |         self._commands = self._get_commands(raw) | ||||||
|         await self._recover_last_command_states() |         await self._recover_last_command_states() | ||||||
|  |  | ||||||
|     async def load_attributes(self): |     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.pop("shadow").get("parameters").items(): | ||||||
|             self._attributes.setdefault("parameters", {})[name] = values["parNewVal"] |             self._attributes.setdefault("parameters", {})[name] = values["parNewVal"] | ||||||
|  |  | ||||||
|     async def load_statistics(self): |     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): |     async def update(self): | ||||||
|         await self.load_attributes() |         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 |     @property | ||||||
|     def parameters(self): |     def command_parameters(self): | ||||||
|         result = {} |         return {n: c.parameter_value for n, c in self._commands.items()} | ||||||
|         for name, command in self._commands.items(): |  | ||||||
|             for key, parameter in command.parameters.items(): |  | ||||||
|                 result.setdefault(name, {})[key] = parameter.value |  | ||||||
|         return result |  | ||||||
|  |  | ||||||
|     @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 in command.setting_keys: |             for key in command.setting_keys: | ||||||
|                 setting = command.settings.get(key, HonParameter(key, {}, name)) |                 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 | ||||||
|  |     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 |     @property | ||||||
|     def data(self): |     def data(self): | ||||||
|         result = { |         result = { | ||||||
| @ -226,20 +250,27 @@ class HonAppliance: | |||||||
|             "appliance": self.info, |             "appliance": self.info, | ||||||
|             "statistics": self.statistics, |             "statistics": self.statistics, | ||||||
|             "additional_data": self._additional_data, |             "additional_data": self._additional_data, | ||||||
|             **self.parameters, |             **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, "xx-xx-xx-xx-xx-xx") |         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.base import HonParameter | ||||||
| from pyhon.parameter.enum import HonParameterEnum | from pyhon.parameter.enum import HonParameterEnum | ||||||
| @ -16,12 +16,11 @@ class HonCommand: | |||||||
|         self, |         self, | ||||||
|         name: str, |         name: str, | ||||||
|         attributes: Dict[str, Any], |         attributes: Dict[str, Any], | ||||||
|         api: "HonAPI", |  | ||||||
|         appliance: "HonAppliance", |         appliance: "HonAppliance", | ||||||
|         categories: Optional[Dict[str, "HonCommand"]] = None, |         categories: Optional[Dict[str, "HonCommand"]] = None, | ||||||
|         category_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._categories: Optional[Dict[str, "HonCommand"]] = categories |         self._categories: Optional[Dict[str, "HonCommand"]] = categories | ||||||
| @ -29,7 +28,8 @@ class HonCommand: | |||||||
|         self._description: str = attributes.pop("description", "") |         self._description: str = attributes.pop("description", "") | ||||||
|         self._protocol_type: str = attributes.pop("protocolType", "") |         self._protocol_type: str = attributes.pop("protocolType", "") | ||||||
|         self._parameters: Dict[str, HonParameter] = {} |         self._parameters: Dict[str, HonParameter] = {} | ||||||
|         self._data = {} |         self._data: Dict[str, Any] = {} | ||||||
|  |         self._available_settings: Dict[str, HonParameter] = {} | ||||||
|         self._load_parameters(attributes) |         self._load_parameters(attributes) | ||||||
|  |  | ||||||
|     def __repr__(self) -> str: |     def __repr__(self) -> str: | ||||||
| @ -47,6 +47,21 @@ class HonCommand: | |||||||
|     def parameters(self) -> Dict[str, HonParameter]: |     def parameters(self) -> Dict[str, HonParameter]: | ||||||
|         return self._parameters |         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): |     def _load_parameters(self, attributes): | ||||||
|         for key, items in attributes.items(): |         for key, items in attributes.items(): | ||||||
|             for name, data in items.items(): |             for name, data in items.items(): | ||||||
| @ -66,17 +81,12 @@ class HonCommand: | |||||||
|                 self._data[name] = data |                 self._data[name] = data | ||||||
|                 return |                 return | ||||||
|         if self._category_name: |         if self._category_name: | ||||||
|             if not self._categories: |             name = "program" if "PROGRAM" in self._category_name else "category" | ||||||
|                 self._parameters["program"] = HonParameterProgram("program", self, name) |             self._parameters[name] = HonParameterProgram(name, self, "custom") | ||||||
|  |  | ||||||
|     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: |     async def send(self) -> bool: | ||||||
|         params = self._parameters_by_group("parameters") |         params = self.parameter_groups.get("parameters", {}) | ||||||
|         ancillary_params = self._parameters_by_group("ancillary_parameters") |         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 | ||||||
|         ) |         ) | ||||||
| @ -101,12 +111,22 @@ class HonCommand: | |||||||
|             {param for cmd in self.categories.values() for param in cmd.parameters} |             {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 |     @property | ||||||
|     def settings(self) -> Dict[str, HonParameter]: |     def available_settings(self) -> Dict[str, HonParameter]: | ||||||
|         result = {} |         result: Dict[str, HonParameter] = {} | ||||||
|         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: | ||||||
|                     continue |                     result[name] = self._more_options(result[name], parameter) | ||||||
|                 result[name] = parameter |                 result[name] = parameter | ||||||
|         return result |         return result | ||||||
|  | |||||||
| @ -1,6 +1,3 @@ | |||||||
| from pyhon.parameter.base import HonParameter |  | ||||||
|  |  | ||||||
|  |  | ||||||
| def key_print(data, key="", start=True): | def key_print(data, key="", start=True): | ||||||
|     result = "" |     result = "" | ||||||
|     if isinstance(data, list): |     if isinstance(data, list): | ||||||
| @ -47,21 +44,12 @@ def pretty_print(data, key="", intend=0, is_list=False, whitespace="  "): | |||||||
|     return result |     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): | def create_command(commands, concat=False): | ||||||
|     result = {} |     result = {} | ||||||
|     for name, command in commands.items(): |     for name, command in commands.items(): | ||||||
|         if not concat: |         if not concat: | ||||||
|             result[name] = {} |             result[name] = {} | ||||||
|         for parameter in command.setting_keys: |         for parameter, data in command.available_settings.items(): | ||||||
|             data = get_parameter(command, parameter) |  | ||||||
|             if data.typology == "enum": |             if data.typology == "enum": | ||||||
|                 value = data.values |                 value = data.values | ||||||
|             elif data.typology == "range": |             elif data.typology == "range": | ||||||
|  | |||||||
| @ -12,7 +12,10 @@ class HonParameterProgram(HonParameterEnum): | |||||||
|     def __init__(self, key: str, command: "HonCommand", group: str) -> None: |     def __init__(self, key: str, command: "HonCommand", group: str) -> None: | ||||||
|         super().__init__(key, {}, group) |         super().__init__(key, {}, group) | ||||||
|         self._command = command |         self._command = command | ||||||
|         self._value: str = command.category |         if "PROGRAM" in command.category: | ||||||
|  |             self._value = command.category.split(".")[-1].lower() | ||||||
|  |         else: | ||||||
|  |             self._value = command.category | ||||||
|         self._programs: Dict[str, "HonCommand"] = command.categories |         self._programs: Dict[str, "HonCommand"] = command.categories | ||||||
|         self._typology: str = "enum" |         self._typology: str = "enum" | ||||||
|  |  | ||||||
| @ -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}") | ||||||
|  |  | ||||||
|  | |||||||
							
								
								
									
										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.10.0b0", |     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
	