diff --git a/.flake8 b/.flake8 new file mode 100644 index 0000000..3127c2b --- /dev/null +++ b/.flake8 @@ -0,0 +1,3 @@ +[flake8] +max-line-length = 88 +max-complexity = 7 \ No newline at end of file diff --git a/.github/workflows/python-check.yml b/.github/workflows/python-check.yml index 53a37b0..3d6f17c 100644 --- a/.github/workflows/python-check.yml +++ b/.github/workflows/python-check.yml @@ -28,9 +28,7 @@ jobs: python -m pip install -r requirements_dev.txt - name: Lint with flake8 run: | - # 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 + flake8 . --count --statistics - name: Type check with mypy run: | mypy pyhon/ diff --git a/.pylintrc b/.pylintrc index f350fd7..5071b8a 100644 --- a/.pylintrc +++ b/.pylintrc @@ -1,7 +1,9 @@ [MESSAGES CONTROL] -disable=C,R +disable=missing-docstring [FORMAT] +max-args=6 +max-attributes=8 max-line-length=88 diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 0000000..035f5c7 --- /dev/null +++ b/MANIFEST.in @@ -0,0 +1 @@ +include pyhon/py.typed diff --git a/mypy.ini b/mypy.ini index 9a76b78..52281b7 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1,9 +1,24 @@ [mypy] -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 +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 diff --git a/pyhon/__main__.py b/pyhon/__main__.py index c6460f5..9e9b25b 100755 --- a/pyhon/__main__.py +++ b/pyhon/__main__.py @@ -11,6 +11,7 @@ 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__) @@ -91,11 +92,9 @@ async def main() -> None: data = device.data.copy() attr = "get" if args.get("all") else "pop" print( - printer.key_print( - data["attributes"].__getattribute__(attr)("parameters") - ) + printer.key_print(getattr(data["attributes"], attr)("parameters")) ) - print(printer.key_print(data.__getattribute__(attr)("appliance"))) + print(printer.key_print(getattr(data, attr)("appliance"))) print(printer.key_print(data)) print( printer.pretty_print( diff --git a/pyhon/appliance.py b/pyhon/appliance.py index 05b4b8d..996c463 100644 --- a/pyhon/appliance.py +++ b/pyhon/appliance.py @@ -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 +from typing import Optional, Dict, Any, TYPE_CHECKING, List, TypeVar, overload from pyhon import diagnose, exceptions from pyhon.appliances.base import ApplianceBase @@ -20,7 +20,10 @@ 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 @@ -48,24 +51,35 @@ 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: - 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 + return self._get_nested_item(item) 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] - 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: return self[item] except (KeyError, IndexError): @@ -188,12 +202,8 @@ class HonAppliance: async def update(self, force: bool = False) -> None: now = datetime.now() - if ( - force - or not self._last_update - or self._last_update - < now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL) - ): + min_age = now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL) + if force or not self._last_update or self._last_update < min_age: self._last_update = now await self.load_attributes() self.sync_params_to_command("settings") @@ -253,7 +263,9 @@ class HonAppliance: 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: + if ( + new := self.attributes.get("parameters", {}).get(key) + ) is None or new.value == "": continue setting = command.settings[key] try: @@ -272,18 +284,23 @@ class HonAppliance: for command, data in self.commands.items(): if command == main or target and command not in target: continue - 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 + + 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 diff --git a/pyhon/appliances/dw.py b/pyhon/appliances/dw.py index 8c08da3..5d0465d 100644 --- a/pyhon/appliances/dw.py +++ b/pyhon/appliances/dw.py @@ -1,3 +1,4 @@ +# pylint: disable=duplicate-code from typing import Any, Dict from pyhon.appliances.base import ApplianceBase diff --git a/pyhon/appliances/ref.py b/pyhon/appliances/ref.py index 57facb3..787dd84 100644 --- a/pyhon/appliances/ref.py +++ b/pyhon/appliances/ref.py @@ -10,12 +10,12 @@ class Appliance(ApplianceBase): data["modeZ1"] = "holiday" elif data["parameters"]["intelligenceMode"] == "1": data["modeZ1"] = "auto_set" - elif data["parameters"]["quickModeZ1"] == "1": + elif data["parameters"].get("quickModeZ1") == "1": data["modeZ1"] = "super_cool" else: data["modeZ1"] = "no_mode" - if data["parameters"]["quickModeZ2"] == "1": + if data["parameters"].get("quickModeZ2") == "1": data["modeZ2"] = "super_freeze" elif data["parameters"]["intelligenceMode"] == "1": data["modeZ2"] = "auto_set" diff --git a/pyhon/appliances/td.py b/pyhon/appliances/td.py index 1b623f8..1f33044 100644 --- a/pyhon/appliances/td.py +++ b/pyhon/appliances/td.py @@ -1,3 +1,4 @@ +# pylint: disable=duplicate-code from typing import Any, Dict from pyhon.appliances.base import ApplianceBase diff --git a/pyhon/appliances/wd.py b/pyhon/appliances/wd.py index e2a274b..6ae0388 100644 --- a/pyhon/appliances/wd.py +++ b/pyhon/appliances/wd.py @@ -1,3 +1,4 @@ +# pylint: disable=duplicate-code from typing import Dict, Any from pyhon.appliances.base import ApplianceBase diff --git a/pyhon/appliances/wm.py b/pyhon/appliances/wm.py index 1fc95c0..27572fa 100644 --- a/pyhon/appliances/wm.py +++ b/pyhon/appliances/wm.py @@ -1,3 +1,4 @@ +# pylint: disable=duplicate-code from typing import Any, Dict from pyhon.appliances.base import ApplianceBase diff --git a/pyhon/command_loader.py b/pyhon/command_loader.py index 127dd08..94b0277 100644 --- a/pyhon/command_loader.py +++ b/pyhon/command_loader.py @@ -17,12 +17,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] = {} @@ -184,22 +184,44 @@ class HonCommandLoader: def _add_favourites(self) -> None: """Patch program categories with favourites""" for favourite in self._favourites: - name = favourite.get("favouriteName", {}) - command = favourite.get("command", {}) - command_name = command.get("commandName", "") - program_name = self._clean_name(command.get("programName", "")) - base: HonCommand = copy( - self.commands[command_name].categories[program_name] - ) - for data in command.values(): - if isinstance(data, str): + 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_name: str = command.get("commandName", "") + program_name = self._clean_name(command.get("programName", "")) + base_command = self.commands[command_name].categories.get(program_name) + return name, command_name, base_command + + def _update_base_command_with_data( + self, base_command: HonCommand, command: Dict[str, Any] + ) -> None: + for data in command.values(): + if isinstance(data, str): + continue + for key, value in data.items(): + if not (parameter := base_command.parameters.get(key)): continue - for key, value in data.items(): - if parameter := base.parameters.get(key): - with suppress(ValueError): - parameter.value = value - extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") - base.parameters.update(favourite=extra_param) - if isinstance(program := base.parameters["program"], HonParameterProgram): - program.set_value(name) - self.commands[command_name].categories[name] = base + with suppress(ValueError): + parameter.value = value + + def _update_base_command_with_favourite(self, base_command: HonCommand) -> None: + extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") + base_command.parameters.update(favourite=extra_param) + + 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) + self.commands[command_name].categories[name] = base_command diff --git a/pyhon/commands.py b/pyhon/commands.py index cea218b..b671e55 100644 --- a/pyhon/commands.py +++ b/pyhon/commands.py @@ -27,17 +27,16 @@ class HonCommand: categories: Optional[Dict[str, "HonCommand"]] = None, category_name: str = "", ): - self._api: Optional[HonAPI] = appliance.api - self._appliance: "HonAppliance" = appliance self._name: str = name + self._api: Optional[HonAPI] = None + self._appliance: "HonAppliance" = appliance self._categories: Optional[Dict[str, "HonCommand"]] = categories self._category_name: str = category_name - self._description: str = attributes.pop("description", "") - self._protocol_type: str = attributes.pop("protocolType", "") - self._parameters: Dict[str, HonParameter] = {} + self._parameters: Dict[str, Parameter] = {} 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: @@ -49,6 +48,8 @@ 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 diff --git a/pyhon/connection/api.py b/pyhon/connection/api.py index 364bbb6..61bfe1f 100644 --- a/pyhon/connection/api.py +++ b/pyhon/connection/api.py @@ -278,10 +278,12 @@ class TestAPI(HonAPI): async def load_appliances(self) -> List[Dict[str, Any]]: result = [] for appliance in self._path.glob("*/"): - with open( - appliance / "appliance_data.json", "r", encoding="utf-8" - ) as json_file: - result.append(json.loads(json_file.read())) + 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) return result async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]: @@ -318,4 +320,5 @@ class TestAPI(HonAPI): parameters: Dict[str, Any], ancillary_parameters: Dict[str, Any], ) -> bool: + _LOGGER.info("%s - %s", str(parameters), str(ancillary_parameters)) return True diff --git a/pyhon/connection/auth.py b/pyhon/connection/auth.py index 2629dfb..cf75410 100644 --- a/pyhon/connection/auth.py +++ b/pyhon/connection/auth.py @@ -30,6 +30,14 @@ 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 @@ -46,28 +54,25 @@ 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._cognito_token + return self._auth.cognito_token @property def id_token(self) -> str: - return self._id_token + return self._auth.id_token @property def access_token(self) -> str: - return self._access_token + return self._auth.access_token @property def refresh_token(self) -> str: - return self._refresh_token + return self._auth.refresh_token def _check_token_expiration(self, hours: int) -> bool: return datetime.utcnow() >= self._expires + timedelta(hours=hours) @@ -192,12 +197,12 @@ class HonAuth: def _parse_token_data(self, text: str) -> bool: 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): - self._refresh_token = refresh_token[0] + self._auth.refresh_token = refresh_token[0] if id_token := re.findall("id_token=(.*?)&", text): - self._id_token = id_token[0] - return True if access_token and refresh_token and id_token else False + self._auth.id_token = id_token[0] + return bool(access_token and refresh_token and id_token) async def _get_token(self, url: str) -> bool: async with self._request.get(url) as response: @@ -229,7 +234,7 @@ class HonAuth: return True 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() async with self._request.post( f"{const.API_URL}/auth/v1/login", headers=post_headers, json=data @@ -239,8 +244,8 @@ class HonAuth: except json.JSONDecodeError: await self._error_logger(response) return False - self._cognito_token = json_data.get("cognitoUser", {}).get("Token", "") - if not self._cognito_token: + self._auth.cognito_token = json_data.get("cognitoUser", {}).get("Token", "") + if not self._auth.cognito_token: _LOGGER.error(json_data) raise exceptions.HonAuthenticationError() return True @@ -262,7 +267,7 @@ class HonAuth: async def refresh(self) -> bool: params = { "client_id": const.CLIENT_ID, - "refresh_token": self._refresh_token, + "refresh_token": self._auth.refresh_token, "grant_type": "refresh_token", } async with self._request.post( @@ -273,14 +278,14 @@ class HonAuth: return False data = await response.json() self._expires = datetime.utcnow() - self._id_token = data["id_token"] - self._access_token = data["access_token"] + self._auth.id_token = data["id_token"] + self._auth.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._cognito_token = "" - self._id_token = "" - self._access_token = "" - self._refresh_token = "" + self._auth.cognito_token = "" + self._auth.id_token = "" + self._auth.access_token = "" + self._auth.refresh_token = "" diff --git a/pyhon/connection/handler/base.py b/pyhon/connection/handler/base.py index 9e5c77b..6e169e1 100644 --- a/pyhon/connection/handler/base.py +++ b/pyhon/connection/handler/base.py @@ -60,7 +60,8 @@ class ConnectionHandler: if self._session is None: raise exceptions.NoSessionException() 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 @asynccontextmanager @@ -70,7 +71,8 @@ class ConnectionHandler: if self._session is None: raise exceptions.NoSessionException() 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 async def close(self) -> None: diff --git a/pyhon/const.py b/pyhon/const.py index 49af0bf..4f97435 100644 --- a/pyhon/const.py +++ b/pyhon/const.py @@ -2,8 +2,10 @@ AUTH_API = "https://account2.hon-smarthome.com" API_URL = "https://api-iot.he.services" API_KEY = "GRCqFhC6Gk@ikWXm1RmnSmX1cm,MxY-configuration" APP = "hon" -# 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" +CLIENT_ID = ( + "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9." + "HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6" +) APP_VERSION = "2.1.2" OS_VERSION = 31 OS = "android" diff --git a/pyhon/hon.py b/pyhon/hon.py index b207b87..d6c73e2 100644 --- a/pyhon/hon.py +++ b/pyhon/hon.py @@ -7,9 +7,10 @@ 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__) @@ -43,7 +44,7 @@ class Hon: @property def api(self) -> HonAPI: if self._api is None: - raise exceptions.NoAuthenticationException + raise NoAuthenticationException return self._api @property diff --git a/pyhon/parameter/program.py b/pyhon/parameter/program.py index 4ed2004..83b843f 100644 --- a/pyhon/parameter/program.py +++ b/pyhon/parameter/program.py @@ -41,13 +41,15 @@ class HonParameterProgram(HonParameterEnum): @property def ids(self) -> Dict[int, str]: - 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") - } + 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 return dict(sorted(values.items())) def set_value(self, value: str) -> None: diff --git a/pyhon/parameter/range.py b/pyhon/parameter/range.py index c51642c..6eef53e 100644 --- a/pyhon/parameter/range.py +++ b/pyhon/parameter/range.py @@ -16,9 +16,9 @@ class HonParameterRange(HonParameter): def _set_attributes(self) -> None: super()._set_attributes() - self._min = str_to_float(self._attributes["minimumValue"]) - self._max = str_to_float(self._attributes["maximumValue"]) - self._step = str_to_float(self._attributes["incrementValue"]) + 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 @@ -64,9 +64,8 @@ class HonParameterRange(HonParameter): self._value = value self.check_trigger(value) else: - raise ValueError( - f"Allowed: min {self.min} max {self.max} step {self.step} But was: {value}" - ) + allowed = f"min {self.min} max {self.max} step {self.step}" + raise ValueError(f"Allowed: {allowed} But was: {value}") @property def values(self) -> List[str]: diff --git a/pyhon/printer.py b/pyhon/printer.py index 395c531..9c84815 100644 --- a/pyhon/printer.py +++ b/pyhon/printer.py @@ -29,33 +29,26 @@ 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): - 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 - ) + 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, + ) else: - result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n" + result += f"{space}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n" return result diff --git a/pyhon/py.typed b/pyhon/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/pyhon/rules.py b/pyhon/rules.py index eb580e1..57d6de5 100644 --- a/pyhon/rules.py +++ b/pyhon/rules.py @@ -3,6 +3,7 @@ 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 @@ -24,6 +25,10 @@ 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) @@ -83,28 +88,42 @@ 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 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 + 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) diff --git a/pyhon/typedefs.py b/pyhon/typedefs.py index 2a22319..51a137b 100644 --- a/pyhon/typedefs.py +++ b/pyhon/typedefs.py @@ -11,7 +11,7 @@ if TYPE_CHECKING: from pyhon.parameter.range import HonParameterRange -class Callback(Protocol): +class Callback(Protocol): # pylint: disable=too-few-public-methods def __call__( self, url: str | URL, *args: Any, **kwargs: Any ) -> aiohttp.client._RequestContextManager: diff --git a/requirements.txt b/requirements.txt index 5c21de6..1e9d351 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,3 @@ -aiohttp==3.8.4 -yarl==1.8.2 +aiohttp==3.8.5 +yarl==1.9.2 typing-extensions==4.7.1 diff --git a/requirements_dev.txt b/requirements_dev.txt index 9bb012e..a28b33e 100644 --- a/requirements_dev.txt +++ b/requirements_dev.txt @@ -1,4 +1,4 @@ -black==23.3.0 +black==23.7.0 flake8==6.0.0 -mypy==1.2.0 -pylint==2.17.2 +mypy==1.4.1 +pylint==2.17.4 diff --git a/setup.py b/setup.py index 5ee66ab..2735f63 100644 --- a/setup.py +++ b/setup.py @@ -7,7 +7,7 @@ with open("README.md", "r", encoding="utf-8") as f: setup( name="pyhOn", - version="0.14.9", + version="0.15.5", 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.4", "typing-extensions==4.7.1"], + install_requires=["aiohttp==3.8.5", "typing-extensions==4.7.1"], classifiers=[ "Development Status :: 4 - Beta", "Environment :: Console",