Compare commits
	
		
			37 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| ccff32e6c1 | |||
| 22cbd7474a | |||
| dd61b24eed | |||
| ea8f481b01 | |||
| 7dcb34559b | |||
| 5db13a90e7 | |||
| 9ee5dbc956 | |||
| d4c6ccdce3 | |||
| 9594b9ebd8 | |||
| b011d98e07 | |||
| ad864286fc | |||
| 13cff8caa0 | |||
| 5fc6245806 | |||
| 1dad0e14b8 | |||
| b04c601ad6 | |||
| 3ec0f5a006 | |||
| 78bc85132f | |||
| 191928287f | |||
| c0aab8b99d | |||
| b37715d0ca | |||
| 411effd814 | |||
| 04f19c4609 | |||
| a68dcac379 | |||
| 69be63df2a | |||
| 6c44aa895d | |||
| 8372c75e30 | |||
| 40cc0013a5 | |||
| f2aa3dc8fd | |||
| e887371bec | |||
| 5b91747ec1 | |||
| 8da2018302 | |||
| 03187745bf | |||
| 461a247ad3 | |||
| 834f25a639 | |||
| 46ff9be4a2 | |||
| a1618bb18a | |||
| 18b0ecdd68 | 
							
								
								
									
										2
									
								
								.github/workflows/python-check.yml
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										2
									
								
								.github/workflows/python-check.yml
									
									
									
									
										vendored
									
									
								
							| @ -25,7 +25,7 @@ jobs: | |||||||
|       run: | |       run: | | ||||||
|         python -m pip install --upgrade pip |         python -m pip install --upgrade pip | ||||||
|         python -m pip install -r requirements.txt |         python -m pip install -r requirements.txt | ||||||
|         python -m pip install flake8 pylint black mypy |         python -m pip install -r requirements_dev.txt | ||||||
|     - name: Lint with flake8 |     - name: Lint with flake8 | ||||||
|       run: | |       run: | | ||||||
|         # stop the build if there are Python syntax errors or undefined names |         # stop the build if there are Python syntax errors or undefined names | ||||||
|  | |||||||
							
								
								
									
										1
									
								
								.gitignore
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										1
									
								
								.gitignore
									
									
									
									
										vendored
									
									
								
							| @ -4,3 +4,4 @@ __pycache__/ | |||||||
