Compare commits
	
		
			42 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| cf137dd176 | |||
| ab18e45f97 | |||
| e44b9b6773 | |||
| 5c650d722b | |||
| 6ae40481e3 | |||
| ff8ae160bb | |||
| 658e80a8f4 | |||
| ffba85bf0d | |||
| 10c8d961c4 | |||
| 61dd470588 | |||
| 1ed81c2a77 | |||
| e4dc3cb1d0 | |||
| 2523069ce9 | |||
| eeb458cb1b | |||
| 2764700bc7 | |||
| e6c796e822 | |||
| 454f2d8916 | |||
| 59ca8b6caf | |||
| 44c55c681d | |||
| cfee10df5f | |||
| e0774677eb | |||
| fc60d15e60 | |||
| 8ef8c0405d | |||
| 5575b240e1 | |||
| 22367242a2 | |||
| 4f7d4860db | |||
| 5a778373b6 | |||
| e1c8bc5835 | |||
| 442e7a07dd | |||
| b0e3b15ff0 | |||
| 2788a3ac91 | |||
| bc7e8994c9 | |||
| 8ca40d7ad0 | |||
| 9a6a07fd46 | |||
| f1818bbc5d | |||
| 3d5c8405ea | |||
| e234ef3bbb | |||
| e00e147ecd | |||
| 26bc35c8a6 | |||
| 17d73cdeb8 | |||
| a10ab4423e | |||
| 0553e6c17d | 
							
								
								
									
										12
									
								
								.github/workflows/python-check.yml
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										12
									
								
								.github/workflows/python-check.yml
									
									
									
									
										vendored
									
									
								
							| @ -13,7 +13,7 @@ jobs: | |||||||
|     strategy: |     strategy: | ||||||
|       fail-fast: false |       fail-fast: false | ||||||
|       matrix: |       matrix: | ||||||
|         python-version: ["3.10", "3.11"] |         python-version: ["3.10", "3.11", "3.12"] | ||||||
|  |  | ||||||
|     steps: |     steps: | ||||||
|     - uses: actions/checkout@v3 |     - uses: actions/checkout@v3 | ||||||
| @ -28,15 +28,13 @@ 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: | | ||||||
|         # stop the build if there are Python syntax errors or undefined names |         flake8 . --count --statistics | ||||||
|         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 --max-line-length 88 $(git ls-files '*.py') |         pylint $(git ls-files '*.py') | ||||||
|     - name: Check black style |     - name: Check black style | ||||||
|       run: | |       run: | | ||||||
|         black . --check |         black . --check | ||||||
|  | |||||||
							
								
								
									
										9
									
								
								.pylintrc
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										9
									
								
								.pylintrc
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,9 @@ | |||||||
|  | [MESSAGES CONTROL] | ||||||
|  |  | ||||||
|  | disable=missing-docstring | ||||||
|  |  | ||||||
|  | [FORMAT] | ||||||
|  |  | ||||||
|  | max-args=6 | ||||||
|  | max-attributes=8 | ||||||
|  | max-line-length=88 | ||||||
							
								
								
									
										1
									
								
								MANIFEST.in
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										1
									
								
								MANIFEST.in
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1 @@ | |||||||
|  | include pyhon/py.typed | ||||||
							
								
								
									
										26
									
								
								mypy.ini
									
									
									
									
									
								
							
							
						
						
									
										26
									
								
								mypy.ini
									
									
									
									
									
								
							| @ -1,4 +1,24 @@ | |||||||
