Compare commits
	
		
			7 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| 022da71800 | |||
| 8e16b4a215 | |||
| 7bd3aac7c5 | |||
| 365a37b42d | |||
| 2bde6bb61c | |||
| ccff32e6c1 | |||
| 22cbd7474a | 
| @ -1,6 +1,7 @@ | |||||||
| 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 | ||||||
|  |  | ||||||
| @ -16,6 +17,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: | ||||||
| @ -30,6 +33,8 @@ class HonAppliance: | |||||||
|         self._attributes: Dict = {} |         self._attributes: Dict = {} | ||||||
|         self._zone: int = zone |         self._zone: int = zone | ||||||
|         self._additional_data: Dict[str, Any] = {} |         self._additional_data: Dict[str, Any] = {} | ||||||
|  |         self._last_update = None | ||||||
|  |         self._default_setting = HonParameter("", {}, "") | ||||||
|  |  | ||||||
|         try: |         try: | ||||||
|             self._extra = importlib.import_module( |             self._extra = importlib.import_module( | ||||||
| @ -139,8 +144,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 ( | ||||||
| @ -192,7 +202,7 @@ class HonAppliance: | |||||||
|     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", None) | ||||||
|         self._commands = self._get_commands(raw) |         self._commands = self._get_commands(raw) | ||||||
|         await self._recover_last_command_states() |         await self._recover_last_command_states() | ||||||
|  |  | ||||||
| @ -205,6 +215,11 @@ class HonAppliance: | |||||||
|         self._statistics = await self.api.load_statistics(self) |         self._statistics = await self.api.load_statistics(self) | ||||||
|  |  | ||||||
|     async def update(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() |             await self.load_attributes() | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
| @ -216,12 +231,20 @@ class HonAppliance: | |||||||
|         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, self._default_setting) | ||||||
|                 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 = { | ||||||
| @ -235,18 +258,21 @@ class HonAppliance: | |||||||
|             return self._extra.data(result) |             return self._extra.data(result) | ||||||
|         return result |         return result | ||||||
|  |  | ||||||
|     def diagnose(self, whitespace="\u200B \u200B "): |     def diagnose(self, whitespace="\u200B \u200B ", command_only=False): | ||||||
|         data = { |         data = { | ||||||
|             "attributes": self.attributes.copy(), |             "attributes": self.attributes.copy(), | ||||||
|             "appliance": self.info, |             "appliance": self.info, | ||||||
|             "additional_data": self._additional_data, |             "additional_data": self._additional_data, | ||||||
|         } |         } | ||||||
|  |         if command_only: | ||||||
|  |             data.pop("attributes") | ||||||
|  |             data.pop("appliance") | ||||||
|         data |= {n: c.parameter_groups for n, c in self._commands.items()} |         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} |         extra = {n: c.data for n, c in self._commands.items() if c.data} | ||||||
|         if extra: |         if extra: | ||||||
|             data |= {"extra_command_data": 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.get("appliance", {}).pop(sensible, None) | ||||||
|         result = helper.pretty_print({"data": data}, whitespace=whitespace) |         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)}, | ||||||
|  | |||||||
| @ -29,6 +29,7 @@ class HonCommand: | |||||||
|         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: Dict[str, Any] = {} |         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: | ||||||
| @ -46,6 +47,10 @@ 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 |     @property | ||||||
|     def parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]: |     def parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]: | ||||||
|         result: Dict[str, Dict[str, Union[str, float]]] = {} |         result: Dict[str, Dict[str, Union[str, float]]] = {} | ||||||
| @ -76,14 +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( |             self._parameters[name] = HonParameterProgram(name, self, "custom") | ||||||
|                     "program", self, "custom" |  | ||||||
|                 ) |  | ||||||
|  |  | ||||||
|     async def send(self) -> bool: |     async def send(self) -> bool: | ||||||
|         params = self.parameter_groups["parameters"] |         params = self.parameter_groups.get("parameters", {}) | ||||||
|         ancillary_params = self.parameter_groups["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 | ||||||
|         ) |         ) | ||||||
| @ -100,6 +103,7 @@ class HonCommand: | |||||||
|  |  | ||||||
|     @category.setter |     @category.setter | ||||||
|     def category(self, category: str) -> None: |     def category(self, category: str) -> None: | ||||||
|  |         if category in self.categories: | ||||||
|             self._appliance.commands[self._name] = self.categories[category] |             self._appliance.commands[self._name] = self.categories[category] | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
| @ -119,7 +123,7 @@ class HonCommand: | |||||||
|         return first |         return first | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def settings(self) -> Dict[str, HonParameter]: |     def available_settings(self) -> Dict[str, HonParameter]: | ||||||
|         result: Dict[str, HonParameter] = {} |         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(): | ||||||
|  | |||||||
| @ -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" | ||||||
|  |  | ||||||
|  | |||||||
| @ -32,6 +32,8 @@ class HonParameterRange(HonParameter): | |||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def step(self) -> float: |     def step(self) -> float: | ||||||
|  |         if not self._step: | ||||||
|  |             return 1 | ||||||
|         return self._step |         return self._step | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|  | |||||||
							
								
								
									
										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.0", |     version="0.10.6", | ||||||
|     author="Andre Basche", |     author="Andre Basche", | ||||||
|     description="Control hOn devices with python", |     description="Control hOn devices with python", | ||||||
|     long_description=long_description, |     long_description=long_description, | ||||||
|  | |||||||
		Reference in New Issue
	
	Block a user
	