| dist/ | dist/ | ||||||
| **/*.egg-info/ | **/*.egg-info/ | ||||||
| test* | test* | ||||||
|  | build/ | ||||||
|  | |||||||
| @ -6,7 +6,7 @@ | |||||||
| [](https://www.python.org/) | [](https://www.python.org/) | ||||||
| [](https://github.com/Andre0512/pyhOn/blob/main/LICENSE) | [](https://github.com/Andre0512/pyhOn/blob/main/LICENSE) | ||||||
| [](https://pypistats.org/packages/pyhon)   | [](https://pypistats.org/packages/pyhon)   | ||||||
| Control your Haier appliances with python! | Control your Haier, Candy and Hoover appliances with python! | ||||||
| The idea behind this library is, to make the use of all available commands as simple as possible. | The idea behind this library is, to make the use of all available commands as simple as possible. | ||||||
|  |  | ||||||
| ## Installation | ## Installation | ||||||
|  | |||||||
| @ -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 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: | ||||||
| @ -24,12 +30,14 @@ class HonAppliance: | |||||||
|         self._commands: Dict = {} |         self._commands: Dict = {} | ||||||
|         self._statistics: Dict = {} |         self._statistics: Dict = {} | ||||||
|         self._attributes: Dict = {} |         self._attributes: Dict = {} | ||||||
|         self._zone = 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,9 +114,23 @@ class HonAppliance: | |||||||
|     def info(self): |     def info(self): | ||||||
|         return self._info |         return self._info | ||||||
|  |  | ||||||
|     async def _recover_last_command_states(self, commands): |     @property | ||||||
|         command_history = await self._api.command_history(self) |     def additional_data(self): | ||||||
|         for name, command in commands.items(): |         return self._additional_data | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     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) | ||||||
|  |         for name, command in self._commands.items(): | ||||||
|             last = next( |             last = next( | ||||||
|                 ( |                 ( | ||||||
|                     index |                     index | ||||||
| @ -120,8 +142,11 @@ 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._multi and parameters.get("program"): |             if command.categories: | ||||||
|                 command.set_program(parameters.pop("program").split(".")[-1].lower()) |                 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 ( | ||||||
| @ -131,76 +156,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 = [] | ||||||
|                         command, attr2, self._api, self, multi=multi, program=program |         for command, value in data.items(): | ||||||
|  |             commands += self._get_command(value, command, "") | ||||||
|  |         return {c.name: c for c in commands} | ||||||
|  |  | ||||||
|  |     def _get_command(self, data, command="", category="", categories=None): | ||||||
|  |         commands = [] | ||||||
|  |         if isinstance(data, dict): | ||||||
|  |             if data.get("description") and data.get("protocolType", None): | ||||||
|  |                 commands += [ | ||||||
|  |                     HonCommand( | ||||||
|  |                         command, | ||||||
|  |                         data, | ||||||
|  |                         self, | ||||||
|  |                         category_name=category, | ||||||
|  |                         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 command.parameters.items(): |             for key in command.setting_keys: | ||||||
|                 result.setdefault(name, {})[key] = parameter.value |                 result.append(f"{name}.{key}") | ||||||
|         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") | ||||||
|  | |||||||
							
								
								
									
										12
									
								
								pyhon/appliances/dw.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										12
									
								
								pyhon/appliances/dw.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,12 @@ | |||||||
|  | class Appliance: | ||||||
|  |     def __init__(self, appliance): | ||||||
|  |         self.parent = appliance | ||||||
|  |  | ||||||
|  |     def data(self, data): | ||||||
|  |         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||||
|  |             data["attributes"]["parameters"]["machMode"] = "0" | ||||||
|  |         data["active"] = bool(data.get("attributes", {}).get("activity")) | ||||||
|  |         return data | ||||||
|  |  | ||||||
|  |     def settings(self, settings): | ||||||
|  |         return settings | ||||||
| @ -1,23 +1,17 @@ | |||||||
| from pyhon.parameter import HonParameterEnum |  | ||||||
|  |  | ||||||
|  |  | ||||||
| class Appliance: | class Appliance: | ||||||
|     _FILTERS = { |     def __init__(self, appliance): | ||||||
|         "default": "^(?!iot_(?:recipe|guided))\\S+$", |         self.parent = appliance | ||||||
|         "recipe": "iot_recipe_", |  | ||||||
|         "guided": "iot_guided_", |  | ||||||
|     } |  | ||||||
|  |  | ||||||
|     def __init__(self): |  | ||||||
|         filters = list(self._FILTERS.values()) |  | ||||||
|         data = {"defaultValue": filters[0], "enumValues": filters} |  | ||||||
|         self._program_filter = HonParameterEnum("program_filter", data) |  | ||||||
|  |  | ||||||
|     def data(self, data): |     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 |         return data | ||||||
|  |  | ||||||
|     def settings(self, settings): |     def settings(self, settings): | ||||||
|         settings["program_filter"] = self._program_filter |  | ||||||
|         value = self._FILTERS[self._program_filter.value] |  | ||||||
|         settings["startProgram.program"].filter = value |  | ||||||
|         return 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,92 +1,132 @@ | |||||||
| from pyhon.parameter import ( | from typing import Optional, Dict, Any, List, TYPE_CHECKING, Union | ||||||
|     HonParameterFixed, |  | ||||||
|     HonParameterEnum, | from pyhon.parameter.base import HonParameter | ||||||
|     HonParameterRange, | from pyhon.parameter.enum import HonParameterEnum | ||||||
|     HonParameterProgram, | from pyhon.parameter.fixed import HonParameterFixed | ||||||
| ) | from pyhon.parameter.program import HonParameterProgram | ||||||
|  | from pyhon.parameter.range import HonParameterRange | ||||||
|  |  | ||||||
|  | if TYPE_CHECKING: | ||||||
|  |     from pyhon import HonAPI | ||||||
|  |     from pyhon.appliance import HonAppliance | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonCommand: | class HonCommand: | ||||||
|     def __init__(self, name, attributes, connector, device, multi=None, program=""): |     def __init__( | ||||||
|         self._connector = connector |         self, | ||||||
|         self._device = device |         name: str, | ||||||
|         self._name = name |         attributes: Dict[str, Any], | ||||||
|         self._multi = multi or {} |         appliance: "HonAppliance", | ||||||
|         self._program = program |         categories: Optional[Dict[str, "HonCommand"]] = None, | ||||||
|         self._description = attributes.get("description", "") |         category_name: str = "", | ||||||
|         self._parameters = self._create_parameters(attributes.get("parameters", {})) |     ): | ||||||
|         self._ancillary_parameters = self._create_parameters( |         self._api: HonAPI = appliance.api | ||||||
|             attributes.get("ancillaryParameters", {}) |         self._appliance: "HonAppliance" = appliance | ||||||
|         ) |         self._name: str = name | ||||||
|  |         self._categories: Optional[Dict[str, "HonCommand"]] = categories | ||||||
|  |         self._category_name: str = category_name | ||||||
|  |         self._description: str = attributes.pop("description", "") | ||||||
|  |         self._protocol_type: str = attributes.pop("protocolType", "") | ||||||
|  |         self._parameters: Dict[str, HonParameter] = {} | ||||||
|  |         self._data: Dict[str, Any] = {} | ||||||
|  |         self._available_settings: Dict[str, HonParameter] = {} | ||||||
|  |         self._load_parameters(attributes) | ||||||
|  |  | ||||||
|     def __repr__(self): |     def __repr__(self) -> str: | ||||||
|         return f"{self._name} command" |         return f"{self._name} command" | ||||||
|  |  | ||||||
|     def _create_parameters(self, parameters): |     @property | ||||||
|         result = {} |     def name(self): | ||||||
|         for parameter, attributes in parameters.items(): |         return self._name | ||||||
|             if parameter == "zoneMap" and self._device.zone: |  | ||||||
|                 attributes["default"] = self._device.zone |  | ||||||
|             match attributes.get("typology"): |  | ||||||
|                 case "range": |  | ||||||
|                     result[parameter] = HonParameterRange(parameter, attributes) |  | ||||||
|                 case "enum": |  | ||||||
|                     result[parameter] = HonParameterEnum(parameter, attributes) |  | ||||||
|                 case "fixed": |  | ||||||
|                     result[parameter] = HonParameterFixed(parameter, attributes) |  | ||||||
|         if self._multi: |  | ||||||
|             result["program"] = HonParameterProgram("program", self) |  | ||||||
|         return result |  | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def parameters(self): |     def data(self): | ||||||
|  |         return self._data | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def parameters(self) -> Dict[str, HonParameter]: | ||||||
|         return self._parameters |         return self._parameters | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def ancillary_parameters(self): |     def settings(self) -> Dict[str, HonParameter]: | ||||||
|         return { |         return self._parameters | ||||||
|             key: parameter.value |  | ||||||
|             for key, parameter in self._ancillary_parameters.items() |  | ||||||
|         } |  | ||||||
|  |  | ||||||
|     async def send(self): |     @property | ||||||
|         parameters = { |     def parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]: | ||||||
|             name: parameter.value for name, parameter in self._parameters.items() |         result: Dict[str, Dict[str, Union[str, float]]] = {} | ||||||
|         } |         for name, parameter in self._parameters.items(): | ||||||
|         return await self._connector.send_command( |             result.setdefault(parameter.group, {})[name] = parameter.value | ||||||
|             self._device, self._name, parameters, self.ancillary_parameters |         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: | ||||||
|  |         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 | ||||||
|         ) |         ) | ||||||
|  |  | ||||||
|     def get_programs(self): |     @property | ||||||
|         return self._multi |     def categories(self) -> Dict[str, "HonCommand"]: | ||||||
|  |         if self._categories is None: | ||||||
|     def set_program(self, program): |             return {"_": self} | ||||||
|         self._device.commands[self._name] = self._multi[program] |         return self._categories | ||||||
|  |  | ||||||
|     def _get_settings_keys(self, command=None): |  | ||||||
|         command = command or self |  | ||||||
|         keys = [] |  | ||||||
|         for key, parameter in command._parameters.items(): |  | ||||||
|             if isinstance(parameter, HonParameterFixed): |  | ||||||
|                 continue |  | ||||||
|             if key not in keys: |  | ||||||
|                 keys.append(key) |  | ||||||
|         return keys |  | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def setting_keys(self): |     def category(self) -> str: | ||||||
|         if not self._multi: |         return self._category_name | ||||||
|             return self._get_settings_keys() |  | ||||||
|         result = [ |     @category.setter | ||||||
|             key for cmd in self._multi.values() for key in self._get_settings_keys(cmd) |     def category(self, category: str) -> None: | ||||||
|         ] |         self._appliance.commands[self._name] = self.categories[category] | ||||||
|         return list(set(result + ["program"])) |  | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def settings(self): |     def setting_keys(self) -> List[str]: | ||||||
|         """Parameters with typology enum and range""" |         return list( | ||||||
|         return { |             {param for cmd in self.categories.values() for param in cmd.parameters} | ||||||
|             s: self._parameters.get(s) |         ) | ||||||
|             for s in self.setting_keys |  | ||||||
|             if self._parameters.get(s) is not None |     @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 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: | ||||||
|  |                     result[name] = self._more_options(result[name], parameter) | ||||||
|  |                 result[name] = parameter | ||||||
|  |         return result | ||||||
|  | |||||||
| @ -2,17 +2,17 @@ import json | |||||||
| import logging | import logging | ||||||
| from datetime import datetime | from datetime import datetime | ||||||
| from typing import Dict, Optional | from typing import Dict, Optional | ||||||
| from typing_extensions import Self |  | ||||||
|  |  | ||||||
| from aiohttp import ClientSession | from aiohttp import ClientSession | ||||||
|  | from typing_extensions import Self | ||||||
|  |  | ||||||
| from pyhon import const, exceptions | from pyhon import const, exceptions | ||||||
| from pyhon.appliance import HonAppliance | from pyhon.appliance import HonAppliance | ||||||
| from pyhon.connection.auth import HonAuth | from pyhon.connection.auth import HonAuth | ||||||
| from pyhon.connection.handler.hon import HonConnectionHandler |  | ||||||
| from pyhon.connection.handler.anonym import HonAnonymousConnectionHandler | from pyhon.connection.handler.anonym import HonAnonymousConnectionHandler | ||||||
|  | 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")): | ||||||
|  | |||||||
| @ -6,8 +6,7 @@ import urllib | |||||||
| from contextlib import suppress | from contextlib import suppress | ||||||
| from dataclasses import dataclass | from dataclasses import dataclass | ||||||
| from datetime import datetime, timedelta | from datetime import datetime, timedelta | ||||||
| from pprint import pformat | from typing import Dict, Optional | ||||||
| from typing import Dict, Optional, List |  | ||||||
| from urllib import parse | from urllib import parse | ||||||
| from urllib.parse import quote | from urllib.parse import quote | ||||||
|  |  | ||||||
| @ -183,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: | ||||||
| @ -215,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: | ||||||
| @ -229,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: | ||||||
|  | |||||||
| @ -1,7 +1,7 @@ | |||||||
| import logging | import logging | ||||||
| from collections.abc import AsyncIterator | from collections.abc import AsyncIterator | ||||||
| from contextlib import asynccontextmanager | from contextlib import asynccontextmanager | ||||||
| from typing import Optional, Callable, Dict, Any | from typing import Optional, Callable, Dict | ||||||
|  |  | ||||||
| import aiohttp | import aiohttp | ||||||
| from typing_extensions import Self | from typing_extensions import Self | ||||||
|  | |||||||
| @ -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.parameters.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": | ||||||
|  | |||||||
							
								
								
									
										30
									
								
								pyhon/hon.py
									
									
									
									
									
								
							
							
						
						
									
										30
									
								
								pyhon/hon.py
									
									
									
									
									
								
							| @ -1,9 +1,9 @@ | |||||||
| import asyncio | import asyncio | ||||||
| import copy | from types import TracebackType | ||||||
| from typing import List, Optional, Dict, Any | from typing import List, Optional, Dict, Any, Type | ||||||
| from typing_extensions import Self |  | ||||||
|  |  | ||||||
| from aiohttp import ClientSession | from aiohttp import ClientSession | ||||||
|  | from typing_extensions import Self | ||||||
|  |  | ||||||
| from pyhon import HonAPI, exceptions | from pyhon import HonAPI, exceptions | ||||||
| from pyhon.appliance import HonAppliance | from pyhon.appliance import HonAppliance | ||||||
| @ -17,10 +17,15 @@ class Hon: | |||||||
|         self._appliances: List[HonAppliance] = [] |         self._appliances: List[HonAppliance] = [] | ||||||
|         self._api: Optional[HonAPI] = None |         self._api: Optional[HonAPI] = None | ||||||
|  |  | ||||||
|     async def __aenter__(self): |     async def __aenter__(self) -> Self: | ||||||
|         return await self.create() |         return await self.create() | ||||||
|  |  | ||||||
|     async def __aexit__(self, exc_type, exc_val, exc_tb): |     async def __aexit__( | ||||||
|  |         self, | ||||||
|  |         exc_type: Optional[Type[BaseException]], | ||||||
|  |         exc: Optional[BaseException], | ||||||
|  |         traceback: Optional[TracebackType], | ||||||
|  |     ) -> None: | ||||||
|         await self.close() |         await self.close() | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
| @ -42,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( | ||||||
|             *[ |             *[ | ||||||
| @ -53,12 +58,13 @@ class Hon: | |||||||
|         ) |         ) | ||||||
|         self._appliances.append(appliance) |         self._appliances.append(appliance) | ||||||
|  |  | ||||||
|     async def setup(self): |     async def setup(self) -> None: | ||||||
|         appliance: Dict |         appliance: Dict | ||||||
|         for appliance in (await self._api.load_appliances())["payload"]["appliances"]: |         for appliance in (await self.api.load_appliances())["payload"]["appliances"]: | ||||||
|             for zone in range(int(appliance.get("zone", "0"))): |             if (zones := int(appliance.get("zone", "0"))) > 1: | ||||||
|                 await self._create_appliance(appliance.copy(), zone=zone + 1) |                 for zone in range(zones): | ||||||
|  |                     await self._create_appliance(appliance.copy(), zone=zone + 1) | ||||||
|             await self._create_appliance(appliance) |             await self._create_appliance(appliance) | ||||||
|  |  | ||||||
|     async def close(self): |     async def close(self) -> None: | ||||||
|         await self._api.close() |         await self.api.close() | ||||||
|  | |||||||
| @ -1,157 +0,0 @@ | |||||||
| import re |  | ||||||
|  |  | ||||||
|  |  | ||||||
| def str_to_float(string): |  | ||||||
|     try: |  | ||||||
|         return int(string) |  | ||||||
|     except ValueError: |  | ||||||
|         return float(str(string).replace(",", ".")) |  | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonParameter: |  | ||||||
|     def __init__(self, key, attributes): |  | ||||||
|         self._key = key |  | ||||||
|         self._category = attributes.get("category") |  | ||||||
|         self._typology = attributes.get("typology") |  | ||||||
|         self._mandatory = attributes.get("mandatory") |  | ||||||
|         self._value = "" |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def key(self): |  | ||||||
|         return self._key |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def value(self): |  | ||||||
|         return self._value if self._value is not None else "0" |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def category(self): |  | ||||||
|         return self._category |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def typology(self): |  | ||||||
|         return self._typology |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def mandatory(self): |  | ||||||
|         return self._mandatory |  | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonParameterFixed(HonParameter): |  | ||||||
|     def __init__(self, key, attributes): |  | ||||||
|         super().__init__(key, attributes) |  | ||||||
|         self._value = attributes.get("fixedValue", None) |  | ||||||
|  |  | ||||||
|     def __repr__(self): |  | ||||||
|         return f"{self.__class__} (<{self.key}> fixed)" |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def value(self): |  | ||||||
|         return self._value if self._value is not None else "0" |  | ||||||
|  |  | ||||||
|     @value.setter |  | ||||||
|     def value(self, value): |  | ||||||
|         if not value == self._value: |  | ||||||
|             raise ValueError("Can't change fixed value") |  | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonParameterRange(HonParameter): |  | ||||||
|     def __init__(self, key, attributes): |  | ||||||
|         super().__init__(key, attributes) |  | ||||||
|         self._min = str_to_float(attributes["minimumValue"]) |  | ||||||
|         self._max = str_to_float(attributes["maximumValue"]) |  | ||||||
|         self._step = str_to_float(attributes["incrementValue"]) |  | ||||||
|         self._default = str_to_float(attributes.get("defaultValue", self._min)) |  | ||||||
|         self._value = self._default |  | ||||||
|  |  | ||||||
|     def __repr__(self): |  | ||||||
|         return f"{self.__class__} (<{self.key}> [{self._min} - {self._max}])" |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def min(self): |  | ||||||
|         return self._min |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def max(self): |  | ||||||
|         return self._max |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def step(self): |  | ||||||
|         return self._step |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def value(self): |  | ||||||
|         return self._value if self._value is not None else self._min |  | ||||||
|  |  | ||||||
|     @value.setter |  | ||||||
|     def value(self, value): |  | ||||||
|         value = str_to_float(value) |  | ||||||
|         if self._min <= value <= self._max and not value % self._step: |  | ||||||
|             self._value = value |  | ||||||
|         else: |  | ||||||
|             raise ValueError( |  | ||||||
|                 f"Allowed: min {self._min} max {self._max} step {self._step}" |  | ||||||
|             ) |  | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonParameterEnum(HonParameter): |  | ||||||
|     def __init__(self, key, attributes): |  | ||||||
|         super().__init__(key, attributes) |  | ||||||
|         self._default = attributes.get("defaultValue") |  | ||||||
|         self._value = self._default or "0" |  | ||||||
|         self._values = attributes.get("enumValues") |  | ||||||
|  |  | ||||||
|     def __repr__(self): |  | ||||||
|         return f"{self.__class__} (<{self.key}> {self.values})" |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def values(self): |  | ||||||
|         return [str(value) for value in self._values] |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def value(self): |  | ||||||
|         return self._value if self._value is not None else self.values[0] |  | ||||||
|  |  | ||||||
|     @value.setter |  | ||||||
|     def value(self, value): |  | ||||||
|         if value in self.values: |  | ||||||
|             self._value = value |  | ||||||
|         else: |  | ||||||
|             raise ValueError(f"Allowed values {self._value}") |  | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonParameterProgram(HonParameterEnum): |  | ||||||
|     def __init__(self, key, command): |  | ||||||
|         super().__init__(key, {}) |  | ||||||
|         self._command = command |  | ||||||
|         self._value = command._program |  | ||||||
|         self._values = command._multi |  | ||||||
|         self._typology = "enum" |  | ||||||
|         self._filter = "" |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def value(self): |  | ||||||
|         return self._value |  | ||||||
|  |  | ||||||
|     @value.setter |  | ||||||
|     def value(self, value): |  | ||||||
|         if value in self.values: |  | ||||||
|             self._command.set_program(value) |  | ||||||
|         else: |  | ||||||
|             raise ValueError(f"Allowed values {self._values}") |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def filter(self): |  | ||||||
|         return self._filter |  | ||||||
|  |  | ||||||
|     @filter.setter |  | ||||||
|     def filter(self, filter): |  | ||||||
|         self._filter = filter |  | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def values(self): |  | ||||||
|         values = [] |  | ||||||
|         for value in self._values: |  | ||||||
|             if not self._filter or re.findall(self._filter, str(value)): |  | ||||||
|                 values.append(str(value)) |  | ||||||
|         return sorted(values) |  | ||||||
							
								
								
									
										0
									
								
								pyhon/parameter/__init__.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										0
									
								
								pyhon/parameter/__init__.py
									
									
									
									
									
										Normal file
									
								
							
							
								
								
									
										39
									
								
								pyhon/parameter/base.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										39
									
								
								pyhon/parameter/base.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,39 @@ | |||||||
|  | from typing import Dict, Any, List | ||||||
|  |  | ||||||
|  |  | ||||||
|  | class HonParameter: | ||||||
|  |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|  |         self._key = key | ||||||
|  |         self._category: str = attributes.get("category", "") | ||||||
|  |         self._typology: str = attributes.get("typology", "") | ||||||
|  |         self._mandatory: int = attributes.get("mandatory", 0) | ||||||
|  |         self._value: str | float = "" | ||||||
|  |         self._group: str = group | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def key(self) -> str: | ||||||
|  |         return self._key | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def value(self) -> str | float: | ||||||
|  |         return self._value if self._value is not None else "0" | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def values(self) -> List[str]: | ||||||
|  |         return [str(self.value)] | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def category(self) -> str: | ||||||
|  |         return self._category | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def typology(self) -> str: | ||||||
|  |         return self._typology | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def mandatory(self) -> int: | ||||||
|  |         return self._mandatory | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def group(self) -> str: | ||||||
|  |         return self._group | ||||||
							
								
								
									
										29
									
								
								pyhon/parameter/enum.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										29
									
								
								pyhon/parameter/enum.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,29 @@ | |||||||
|  | from typing import Dict, Any, List | ||||||
|  |  | ||||||
|  | from pyhon.parameter.base import HonParameter | ||||||
|  |  | ||||||
|  |  | ||||||
|  | class HonParameterEnum(HonParameter): | ||||||
|  |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|  |         super().__init__(key, attributes, group) | ||||||
|  |         self._default = attributes.get("defaultValue") | ||||||
|  |         self._value = self._default or "0" | ||||||
|  |         self._values: List[str] = attributes.get("enumValues", []) | ||||||
|  |  | ||||||
|  |     def __repr__(self) -> str: | ||||||
|  |         return f"{self.__class__} (<{self.key}> {self.values})" | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def values(self) -> List[str]: | ||||||
|  |         return [str(value) for value in self._values] | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def value(self) -> str | float: | ||||||
|  |         return self._value if self._value is not None else self.values[0] | ||||||
|  |  | ||||||
|  |     @value.setter | ||||||
|  |     def value(self, value: str) -> None: | ||||||
|  |         if value in self.values: | ||||||
|  |             self._value = value | ||||||
|  |         else: | ||||||
|  |             raise ValueError(f"Allowed values {self._value}") | ||||||
							
								
								
									
										21
									
								
								pyhon/parameter/fixed.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										21
									
								
								pyhon/parameter/fixed.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,21 @@ | |||||||
|  | from typing import Dict, Any | ||||||
|  |  | ||||||
|  | from pyhon.parameter.base import HonParameter | ||||||
|  |  | ||||||
|  |  | ||||||
|  | class HonParameterFixed(HonParameter): | ||||||
|  |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|  |         super().__init__(key, attributes, group) | ||||||
|  |         self._value = attributes.get("fixedValue", None) | ||||||
|  |  | ||||||
|  |     def __repr__(self) -> str: | ||||||
|  |         return f"{self.__class__} (<{self.key}> fixed)" | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def value(self) -> str | float: | ||||||
|  |         return self._value if self._value is not None else "0" | ||||||
|  |  | ||||||
|  |     @value.setter | ||||||
|  |     def value(self, value: str | float) -> None: | ||||||
|  |         # Fixed values seems being not so fixed as thought | ||||||
|  |         self._value = value | ||||||
							
								
								
									
										45
									
								
								pyhon/parameter/program.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										45
									
								
								pyhon/parameter/program.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,45 @@ | |||||||
|  | from typing import List, TYPE_CHECKING, Dict | ||||||
|  |  | ||||||
|  | from pyhon.parameter.enum import HonParameterEnum | ||||||
|  |  | ||||||
|  | if TYPE_CHECKING: | ||||||
|  |     from pyhon.commands import HonCommand | ||||||
|  |  | ||||||
|  |  | ||||||
|  | class HonParameterProgram(HonParameterEnum): | ||||||
|  |     _FILTER = ["iot_recipe", "iot_guided"] | ||||||
|  |  | ||||||
|  |     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" | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def value(self) -> str | float: | ||||||
|  |         return self._value | ||||||
|  |  | ||||||
|  |     @value.setter | ||||||
|  |     def value(self, value: str) -> None: | ||||||
|  |         if value in self.values: | ||||||
|  |             self._command.category = value | ||||||
|  |         else: | ||||||
|  |             raise ValueError(f"Allowed values {self.values}") | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def values(self) -> List[str]: | ||||||
|  |         values = [v for v in self._programs if all(f not in v for f in self._FILTER)] | ||||||
|  |         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())) | ||||||
							
								
								
									
										53
									
								
								pyhon/parameter/range.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										53
									
								
								pyhon/parameter/range.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,53 @@ | |||||||
|  | from typing import Dict, Any, List | ||||||
|  |  | ||||||
|  | from pyhon.parameter.base import HonParameter | ||||||
|  |  | ||||||
|  |  | ||||||
|  | def str_to_float(string: str | float) -> float: | ||||||
|  |     try: | ||||||
|  |         return int(string) | ||||||
|  |     except ValueError: | ||||||
|  |         return float(str(string).replace(",", ".")) | ||||||
|  |  | ||||||
|  |  | ||||||
|  | class HonParameterRange(HonParameter): | ||||||
|  |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|  |         super().__init__(key, attributes, group) | ||||||
|  |         self._min: float = str_to_float(attributes["minimumValue"]) | ||||||
|  |         self._max: float = str_to_float(attributes["maximumValue"]) | ||||||
|  |         self._step: float = str_to_float(attributes["incrementValue"]) | ||||||
|  |         self._default: float = str_to_float(attributes.get("defaultValue", self._min)) | ||||||
|  |         self._value: float = self._default | ||||||
|  |  | ||||||
|  |     def __repr__(self): | ||||||
|  |         return f"{self.__class__} (<{self.key}> [{self._min} - {self._max}])" | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def min(self) -> float: | ||||||
|  |         return self._min | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def max(self) -> float: | ||||||
|  |         return self._max | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def step(self) -> float: | ||||||
|  |         return self._step | ||||||
|  |  | ||||||
|  |     @property | ||||||
|  |     def value(self) -> float: | ||||||
|  |         return self._value if self._value is not None else self._min | ||||||
|  |  | ||||||
|  |     @value.setter | ||||||
|  |     def value(self, value: float) -> None: | ||||||
|  |         value = str_to_float(value) | ||||||
|  |         if self._min <= value <= self._max and not value % self._step: | ||||||
|  |             self._value = value | ||||||
|  |         else: | ||||||
|  |             raise ValueError( | ||||||
|  |                 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))] | ||||||
| @ -1 +1,2 @@ | |||||||
| aiohttp | aiohttp==3.8.4 | ||||||
|  | yarl==1.8.2 | ||||||
|  | |||||||
							
								
								
									
										4
									
								
								requirements_dev.txt
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										4
									
								
								requirements_dev.txt
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,4 @@ | |||||||
|  | black==23.3.0 | ||||||
|  | flake8==6.0.0 | ||||||
|  | mypy==1.2.0 | ||||||
|  | pylint==2.17.2 | ||||||
							
								
								
									
										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.0b3", |     version="0.10.2", | ||||||
|     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
	