| [mypy] | [mypy] | ||||||
| check_untyped_defs = True | check_untyped_defs = true | ||||||
| disallow_any_generics = True | disallow_any_generics = true | ||||||
| disallow_untyped_defs = True | disallow_any_unimported = true | ||||||
|  | disallow_incomplete_defs = true | ||||||
|  | disallow_subclassing_any = true | ||||||
|  | disallow_untyped_calls = true | ||||||
|  | disallow_untyped_decorators = true | ||||||
|  | disallow_untyped_defs = 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,6 +11,7 @@ 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__) | ||||||
| @ -30,13 +31,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()) | ||||||
|     translate = subparser.add_parser( |     translation = subparser.add_parser( | ||||||
|         "translate", help="print available translation keys" |         "translate", help="print available translation keys" | ||||||
|     ) |     ) | ||||||
|     translate.add_argument( |     translation.add_argument( | ||||||
|         "translate", help="language (de, en, fr...)", metavar="LANGUAGE" |         "translate", help="language (de, en, fr...)", metavar="LANGUAGE" | ||||||
|     ) |     ) | ||||||
|     translate.add_argument("--json", help="print as json", action="store_true") |     translation.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() | ||||||
|     ) |     ) | ||||||
| @ -91,15 +92,13 @@ 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( |                     printer.key_print(getattr(data["attributes"], attr)("parameters")) | ||||||
|                         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( | ||||||
|                         printer.create_command(device.commands, concat=True) |                         printer.create_commands(device.commands, concat=True) | ||||||
|                     ) |                     ) | ||||||
|                 ) |                 ) | ||||||
|             else: |             else: | ||||||
|  | |||||||
| @ -3,13 +3,15 @@ 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 | from typing import Optional, Dict, Any, TYPE_CHECKING, List, TypeVar, overload | ||||||
|  |  | ||||||
| from pyhon import diagnose, exceptions | from pyhon import diagnose, exceptions | ||||||
|  | from pyhon.appliances.base import ApplianceBase | ||||||
| from pyhon.attributes import HonAttribute | from pyhon.attributes import HonAttribute | ||||||
| from pyhon.command_loader import HonCommandLoader | from pyhon.command_loader import HonCommandLoader | ||||||
| from pyhon.commands import HonCommand | from pyhon.commands import HonCommand | ||||||
| from pyhon.parameter.base import HonParameter | from pyhon.parameter.base import HonParameter | ||||||
|  | from pyhon.parameter.enum import HonParameterEnum | ||||||
| from pyhon.parameter.range import HonParameterRange | from pyhon.parameter.range import HonParameterRange | ||||||
| from pyhon.typedefs import Parameter | from pyhon.typedefs import Parameter | ||||||
|  |  | ||||||
| @ -18,7 +20,10 @@ 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 | ||||||
|  |  | ||||||
| @ -40,30 +45,41 @@ class HonAppliance: | |||||||
|         self._default_setting = HonParameter("", {}, "") |         self._default_setting = HonParameter("", {}, "") | ||||||
|  |  | ||||||
|         try: |         try: | ||||||
|             self._extra = importlib.import_module( |             self._extra: Optional[ApplianceBase] = importlib.import_module( | ||||||
|                 f"pyhon.appliances.{self.appliance_type.lower()}" |                 f"pyhon.appliances.{self.appliance_type.lower()}" | ||||||
|             ).Appliance(self) |             ).Appliance(self) | ||||||
|         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: | ||||||
|             result = self.data |             return self._get_nested_item(item) | ||||||
|             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] | ||||||
|  |  | ||||||
|     def get(self, item: str, default: Any = None) -> Any: |     @overload | ||||||
|  |     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): | ||||||
| @ -71,21 +87,22 @@ class HonAppliance: | |||||||
|  |  | ||||||
|     def _check_name_zone(self, name: str, frontend: bool = True) -> str: |     def _check_name_zone(self, name: str, frontend: bool = True) -> str: | ||||||
|         zone = " Z" if frontend else "_z" |         zone = " Z" if frontend else "_z" | ||||||
|         if (attribute := self._info.get(name, "")) and self._zone: |         attribute: str = self._info.get(name, "") | ||||||
|  |         if attribute and self._zone: | ||||||
|             return f"{attribute}{zone}{self._zone}" |             return f"{attribute}{zone}{self._zone}" | ||||||
|         return attribute |         return attribute | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def appliance_model_id(self) -> str: |     def appliance_model_id(self) -> str: | ||||||
|         return self._info.get("applianceModelId", "") |         return str(self._info.get("applianceModelId", "")) | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def appliance_type(self) -> str: |     def appliance_type(self) -> str: | ||||||
|         return self._info.get("applianceTypeName", "") |         return str(self._info.get("applianceTypeName", "")) | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def mac_address(self) -> str: |     def mac_address(self) -> str: | ||||||
|         return self.info.get("macAddress", "") |         return str(self.info.get("macAddress", "")) | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def unique_id(self) -> str: |     def unique_id(self) -> str: | ||||||
| @ -101,29 +118,31 @@ class HonAppliance: | |||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def brand(self) -> str: |     def brand(self) -> str: | ||||||
|         return self._check_name_zone("brand") |         brand = self._check_name_zone("brand") | ||||||
|  |         return brand[0].upper() + brand[1:] | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def nick_name(self) -> str: |     def nick_name(self) -> str: | ||||||
|         result = self._check_name_zone("nickName") |         result = self._check_name_zone("nickName") | ||||||
|         if not result or re.findall("^[xX\s]+$", result): |         if not result or re.findall("^[xX1\\s-]+$", result): | ||||||
|             return self.model_name |             return self.model_name | ||||||
|         return result |         return result | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def code(self) -> str: |     def code(self) -> str: | ||||||
|         if code := self.info.get("code"): |         code: str = self.info.get("code", "") | ||||||
|  |         if code: | ||||||
|             return code |             return code | ||||||
|         serial_number = self.info.get("serialNumber", "") |         serial_number: str = self.info.get("serialNumber", "") | ||||||
|         return serial_number[:8] if len(serial_number) < 18 else serial_number[:11] |         return serial_number[:8] if len(serial_number) < 18 else serial_number[:11] | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def model_id(self) -> int: |     def model_id(self) -> int: | ||||||
|         return self._info.get("applianceModelId", 0) |         return int(self._info.get("applianceModelId", 0)) | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def options(self) -> Dict[str, Any]: |     def options(self) -> Dict[str, Any]: | ||||||
|         return self._appliance_model.get("options", {}) |         return dict(self._appliance_model.get("options", {})) | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def commands(self) -> Dict[str, HonCommand]: |     def commands(self) -> Dict[str, HonCommand]: | ||||||
| @ -162,18 +181,18 @@ class HonAppliance: | |||||||
|         self._commands = command_loader.commands |         self._commands = command_loader.commands | ||||||
|         self._additional_data = command_loader.additional_data |         self._additional_data = command_loader.additional_data | ||||||
|         self._appliance_model = command_loader.appliance_data |         self._appliance_model = command_loader.appliance_data | ||||||
|  |         self.sync_params_to_command("settings") | ||||||
|  |  | ||||||
|     async def load_attributes(self) -> None: |     async def load_attributes(self) -> None: | ||||||
|         self._attributes = await self.api.load_attributes(self) |         attributes = await self.api.load_attributes(self) | ||||||
|         for name, values in ( |         for name, values in attributes.pop("shadow", {}).get("parameters", {}).items(): | ||||||
|             self._attributes.pop("shadow", {}).get("parameters", {}).items() |  | ||||||
|         ): |  | ||||||
|             if name in self._attributes.get("parameters", {}): |             if name in self._attributes.get("parameters", {}): | ||||||
|                 self._attributes["parameters"][name].update(values) |                 self._attributes["parameters"][name].update(values) | ||||||
|             else: |             else: | ||||||
|                 self._attributes.setdefault("parameters", {})[name] = HonAttribute( |                 self._attributes.setdefault("parameters", {})[name] = HonAttribute( | ||||||
|                     values |                     values | ||||||
|                 ) |                 ) | ||||||
|  |         self._attributes |= attributes | ||||||
|         if self._extra: |         if self._extra: | ||||||
|             self._attributes = self._extra.attributes(self._attributes) |             self._attributes = self._extra.attributes(self._attributes) | ||||||
|  |  | ||||||
| @ -183,14 +202,11 @@ class HonAppliance: | |||||||
|  |  | ||||||
|     async def update(self, force: bool = False) -> None: |     async def update(self, force: bool = False) -> None: | ||||||
|         now = datetime.now() |         now = datetime.now() | ||||||
|         if ( |         min_age = now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL) | ||||||
|             force |         if force or not self._last_update or self._last_update < min_age: | ||||||
|             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") | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def command_parameters(self) -> Dict[str, Dict[str, str | float]]: |     def command_parameters(self) -> Dict[str, Dict[str, str | float]]: | ||||||
| @ -198,7 +214,7 @@ class HonAppliance: | |||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def settings(self) -> Dict[str, Parameter]: |     def settings(self) -> Dict[str, Parameter]: | ||||||
|         result = {} |         result: Dict[str, Parameter] = {} | ||||||
|         for name, command in self._commands.items(): |         for name, command in self._commands.items(): | ||||||
|             for key in command.setting_keys: |             for key in command.setting_keys: | ||||||
|                 setting = command.settings.get(key, self._default_setting) |                 setting = command.settings.get(key, self._default_setting) | ||||||
| @ -234,32 +250,67 @@ class HonAppliance: | |||||||
|     async def data_archive(self, path: Path) -> str: |     async def data_archive(self, path: Path) -> str: | ||||||
|         return await diagnose.zip_archive(self, path, anonymous=True) |         return await diagnose.zip_archive(self, path, anonymous=True) | ||||||
|  |  | ||||||
|     def sync_to_params(self, command_name: str) -> None: |     def sync_command_to_params(self, command_name: str) -> None: | ||||||
|         if not (command := self.commands.get(command_name)): |         if not (command := self.commands.get(command_name)): | ||||||
|             return |             return | ||||||
|         for key, value in self.attributes.get("parameters", {}).items(): |         for key in self.attributes.get("parameters", {}): | ||||||
|             if isinstance(value, str) and (new := command.parameters.get(key)): |             if new := command.parameters.get(key): | ||||||
|                 self.attributes["parameters"][key].update( |                 self.attributes["parameters"][key].update( | ||||||
|                     str(new.intern_value), shield=True |                     str(new.intern_value), shield=True | ||||||
|                 ) |                 ) | ||||||
|  |  | ||||||
|     def sync_command(self, main: str, target: Optional[List[str]] = None) -> None: |     def sync_params_to_command(self, command_name: str) -> None: | ||||||
|  |         if not (command := self.commands.get(command_name)): | ||||||
|  |             return | ||||||
|  |         for key in command.setting_keys: | ||||||
|  |             if ( | ||||||
|  |                 new := self.attributes.get("parameters", {}).get(key) | ||||||
|  |             ) is None or new.value == "": | ||||||
|  |                 continue | ||||||
|  |             setting = command.settings[key] | ||||||
|  |             try: | ||||||
|  |                 if not isinstance(setting, HonParameterRange): | ||||||
|  |                     command.settings[key].value = str(new.value) | ||||||
|  |                 else: | ||||||
|  |                     command.settings[key].value = float(new.value) | ||||||
|  |             except ValueError as error: | ||||||
|  |                 _LOGGER.info("Can't set %s - %s", key, error) | ||||||
|  |                 continue | ||||||
|  |  | ||||||
|  |     def sync_command( | ||||||
|  |         self, | ||||||
|  |         main: str, | ||||||
|  |         target: Optional[List[str] | str] = None, | ||||||
|  |         to_sync: Optional[List[str] | bool] = None, | ||||||
|  |     ) -> None: | ||||||
|         base: Optional[HonCommand] = self.commands.get(main) |         base: Optional[HonCommand] = self.commands.get(main) | ||||||
|         if not base: |         if not base: | ||||||
|             return |             return | ||||||
|         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(): |  | ||||||
|                 if base_value := base.parameters.get(name): |             for name, target_param in data.parameters.items(): | ||||||
|                     if isinstance(base_value, HonParameterRange) and isinstance( |                 if not (base_param := base.parameters.get(name)): | ||||||
|                         parameter, HonParameterRange |                     continue | ||||||
|  |                 if to_sync and ( | ||||||
|  |                     (isinstance(to_sync, list) and name not in to_sync) | ||||||
|  |                     or not base_param.mandatory | ||||||
|                 ): |                 ): | ||||||
|                         parameter.max = base_value.max |                     continue | ||||||
|                         parameter.min = base_value.min |                 self.sync_parameter(base_param, target_param) | ||||||
|                         parameter.step = base_value.step |  | ||||||
|                     elif isinstance(parameter, HonParameterRange): |     def sync_parameter(self, main: Parameter, target: Parameter) -> None: | ||||||
|                         parameter.max = int(base_value.value) |         if isinstance(main, HonParameterRange) and isinstance( | ||||||
|                         parameter.min = int(base_value.value) |             target, HonParameterRange | ||||||
|                         parameter.step = 1 |         ): | ||||||
|                     parameter.value = base_value.value |             target.max = main.max | ||||||
|  |             target.min = main.min | ||||||
|  |             target.step = main.step | ||||||
|  |         elif isinstance(target, HonParameterRange): | ||||||
|  |             target.max = int(main.value) | ||||||
|  |             target.min = int(main.value) | ||||||
|  |             target.step = 1 | ||||||
|  |         elif isinstance(target, HonParameterEnum): | ||||||
|  |             target.values = main.values | ||||||
|  |         target.value = main.value | ||||||
|  | |||||||
| @ -1,3 +1,4 @@ | |||||||
|  | # 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,7 +1,6 @@ | |||||||
| 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): | ||||||
| @ -14,11 +13,4 @@ class Appliance(ApplianceBase): | |||||||
|             data["parameters"]["remainingTimeMM"].value = "0" |             data["parameters"]["remainingTimeMM"].value = "0" | ||||||
|  |  | ||||||
|         data["active"] = data["parameters"]["onOffStatus"] == "1" |         data["active"] = data["parameters"]["onOffStatus"] == "1" | ||||||
|  |  | ||||||
|         if program := int(data["parameters"]["prCode"].value): |  | ||||||
|             if (setting := self.parent.settings["startProgram.program"]) and isinstance( |  | ||||||
|                 setting, HonParameterProgram |  | ||||||
|             ): |  | ||||||
|                 data["programName"] = setting.ids.get(program, "") |  | ||||||
|  |  | ||||||
|         return data |         return data | ||||||
|  | |||||||
| @ -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"]["quickModeZ1"] == "1": |         elif data["parameters"].get("quickModeZ1") == "1": | ||||||
|             data["modeZ1"] = "super_cool" |             data["modeZ1"] = "super_cool" | ||||||
|         else: |         else: | ||||||
|             data["modeZ1"] = "no_mode" |             data["modeZ1"] = "no_mode" | ||||||
|  |  | ||||||
|         if data["parameters"]["quickModeZ2"] == "1": |         if data["parameters"].get("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,3 +1,4 @@ | |||||||
|  | # 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,3 +1,4 @@ | |||||||
|  | # 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 | ||||||
|  | |||||||
							
								
								
									
										16
									
								
								pyhon/appliances/wh.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										16
									
								
								pyhon/appliances/wh.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,16 @@ | |||||||
|  | from typing import Any, Dict | ||||||
|  |  | ||||||
|  | from pyhon.appliances.base import ApplianceBase | ||||||
|  | from pyhon.parameter.base import HonParameter | ||||||
|  |  | ||||||
|  |  | ||||||
|  | class Appliance(ApplianceBase): | ||||||
|  |     def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: | ||||||
|  |         data = super().attributes(data) | ||||||
|  |         parameter = data["parameters"]["onOffStatus"] | ||||||
|  |         is_class = isinstance(parameter, HonParameter) | ||||||
|  |         data["active"] = parameter.value == 1 if is_class else parameter == 1 | ||||||
|  |         return data | ||||||
|  |  | ||||||
|  |     def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]: | ||||||
|  |         return settings | ||||||
| @ -1,3 +1,4 @@ | |||||||
|  | # 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,14 +1,15 @@ | |||||||
| 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, Collection | from typing import Dict, Any, Optional, TYPE_CHECKING, List | ||||||
|  |  | ||||||
| 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, exceptions |     from pyhon import HonAPI | ||||||
|     from pyhon.appliance import HonAppliance |     from pyhon.appliance import HonAppliance | ||||||
|  |  | ||||||
|  |  | ||||||
| @ -16,12 +17,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] = {} | ||||||
|  |  | ||||||
| @ -29,7 +30,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 exceptions.NoAuthenticationException("Missing hOn login") |             raise NoAuthenticationException("Missing hOn login") | ||||||
|         return self._api |         return self._api | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
| @ -183,22 +184,44 @@ 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 = favourite.get("favouriteName", {}) |             name, command_name, base = self._get_favourite_info(favourite) | ||||||
|  |             if not base: | ||||||
|  |                 continue | ||||||
|  |             base_command: HonCommand = copy(base) | ||||||
|  |             self._update_base_command_with_data(base_command, favourite) | ||||||
|  |             self._update_base_command_with_favourite(base_command) | ||||||
|  |             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 = favourite.get("command", {}) | ||||||
|             command_name = command.get("commandName", "") |         command_name: str = command.get("commandName", "") | ||||||
|         program_name = self._clean_name(command.get("programName", "")) |         program_name = self._clean_name(command.get("programName", "")) | ||||||
|             base: HonCommand = copy( |         base_command = self.commands[command_name].categories.get(program_name) | ||||||
|                 self.commands[command_name].categories[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(): |         for data in command.values(): | ||||||
|             if isinstance(data, str): |             if isinstance(data, str): | ||||||
|                 continue |                 continue | ||||||
|             for key, value in data.items(): |             for key, value in data.items(): | ||||||
|                     if parameter := base.parameters.get(key): |                 if not (parameter := base_command.parameters.get(key)): | ||||||
|  |                     continue | ||||||
|                 with suppress(ValueError): |                 with suppress(ValueError): | ||||||
|                     parameter.value = value |                     parameter.value = value | ||||||
|  |  | ||||||
|  |     def _update_base_command_with_favourite(self, base_command: HonCommand) -> None: | ||||||
|         extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") |         extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") | ||||||
|             base.parameters.update(favourite=extra_param) |         base_command.parameters.update(favourite=extra_param) | ||||||
|             if isinstance(program := base.parameters["program"], HonParameterProgram): |  | ||||||
|  |     def _update_program_categories( | ||||||
|  |         self, command_name: str, name: str, base_command: HonCommand | ||||||
|  |     ) -> None: | ||||||
|  |         program = base_command.parameters["program"] | ||||||
|  |         if isinstance(program, HonParameterProgram): | ||||||
|             program.set_value(name) |             program.set_value(name) | ||||||
|             self.commands[command_name].categories[name] = base |         self.commands[command_name].categories[name] = base_command | ||||||
|  | |||||||
| @ -27,17 +27,16 @@ class HonCommand: | |||||||
|         categories: Optional[Dict[str, "HonCommand"]] = None, |         categories: Optional[Dict[str, "HonCommand"]] = None, | ||||||
|         category_name: str = "", |         category_name: str = "", | ||||||
|     ): |     ): | ||||||
|         self._api: Optional[HonAPI] = appliance.api |  | ||||||
|         self._appliance: "HonAppliance" = appliance |  | ||||||
|         self._name: str = name |         self._name: str = name | ||||||
|  |         self._api: Optional[HonAPI] = None | ||||||
|  |         self._appliance: "HonAppliance" = appliance | ||||||
|         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._description: str = attributes.pop("description", "") |         self._parameters: Dict[str, Parameter] = {} | ||||||
|         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: | ||||||
| @ -49,6 +48,8 @@ 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 | ||||||
| @ -76,6 +77,14 @@ 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()} | ||||||
| @ -93,10 +102,12 @@ 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" not in data: |             if "fixedValue" in data: | ||||||
|                 _LOGGER.error("Rule not supported: %s", data) |  | ||||||
|             else: |  | ||||||
|                 self._rules.append(HonRuleSet(self, data["fixedValue"])) |                 self._rules.append(HonRuleSet(self, data["fixedValue"])) | ||||||
|  |             elif "enumValues" in data: | ||||||
|  |                 self._rules.append(HonRuleSet(self, data["enumValues"])) | ||||||
|  |             else: | ||||||
|  |                 _LOGGER.warning("Rule not supported: %s", data) | ||||||
|         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) | ||||||
| @ -111,11 +122,26 @@ 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) -> bool: |     async def send(self, only_mandatory: bool = False) -> bool: | ||||||
|         params = self.parameter_groups.get("parameters", {}) |         grouped_params = ( | ||||||
|  |             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 or parameter.mandatory: | ||||||
|  |                 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_to_params(self.name) |         if "prStr" in params: | ||||||
|  |             params["prStr"] = self._category_name.upper() | ||||||
|  |         self.appliance.sync_command_to_params(self.name) | ||||||
|         try: |         try: | ||||||
|             result = await self.api.send_command( |             result = await self.api.send_command( | ||||||
|                 self._appliance, self._name, params, ancillary_params |                 self._appliance, self._name, params, ancillary_params | ||||||
| @ -169,3 +195,7 @@ 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() | ||||||
|  | |||||||
| @ -75,8 +75,12 @@ class HonAPI: | |||||||
|  |  | ||||||
|     async def load_appliances(self) -> List[Dict[str, Any]]: |     async def load_appliances(self) -> List[Dict[str, Any]]: | ||||||
|         async with self._hon.get(f"{const.API_URL}/commands/v1/appliance") as resp: |         async with self._hon.get(f"{const.API_URL}/commands/v1/appliance") as resp: | ||||||
|             if result := await resp.json(): |             result = await resp.json() | ||||||
|                 return result.get("payload", {}).get("appliances", {}) |         if result: | ||||||
|  |             appliances: List[Dict[str, Any]] = result.get("payload", {}).get( | ||||||
|  |                 "appliances", {} | ||||||
|  |             ) | ||||||
|  |             return appliances | ||||||
|         return [] |         return [] | ||||||
|  |  | ||||||
|     async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]: |     async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]: | ||||||
| @ -112,7 +116,8 @@ class HonAPI: | |||||||
|             result: Dict[str, Any] = await response.json() |             result: Dict[str, Any] = await response.json() | ||||||
|         if not result or not result.get("payload"): |         if not result or not result.get("payload"): | ||||||
|             return [] |             return [] | ||||||
|             return result["payload"]["history"] |         command_history: List[Dict[str, Any]] = result["payload"]["history"] | ||||||
|  |         return command_history | ||||||
|  |  | ||||||
|     async def load_favourites(self, appliance: HonAppliance) -> List[Dict[str, Any]]: |     async def load_favourites(self, appliance: HonAppliance) -> List[Dict[str, Any]]: | ||||||
|         url: str = ( |         url: str = ( | ||||||
| @ -122,14 +127,17 @@ class HonAPI: | |||||||
|             result: Dict[str, Any] = await response.json() |             result: Dict[str, Any] = await response.json() | ||||||
|         if not result or not result.get("payload"): |         if not result or not result.get("payload"): | ||||||
|             return [] |             return [] | ||||||
|             return result["payload"]["favourites"] |         favourites: List[Dict[str, Any]] = result["payload"]["favourites"] | ||||||
|  |         return favourites | ||||||
|  |  | ||||||
|     async def load_last_activity(self, appliance: HonAppliance) -> Dict[str, Any]: |     async def load_last_activity(self, appliance: HonAppliance) -> Dict[str, Any]: | ||||||
|         url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity" |         url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity" | ||||||
|         params: Dict[str, str] = {"macAddress": appliance.mac_address} |         params: Dict[str, str] = {"macAddress": appliance.mac_address} | ||||||
|         async with self._hon.get(url, params=params) as response: |         async with self._hon.get(url, params=params) as response: | ||||||
|             result: Dict[str, Any] = await response.json() |             result: Dict[str, Any] = await response.json() | ||||||
|             if result and (activity := result.get("attributes")): |             if result: | ||||||
|  |                 activity: Dict[str, Any] = result.get("attributes", "") | ||||||
|  |                 if activity: | ||||||
|                     return activity |                     return activity | ||||||
|         return {} |         return {} | ||||||
|  |  | ||||||
| @ -142,7 +150,10 @@ class HonAPI: | |||||||
|         async with self._hon.get(url, params=params) as response: |         async with self._hon.get(url, params=params) as response: | ||||||
|             result: Dict[str, Any] = await response.json() |             result: Dict[str, Any] = await response.json() | ||||||
|             if result: |             if result: | ||||||
|                 return result.get("payload", {}).get("applianceModel", {}) |                 appliance_data: Dict[str, Any] = result.get("payload", {}).get( | ||||||
|  |                     "applianceModel", {} | ||||||
|  |                 ) | ||||||
|  |                 return appliance_data | ||||||
|         return {} |         return {} | ||||||
|  |  | ||||||
|     async def load_attributes(self, appliance: HonAppliance) -> Dict[str, Any]: |     async def load_attributes(self, appliance: HonAppliance) -> Dict[str, Any]: | ||||||
| @ -153,7 +164,8 @@ class HonAPI: | |||||||
|         } |         } | ||||||
|         url: str = f"{const.API_URL}/commands/v1/context" |         url: str = f"{const.API_URL}/commands/v1/context" | ||||||
|         async with self._hon.get(url, params=params) as response: |         async with self._hon.get(url, params=params) as response: | ||||||
|             return (await response.json()).get("payload", {}) |             attributes: Dict[str, Any] = (await response.json()).get("payload", {}) | ||||||
|  |         return attributes | ||||||
|  |  | ||||||
|     async def load_statistics(self, appliance: HonAppliance) -> Dict[str, Any]: |     async def load_statistics(self, appliance: HonAppliance) -> Dict[str, Any]: | ||||||
|         params: Dict[str, str] = { |         params: Dict[str, str] = { | ||||||
| @ -162,13 +174,15 @@ class HonAPI: | |||||||
|         } |         } | ||||||
|         url: str = f"{const.API_URL}/commands/v1/statistics" |         url: str = f"{const.API_URL}/commands/v1/statistics" | ||||||
|         async with self._hon.get(url, params=params) as response: |         async with self._hon.get(url, params=params) as response: | ||||||
|             return (await response.json()).get("payload", {}) |             statistics: Dict[str, Any] = (await response.json()).get("payload", {}) | ||||||
|  |         return statistics | ||||||
|  |  | ||||||
|     async def load_maintenance(self, appliance: HonAppliance) -> Dict[str, Any]: |     async def load_maintenance(self, appliance: HonAppliance) -> Dict[str, Any]: | ||||||
|         url = f"{const.API_URL}/commands/v1/maintenance-cycle" |         url = f"{const.API_URL}/commands/v1/maintenance-cycle" | ||||||
|         params = {"macAddress": appliance.mac_address} |         params = {"macAddress": appliance.mac_address} | ||||||
|         async with self._hon.get(url, params=params) as response: |         async with self._hon.get(url, params=params) as response: | ||||||
|             return (await response.json()).get("payload", {}) |             maintenance: Dict[str, Any] = (await response.json()).get("payload", {}) | ||||||
|  |         return maintenance | ||||||
|  |  | ||||||
|     async def send_command( |     async def send_command( | ||||||
|         self, |         self, | ||||||
| @ -207,9 +221,8 @@ class HonAPI: | |||||||
|         url: str = f"{const.API_URL}/config/v1/program-list-rules" |         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[str, Any] = await response.json() |             result: Dict[str, Any] = await response.json() | ||||||
|             if result and (data := result.get("payload")): |         data: Dict[str, Any] = result.get("payload", {}) | ||||||
|         return data |         return data | ||||||
|         return {} |  | ||||||
|  |  | ||||||
|     async def app_config( |     async def app_config( | ||||||
|         self, language: str = "en", beta: bool = True |         self, language: str = "en", beta: bool = True | ||||||
| @ -223,17 +236,17 @@ class HonAPI: | |||||||
|         } |         } | ||||||
|         payload: str = json.dumps(payload_data, separators=(",", ":")) |         payload: str = json.dumps(payload_data, separators=(",", ":")) | ||||||
|         async with self._hon_anonymous.post(url, data=payload) as response: |         async with self._hon_anonymous.post(url, data=payload) as response: | ||||||
|             if (result := await response.json()) and (data := result.get("payload")): |             result = await response.json() | ||||||
|  |         data: Dict[str, Any] = result.get("payload", {}) | ||||||
|         return data |         return data | ||||||
|         return {} |  | ||||||
|  |  | ||||||
|     async def translation_keys(self, language: str = "en") -> Dict[str, Any]: |     async def translation_keys(self, language: str = "en") -> Dict[str, Any]: | ||||||
|         config = await self.app_config(language=language) |         config = await self.app_config(language=language) | ||||||
|         if url := config.get("language", {}).get("jsonPath"): |         if not (url := config.get("language", {}).get("jsonPath")): | ||||||
|             async with self._hon_anonymous.get(url) as response: |  | ||||||
|                 if result := await response.json(): |  | ||||||
|                     return result |  | ||||||
|             return {} |             return {} | ||||||
|  |         async with self._hon_anonymous.get(url) as response: | ||||||
|  |             result: Dict[str, Any] = await response.json() | ||||||
|  |         return result | ||||||
|  |  | ||||||
|     async def close(self) -> None: |     async def close(self) -> None: | ||||||
|         if self._hon_handler is not None: |         if self._hon_handler is not None: | ||||||
| @ -250,19 +263,27 @@ class TestAPI(HonAPI): | |||||||
|  |  | ||||||
|     def _load_json(self, appliance: HonAppliance, file: str) -> Dict[str, Any]: |     def _load_json(self, appliance: HonAppliance, file: str) -> Dict[str, Any]: | ||||||
|         directory = f"{appliance.appliance_type}_{appliance.appliance_model_id}".lower() |         directory = f"{appliance.appliance_type}_{appliance.appliance_model_id}".lower() | ||||||
|         if (path := self._path / directory / f"{file}.json").exists(): |         if not (path := self._path / directory / f"{file}.json").exists(): | ||||||
|  |             _LOGGER.warning("Can't open %s", str(path)) | ||||||
|  |             return {} | ||||||
|         with open(path, "r", encoding="utf-8") as json_file: |         with open(path, "r", encoding="utf-8") as json_file: | ||||||
|                 return json.loads(json_file.read()) |             text = json_file.read() | ||||||
|         _LOGGER.warning(f"Can't open {str(path)}") |         try: | ||||||
|  |             data: Dict[str, Any] = json.loads(text) | ||||||
|  |             return data | ||||||
|  |         except json.decoder.JSONDecodeError as error: | ||||||
|  |             _LOGGER.error("%s - %s", str(path), error) | ||||||
|             return {} |             return {} | ||||||
|  |  | ||||||
|     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("*/"): | ||||||
|             with open( |             file = appliance / "appliance_data.json" | ||||||
|                 appliance / "appliance_data.json", "r", encoding="utf-8" |             with open(file, "r", encoding="utf-8") as json_file: | ||||||
|             ) as json_file: |                 try: | ||||||
|                     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]: | ||||||
| @ -299,4 +320,5 @@ 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 | ||||||
|  | |||||||
| @ -6,7 +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 typing import Dict, Optional, Any | from typing import Dict, Optional, Any, List | ||||||
| from urllib import parse | from urllib import parse | ||||||
| from urllib.parse import quote | from urllib.parse import quote | ||||||
|  |  | ||||||
| @ -30,6 +30,14 @@ 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 | ||||||
| @ -46,28 +54,25 @@ 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._cognito_token |         return self._auth.cognito_token | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def id_token(self) -> str: |     def id_token(self) -> str: | ||||||
|         return self._id_token |         return self._auth.id_token | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def access_token(self) -> str: |     def access_token(self) -> str: | ||||||
|         return self._access_token |         return self._auth.access_token | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def refresh_token(self) -> str: |     def refresh_token(self) -> str: | ||||||
|         return self._refresh_token |         return self._auth.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) | ||||||
| @ -115,7 +120,8 @@ class HonAuth: | |||||||
|         async with self._request.get(url) as response: |         async with self._request.get(url) as response: | ||||||
|             text = await response.text() |             text = await response.text() | ||||||
|             self._expires = datetime.utcnow() |             self._expires = datetime.utcnow() | ||||||
|             if not (login_url := re.findall("url = '(.+?)'", text)): |             login_url: List[str] = re.findall("url = '(.+?)'", text) | ||||||
|  |             if not login_url: | ||||||
|                 if "oauth/done#access_token=" in text: |                 if "oauth/done#access_token=" in text: | ||||||
|                     self._parse_token_data(text) |                     self._parse_token_data(text) | ||||||
|                     raise exceptions.HonNoAuthenticationNeeded() |                     raise exceptions.HonNoAuthenticationNeeded() | ||||||
| @ -184,18 +190,19 @@ class HonAuth: | |||||||
|             if response.status == 200: |             if response.status == 200: | ||||||
|                 with suppress(json.JSONDecodeError, KeyError): |                 with suppress(json.JSONDecodeError, KeyError): | ||||||
|                     result = await response.json() |                     result = await response.json() | ||||||
|                     return result["events"][0]["attributes"]["values"]["url"] |                     url: str = result["events"][0]["attributes"]["values"]["url"] | ||||||
|  |                     return url | ||||||
|             await self._error_logger(response) |             await self._error_logger(response) | ||||||
|             return "" |             return "" | ||||||
|  |  | ||||||
|     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._access_token = access_token[0] |             self._auth.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._auth.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._auth.id_token = id_token[0] | ||||||
|         return True if access_token and refresh_token and id_token else False |         return bool(access_token and refresh_token and id_token) | ||||||
|  |  | ||||||
|     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: | ||||||
| @ -227,7 +234,7 @@ class HonAuth: | |||||||
|         return True |         return True | ||||||
|  |  | ||||||
|     async def _api_auth(self) -> bool: |     async def _api_auth(self) -> bool: | ||||||
|         post_headers = {"id-token": self._id_token} |         post_headers = {"id-token": self._auth.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 | ||||||
| @ -237,8 +244,8 @@ 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.get("cognitoUser", {}).get("Token", "") |             self._auth.cognito_token = json_data.get("cognitoUser", {}).get("Token", "") | ||||||
|             if not self._cognito_token: |             if not self._auth.cognito_token: | ||||||
|                 _LOGGER.error(json_data) |                 _LOGGER.error(json_data) | ||||||
|                 raise exceptions.HonAuthenticationError() |                 raise exceptions.HonAuthenticationError() | ||||||
|         return True |         return True | ||||||
| @ -260,7 +267,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._refresh_token, |             "refresh_token": self._auth.refresh_token, | ||||||
|             "grant_type": "refresh_token", |             "grant_type": "refresh_token", | ||||||
|         } |         } | ||||||
|         async with self._request.post( |         async with self._request.post( | ||||||
| @ -271,14 +278,14 @@ class HonAuth: | |||||||
|                 return False |                 return False | ||||||
|             data = await response.json() |             data = await response.json() | ||||||
|         self._expires = datetime.utcnow() |         self._expires = datetime.utcnow() | ||||||
|         self._id_token = data["id_token"] |         self._auth.id_token = data["id_token"] | ||||||
|         self._access_token = data["access_token"] |         self._auth.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._cognito_token = "" |         self._auth.cognito_token = "" | ||||||
|         self._id_token = "" |         self._auth.id_token = "" | ||||||
|         self._access_token = "" |         self._auth.access_token = "" | ||||||
|         self._refresh_token = "" |         self._auth.refresh_token = "" | ||||||
|  | |||||||
| @ -21,7 +21,7 @@ class HonDevice: | |||||||
|         return self._os_version |         return self._os_version | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def os(self) -> str: |     def os_type(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, |             "os": self.os_type, | ||||||
|             "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, Protocol | from typing import Optional, Dict, Type, Any | ||||||
|  |  | ||||||
| import aiohttp | import aiohttp | ||||||
| from typing_extensions import Self | from typing_extensions import Self | ||||||
| @ -47,10 +47,11 @@ class ConnectionHandler: | |||||||
|         return self |         return self | ||||||
|  |  | ||||||
|     @asynccontextmanager |     @asynccontextmanager | ||||||
|     def _intercept( |     async 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]: | ||||||
|         raise NotImplementedError |         async with method(url, *args, **kwargs) as response: | ||||||
|  |             yield response | ||||||
|  |  | ||||||
|     @asynccontextmanager |     @asynccontextmanager | ||||||
|     async def get( |     async def get( | ||||||
| @ -59,7 +60,8 @@ class ConnectionHandler: | |||||||
|         if self._session is None: |         if self._session is None: | ||||||
|             raise exceptions.NoSessionException() |             raise exceptions.NoSessionException() | ||||||
|         response: aiohttp.ClientResponse |         response: aiohttp.ClientResponse | ||||||
|         async with self._intercept(self._session.get, *args, **kwargs) as response:  # type: ignore[arg-type] |         args = self._session.get, *args | ||||||
|  |         async with self._intercept(*args, **kwargs) as response: | ||||||
|             yield response |             yield response | ||||||
|  |  | ||||||
|     @asynccontextmanager |     @asynccontextmanager | ||||||
| @ -69,7 +71,8 @@ class ConnectionHandler: | |||||||
|         if self._session is None: |         if self._session is None: | ||||||
|             raise exceptions.NoSessionException() |             raise exceptions.NoSessionException() | ||||||
|         response: aiohttp.ClientResponse |         response: aiohttp.ClientResponse | ||||||
|         async with self._intercept(self._session.post, *args, **kwargs) as response:  # type: ignore[arg-type] |         args = self._session.post, *args | ||||||
|  |         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: |                 except json.JSONDecodeError as exc: | ||||||
|                     _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") |                     raise HonAuthenticationError("Decode Error") from exc | ||||||
|  | |||||||
| @ -2,9 +2,11 @@ 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" | ||||||
| # All seen id's (different accounts, different devices) are the same, so I guess this hash is static | CLIENT_ID = ( | ||||||
| CLIENT_ID = "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9.HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6" |     "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9." | ||||||
| APP_VERSION = "2.0.10" |     "HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6" | ||||||
|  | ) | ||||||
|  | APP_VERSION = "2.3.5" | ||||||
| OS_VERSION = 31 | OS_VERSION = 31 | ||||||
| OS = "android" | OS = "android" | ||||||
| DEVICE_MODEL = "exynos9820" | DEVICE_MODEL = "exynos9820" | ||||||
|  | |||||||
| @ -89,12 +89,11 @@ def yaml_export(appliance: "HonAppliance", anonymous: bool = False) -> str: | |||||||
|     if anonymous: |     if anonymous: | ||||||
|         for sensible in ["serialNumber", "coords"]: |         for sensible in ["serialNumber", "coords"]: | ||||||
|             data.get("appliance", {}).pop(sensible, None) |             data.get("appliance", {}).pop(sensible, None) | ||||||
|     data = { |     result = printer.pretty_print({"data": data}) | ||||||
|         "data": data, |     if commands := printer.create_commands(appliance.commands): | ||||||
|         "commands": printer.create_command(appliance.commands), |         result += printer.pretty_print({"commands": commands}) | ||||||
|         "rules": printer.create_rules(appliance.commands), |     if rules := printer.create_rules(appliance.commands): | ||||||
|     } |         result += printer.pretty_print({"rules": rules}) | ||||||
|     result = printer.pretty_print(data) |  | ||||||
|     if anonymous: |     if anonymous: | ||||||
|         result = anonymize_data(result) |         result = anonymize_data(result) | ||||||
|     return result |     return result | ||||||
|  | |||||||
| @ -7,9 +7,10 @@ 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__) | ||||||
|  |  | ||||||
| @ -43,7 +44,7 @@ class Hon: | |||||||
|     @property |     @property | ||||||
|     def api(self) -> HonAPI: |     def api(self) -> HonAPI: | ||||||
|         if self._api is None: |         if self._api is None: | ||||||
|             raise exceptions.NoAuthenticationException |             raise NoAuthenticationException | ||||||
|         return self._api |         return self._api | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|  | |||||||
| @ -7,14 +7,21 @@ 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._category: str = attributes.get("category", "") |         self._attributes = attributes | ||||||
|         self._typology: str = attributes.get("typology", "") |         self._category: str = "" | ||||||
|         self._mandatory: int = attributes.get("mandatory", 0) |         self._typology: str = "" | ||||||
|  |         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: | ||||||
| @ -61,8 +68,9 @@ 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: | ||||||
|         if str(value) in self._triggers: |         triggers = {str(k).lower(): v for k, v in self._triggers.items()} | ||||||
|             for trigger in self._triggers[str(value)]: |         if str(value).lower() in triggers: | ||||||
|  |             for trigger in triggers[str(value)]: | ||||||
|                 func, args = trigger |                 func, args = trigger | ||||||
|                 func(args) |                 func(args) | ||||||
|  |  | ||||||
| @ -85,3 +93,6 @@ 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,12 +10,19 @@ 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 = attributes.get("defaultValue") |         self._default: str | float = "" | ||||||
|         self._value = self._default or "0" |         self._value: str | float = "" | ||||||
|         self._values: List[str] = attributes.get("enumValues", []) |         self._values: List[str] = [] | ||||||
|  |         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})" | ||||||
|  |  | ||||||
| @ -41,4 +48,4 @@ class HonParameterEnum(HonParameter): | |||||||
|             self._value = value |             self._value = value | ||||||
|             self.check_trigger(value) |             self.check_trigger(value) | ||||||
|         else: |         else: | ||||||
|             raise ValueError(f"Allowed values {self._values}") |             raise ValueError(f"Allowed values: {self._values} But was: {value}") | ||||||
|  | |||||||
| @ -6,14 +6,19 @@ 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 = attributes.get("fixedValue", None) |         self._value: str | float = "" | ||||||
|  |         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)" | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def value(self) -> str | float: |     def value(self) -> str | float: | ||||||
|         return self._value if self._value is not None else "0" |         return self._value if self._value != "" else "0" | ||||||
|  |  | ||||||
|     @value.setter |     @value.setter | ||||||
|     def value(self, value: str | float) -> None: |     def value(self, value: str | float) -> None: | ||||||
|  | |||||||
| @ -28,7 +28,7 @@ class HonParameterProgram(HonParameterEnum): | |||||||
|         if value in self.values: |         if value in self.values: | ||||||
|             self._command.category = value |             self._command.category = value | ||||||
|         else: |         else: | ||||||
|             raise ValueError(f"Allowed values {self.values}") |             raise ValueError(f"Allowed values: {self.values} But was: {value}") | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def values(self) -> List[str]: |     def values(self) -> List[str]: | ||||||
| @ -37,17 +37,19 @@ class HonParameterProgram(HonParameterEnum): | |||||||
|  |  | ||||||
|     @values.setter |     @values.setter | ||||||
|     def values(self, values: List[str]) -> None: |     def values(self, values: List[str]) -> None: | ||||||
|         return |         raise ValueError("Cant set values {values}") | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def ids(self) -> Dict[int, str]: |     def ids(self) -> Dict[int, str]: | ||||||
|         values = { |         values: Dict[int, str] = {} | ||||||
|             int(p.parameters["prCode"].value): n |         for name, parameter in self._programs.items(): | ||||||
|             for i, (n, p) in enumerate(self._programs.items()) |             if "iot_" in name: | ||||||
|             if "iot_" not in n |                 continue | ||||||
|             and p.parameters.get("prCode") |             if parameter.parameters.get("prCode"): | ||||||
|             and not ((fav := p.parameters.get("favourite")) and fav.value == "1") |                 continue | ||||||
|         } |             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,11 +7,20 @@ 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 = str_to_float(attributes["minimumValue"]) |         self._min: float = 0 | ||||||
|         self._max: float = str_to_float(attributes["maximumValue"]) |         self._max: float = 0 | ||||||
|         self._step: float = str_to_float(attributes["incrementValue"]) |         self._step: float = 0 | ||||||
|         self._default: float = str_to_float(attributes.get("defaultValue", self.min)) |         self._default: float = 0 | ||||||
|         self._value: float = self._default |         self._value: float = 0 | ||||||
|  |         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}])" | ||||||
| @ -49,11 +58,14 @@ class HonParameterRange(HonParameter): | |||||||
|     @value.setter |     @value.setter | ||||||
|     def value(self, value: str | float) -> None: |     def value(self, value: str | float) -> None: | ||||||
|         value = str_to_float(value) |         value = str_to_float(value) | ||||||
|         if self.min <= value <= self.max and not (value - self.min) % self.step: |         if self.min <= value <= self.max and not ((value - self.min) * 100) % ( | ||||||
|  |             self.step * 100 | ||||||
|  |         ): | ||||||
|             self._value = value |             self._value = value | ||||||
|             self.check_trigger(value) |             self.check_trigger(value) | ||||||
|         else: |         else: | ||||||
|             raise ValueError(f"Allowed: min {self.min} max {self.max} step {self.step}") |             allowed = f"min {self.min} max {self.max} step {self.step}" | ||||||
|  |             raise ValueError(f"Allowed: {allowed} But was: {value}") | ||||||
|  |  | ||||||
|     @property |     @property | ||||||
|     def values(self) -> List[str]: |     def values(self) -> List[str]: | ||||||
|  | |||||||
| @ -29,37 +29,30 @@ def pretty_print( | |||||||
|     whitespace: str = "  ", |     whitespace: str = "  ", | ||||||
| ) -> str: | ) -> str: | ||||||
|     result = "" |     result = "" | ||||||
|     if isinstance(data, list): |     space = whitespace * intend | ||||||
|         if key: |     if isinstance(data, (dict, list)) and key: | ||||||
|             result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n" |         result += f"{space}{'- ' if is_list else ''}{key}:\n" | ||||||
|         intend += 1 |         intend += 1 | ||||||
|  |     if isinstance(data, list): | ||||||
|         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): | ||||||
|         if key: |         for i, (list_key, value) in enumerate(sorted(data.items())): | ||||||
|             result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n" |  | ||||||
|             intend += 1 |  | ||||||
|         for i, (key, value) in enumerate(sorted(data.items())): |  | ||||||
|             if is_list and not i: |  | ||||||
|             result += pretty_print( |             result += pretty_print( | ||||||
|                     value, key=key, intend=intend, is_list=True, whitespace=whitespace |                 value, | ||||||
|                 ) |                 key=list_key, | ||||||
|             elif is_list: |                 intend=intend + (is_list if i else 0), | ||||||
|                 result += pretty_print( |                 is_list=is_list and not i, | ||||||
|                     value, key=key, intend=intend + 1, whitespace=whitespace |                 whitespace=whitespace, | ||||||
|             ) |             ) | ||||||
|     else: |     else: | ||||||
|                 result += pretty_print( |         result += f"{space}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n" | ||||||
|                     value, key=key, intend=intend, whitespace=whitespace |  | ||||||
|                 ) |  | ||||||
|     else: |  | ||||||
|         result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n" |  | ||||||
|     return result |     return result | ||||||
|  |  | ||||||
|  |  | ||||||
| def create_command( | def create_commands( | ||||||
|     commands: Dict[str, "HonCommand"], concat: bool = False |     commands: Dict[str, "HonCommand"], concat: bool = False | ||||||
| ) -> Dict[str, Any]: | ) -> Dict[str, Any]: | ||||||
|     result: Dict[str, Any] = {} |     result: Dict[str, Any] = {} | ||||||
|  | |||||||
							
								
								
									
										0
									
								
								pyhon/py.typed
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										0
									
								
								pyhon/py.typed
									
									
									
									
									
										Normal file
									
								
							| @ -3,6 +3,7 @@ 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 | ||||||
