Compare commits
	
		
			6 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| 715d8a96fc | |||
| ec358034e1 | |||
| 6cdac99ce8 | |||
| 0c1670733f | |||
| 744094dc07 | |||
| bf2015fcb9 | 
							
								
								
									
										10
									
								
								.github/workflows/python-check.yml
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										10
									
								
								.github/workflows/python-check.yml
									
									
									
									
										vendored
									
									
								
							| @ -28,13 +28,15 @@ jobs: | |||||||
|         python -m pip install -r requirements_dev.txt |         python -m pip install -r requirements_dev.txt | ||||||
|     - name: Lint with flake8 |     - name: Lint with flake8 | ||||||
|       run: | |       run: | | ||||||
|         flake8 . --count --statistics |         # stop the build if there are Python syntax errors or undefined names | ||||||
|  |         flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics | ||||||
|  |         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=88 --statistics | ||||||
|     - name: Type check with mypy |     - name: Type check with mypy | ||||||
|       run: | |       run: | | ||||||
|         mypy pyhon/ |         mypy pyhon/ | ||||||
|     - name: Analysing the code with pylint |     # - name: Analysing the code with pylint | ||||||
|       run: | |     #   run: | | ||||||
|         pylint $(git ls-files '*.py') |     #     pylint --max-line-length 88 $(git ls-files '*.py') | ||||||
|     - name: Check black style |     - name: Check black style | ||||||
|       run: | |       run: | | ||||||
|         black . --check |         black . --check | ||||||
|  | |||||||
| @ -1,9 +0,0 @@ | |||||||
| [MESSAGES CONTROL] |  | ||||||
|  |  | ||||||
| disable=missing-docstring |  | ||||||
|  |  | ||||||
| [FORMAT] |  | ||||||
|  |  | ||||||
| max-args=6 |  | ||||||
| max-attributes=8 |  | ||||||
| max-line-length=88 |  | ||||||
| @ -1 +0,0 @@ | |||||||
| include pyhon/py.typed |  | ||||||
							
								
								
									
										31
									
								
								mypy.ini
									
									
									
									
									
								
							
							
						
						
									
										31
									
								
								mypy.ini
									
									
									
									
									
								
							| @ -1,24 +1,9 @@ | |||||||
