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