| @ -24,6 +25,10 @@ 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) | ||||||
| @ -51,6 +56,11 @@ 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, | ||||||
| @ -83,29 +93,47 @@ class HonRuleSet: | |||||||
|             for rule in rules: |             for rule in rules: | ||||||
|                 self._rules.setdefault(key, []).append(rule) |                 self._rules.setdefault(key, []).append(rule) | ||||||
|  |  | ||||||
|     def _add_trigger(self, parameter: "HonParameter", data: HonRule) -> None: |     def _extra_rules_matches(self, rule: HonRule) -> bool: | ||||||
|         def apply(rule: HonRule) -> None: |         if rule.extras: | ||||||
|             if rule.extras is not None: |  | ||||||
|             for key, value in rule.extras.items(): |             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): |                 if str(self._command.parameters.get(key)) != str(value): | ||||||
|                         return |                     return False | ||||||
|             if param := self._command.parameters.get(rule.param_key): |         return True | ||||||
|                 if value := rule.param_data.get("fixedValue", ""): |  | ||||||
|                     if isinstance(param, HonParameterEnum) and set(param.values) != { |     def _apply_fixed(self, param: Parameter, value: str | float) -> None: | ||||||
|                         str(value) |         if isinstance(param, HonParameterEnum) and set(param.values) != {str(value)}: | ||||||
|                     }: |  | ||||||
|             param.values = [str(value)] |             param.values = [str(value)] | ||||||
|  |             param.value = str(value) | ||||||
|         elif isinstance(param, HonParameterRange): |         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) |             param.value = float(value) | ||||||
|             return |             return | ||||||
|         param.value = str(value) |         param.value = str(value) | ||||||
|                 elif rule.param_data.get("typology") == "enum": |  | ||||||
|                     if isinstance(param, HonParameterEnum): |     def _apply_enum(self, param: Parameter, rule: HonRule) -> None: | ||||||
|  |         if not isinstance(param, HonParameterEnum): | ||||||
|  |             return | ||||||
|         if enum_values := rule.param_data.get("enumValues"): |         if enum_values := rule.param_data.get("enumValues"): | ||||||
|             param.values = enum_values.split("|") |             param.values = enum_values.split("|") | ||||||
|         if default_value := rule.param_data.get("defaultValue"): |         if default_value := rule.param_data.get("defaultValue"): | ||||||
|             param.value = default_value |             param.value = default_value | ||||||
|  |  | ||||||
|  |     def _add_trigger(self, parameter: "HonParameter", data: HonRule) -> None: | ||||||
|  |         def apply(rule: HonRule) -> None: | ||||||
|  |             if not self._extra_rules_matches(rule): | ||||||
|  |                 return | ||||||
|  |             if not (param := self._command.parameters.get(rule.param_key)): | ||||||
|  |                 return | ||||||
|  |             if fixed_value := rule.param_data.get("fixedValue", ""): | ||||||
|  |                 self._apply_fixed(param, fixed_value) | ||||||
|  |             elif rule.param_data.get("typology") == "enum": | ||||||
|  |                 self._apply_enum(param, rule) | ||||||
|  |  | ||||||
|         parameter.add_trigger(data.trigger_value, apply, data) |         parameter.add_trigger(data.trigger_value, apply, data) | ||||||
|  |  | ||||||
|     def patch(self) -> None: |     def patch(self) -> None: | ||||||
|  | |||||||
| @ -11,7 +11,7 @@ if TYPE_CHECKING: | |||||||
|     from pyhon.parameter.range import HonParameterRange |     from pyhon.parameter.range import HonParameterRange | ||||||
|  |  | ||||||
|  |  | ||||||
| class Callback(Protocol): | class Callback(Protocol):  # pylint: disable=too-few-public-methods | ||||||
|     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,2 +1,3 @@ | |||||||
| aiohttp==3.8.4 | aiohttp>=3.8 | ||||||
| yarl==1.8.2 | yarl>=1.8 | ||||||
|  | typing-extensions>=4.8 | ||||||
|  | |||||||
| @ -1,4 +1,5 @@ | |||||||
| black==23.3.0 | black>=23.11 | ||||||
| flake8==6.0.0 | flake8>=6.1 | ||||||
| mypy==1.2.0 | mypy>=1.7 | ||||||
| pylint==2.17.2 | pylint>=3.0 | ||||||
|  | setuptools>=68.2 | ||||||
|  | |||||||
							
								
								
									
										7
									
								
								setup.py
									
									
									
									
									
								
							
							
						
						
									
										7
									
								
								setup.py
									
									
									
									
									
								
							| @ -2,12 +2,12 @@ | |||||||