| [mypy] | [mypy] | ||||||
| check_untyped_defs = true | check_untyped_defs = True | ||||||
| disallow_any_generics = true | disallow_any_generics = True | ||||||
| disallow_any_unimported = true | disallow_untyped_defs = True | ||||||
| disallow_incomplete_defs = true | disallow_any_unimported = True | ||||||
| disallow_subclassing_any = true | no_implicit_optional = True | ||||||
| disallow_untyped_calls = true | warn_return_any = True | ||||||
| disallow_untyped_decorators = true | show_error_codes = True | ||||||
| disallow_untyped_defs = true | warn_unused_ignores = True | ||||||
| disable_error_code = annotation-unchecked |  | ||||||
| enable_error_code = ignore-without-code, redundant-self, truthy-iterable |  | ||||||
| follow_imports = silent |  | ||||||
| local_partial_types = true |  | ||||||
| no_implicit_optional = true |  | ||||||
| no_implicit_reexport = true |  | ||||||
| show_error_codes = true |  | ||||||
| strict_concatenate = false |  | ||||||
| strict_equality = true |  | ||||||
| warn_incomplete_stub = true |  | ||||||
| warn_redundant_casts = true |  | ||||||
| warn_return_any = true |  | ||||||
| warn_unreachable = true |  | ||||||
| warn_unused_configs = true |  | ||||||
| warn_unused_ignores = true |  | ||||||
|  | |||||||
| @ -11,7 +11,6 @@ from typing import Tuple, Dict, Any | |||||||
| if __name__ == "__main__": | if __name__ == "__main__": | ||||||
|     sys.path.insert(0, str(Path(__file__).parent.parent)) |     sys.path.insert(0, str(Path(__file__).parent.parent)) | ||||||
|  |  | ||||||
| # pylint: disable=wrong-import-position |  | ||||||
| from pyhon import Hon, HonAPI, diagnose, printer | from pyhon import Hon, HonAPI, diagnose, printer | ||||||
|  |  | ||||||
| _LOGGER = logging.getLogger(__name__) | _LOGGER = logging.getLogger(__name__) | ||||||
| @ -31,13 +30,13 @@ def get_arguments() -> Dict[str, Any]: | |||||||
|     export.add_argument("--zip", help="create zip archive", action="store_true") |     export.add_argument("--zip", help="create zip archive", action="store_true") | ||||||
|     export.add_argument("--anonymous", help="anonymize data", action="store_true") |     export.add_argument("--anonymous", help="anonymize data", action="store_true") | ||||||
|     export.add_argument("directory", nargs="?", default=Path().cwd()) |     export.add_argument("directory", nargs="?", default=Path().cwd()) | ||||||
|     translation = subparser.add_parser( |     translate = subparser.add_parser( | ||||||
|         "translate", help="print available translation keys" |         "translate", help="print available translation keys" | ||||||
|     ) |     ) | ||||||
|     translation.add_argument( |     translate.add_argument( | ||||||
|         "translate", help="language (de, en, fr...)", metavar="LANGUAGE" |         "translate", help="language (de, en, fr...)", metavar="LANGUAGE" | ||||||
|     ) |     ) | ||||||
|     translation.add_argument("--json", help="print as json", action="store_true") |     translate.add_argument("--json", help="print as json", action="store_true") | ||||||
|     parser.add_argument( |     parser.add_argument( | ||||||
|         "-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd() |         "-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd() | ||||||
|     ) |     ) | ||||||
| @ -92,9 +91,11 @@ async def main() -> None: | |||||||
|                 data = device.data.copy() |                 data = device.data.copy() | ||||||
|                 attr = "get" if args.get("all") else "pop" |                 attr = "get" if args.get("all") else "pop" | ||||||
|                 print( |                 print( | ||||||
|                     printer.key_print(getattr(data["attributes"], attr)("parameters")) |                     printer.key_print( | ||||||
|  |                         data["attributes"].__getattribute__(attr)("parameters") | ||||||
|  |                     ) | ||||||
|                 ) |                 ) | ||||||
|                 print(printer.key_print(getattr(data, attr)("appliance"))) |                 print(printer.key_print(data.__getattribute__(attr)("appliance"))) | ||||||
|                 print(printer.key_print(data)) |                 print(printer.key_print(data)) | ||||||
|                 print( |                 print( | ||||||
|                     printer.pretty_print( |                     printer.pretty_print( | ||||||
|  | |||||||
| @ -3,7 +3,7 @@ import logging | |||||||
| import re | import re | ||||||
| from datetime import datetime, timedelta | from datetime import datetime, timedelta | ||||||
| from pathlib import Path | from pathlib import Path | ||||||
| from typing import Optional, Dict, Any, TYPE_CHECKING, List, TypeVar, overload | from typing import Optional, Dict, Any, TYPE_CHECKING, List | ||||||
|  |  | ||||||
| from pyhon import diagnose, exceptions | from pyhon import diagnose, exceptions | ||||||
| from pyhon.appliances.base import ApplianceBase | from pyhon.appliances.base import ApplianceBase | ||||||
| @ -20,10 +20,7 @@ if TYPE_CHECKING: | |||||||
|  |  | ||||||
| _LOGGER = logging.getLogger(__name__) | _LOGGER = logging.getLogger(__name__) | ||||||
|  |  | ||||||
| T = TypeVar("T") |  | ||||||
|  |  | ||||||
|  |  | ||||||
| # pylint: disable=too-many-public-methods,too-many-instance-attributes |  | ||||||
| class HonAppliance: | class HonAppliance: | ||||||
|     _MINIMAL_UPDATE_INTERVAL = 5  # seconds |     _MINIMAL_UPDATE_INTERVAL = 5  # seconds | ||||||
|  |  | ||||||
| @ -51,35 +48,24 @@ class HonAppliance: | |||||||
|         except ModuleNotFoundError: |         except ModuleNotFoundError: | ||||||
|             self._extra = None |             self._extra = None | ||||||
|  |  | ||||||
|     def _get_nested_item(self, item: str) -> Any: |  | ||||||
|         result: List[Any] | Dict[str, Any] = self.data |  | ||||||
|         for key in item.split("."): |  | ||||||
|             if all(k in "0123456789" for k in key) and isinstance(result, list): |  | ||||||
|                 result = result[int(key)] |  | ||||||
|             elif isinstance(result, dict): |  | ||||||
|                 result = result[key] |  | ||||||
|         return result |  | ||||||
|  |  | ||||||
|     def __getitem__(self, item: str) -> Any: |     def __getitem__(self, item: str) -> Any: | ||||||
|         if self._zone: |         if self._zone: | ||||||
|             item += f"Z{self._zone}" |             item += f"Z{self._zone}" | ||||||
|         if "." in item: |         if "." in item: | ||||||
|             return self._get_nested_item(item) |             result = self.data | ||||||
|  |             for key in item.split("."): | ||||||
|  |                 if all(k in "0123456789" for k in key) and isinstance(result, list): | ||||||
|  |                     result = result[int(key)] | ||||||
|  |                 else: | ||||||
|  |                     result = result[key] | ||||||
|  |             return result | ||||||
|         if item in self.data: |         if item in self.data: | ||||||
|             return self.data[item] |             return self.data[item] | ||||||
|         if item in self.attributes["parameters"]: |         if item in self.attributes["parameters"]: | ||||||
|             return self.attributes["parameters"][item].value |             return self.attributes["parameters"][item].value | ||||||
|         return self.info[item] |         return self.info[item] | ||||||
|  |  | ||||||
|     @overload |     def get(self, item: str, default: Any = None) -> Any: | ||||||
|     def get(self, item: str, default: None = None) -> Any: |  | ||||||
|         ... |  | ||||||
|  |  | ||||||
|     @overload |  | ||||||
|     def get(self, item: str, default: T) -> T: |  | ||||||
|         ... |  | ||||||
|  |  | ||||||
|     def get(self, item: str, default: Optional[T] = None) -> Any: |  | ||||||
|         try: |         try: | ||||||
|             return self[item] |             return self[item] | ||||||
|         except (KeyError, IndexError): |         except (KeyError, IndexError): | ||||||
| @ -202,8 +188,12 @@ class HonAppliance: | |||||||
|  |  | ||||||
|     async def update(self, force: bool = False) -> None: |     async def update(self, force: bool = False) -> None: | ||||||
|         now = datetime.now() |         now = datetime.now() | ||||||
|         min_age = now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL) |         if ( | ||||||
|         if force or not self._last_update or self._last_update < min_age: |             force | ||||||
|  |             or not self._last_update | ||||||
|  |             or self._last_update | ||||||
|  |             < now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL) | ||||||
|  |         ): | ||||||
|             self._last_update = now |             self._last_update = now | ||||||
|             await self.load_attributes() |             await self.load_attributes() | ||||||
|             self.sync_params_to_command("settings") |             self.sync_params_to_command("settings") | ||||||
| @ -284,23 +274,18 @@ class HonAppliance: | |||||||
|         for command, data in self.commands.items(): |         for command, data in self.commands.items(): | ||||||
|             if command == main or target and command not in target: |             if command == main or target and command not in target: | ||||||
|                 continue |                 continue | ||||||
|  |             for name, parameter in data.parameters.items(): | ||||||
|             for name, target_param in data.parameters.items(): |                 if base_value := base.parameters.get(name): | ||||||
|                 if not (base_param := base.parameters.get(name)): |                     if isinstance(base_value, HonParameterRange) and isinstance( | ||||||
|                     return |                         parameter, HonParameterRange | ||||||
|                 self.sync_parameter(base_param, target_param) |                     ): | ||||||
|  |                         parameter.max = base_value.max | ||||||
|     def sync_parameter(self, main: Parameter, target: Parameter) -> None: |                         parameter.min = base_value.min | ||||||
|         if isinstance(main, HonParameterRange) and isinstance( |                         parameter.step = base_value.step | ||||||
|             target, HonParameterRange |                     elif isinstance(parameter, HonParameterRange): | ||||||
|         ): |                         parameter.max = int(base_value.value) | ||||||
|             target.max = main.max |                         parameter.min = int(base_value.value) | ||||||
|             target.min = main.min |                         parameter.step = 1 | ||||||
|             target.step = main.step |                     elif isinstance(parameter, HonParameterEnum): | ||||||
|         elif isinstance(target, HonParameterRange): |                         parameter.values = base_value.values | ||||||
|             target.max = int(main.value) |                     parameter.value = base_value.value | ||||||
|             target.min = int(main.value) |  | ||||||
|             target.step = 1 |  | ||||||
|         elif isinstance(target, HonParameterEnum): |  | ||||||
|             target.values = main.values |  | ||||||
|         target.value = main.value |  | ||||||
|  | |||||||
| @ -1,4 +1,3 @@ | |||||||
| # pylint: disable=duplicate-code |  | ||||||
| from typing import Any, Dict | from typing import Any, Dict | ||||||
|  |  | ||||||
| from pyhon.appliances.base import ApplianceBase | from pyhon.appliances.base import ApplianceBase | ||||||
|  | |||||||
| @ -1,6 +1,7 @@ | |||||||
| from typing import Any, Dict | from typing import Any, Dict | ||||||
|  |  | ||||||
| from pyhon.appliances.base import ApplianceBase | from pyhon.appliances.base import ApplianceBase | ||||||
|  | from pyhon.parameter.program import HonParameterProgram | ||||||
|  |  | ||||||
|  |  | ||||||
| class Appliance(ApplianceBase): | class Appliance(ApplianceBase): | ||||||
|  | |||||||
| @ -10,12 +10,12 @@ class Appliance(ApplianceBase): | |||||||
|             data["modeZ1"] = "holiday" |             data["modeZ1"] = "holiday" | ||||||
|         elif data["parameters"]["intelligenceMode"] == "1": |         elif data["parameters"]["intelligenceMode"] == "1": | ||||||
|             data["modeZ1"] = "auto_set" |             data["modeZ1"] = "auto_set" | ||||||
|         elif data["parameters"].get("quickModeZ1") == "1": |         elif data["parameters"]["quickModeZ1"] == "1": | ||||||
|             data["modeZ1"] = "super_cool" |             data["modeZ1"] = "super_cool" | ||||||
|         else: |         else: | ||||||
|             data["modeZ1"] = "no_mode" |             data["modeZ1"] = "no_mode" | ||||||
|  |  | ||||||
|         if data["parameters"].get("quickModeZ2") == "1": |         if data["parameters"]["quickModeZ2"] == "1": | ||||||
|             data["modeZ2"] = "super_freeze" |             data["modeZ2"] = "super_freeze" | ||||||
|         elif data["parameters"]["intelligenceMode"] == "1": |         elif data["parameters"]["intelligenceMode"] == "1": | ||||||
|             data["modeZ2"] = "auto_set" |             data["modeZ2"] = "auto_set" | ||||||
|  | |||||||
| @ -1,4 +1,3 @@ | |||||||
| # pylint: disable=duplicate-code |  | ||||||
| from typing import Any, Dict | from typing import Any, Dict | ||||||
|  |  | ||||||
| from pyhon.appliances.base import ApplianceBase | from pyhon.appliances.base import ApplianceBase | ||||||
|  | |||||||
| @ -1,4 +1,3 @@ | |||||||
| # pylint: disable=duplicate-code |  | ||||||
| from typing import Dict, Any | from typing import Dict, Any | ||||||
|  |  | ||||||
| from pyhon.appliances.base import ApplianceBase | from pyhon.appliances.base import ApplianceBase | ||||||
|  | |||||||
| @ -1,11 +0,0 @@ | |||||||
| from typing import Any, Dict |  | ||||||
|  |  | ||||||
| from pyhon.appliances.base import ApplianceBase |  | ||||||
|  |  | ||||||
|  |  | ||||||
| class Appliance(ApplianceBase): |  | ||||||
|     def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: |  | ||||||
|         data = super().attributes(data) |  | ||||||
|         data["active"] = data["parameters"]["onOffStatus"] == "1" |  | ||||||
|  |  | ||||||
|         return data |  | ||||||
| @ -1,4 +1,3 @@ | |||||||
| # pylint: disable=duplicate-code |  | ||||||
| from typing import Any, Dict | from typing import Any, Dict | ||||||
|  |  | ||||||
| from pyhon.appliances.base import ApplianceBase | from pyhon.appliances.base import ApplianceBase | ||||||
|  | |||||||
| @ -1,15 +1,14 @@ | |||||||
| import asyncio | import asyncio | ||||||
| from contextlib import suppress | from contextlib import suppress | ||||||
| from copy import copy | from copy import copy | ||||||
| from typing import Dict, Any, Optional, TYPE_CHECKING, List | from typing import Dict, Any, Optional, TYPE_CHECKING, List, Collection | ||||||
|  |  | ||||||
| from pyhon.commands import HonCommand | from pyhon.commands import HonCommand | ||||||
| from pyhon.exceptions import NoAuthenticationException |  | ||||||
| from pyhon.parameter.fixed import HonParameterFixed | from pyhon.parameter.fixed import HonParameterFixed | ||||||
| from pyhon.parameter.program import HonParameterProgram | from pyhon.parameter.program import HonParameterProgram | ||||||
|  |  | ||||||
| if TYPE_CHECKING: | if TYPE_CHECKING: | ||||||
|     from pyhon import HonAPI |     from pyhon import HonAPI, exceptions | ||||||
|     from pyhon.appliance import HonAppliance |     from pyhon.appliance import HonAppliance | ||||||
|  |  | ||||||
|  |  | ||||||
| @ -17,12 +16,12 @@ class HonCommandLoader: | |||||||
|     """Loads and parses hOn command data""" |     """Loads and parses hOn command data""" | ||||||
|  |  | ||||||
|     def __init__(self, api: "HonAPI", appliance: "HonAppliance") -> None: |     def __init__(self, api: "HonAPI", appliance: "HonAppliance") -> None: | ||||||
|         self._api: "HonAPI" = api |  | ||||||
|         self._appliance: "HonAppliance" = appliance |  | ||||||
|         self._api_commands: Dict[str, Any] = {} |         self._api_commands: Dict[str, Any] = {} | ||||||
|         self._favourites: List[Dict[str, Any]] = [] |         self._favourites: List[Dict[str, Any]] = [] | ||||||
|         self._command_history: List[Dict[str, Any]] = [] |         self._command_history: List[Dict[str, Any]] = [] | ||||||
|         self._commands: Dict[str, HonCommand] = {} |         self._commands: Dict[str, HonCommand] = {} | ||||||
|  |         self._api: "HonAPI" = api | ||||||
|  |         self._appliance: "HonAppliance" = appliance | ||||||
|         self._appliance_data: Dict[str, Any] = {} |         self._appliance_data: Dict[str, Any] = {} | ||||||
|         self._additional_data: Dict[str, Any] = {} |         self._additional_data: Dict[str, Any] = {} | ||||||
|  |  | ||||||
| @ -30,7 +29,7 @@ class HonCommandLoader: | |||||||
|     def api(self) -> "HonAPI": |     def api(self) -> "HonAPI": | ||||||
|         """api connection object""" |         """api connection object""" | ||||||
|         if self._api is None: |         if self._api is None: | ||||||
|             raise NoAuthenticationException("Missing hOn login") |             raise exceptions.NoAuthenticationException("Missing hOn login") | ||||||
|         return self._api |         return self._api | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
| @ -184,44 +183,23 @@ class HonCommandLoader: | |||||||
|     def _add_favourites(self) -> None: |     def _add_favourites(self) -> None: | ||||||
|         """Patch program categories with favourites""" |         """Patch program categories with favourites""" | ||||||
|         for favourite in self._favourites: |         for favourite in self._favourites: | ||||||
|             name, command_name, base = self._get_favourite_info(favourite) |             name = favourite.get("favouriteName", {}) | ||||||
|             if not base: |             command = favourite.get("command", {}) | ||||||
|  |             command_name = command.get("commandName", "") | ||||||
|  |             program_name = self._clean_name(command.get("programName", "")) | ||||||
|  |             if not (base := self.commands[command_name].categories.get(program_name)): | ||||||
|                 continue |                 continue | ||||||
|             base_command: HonCommand = copy(base) |             base_command: HonCommand = copy(base) | ||||||
|             self._update_base_command_with_data(base_command, favourite) |             for data in command.values(): | ||||||
|             self._update_base_command_with_favourite(base_command) |                 if isinstance(data, str): | ||||||
|             self._update_program_categories(command_name, name, base_command) |  | ||||||
|  |  | ||||||
|     def _get_favourite_info( |  | ||||||
|         self, favourite: Dict[str, Any] |  | ||||||
|     ) -> tuple[str, str, HonCommand | None]: |  | ||||||
|         name: str = favourite.get("favouriteName", {}) |  | ||||||
|         command = favourite.get("command", {}) |  | ||||||
|         command_name: str = command.get("commandName", "") |  | ||||||
|         program_name = self._clean_name(command.get("programName", "")) |  | ||||||
|         base_command = self.commands[command_name].categories.get(program_name) |  | ||||||
|         return name, command_name, base_command |  | ||||||
|  |  | ||||||
|     def _update_base_command_with_data( |  | ||||||
|         self, base_command: HonCommand, command: Dict[str, Any] |  | ||||||
|     ) -> None: |  | ||||||
|         for data in command.values(): |  | ||||||
|             if isinstance(data, str): |  | ||||||
|                 continue |  | ||||||
|             for key, value in data.items(): |  | ||||||
|                 if not (parameter := base_command.parameters.get(key)): |  | ||||||
|                     continue |                     continue | ||||||
|                 with suppress(ValueError): |                 for key, value in data.items(): | ||||||
|                     parameter.value = value |                     if parameter := base_command.parameters.get(key): | ||||||
|  |                         with suppress(ValueError): | ||||||
|     def _update_base_command_with_favourite(self, base_command: HonCommand) -> None: |                             parameter.value = value | ||||||
|         extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") |             extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") | ||||||
|         base_command.parameters.update(favourite=extra_param) |             base_command.parameters.update(favourite=extra_param) | ||||||
|  |             program = base_command.parameters["program"] | ||||||
|     def _update_program_categories( |             if isinstance(program, HonParameterProgram): | ||||||
|         self, command_name: str, name: str, base_command: HonCommand |                 program.set_value(name) | ||||||
|     ) -> None: |             self.commands[command_name].categories[name] = base_command | ||||||
|         program = base_command.parameters["program"] |  | ||||||
|         if isinstance(program, HonParameterProgram): |  | ||||||
|             program.set_value(name) |  | ||||||
|         self.commands[command_name].categories[name] = base_command |  | ||||||
|  | |||||||
| @ -27,16 +27,17 @@ class HonCommand: | |||||||
|         categories: Optional[Dict[str, "HonCommand"]] = None, |         categories: Optional[Dict[str, "HonCommand"]] = None, | ||||||
|         category_name: str = "", |         category_name: str = "", | ||||||
|     ): |     ): | ||||||
|         self._name: str = name |         self._api: Optional[HonAPI] = appliance.api | ||||||
|         self._api: Optional[HonAPI] = None |  | ||||||
|         self._appliance: "HonAppliance" = appliance |         self._appliance: "HonAppliance" = appliance | ||||||
|  |         self._name: str = name | ||||||
|         self._categories: Optional[Dict[str, "HonCommand"]] = categories |         self._categories: Optional[Dict[str, "HonCommand"]] = categories | ||||||
|         self._category_name: str = category_name |         self._category_name: str = category_name | ||||||
|         self._parameters: Dict[str, Parameter] = {} |         self._description: str = attributes.pop("description", "") | ||||||
|  |         self._protocol_type: str = attributes.pop("protocolType", "") | ||||||
|  |         self._parameters: Dict[str, HonParameter] = {} | ||||||
|         self._data: Dict[str, Any] = {} |         self._data: Dict[str, Any] = {} | ||||||
|  |         self._available_settings: Dict[str, HonParameter] = {} | ||||||
|         self._rules: List[HonRuleSet] = [] |         self._rules: List[HonRuleSet] = [] | ||||||
|         attributes.pop("description", "") |  | ||||||
|         attributes.pop("protocolType", "") |  | ||||||
|         self._load_parameters(attributes) |         self._load_parameters(attributes) | ||||||
|  |  | ||||||
|     def __repr__(self) -> str: |     def __repr__(self) -> str: | ||||||
| @ -48,8 +49,6 @@ class HonCommand: | |||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def api(self) -> "HonAPI": |     def api(self) -> "HonAPI": | ||||||
|         if self._api is None and self._appliance: |  | ||||||
|             self._api = self._appliance.api |  | ||||||
|         if self._api is None: |         if self._api is None: | ||||||
|             raise exceptions.NoAuthenticationException("Missing hOn login") |             raise exceptions.NoAuthenticationException("Missing hOn login") | ||||||
|         return self._api |         return self._api | ||||||
| @ -77,14 +76,6 @@ class HonCommand: | |||||||
|             result.setdefault(parameter.group, {})[name] = parameter.intern_value |             result.setdefault(parameter.group, {})[name] = parameter.intern_value | ||||||
|         return result |         return result | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def mandatory_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(): |  | ||||||
|             if parameter.mandatory: |  | ||||||
|                 result.setdefault(parameter.group, {})[name] = parameter.intern_value |  | ||||||
|         return result |  | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def parameter_value(self) -> Dict[str, Union[str, float]]: |     def parameter_value(self) -> Dict[str, Union[str, float]]: | ||||||
|         return {n: p.value for n, p in self._parameters.items()} |         return {n: p.value for n, p in self._parameters.items()} | ||||||
| @ -102,12 +93,10 @@ class HonCommand: | |||||||
|         if name == "zoneMap" and self._appliance.zone: |         if name == "zoneMap" and self._appliance.zone: | ||||||
|             data["default"] = self._appliance.zone |             data["default"] = self._appliance.zone | ||||||
|         if data.get("category") == "rule": |         if data.get("category") == "rule": | ||||||
|             if "fixedValue" in data: |             if "fixedValue" not in data: | ||||||
|                 self._rules.append(HonRuleSet(self, data["fixedValue"])) |                 _LOGGER.error("Rule not supported: %s", data) | ||||||
|             elif "enumValues" in data: |  | ||||||
|                 self._rules.append(HonRuleSet(self, data["enumValues"])) |  | ||||||
|             else: |             else: | ||||||
|                 _LOGGER.warning("Rule not supported: %s", data) |                 self._rules.append(HonRuleSet(self, data["fixedValue"])) | ||||||
|         match data.get("typology"): |         match data.get("typology"): | ||||||
|             case "range": |             case "range": | ||||||
|                 self._parameters[name] = HonParameterRange(name, data, parameter) |                 self._parameters[name] = HonParameterRange(name, data, parameter) | ||||||
| @ -122,21 +111,8 @@ class HonCommand: | |||||||
|             name = "program" if "PROGRAM" in self._category_name else "category" |             name = "program" if "PROGRAM" in self._category_name else "category" | ||||||
|             self._parameters[name] = HonParameterProgram(name, self, "custom") |             self._parameters[name] = HonParameterProgram(name, self, "custom") | ||||||
|  |  | ||||||
|     async def send(self, only_mandatory: bool = False) -> bool: |     async def send(self) -> bool: | ||||||
|         grouped_params = ( |         params = self.parameter_groups.get("parameters", {}) | ||||||
|             self.mandatory_parameter_groups if only_mandatory else self.parameter_groups |  | ||||||
|         ) |  | ||||||
|         params = grouped_params.get("parameters", {}) |  | ||||||
|         return await self.send_parameters(params) |  | ||||||
|  |  | ||||||
|     async def send_specific(self, param_names: List[str]) -> bool: |  | ||||||
|         params: Dict[str, str | float] = {} |  | ||||||
|         for key, parameter in self._parameters.items(): |  | ||||||
|             if key in param_names: |  | ||||||
|                 params[key] = parameter.value |  | ||||||
|         return await self.send_parameters(params) |  | ||||||
|  |  | ||||||
|     async def send_parameters(self, params: Dict[str, str | float]) -> bool: |  | ||||||
|         ancillary_params = self.parameter_groups.get("ancillaryParameters", {}) |         ancillary_params = self.parameter_groups.get("ancillaryParameters", {}) | ||||||
|         ancillary_params.pop("programRules", None) |         ancillary_params.pop("programRules", None) | ||||||
|         self.appliance.sync_command_to_params(self.name) |         self.appliance.sync_command_to_params(self.name) | ||||||
| @ -193,7 +169,3 @@ class HonCommand: | |||||||
|                 else: |                 else: | ||||||
|                     result[name] = parameter |                     result[name] = parameter | ||||||
|         return result |         return result | ||||||
|  |  | ||||||
|     def reset(self) -> None: |  | ||||||
|         for parameter in self._parameters.values(): |  | ||||||
|             parameter.reset() |  | ||||||
|  | |||||||
| @ -278,12 +278,10 @@ class TestAPI(HonAPI): | |||||||
|     async def load_appliances(self) -> List[Dict[str, Any]]: |     async def load_appliances(self) -> List[Dict[str, Any]]: | ||||||
|         result = [] |         result = [] | ||||||
|         for appliance in self._path.glob("*/"): |         for appliance in self._path.glob("*/"): | ||||||
|             file = appliance / "appliance_data.json" |             with open( | ||||||
|             with open(file, "r", encoding="utf-8") as json_file: |                 appliance / "appliance_data.json", "r", encoding="utf-8" | ||||||
|                 try: |             ) as json_file: | ||||||
|                     result.append(json.loads(json_file.read())) |                 result.append(json.loads(json_file.read())) | ||||||
|                 except json.decoder.JSONDecodeError as error: |  | ||||||
|                     _LOGGER.error("%s - %s", str(file), error) |  | ||||||
|         return result |         return result | ||||||
|  |  | ||||||
|     async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]: |     async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]: | ||||||
| @ -320,5 +318,4 @@ class TestAPI(HonAPI): | |||||||
|         parameters: Dict[str, Any], |         parameters: Dict[str, Any], | ||||||
|         ancillary_parameters: Dict[str, Any], |         ancillary_parameters: Dict[str, Any], | ||||||
|     ) -> bool: |     ) -> bool: | ||||||
|         _LOGGER.info("%s - %s", str(parameters), str(ancillary_parameters)) |  | ||||||
|         return True |         return True | ||||||
|  | |||||||
| @ -30,14 +30,6 @@ class HonLoginData: | |||||||
|     loaded: Optional[Dict[str, Any]] = None |     loaded: Optional[Dict[str, Any]] = None | ||||||
|  |  | ||||||
|  |  | ||||||
| @dataclass |  | ||||||
| class HonAuthData: |  | ||||||
|     access_token: str = "" |  | ||||||
|     refresh_token: str = "" |  | ||||||
|     cognito_token: str = "" |  | ||||||
|     id_token: str = "" |  | ||||||
|  |  | ||||||
|  |  | ||||||
| class HonAuth: | class HonAuth: | ||||||
|     _TOKEN_EXPIRES_AFTER_HOURS = 8 |     _TOKEN_EXPIRES_AFTER_HOURS = 8 | ||||||
|     _TOKEN_EXPIRE_WARNING_HOURS = 7 |     _TOKEN_EXPIRE_WARNING_HOURS = 7 | ||||||
| @ -54,25 +46,28 @@ class HonAuth: | |||||||
|         self._login_data = HonLoginData() |         self._login_data = HonLoginData() | ||||||
|         self._login_data.email = email |         self._login_data.email = email | ||||||
|         self._login_data.password = password |         self._login_data.password = password | ||||||
|  |         self._access_token = "" | ||||||
|  |         self._refresh_token = "" | ||||||
|  |         self._cognito_token = "" | ||||||
|  |         self._id_token = "" | ||||||
|         self._device = device |         self._device = device | ||||||
|         self._expires: datetime = datetime.utcnow() |         self._expires: datetime = datetime.utcnow() | ||||||
|         self._auth = HonAuthData() |  | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def cognito_token(self) -> str: |     def cognito_token(self) -> str: | ||||||
|         return self._auth.cognito_token |         return self._cognito_token | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def id_token(self) -> str: |     def id_token(self) -> str: | ||||||
|         return self._auth.id_token |         return self._id_token | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def access_token(self) -> str: |     def access_token(self) -> str: | ||||||
|         return self._auth.access_token |         return self._access_token | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def refresh_token(self) -> str: |     def refresh_token(self) -> str: | ||||||
|         return self._auth.refresh_token |         return self._refresh_token | ||||||
|  |  | ||||||
|     def _check_token_expiration(self, hours: int) -> bool: |     def _check_token_expiration(self, hours: int) -> bool: | ||||||
|         return datetime.utcnow() >= self._expires + timedelta(hours=hours) |         return datetime.utcnow() >= self._expires + timedelta(hours=hours) | ||||||
| @ -197,12 +192,12 @@ class HonAuth: | |||||||
|  |  | ||||||
|     def _parse_token_data(self, text: str) -> bool: |     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._auth.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._auth.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._auth.id_token = id_token[0] |             self._id_token = id_token[0] | ||||||
|         return bool(access_token and refresh_token and id_token) |         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: | ||||||
| @ -234,7 +229,7 @@ class HonAuth: | |||||||
|         return True |         return True | ||||||
|  |  | ||||||
|     async def _api_auth(self) -> bool: |     async def _api_auth(self) -> bool: | ||||||
|         post_headers = {"id-token": self._auth.id_token} |         post_headers = {"id-token": self._id_token} | ||||||
|         data = self._device.get() |         data = self._device.get() | ||||||
|         async with self._request.post( |         async with self._request.post( | ||||||
|             f"{const.API_URL}/auth/v1/login", headers=post_headers, json=data |             f"{const.API_URL}/auth/v1/login", headers=post_headers, json=data | ||||||
| @ -244,8 +239,8 @@ class HonAuth: | |||||||
|             except json.JSONDecodeError: |             except json.JSONDecodeError: | ||||||
|                 await self._error_logger(response) |                 await self._error_logger(response) | ||||||
|                 return False |                 return False | ||||||
|             self._auth.cognito_token = json_data.get("cognitoUser", {}).get("Token", "") |             self._cognito_token = json_data.get("cognitoUser", {}).get("Token", "") | ||||||
|             if not self._auth.cognito_token: |             if not self._cognito_token: | ||||||
|                 _LOGGER.error(json_data) |                 _LOGGER.error(json_data) | ||||||
|                 raise exceptions.HonAuthenticationError() |                 raise exceptions.HonAuthenticationError() | ||||||
|         return True |         return True | ||||||
| @ -267,7 +262,7 @@ class HonAuth: | |||||||
|     async def refresh(self) -> bool: |     async def refresh(self) -> bool: | ||||||
|         params = { |         params = { | ||||||
|             "client_id": const.CLIENT_ID, |             "client_id": const.CLIENT_ID, | ||||||
|             "refresh_token": self._auth.refresh_token, |             "refresh_token": self._refresh_token, | ||||||
|             "grant_type": "refresh_token", |             "grant_type": "refresh_token", | ||||||
|         } |         } | ||||||
|         async with self._request.post( |         async with self._request.post( | ||||||
| @ -278,14 +273,14 @@ class HonAuth: | |||||||
|                 return False |                 return False | ||||||
|             data = await response.json() |             data = await response.json() | ||||||
|         self._expires = datetime.utcnow() |         self._expires = datetime.utcnow() | ||||||
|         self._auth.id_token = data["id_token"] |         self._id_token = data["id_token"] | ||||||
|         self._auth.access_token = data["access_token"] |         self._access_token = data["access_token"] | ||||||
|         return await self._api_auth() |         return await self._api_auth() | ||||||
|  |  | ||||||
|     def clear(self) -> None: |     def clear(self) -> None: | ||||||
|         self._session.cookie_jar.clear_domain(const.AUTH_API.split("/")[-2]) |         self._session.cookie_jar.clear_domain(const.AUTH_API.split("/")[-2]) | ||||||
|         self._request.called_urls = [] |         self._request.called_urls = [] | ||||||
|         self._auth.cognito_token = "" |         self._cognito_token = "" | ||||||
|         self._auth.id_token = "" |         self._id_token = "" | ||||||
|         self._auth.access_token = "" |         self._access_token = "" | ||||||
|         self._auth.refresh_token = "" |         self._refresh_token = "" | ||||||
|  | |||||||
| @ -21,7 +21,7 @@ class HonDevice: | |||||||
|         return self._os_version |         return self._os_version | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def os_type(self) -> str: |     def os(self) -> str: | ||||||
|         return self._os |         return self._os | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
| @ -36,7 +36,7 @@ class HonDevice: | |||||||
|         result: Dict[str, str | int] = { |         result: Dict[str, str | int] = { | ||||||
|             "appVersion": self.app_version, |             "appVersion": self.app_version, | ||||||
|             "mobileId": self.mobile_id, |             "mobileId": self.mobile_id, | ||||||
|             "os": self.os_type, |             "os": self.os, | ||||||
|             "osVersion": self.os_version, |             "osVersion": self.os_version, | ||||||
|             "deviceModel": self.device_model, |             "deviceModel": self.device_model, | ||||||
|         } |         } | ||||||
|  | |||||||
| @ -2,7 +2,7 @@ import logging | |||||||
| from collections.abc import AsyncIterator | from collections.abc import AsyncIterator | ||||||
| from contextlib import asynccontextmanager | from contextlib import asynccontextmanager | ||||||
| from types import TracebackType | from types import TracebackType | ||||||
| from typing import Optional, Dict, Type, Any | from typing import Optional, Dict, Type, Any, Protocol | ||||||
|  |  | ||||||
| import aiohttp | import aiohttp | ||||||
| from typing_extensions import Self | from typing_extensions import Self | ||||||
| @ -47,11 +47,10 @@ class ConnectionHandler: | |||||||
|         return self |         return self | ||||||
|  |  | ||||||
|     @asynccontextmanager |     @asynccontextmanager | ||||||
|     async def _intercept( |     def _intercept( | ||||||
|         self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any] |         self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any] | ||||||
|     ) -> AsyncIterator[aiohttp.ClientResponse]: |     ) -> AsyncIterator[aiohttp.ClientResponse]: | ||||||
|         async with method(url, *args, **kwargs) as response: |         raise NotImplementedError | ||||||
|             yield response |  | ||||||
|  |  | ||||||
|     @asynccontextmanager |     @asynccontextmanager | ||||||
|     async def get( |     async def get( | ||||||
| @ -60,8 +59,7 @@ class ConnectionHandler: | |||||||
|         if self._session is None: |         if self._session is None: | ||||||
|             raise exceptions.NoSessionException() |             raise exceptions.NoSessionException() | ||||||
|         response: aiohttp.ClientResponse |         response: aiohttp.ClientResponse | ||||||
|         args = self._session.get, *args |         async with self._intercept(self._session.get, *args, **kwargs) as response:  # type: ignore[arg-type] | ||||||
|         async with self._intercept(*args, **kwargs) as response: |  | ||||||
|             yield response |             yield response | ||||||
|  |  | ||||||
|     @asynccontextmanager |     @asynccontextmanager | ||||||
| @ -71,8 +69,7 @@ class ConnectionHandler: | |||||||
|         if self._session is None: |         if self._session is None: | ||||||
|             raise exceptions.NoSessionException() |             raise exceptions.NoSessionException() | ||||||
|         response: aiohttp.ClientResponse |         response: aiohttp.ClientResponse | ||||||
|         args = self._session.post, *args |         async with self._intercept(self._session.post, *args, **kwargs) as response:  # type: ignore[arg-type] | ||||||
|         async with self._intercept(*args, **kwargs) as response: |  | ||||||
|             yield response |             yield response | ||||||
|  |  | ||||||
|     async def close(self) -> None: |     async def close(self) -> None: | ||||||
|  | |||||||
| @ -95,11 +95,11 @@ class HonConnectionHandler(ConnectionHandler): | |||||||
|                 try: |                 try: | ||||||
|                     await response.json() |                     await response.json() | ||||||
|                     yield response |                     yield response | ||||||
|                 except json.JSONDecodeError as exc: |                 except json.JSONDecodeError: | ||||||
|                     _LOGGER.warning( |                     _LOGGER.warning( | ||||||
|                         "%s - JsonDecodeError %s - %s", |                         "%s - JsonDecodeError %s - %s", | ||||||
|                         response.request_info.url, |                         response.request_info.url, | ||||||
|                         response.status, |                         response.status, | ||||||
|                         await response.text(), |                         await response.text(), | ||||||
|                     ) |                     ) | ||||||
|                     raise HonAuthenticationError("Decode Error") from exc |                     raise HonAuthenticationError("Decode Error") | ||||||
|  | |||||||
| @ -2,10 +2,8 @@ AUTH_API = "https://account2.hon-smarthome.com" | |||||||
| API_URL = "https://api-iot.he.services" | API_URL = "https://api-iot.he.services" | ||||||
| API_KEY = "GRCqFhC6Gk@ikWXm1RmnSmX1cm,MxY-configuration" | API_KEY = "GRCqFhC6Gk@ikWXm1RmnSmX1cm,MxY-configuration" | ||||||
| APP = "hon" | APP = "hon" | ||||||
| CLIENT_ID = ( | # All seen id's (different accounts, different devices) are the same, so I guess this hash is static | ||||||
|     "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9." | CLIENT_ID = "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9.HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6" | ||||||
|     "HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6" |  | ||||||
| ) |  | ||||||
| APP_VERSION = "2.1.2" | APP_VERSION = "2.1.2" | ||||||
| OS_VERSION = 31 | OS_VERSION = 31 | ||||||
| OS = "android" | OS = "android" | ||||||
|  | |||||||
| @ -7,10 +7,9 @@ from typing import List, Optional, Dict, Any, Type | |||||||
| from aiohttp import ClientSession | from aiohttp import ClientSession | ||||||
| from typing_extensions import Self | from typing_extensions import Self | ||||||
|  |  | ||||||
|  | from pyhon import HonAPI, exceptions | ||||||
| from pyhon.appliance import HonAppliance | from pyhon.appliance import HonAppliance | ||||||
| from pyhon.connection.api import HonAPI |  | ||||||
| from pyhon.connection.api import TestAPI | from pyhon.connection.api import TestAPI | ||||||
| from pyhon.exceptions import NoAuthenticationException |  | ||||||
|  |  | ||||||
| _LOGGER = logging.getLogger(__name__) | _LOGGER = logging.getLogger(__name__) | ||||||
|  |  | ||||||
| @ -44,7 +43,7 @@ class Hon: | |||||||
|     @property |     @property | ||||||
|     def api(self) -> HonAPI: |     def api(self) -> HonAPI: | ||||||
|         if self._api is None: |         if self._api is None: | ||||||
|             raise NoAuthenticationException |             raise exceptions.NoAuthenticationException | ||||||
|         return self._api |         return self._api | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|  | |||||||
| @ -7,21 +7,14 @@ if TYPE_CHECKING: | |||||||
| class HonParameter: | class HonParameter: | ||||||
|     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|         self._key = key |         self._key = key | ||||||
|         self._attributes = attributes |         self._category: str = attributes.get("category", "") | ||||||
|         self._category: str = "" |         self._typology: str = attributes.get("typology", "") | ||||||
|         self._typology: str = "" |         self._mandatory: int = attributes.get("mandatory", 0) | ||||||
|         self._mandatory: int = 0 |  | ||||||
|         self._value: str | float = "" |         self._value: str | float = "" | ||||||
|         self._group: str = group |         self._group: str = group | ||||||
|         self._triggers: Dict[ |         self._triggers: Dict[ | ||||||
|             str, List[Tuple[Callable[["HonRule"], None], "HonRule"]] |             str, List[Tuple[Callable[["HonRule"], None], "HonRule"]] | ||||||
|         ] = {} |         ] = {} | ||||||
|         self._set_attributes() |  | ||||||
|  |  | ||||||
|     def _set_attributes(self) -> None: |  | ||||||
|         self._category = self._attributes.get("category", "") |  | ||||||
|         self._typology = self._attributes.get("typology", "") |  | ||||||
|         self._mandatory = self._attributes.get("mandatory", 0) |  | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def key(self) -> str: |     def key(self) -> str: | ||||||
| @ -68,9 +61,8 @@ class HonParameter: | |||||||
|         self._triggers.setdefault(value, []).append((func, data)) |         self._triggers.setdefault(value, []).append((func, data)) | ||||||
|  |  | ||||||
|     def check_trigger(self, value: str | float) -> None: |     def check_trigger(self, value: str | float) -> None: | ||||||
|         triggers = {str(k).lower(): v for k, v in self._triggers.items()} |         if str(value) in self._triggers: | ||||||
|         if str(value).lower() in triggers: |             for trigger in self._triggers[str(value)]: | ||||||
|             for trigger in triggers[str(value)]: |  | ||||||
|                 func, args = trigger |                 func, args = trigger | ||||||
|                 func(args) |                 func(args) | ||||||
|  |  | ||||||
| @ -93,6 +85,3 @@ class HonParameter: | |||||||
|                     param[rule.param_key] = rule.param_data.get("defaultValue", "") |                     param[rule.param_key] = rule.param_data.get("defaultValue", "") | ||||||
|  |  | ||||||
|         return result |         return result | ||||||
|  |  | ||||||
|     def reset(self) -> None: |  | ||||||
|         self._set_attributes() |  | ||||||
|  | |||||||
| @ -10,19 +10,12 @@ def clean_value(value: str | float) -> str: | |||||||
| class HonParameterEnum(HonParameter): | class HonParameterEnum(HonParameter): | ||||||
|     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|         super().__init__(key, attributes, group) |         super().__init__(key, attributes, group) | ||||||
|         self._default: str | float = "" |         self._default = attributes.get("defaultValue") | ||||||
|         self._value: str | float = "" |         self._value = self._default or "0" | ||||||
|         self._values: List[str] = [] |         self._values: List[str] = attributes.get("enumValues", []) | ||||||
|         self._set_attributes() |  | ||||||
|         if self._default and clean_value(self._default.strip("[]")) not in self.values: |         if self._default and clean_value(self._default.strip("[]")) not in self.values: | ||||||
|             self._values.append(self._default) |             self._values.append(self._default) | ||||||
|  |  | ||||||
|     def _set_attributes(self) -> None: |  | ||||||
|         super()._set_attributes() |  | ||||||
|         self._default = self._attributes.get("defaultValue", "") |  | ||||||
|         self._value = self._default or "0" |  | ||||||
|         self._values = self._attributes.get("enumValues", []) |  | ||||||
|  |  | ||||||
|     def __repr__(self) -> str: |     def __repr__(self) -> str: | ||||||
|         return f"{self.__class__} (<{self.key}> {self.values})" |         return f"{self.__class__} (<{self.key}> {self.values})" | ||||||
|  |  | ||||||
|  | |||||||
| @ -6,12 +6,7 @@ from pyhon.parameter.base import HonParameter | |||||||
| class HonParameterFixed(HonParameter): | class HonParameterFixed(HonParameter): | ||||||
|     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|         super().__init__(key, attributes, group) |         super().__init__(key, attributes, group) | ||||||
|         self._value: str | float = "" |         self._value = attributes.get("fixedValue", None) | ||||||
|         self._set_attributes() |  | ||||||
|  |  | ||||||
|     def _set_attributes(self) -> None: |  | ||||||
|         super()._set_attributes() |  | ||||||
|         self._value = self._attributes.get("fixedValue", "") |  | ||||||
|  |  | ||||||
|     def __repr__(self) -> str: |     def __repr__(self) -> str: | ||||||
|         return f"{self.__class__} (<{self.key}> fixed)" |         return f"{self.__class__} (<{self.key}> fixed)" | ||||||
|  | |||||||
| @ -37,19 +37,17 @@ class HonParameterProgram(HonParameterEnum): | |||||||
|  |  | ||||||
|     @values.setter |     @values.setter | ||||||
|     def values(self, values: List[str]) -> None: |     def values(self, values: List[str]) -> None: | ||||||
|         raise ValueError("Cant set values {values}") |         return | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def ids(self) -> Dict[int, str]: |     def ids(self) -> Dict[int, str]: | ||||||
|         values: Dict[int, str] = {} |         values = { | ||||||
|         for name, parameter in self._programs.items(): |             int(p.parameters["prCode"].value): n | ||||||
|             if "iot_" in name: |             for i, (n, p) in enumerate(self._programs.items()) | ||||||
|                 continue |             if "iot_" not in n | ||||||
|             if parameter.parameters.get("prCode"): |             and p.parameters.get("prCode") | ||||||
|                 continue |             and not ((fav := p.parameters.get("favourite")) and fav.value == "1") | ||||||
|             if (fav := parameter.parameters.get("favourite")) and fav.value == "1": |         } | ||||||
|                 continue |  | ||||||
|             values[int(parameter.parameters["prCode"].value)] = name |  | ||||||
|         return dict(sorted(values.items())) |         return dict(sorted(values.items())) | ||||||
|  |  | ||||||
|     def set_value(self, value: str) -> None: |     def set_value(self, value: str) -> None: | ||||||
|  | |||||||
| @ -7,20 +7,11 @@ from pyhon.parameter.base import HonParameter | |||||||
| class HonParameterRange(HonParameter): | class HonParameterRange(HonParameter): | ||||||
|     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: |     def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: | ||||||
|         super().__init__(key, attributes, group) |         super().__init__(key, attributes, group) | ||||||
|         self._min: float = 0 |         self._min: float = str_to_float(attributes["minimumValue"]) | ||||||
|         self._max: float = 0 |         self._max: float = str_to_float(attributes["maximumValue"]) | ||||||
|         self._step: float = 0 |         self._step: float = str_to_float(attributes["incrementValue"]) | ||||||
|         self._default: float = 0 |         self._default: float = str_to_float(attributes.get("defaultValue", self.min)) | ||||||
|         self._value: float = 0 |         self._value: float = self._default | ||||||
|         self._set_attributes() |  | ||||||
|  |  | ||||||
|     def _set_attributes(self) -> None: |  | ||||||
|         super()._set_attributes() |  | ||||||
|         self._min = str_to_float(self._attributes.get("minimumValue", 0)) |  | ||||||
|         self._max = str_to_float(self._attributes.get("maximumValue", 0)) |  | ||||||
|         self._step = str_to_float(self._attributes.get("incrementValue", 0)) |  | ||||||
|         self._default = str_to_float(self._attributes.get("defaultValue", self.min)) |  | ||||||
|         self._value = self._default |  | ||||||
|  |  | ||||||
|     def __repr__(self) -> str: |     def __repr__(self) -> str: | ||||||
|         return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])" |         return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])" | ||||||
| @ -64,8 +55,9 @@ class HonParameterRange(HonParameter): | |||||||
|             self._value = value |             self._value = value | ||||||
|             self.check_trigger(value) |             self.check_trigger(value) | ||||||
|         else: |         else: | ||||||
|             allowed = f"min {self.min} max {self.max} step {self.step}" |             raise ValueError( | ||||||
|             raise ValueError(f"Allowed: {allowed} But was: {value}") |                 f"Allowed: min {self.min} max {self.max} step {self.step} But was: {value}" | ||||||
|  |             ) | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def values(self) -> List[str]: |     def values(self) -> List[str]: | ||||||
|  | |||||||
| @ -29,26 +29,33 @@ def pretty_print( | |||||||
|     whitespace: str = "  ", |     whitespace: str = "  ", | ||||||
| ) -> str: | ) -> str: | ||||||
|     result = "" |     result = "" | ||||||
|     space = whitespace * intend |  | ||||||
|     if isinstance(data, (dict, list)) and key: |  | ||||||
|         result += f"{space}{'- ' if is_list else ''}{key}:\n" |  | ||||||
|         intend += 1 |  | ||||||
|     if isinstance(data, list): |     if isinstance(data, list): | ||||||
|  |         if key: | ||||||
|  |             result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n" | ||||||
|  |             intend += 1 | ||||||
|         for i, value in enumerate(data): |         for i, value in enumerate(data): | ||||||
|             result += pretty_print( |             result += pretty_print( | ||||||
|                 value, intend=intend, is_list=True, whitespace=whitespace |                 value, intend=intend, is_list=True, whitespace=whitespace | ||||||
|             ) |             ) | ||||||
|     elif isinstance(data, dict): |     elif isinstance(data, dict): | ||||||
|         for i, (list_key, value) in enumerate(sorted(data.items())): |         if key: | ||||||
|             result += pretty_print( |             result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n" | ||||||
|                 value, |             intend += 1 | ||||||
|                 key=list_key, |         for i, (key, value) in enumerate(sorted(data.items())): | ||||||
|                 intend=intend + (is_list if i else 0), |             if is_list and not i: | ||||||
|                 is_list=is_list and not i, |                 result += pretty_print( | ||||||
|                 whitespace=whitespace, |                     value, key=key, intend=intend, is_list=True, whitespace=whitespace | ||||||
|             ) |                 ) | ||||||
|  |             elif is_list: | ||||||
|  |                 result += pretty_print( | ||||||
|  |                     value, key=key, intend=intend + 1, whitespace=whitespace | ||||||
|  |                 ) | ||||||
|  |             else: | ||||||
|  |                 result += pretty_print( | ||||||
|  |                     value, key=key, intend=intend, whitespace=whitespace | ||||||
|  |                 ) | ||||||
|     else: |     else: | ||||||
|         result += f"{space}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n" |         result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n" | ||||||
|     return result |     return result | ||||||
|  |  | ||||||
|  |  | ||||||
|  | |||||||
| @ -3,7 +3,6 @@ from typing import List, Dict, TYPE_CHECKING, Any, Optional | |||||||
|  |  | ||||||
| from pyhon.parameter.enum import HonParameterEnum | from pyhon.parameter.enum import HonParameterEnum | ||||||
| from pyhon.parameter.range import HonParameterRange | from pyhon.parameter.range import HonParameterRange | ||||||
| from pyhon.typedefs import Parameter |  | ||||||
|  |  | ||||||
| if TYPE_CHECKING: | if TYPE_CHECKING: | ||||||
|     from pyhon.commands import HonCommand |     from pyhon.commands import HonCommand | ||||||
| @ -25,10 +24,6 @@ class HonRuleSet: | |||||||
|         self._rules: Dict[str, List[HonRule]] = {} |         self._rules: Dict[str, List[HonRule]] = {} | ||||||
|         self._parse_rule(rule) |         self._parse_rule(rule) | ||||||
|  |  | ||||||
|     @property |  | ||||||
|     def rules(self) -> Dict[str, List[HonRule]]: |  | ||||||
|         return self._rules |  | ||||||
|  |  | ||||||
|     def _parse_rule(self, rule: Dict[str, Any]) -> None: |     def _parse_rule(self, rule: Dict[str, Any]) -> None: | ||||||
|         for param_key, params in rule.items(): |         for param_key, params in rule.items(): | ||||||
|             param_key = self._command.appliance.options.get(param_key, param_key) |             param_key = self._command.appliance.options.get(param_key, param_key) | ||||||
| @ -56,11 +51,6 @@ class HonRuleSet: | |||||||
|                     extra[trigger_key] = trigger_value |                     extra[trigger_key] = trigger_value | ||||||
|                     for extra_key, extra_data in param_data.items(): |                     for extra_key, extra_data in param_data.items(): | ||||||
|                         self._parse_conditions(param_key, extra_key, extra_data, extra) |                         self._parse_conditions(param_key, extra_key, extra_data, extra) | ||||||
|                 else: |  | ||||||
|                     param_data = {"typology": "fixed", "fixedValue": param_data} |  | ||||||
|                     self._create_rule( |  | ||||||
|                         param_key, trigger_key, trigger_value, param_data, extra |  | ||||||
|                     ) |  | ||||||
|  |  | ||||||
|     def _create_rule( |     def _create_rule( | ||||||
|         self, |         self, | ||||||
| @ -93,46 +83,28 @@ class HonRuleSet: | |||||||
|             for rule in rules: |             for rule in rules: | ||||||
|                 self._rules.setdefault(key, []).append(rule) |                 self._rules.setdefault(key, []).append(rule) | ||||||
|  |  | ||||||
|     def _extra_rules_matches(self, rule: HonRule) -> bool: |  | ||||||
|         if rule.extras: |  | ||||||
|             for key, value in rule.extras.items(): |  | ||||||
|                 if not self._command.parameters.get(key): |  | ||||||
|                     return False |  | ||||||
|                 if str(self._command.parameters.get(key)) != str(value): |  | ||||||
|                     return False |  | ||||||
|         return True |  | ||||||
|  |  | ||||||
|     def _apply_fixed(self, param: Parameter, value: str | float) -> None: |  | ||||||
|         if isinstance(param, HonParameterEnum) and set(param.values) != {str(value)}: |  | ||||||
|             param.values = [str(value)] |  | ||||||
|             param.value = str(value) |  | ||||||
|         elif isinstance(param, HonParameterRange): |  | ||||||
|             if float(value) < param.min: |  | ||||||
|                 param.min = float(value) |  | ||||||
|             elif float(value) > param.max: |  | ||||||
|                 param.max = float(value) |  | ||||||
|             param.value = float(value) |  | ||||||
|             return |  | ||||||
|         param.value = str(value) |  | ||||||
|  |  | ||||||
|     def _apply_enum(self, param: Parameter, rule: HonRule) -> None: |  | ||||||
|         if not isinstance(param, HonParameterEnum): |  | ||||||
|             return |  | ||||||
|         if enum_values := rule.param_data.get("enumValues"): |  | ||||||
|             param.values = enum_values.split("|") |  | ||||||
|         if default_value := rule.param_data.get("defaultValue"): |  | ||||||
|             param.value = default_value |  | ||||||
|  |  | ||||||
|     def _add_trigger(self, parameter: "HonParameter", data: HonRule) -> None: |     def _add_trigger(self, parameter: "HonParameter", data: HonRule) -> None: | ||||||
|         def apply(rule: HonRule) -> None: |         def apply(rule: HonRule) -> None: | ||||||
|             if not self._extra_rules_matches(rule): |             if rule.extras is not None: | ||||||
|                 return |                 for key, value in rule.extras.items(): | ||||||
|             if not (param := self._command.parameters.get(rule.param_key)): |                     if str(self._command.parameters.get(key)) != str(value): | ||||||
|                 return |                         return | ||||||
|             if fixed_value := rule.param_data.get("fixedValue", ""): |             if param := self._command.parameters.get(rule.param_key): | ||||||
|                 self._apply_fixed(param, fixed_value) |                 if value := rule.param_data.get("fixedValue", ""): | ||||||
|             elif rule.param_data.get("typology") == "enum": |                     if isinstance(param, HonParameterEnum) and set(param.values) != { | ||||||
|                 self._apply_enum(param, rule) |                         str(value) | ||||||
|  |                     }: | ||||||
|  |                         param.values = [str(value)] | ||||||
|  |                     elif isinstance(param, HonParameterRange): | ||||||
|  |                         param.value = float(value) | ||||||
|  |                         return | ||||||
|  |                     param.value = str(value) | ||||||
|  |                 elif rule.param_data.get("typology") == "enum": | ||||||
|  |                     if isinstance(param, HonParameterEnum): | ||||||
|  |                         if enum_values := rule.param_data.get("enumValues"): | ||||||
|  |                             param.values = enum_values.split("|") | ||||||
|  |                         if default_value := rule.param_data.get("defaultValue"): | ||||||
|  |                             param.value = default_value | ||||||
|  |  | ||||||
|         parameter.add_trigger(data.trigger_value, apply, data) |         parameter.add_trigger(data.trigger_value, apply, data) | ||||||
|  |  | ||||||
|  | |||||||
| @ -11,7 +11,7 @@ if TYPE_CHECKING: | |||||||
|     from pyhon.parameter.range import HonParameterRange |     from pyhon.parameter.range import HonParameterRange | ||||||
|  |  | ||||||
|  |  | ||||||
| class Callback(Protocol):  # pylint: disable=too-few-public-methods | class Callback(Protocol): | ||||||
|     def __call__( |     def __call__( | ||||||
|         self, url: str | URL, *args: Any, **kwargs: Any |         self, url: str | URL, *args: Any, **kwargs: Any | ||||||
|     ) -> aiohttp.client._RequestContextManager: |     ) -> aiohttp.client._RequestContextManager: | ||||||
|  | |||||||
| @ -1,3 +1,3 @@ | |||||||
| aiohttp~=3.8 | aiohttp~=3.8.5 | ||||||
| yarl~=1.9 | yarl~=1.9.2 | ||||||
| typing-extensions~=4.7 | typing-extensions~=4.7.1 | ||||||
|  | |||||||
| @ -1,4 +1,4 @@ | |||||||
| black==23.7.0 | black==23.3.0 | ||||||
| flake8==6.0.0 | flake8==6.0.0 | ||||||
| mypy==1.4.1 | mypy==1.2.0 | ||||||
| pylint==2.17.4 | pylint==2.17.2 | ||||||
|  | |||||||
							
								
								
									
										6
									
								
								setup.py
									
									
									
									
									
								
							
							
						
						
									
										6
									
								
								setup.py
									
									
									
									
									
								
							| @ -2,12 +2,12 @@ | |||||||
|  |  | ||||||
| from setuptools import setup, find_packages | from setuptools import setup, find_packages | ||||||
|  |  | ||||||
| with open("README.md", "r", encoding="utf-8") as f: | with open("README.md", "r") as f: | ||||||
|     long_description = f.read() |     long_description = f.read() | ||||||
|  |  | ||||||
| setup( | setup( | ||||||
|     name="pyhOn", |     name="pyhOn", | ||||||
|     version="0.15.8", |     version="0.14.12", | ||||||
|     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, | ||||||
| @ -21,7 +21,7 @@ setup( | |||||||
|     packages=find_packages(), |     packages=find_packages(), | ||||||
|     include_package_data=True, |     include_package_data=True, | ||||||
|     python_requires=">=3.10", |     python_requires=">=3.10", | ||||||
|     install_requires=["aiohttp~=3.8", "typing-extensions~=4.7", "yarl~=1.9"], |     install_requires=["aiohttp~=3.8.5", "typing-extensions~=4.7.1", "yarl~=1.9.2"], | ||||||
|     classifiers=[ |     classifiers=[ | ||||||
|         "Development Status :: 4 - Beta", |         "Development Status :: 4 - Beta", | ||||||
|         "Environment :: Console", |         "Environment :: Console", | ||||||
|  | |||||||
		Reference in New Issue
	
	Block a user
	