|  |  | ||||||
| from setuptools import setup, find_packages | from setuptools import setup, find_packages | ||||||
|  |  | ||||||
| with open("README.md", "r") as f: | with open("README.md", "r", encoding="utf-8") as f: | ||||||
|     long_description = f.read() |     long_description = f.read() | ||||||
|  |  | ||||||
| setup( | setup( | ||||||
|     name="pyhOn", |     name="pyhOn", | ||||||
|     version="0.14.6", |     version="0.15.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"], |     install_requires=["aiohttp>=3.8", "typing-extensions>=4.8", "yarl>=1.8"], | ||||||
|     classifiers=[ |     classifiers=[ | ||||||
|         "Development Status :: 4 - Beta", |         "Development Status :: 4 - Beta", | ||||||
|         "Environment :: Console", |         "Environment :: Console", | ||||||
| @ -30,6 +30,7 @@ setup( | |||||||
|         "Operating System :: OS Independent", |         "Operating System :: OS Independent", | ||||||
|         "Programming Language :: Python :: 3.10", |         "Programming Language :: Python :: 3.10", | ||||||
|         "Programming Language :: Python :: 3.11", |         "Programming Language :: Python :: 3.11", | ||||||
|  |         "Programming Language :: Python :: 3.12", | ||||||
|         "Topic :: Software Development :: Libraries :: Python Modules", |         "Topic :: Software Development :: Libraries :: Python Modules", | ||||||
|     ], |     ], | ||||||
|     entry_points={ |     entry_points={ | ||||||
|  | |||||||
		Reference in New Issue
	
	Block a user
	