diff --git a/mypy.ini b/mypy.ini new file mode 100644 index 0000000..8e8f27d --- /dev/null +++ b/mypy.ini @@ -0,0 +1,4 @@ +[mypy] +check_untyped_defs = True +disallow_any_generics = True +disallow_untyped_defs = True \ No newline at end of file diff --git a/pyhon/__main__.py b/pyhon/__main__.py index f8d810f..713e5a1 100755 --- a/pyhon/__main__.py +++ b/pyhon/__main__.py @@ -6,16 +6,17 @@ import logging import sys from getpass import getpass from pathlib import Path +from typing import Tuple, Dict, Any if __name__ == "__main__": sys.path.insert(0, str(Path(__file__).parent.parent)) -from pyhon import Hon, HonAPI, helper, diagnose +from pyhon import Hon, HonAPI, diagnose, printer _LOGGER = logging.getLogger(__name__) -def get_arguments(): +def get_arguments() -> Dict[str, Any]: """Get parsed arguments.""" parser = argparse.ArgumentParser(description="pyhOn: Command Line Utility") parser.add_argument("-u", "--user", help="user for haier hOn account") @@ -39,7 +40,7 @@ def get_arguments(): return vars(parser.parse_args()) -async def translate(language, json_output=False): +async def translate(language: str, json_output: bool = False) -> None: async with HonAPI(anonymous=True) as hon: keys = await hon.translation_keys(language) if json_output: @@ -52,10 +53,10 @@ async def translate(language, json_output=False): .replace("\\r", "") ) keys = json.loads(clean_keys) - print(helper.pretty_print(keys)) + print(printer.pretty_print(keys)) -def get_login_data(args): +def get_login_data(args: Dict[str, str]) -> Tuple[str, str]: if not (user := args["user"]): user = input("User for hOn account: ") if not (password := args["password"]): @@ -63,44 +64,44 @@ def get_login_data(args): return user, password -async def main(): +async def main() -> None: args = get_arguments() if language := args.get("translate"): - await translate(language, json_output=args.get("json")) + await translate(language, json_output=args.get("json", "")) return async with Hon(*get_login_data(args)) as hon: for device in hon.appliances: if args.get("export"): anonymous = args.get("anonymous", False) - path = Path(args.get("directory")) + path = Path(args.get("directory", ".")) if not args.get("zip"): for file in await diagnose.appliance_data(device, path, anonymous): print(f"Created {file}") else: - file = await diagnose.zip_archive(device, path, anonymous) - print(f"Created {file}") + archive = await diagnose.zip_archive(device, path, anonymous) + print(f"Created {archive}") continue print("=" * 10, device.appliance_type, "-", device.nick_name, "=" * 10) if args.get("keys"): data = device.data.copy() attr = "get" if args.get("all") else "pop" print( - helper.key_print( + printer.key_print( data["attributes"].__getattribute__(attr)("parameters") ) ) - print(helper.key_print(data.__getattribute__(attr)("appliance"))) - print(helper.key_print(data)) + print(printer.key_print(data.__getattribute__(attr)("appliance"))) + print(printer.key_print(data)) print( - helper.pretty_print( - helper.create_command(device.commands, concat=True) + printer.pretty_print( + printer.create_command(device.commands, concat=True) ) ) else: print(diagnose.yaml_export(device)) -def start(): +def start() -> None: try: asyncio.run(main()) except KeyboardInterrupt: diff --git a/pyhon/appliance.py b/pyhon/appliance.py index 3eaa525..a368cfa 100644 --- a/pyhon/appliance.py +++ b/pyhon/appliance.py @@ -2,14 +2,15 @@ import importlib import logging from datetime import datetime, timedelta from pathlib import Path -from typing import Optional, Dict, Any, TYPE_CHECKING +from typing import Optional, Dict, Any, TYPE_CHECKING, List -from pyhon import diagnose +from pyhon import diagnose, exceptions from pyhon.attributes import HonAttribute from pyhon.command_loader import HonCommandLoader from pyhon.commands import HonCommand from pyhon.parameter.base import HonParameter from pyhon.parameter.range import HonParameterRange +from pyhon.typedefs import Parameter if TYPE_CHECKING: from pyhon import HonAPI @@ -25,16 +26,16 @@ class HonAppliance: ) -> None: if attributes := info.get("attributes"): info["attributes"] = {v["parName"]: v["parValue"] for v in attributes} - self._info: Dict = info + self._info: Dict[str, Any] = info self._api: Optional[HonAPI] = api - self._appliance_model: Dict = {} + self._appliance_model: Dict[str, Any] = {} self._commands: Dict[str, HonCommand] = {} - self._statistics: Dict = {} - self._attributes: Dict = {} + self._statistics: Dict[str, Any] = {} + self._attributes: Dict[str, Any] = {} self._zone: int = zone self._additional_data: Dict[str, Any] = {} - self._last_update = None + self._last_update: Optional[datetime] = None self._default_setting = HonParameter("", {}, "") try: @@ -44,7 +45,7 @@ class HonAppliance: except ModuleNotFoundError: self._extra = None - def __getitem__(self, item): + def __getitem__(self, item: str) -> Any: if self._zone: item += f"Z{self._zone}" if "." in item: @@ -61,7 +62,7 @@ class HonAppliance: return self.attributes["parameters"][item].value return self.info[item] - def get(self, item, default=None): + def get(self, item: str, default: Any = None) -> Any: try: return self[item] except (KeyError, IndexError): @@ -113,7 +114,7 @@ class HonAppliance: return self._info.get("applianceModelId", 0) @property - def options(self): + def options(self) -> Dict[str, Any]: return self._appliance_model.get("options", {}) @property @@ -121,19 +122,19 @@ class HonAppliance: return self._commands @property - def attributes(self): + def attributes(self) -> Dict[str, Any]: return self._attributes @property - def statistics(self): + def statistics(self) -> Dict[str, Any]: return self._statistics @property - def info(self): + def info(self) -> Dict[str, Any]: return self._info @property - def additional_data(self): + def additional_data(self) -> Dict[str, Any]: return self._additional_data @property @@ -141,17 +142,20 @@ class HonAppliance: return self._zone @property - def api(self) -> Optional["HonAPI"]: + def api(self) -> "HonAPI": + """api connection object""" + if self._api is None: + raise exceptions.NoAuthenticationException("Missing hOn login") return self._api - async def load_commands(self): + async def load_commands(self) -> None: command_loader = HonCommandLoader(self.api, self) await command_loader.load_commands() self._commands = command_loader.commands self._additional_data = command_loader.additional_data self._appliance_model = command_loader.appliance_data - async def load_attributes(self): + async def load_attributes(self) -> None: self._attributes = await self.api.load_attributes(self) for name, values in self._attributes.pop("shadow").get("parameters").items(): if name in self._attributes.get("parameters", {}): @@ -163,11 +167,11 @@ class HonAppliance: if self._extra: self._attributes = self._extra.attributes(self._attributes) - async def load_statistics(self): + async def load_statistics(self) -> None: self._statistics = await self.api.load_statistics(self) self._statistics |= await self.api.load_maintenance(self) - async def update(self, force=False): + async def update(self, force: bool = False) -> None: now = datetime.now() if ( force @@ -179,11 +183,11 @@ class HonAppliance: await self.load_attributes() @property - def command_parameters(self): + def command_parameters(self) -> Dict[str, Dict[str, str | float]]: return {n: c.parameter_value for n, c in self._commands.items()} @property - def settings(self): + def settings(self) -> Dict[str, Parameter]: result = {} for name, command in self._commands.items(): for key in command.setting_keys: @@ -194,7 +198,7 @@ class HonAppliance: return result @property - def available_settings(self): + def available_settings(self) -> List[str]: result = [] for name, command in self._commands.items(): for key in command.setting_keys: @@ -202,7 +206,7 @@ class HonAppliance: return result @property - def data(self): + def data(self) -> Dict[str, Any]: result = { "attributes": self.attributes, "appliance": self.info, @@ -220,15 +224,16 @@ class HonAppliance: async def data_archive(self, path: Path) -> str: return await diagnose.zip_archive(self, path, anonymous=True) - def sync_to_params(self, command_name): - command: HonCommand = self.commands.get(command_name) + def sync_to_params(self, command_name: str) -> None: + if not (command := self.commands.get(command_name)): + return for key, value in self.attributes.get("parameters", {}).items(): if isinstance(value, str) and (new := command.parameters.get(key)): self.attributes["parameters"][key].update( str(new.intern_value), shield=True ) - def sync_command(self, main, target=None) -> None: + def sync_command(self, main: str, target: Optional[List[str]] = None) -> None: base: Optional[HonCommand] = self.commands.get(main) if not base: return diff --git a/pyhon/appliances/base.py b/pyhon/appliances/base.py index dfb4c51..dfe95f3 100644 --- a/pyhon/appliances/base.py +++ b/pyhon/appliances/base.py @@ -1,15 +1,25 @@ +from typing import Dict, Any, TYPE_CHECKING + +from pyhon.parameter.program import HonParameterProgram + +if TYPE_CHECKING: + from pyhon.appliance import HonAppliance + + class ApplianceBase: - def __init__(self, appliance): + def __init__(self, appliance: "HonAppliance"): self.parent = appliance - def attributes(self, data): + def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: program_name = "No Program" if program := int(str(data.get("parameters", {}).get("prCode", "0"))): if start_cmd := self.parent.settings.get("startProgram.program"): - if ids := start_cmd.ids: + if isinstance(start_cmd, HonParameterProgram) and ( + ids := start_cmd.ids + ): program_name = ids.get(program, program_name) data["programName"] = program_name return data - def settings(self, settings): + def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]: return settings diff --git a/pyhon/appliances/dw.py b/pyhon/appliances/dw.py index 945ab25..8c08da3 100644 --- a/pyhon/appliances/dw.py +++ b/pyhon/appliances/dw.py @@ -1,8 +1,10 @@ +from typing import Any, Dict + from pyhon.appliances.base import ApplianceBase class Appliance(ApplianceBase): - def attributes(self, data): + def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: data = super().attributes(data) if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": data["parameters"]["machMode"].value = "0" diff --git a/pyhon/appliances/ov.py b/pyhon/appliances/ov.py index 0af5a28..a871fb7 100644 --- a/pyhon/appliances/ov.py +++ b/pyhon/appliances/ov.py @@ -1,8 +1,11 @@ +from typing import Any, Dict + from pyhon.appliances.base import ApplianceBase +from pyhon.parameter.program import HonParameterProgram class Appliance(ApplianceBase): - def attributes(self, data): + def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: data = super().attributes(data) if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": data["parameters"]["temp"].value = "0" @@ -13,7 +16,9 @@ class Appliance(ApplianceBase): data["active"] = data["parameters"]["onOffStatus"] == "1" if program := int(data["parameters"]["prCode"]): - ids = self.parent.settings["startProgram.program"].ids - data["programName"] = ids.get(program, "") + if (setting := self.parent.settings["startProgram.program"]) and isinstance( + setting, HonParameterProgram + ): + data["programName"] = setting.ids.get(program, "") return data diff --git a/pyhon/appliances/ref.py b/pyhon/appliances/ref.py index c83982f..57facb3 100644 --- a/pyhon/appliances/ref.py +++ b/pyhon/appliances/ref.py @@ -1,8 +1,10 @@ +from typing import Dict, Any + from pyhon.appliances.base import ApplianceBase class Appliance(ApplianceBase): - def attributes(self, data): + def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: data = super().attributes(data) if data["parameters"]["holidayMode"] == "1": data["modeZ1"] = "holiday" diff --git a/pyhon/appliances/td.py b/pyhon/appliances/td.py index 495699a..1b623f8 100644 --- a/pyhon/appliances/td.py +++ b/pyhon/appliances/td.py @@ -1,9 +1,11 @@ +from typing import Any, Dict + from pyhon.appliances.base import ApplianceBase from pyhon.parameter.fixed import HonParameterFixed class Appliance(ApplianceBase): - def attributes(self, data): + def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: data = super().attributes(data) if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": data["parameters"]["machMode"].value = "0" @@ -11,7 +13,7 @@ class Appliance(ApplianceBase): data["pause"] = data["parameters"]["machMode"] == "3" return data - def settings(self, settings): + def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]: dry_level = settings.get("startProgram.dryLevel") if isinstance(dry_level, HonParameterFixed) and dry_level.value == "11": settings.pop("startProgram.dryLevel", None) diff --git a/pyhon/appliances/wd.py b/pyhon/appliances/wd.py index ff06f57..e2a274b 100644 --- a/pyhon/appliances/wd.py +++ b/pyhon/appliances/wd.py @@ -1,8 +1,10 @@ +from typing import Dict, Any + from pyhon.appliances.base import ApplianceBase class Appliance(ApplianceBase): - def attributes(self, data): + def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: data = super().attributes(data) if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": data["parameters"]["machMode"].value = "0" @@ -10,5 +12,5 @@ class Appliance(ApplianceBase): data["pause"] = data["parameters"]["machMode"] == "3" return data - def settings(self, settings): + def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]: return settings diff --git a/pyhon/appliances/wm.py b/pyhon/appliances/wm.py index ff06f57..1fc95c0 100644 --- a/pyhon/appliances/wm.py +++ b/pyhon/appliances/wm.py @@ -1,8 +1,10 @@ +from typing import Any, Dict + from pyhon.appliances.base import ApplianceBase class Appliance(ApplianceBase): - def attributes(self, data): + def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: data = super().attributes(data) if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": data["parameters"]["machMode"].value = "0" @@ -10,5 +12,5 @@ class Appliance(ApplianceBase): data["pause"] = data["parameters"]["machMode"] == "3" return data - def settings(self, settings): + def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]: return settings diff --git a/pyhon/attributes.py b/pyhon/attributes.py index 55ec0b7..6d0458f 100644 --- a/pyhon/attributes.py +++ b/pyhon/attributes.py @@ -7,7 +7,7 @@ from pyhon.helper import str_to_float class HonAttribute: _LOCK_TIMEOUT: Final = 10 - def __init__(self, data): + def __init__(self, data: Dict[str, str] | str): self._value: str = "" self._last_update: Optional[datetime] = None self._lock_timestamp: Optional[datetime] = None @@ -22,7 +22,7 @@ class HonAttribute: return self._value @value.setter - def value(self, value) -> None: + def value(self, value: str) -> None: self._value = value @property diff --git a/pyhon/command_loader.py b/pyhon/command_loader.py index e6ab721..1a6a5a4 100644 --- a/pyhon/command_loader.py +++ b/pyhon/command_loader.py @@ -1,7 +1,7 @@ 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.parameter.fixed import HonParameterFixed @@ -15,7 +15,7 @@ if TYPE_CHECKING: class HonCommandLoader: """Loads and parses hOn command data""" - def __init__(self, api, appliance): + def __init__(self, api: "HonAPI", appliance: "HonAppliance") -> None: self._api_commands: Dict[str, Any] = {} self._favourites: List[Dict[str, Any]] = [] self._command_history: List[Dict[str, Any]] = [] @@ -52,7 +52,7 @@ class HonCommandLoader: """Get command additional data""" return self._additional_data - async def load_commands(self): + async def load_commands(self) -> None: """Trigger loading of command data""" await self._load_data() self._appliance_data = self._api_commands.pop("applianceModel") @@ -60,17 +60,17 @@ class HonCommandLoader: self._add_favourites() self._recover_last_command_states() - async def _load_commands(self): + async def _load_commands(self) -> None: self._api_commands = await self._api.load_commands(self._appliance) - async def _load_favourites(self): + async def _load_favourites(self) -> None: self._favourites = await self._api.load_favourites(self._appliance) - async def _load_command_history(self): + async def _load_command_history(self) -> None: self._command_history = await self._api.load_command_history(self._appliance) - async def _load_data(self): - """Request parallel all relevant data""" + async def _load_data(self) -> None: + """Callback parallel all relevant data""" await asyncio.gather( *[ self._load_commands(), @@ -102,14 +102,24 @@ class HonCommandLoader: self._commands = {c.name: c for c in commands} def _parse_command( - self, data: Dict[str, Any] | str, command_name: str, **kwargs + self, + data: Dict[str, Any] | str, + command_name: str, + categories: Optional[Dict[str, "HonCommand"]] = None, + category_name: str = "", ) -> Optional[HonCommand]: """Try to crate HonCommand object""" if not isinstance(data, dict): self._additional_data[command_name] = data return None if self._is_command(data): - return HonCommand(command_name, data, self._appliance, **kwargs) + return HonCommand( + command_name, + data, + self._appliance, + category_name=category_name, + categories=categories, + ) if category := self._parse_categories(data, command_name): return category return None @@ -120,8 +130,9 @@ class HonCommandLoader: """Parse categories and create reference to other""" categories: Dict[str, HonCommand] = {} for category, value in data.items(): - kwargs = {"category_name": category, "categories": categories} - if command := self._parse_command(value, command_name, **kwargs): + if command := self._parse_command( + value, command_name, category_name=category, categories=categories + ): categories[self._clean_name(category)] = command if categories: # setParameters should be at first place diff --git a/pyhon/commands.py b/pyhon/commands.py index 9b65b1c..7af1f7f 100644 --- a/pyhon/commands.py +++ b/pyhon/commands.py @@ -9,6 +9,7 @@ from pyhon.parameter.fixed import HonParameterFixed from pyhon.parameter.program import HonParameterProgram from pyhon.parameter.range import HonParameterRange from pyhon.rules import HonRuleSet +from pyhon.typedefs import Parameter if TYPE_CHECKING: from pyhon import HonAPI @@ -43,7 +44,7 @@ class HonCommand: return f"{self._name} command" @property - def name(self): + def name(self) -> str: return self._name @property @@ -57,7 +58,7 @@ class HonCommand: return self._appliance @property - def data(self): + def data(self) -> Dict[str, Any]: return self._data @property @@ -79,14 +80,16 @@ class HonCommand: def parameter_value(self) -> Dict[str, Union[str, float]]: return {n: p.value for n, p in self._parameters.items()} - def _load_parameters(self, attributes): + def _load_parameters(self, attributes: Dict[str, Dict[str, Any]]) -> None: for key, items in attributes.items(): for name, data in items.items(): self._create_parameters(data, name, key) for rule in self._rules: rule.patch() - def _create_parameters(self, data: Dict, name: str, parameter: str) -> None: + def _create_parameters( + self, data: Dict[str, Any], name: str, parameter: str + ) -> None: if name == "zoneMap" and self._appliance.zone: data["default"] = self._appliance.zone if data.get("category") == "rule": @@ -147,7 +150,7 @@ class HonCommand: ) @staticmethod - def _more_options(first: HonParameter, second: HonParameter): + def _more_options(first: Parameter, second: Parameter) -> Parameter: if isinstance(first, HonParameterFixed) and not isinstance( second, HonParameterFixed ): @@ -157,8 +160,8 @@ class HonCommand: return first @property - def available_settings(self) -> Dict[str, HonParameter]: - result: Dict[str, HonParameter] = {} + def available_settings(self) -> Dict[str, Parameter]: + result: Dict[str, Parameter] = {} for command in self.categories.values(): for name, parameter in command.parameters.items(): if name in result: diff --git a/pyhon/connection/api.py b/pyhon/connection/api.py index 5819776..c9a4588 100644 --- a/pyhon/connection/api.py +++ b/pyhon/connection/api.py @@ -3,7 +3,8 @@ import logging from datetime import datetime from pathlib import Path from pprint import pformat -from typing import Dict, Optional, Any, List, no_type_check +from types import TracebackType +from typing import Dict, Optional, Any, List, no_type_check, Type from aiohttp import ClientSession from typing_extensions import Self @@ -36,7 +37,12 @@ class HonAPI: async def __aenter__(self) -> Self: return await self.create() - async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: + async def __aexit__( + self, + exc_type: Optional[Type[BaseException]], + exc: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: await self.close() @property @@ -46,13 +52,13 @@ class HonAPI: return self._hon.auth @property - def _hon(self): + def _hon(self) -> HonConnectionHandler: if self._hon_handler is None: raise exceptions.NoAuthenticationException return self._hon_handler @property - def _hon_anonymous(self): + def _hon_anonymous(self) -> HonAnonymousConnectionHandler: if self._hon_anonymous_handler is None: raise exceptions.NoAuthenticationException return self._hon_anonymous_handler @@ -74,7 +80,7 @@ class HonAPI: return [] async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]: - params: Dict = { + params: Dict[str, str | int] = { "applianceType": appliance.appliance_type, "applianceModelId": appliance.appliance_model_id, "macAddress": appliance.mac_address, @@ -90,7 +96,7 @@ class HonAPI: params["series"] = series url: str = f"{const.API_URL}/commands/v1/retrieve" async with self._hon.get(url, params=params) as response: - result: Dict = (await response.json()).get("payload", {}) + result: Dict[str, Any] = (await response.json()).get("payload", {}) if not result or result.pop("resultCode") != "0": _LOGGER.error(await response.json()) return {} @@ -103,7 +109,7 @@ class HonAPI: f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/history" ) async with self._hon.get(url) as response: - result: Dict = await response.json() + result: Dict[str, Any] = await response.json() if not result or not result.get("payload"): return [] return result["payload"]["history"] @@ -113,34 +119,34 @@ class HonAPI: f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/favourite" ) async with self._hon.get(url) as response: - result: Dict = await response.json() + result: Dict[str, Any] = await response.json() if not result or not result.get("payload"): return [] return result["payload"]["favourites"] async def load_last_activity(self, appliance: HonAppliance) -> Dict[str, Any]: url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity" - params: Dict = {"macAddress": appliance.mac_address} + params: Dict[str, str] = {"macAddress": appliance.mac_address} async with self._hon.get(url, params=params) as response: - result: Dict = await response.json() + result: Dict[str, Any] = await response.json() if result and (activity := result.get("attributes")): return activity return {} async def load_appliance_data(self, appliance: HonAppliance) -> Dict[str, Any]: url: str = f"{const.API_URL}/commands/v1/appliance-model" - params: Dict = { + params: Dict[str, str] = { "code": appliance.code, "macAddress": appliance.mac_address, } async with self._hon.get(url, params=params) as response: - result: Dict = await response.json() + result: Dict[str, Any] = await response.json() if result: return result.get("payload", {}).get("applianceModel", {}) return {} async def load_attributes(self, appliance: HonAppliance) -> Dict[str, Any]: - params: Dict = { + params: Dict[str, str] = { "macAddress": appliance.mac_address, "applianceType": appliance.appliance_type, "category": "CYCLE", @@ -150,7 +156,7 @@ class HonAPI: return (await response.json()).get("payload", {}) async def load_statistics(self, appliance: HonAppliance) -> Dict[str, Any]: - params: Dict = { + params: Dict[str, str] = { "macAddress": appliance.mac_address, "applianceType": appliance.appliance_type, } @@ -168,11 +174,11 @@ class HonAPI: self, appliance: HonAppliance, command: str, - parameters: Dict, - ancillary_parameters: Dict, + parameters: Dict[str, Any], + ancillary_parameters: Dict[str, Any], ) -> bool: now: str = datetime.utcnow().isoformat() - data: Dict = { + data: Dict[str, Any] = { "macAddress": appliance.mac_address, "timestamp": f"{now[:-3]}Z", "commandName": command, @@ -190,7 +196,7 @@ class HonAPI: } url: str = f"{const.API_URL}/commands/v1/send" async with self._hon.post(url, json=data) as response: - json_data: Dict = await response.json() + json_data: Dict[str, Any] = await response.json() if json_data.get("payload", {}).get("resultCode") == "0": return True _LOGGER.error(await response.text()) @@ -200,7 +206,7 @@ class HonAPI: async def appliance_configuration(self) -> Dict[str, Any]: url: str = f"{const.API_URL}/config/v1/program-list-rules" async with self._hon_anonymous.get(url) as response: - result: Dict = await response.json() + result: Dict[str, Any] = await response.json() if result and (data := result.get("payload")): return data return {} @@ -209,7 +215,7 @@ class HonAPI: self, language: str = "en", beta: bool = True ) -> Dict[str, Any]: url: str = f"{const.API_URL}/app-config" - payload_data: Dict = { + payload_data: Dict[str, str | int] = { "languageCode": language, "beta": beta, "appVersion": const.APP_VERSION, @@ -237,12 +243,12 @@ class HonAPI: class TestAPI(HonAPI): - def __init__(self, path): + def __init__(self, path: Path): super().__init__() self._anonymous = True self._path: Path = path - def _load_json(self, appliance: HonAppliance, file) -> Dict[str, Any]: + def _load_json(self, appliance: HonAppliance, file: str) -> Dict[str, Any]: directory = f"{appliance.appliance_type}_{appliance.appliance_model_id}".lower() path = f"{self._path}/{directory}/{file}.json" with open(path, "r", encoding="utf-8") as json_file: @@ -288,7 +294,7 @@ class TestAPI(HonAPI): self, appliance: HonAppliance, command: str, - parameters: Dict, - ancillary_parameters: Dict, + parameters: Dict[str, Any], + ancillary_parameters: Dict[str, Any], ) -> bool: return True diff --git a/pyhon/connection/auth.py b/pyhon/connection/auth.py index 4ecc5ab..c011274 100644 --- a/pyhon/connection/auth.py +++ b/pyhon/connection/auth.py @@ -6,14 +6,16 @@ import urllib from contextlib import suppress from dataclasses import dataclass from datetime import datetime, timedelta -from typing import Dict, Optional +from typing import Dict, Optional, Any from urllib import parse from urllib.parse import quote +import aiohttp from aiohttp import ClientResponse from yarl import URL from pyhon import const, exceptions +from pyhon.connection.device import HonDevice from pyhon.connection.handler.auth import HonAuthConnectionHandler _LOGGER = logging.getLogger(__name__) @@ -25,14 +27,20 @@ class HonLoginData: email: str = "" password: str = "" fw_uid: str = "" - loaded: Optional[Dict] = None + loaded: Optional[Dict[str, Any]] = None class HonAuth: _TOKEN_EXPIRES_AFTER_HOURS = 8 _TOKEN_EXPIRE_WARNING_HOURS = 7 - def __init__(self, session, email, password, device) -> None: + def __init__( + self, + session: aiohttp.ClientSession, + email: str, + password: str, + device: HonDevice, + ) -> None: self._session = session self._request = HonAuthConnectionHandler(session) self._login_data = HonLoginData() @@ -120,7 +128,7 @@ class HonAuth: await self._error_logger(response) return new_location - async def _handle_redirects(self, login_url) -> str: + async def _handle_redirects(self, login_url: str) -> str: redirect1 = await self._manual_redirect(login_url) redirect2 = await self._manual_redirect(redirect1) return f"{redirect2}&System=IoT_Mobile_App&RegistrationSubChannel=hOn" diff --git a/pyhon/connection/device.py b/pyhon/connection/device.py index ea7f30e..6d680c0 100644 --- a/pyhon/connection/device.py +++ b/pyhon/connection/device.py @@ -32,12 +32,14 @@ class HonDevice: def mobile_id(self) -> str: return self._mobile_id - def get(self, mobile: bool = False) -> Dict: - result = { + def get(self, mobile: bool = False) -> Dict[str, str | int]: + result: Dict[str, str | int] = { "appVersion": self.app_version, "mobileId": self.mobile_id, "os": self.os, "osVersion": self.os_version, "deviceModel": self.device_model, } - return (result | {"mobileOs": result.pop("os")}) if mobile else result + if mobile: + result |= {"mobileOs": result.pop("os", "")} + return result diff --git a/pyhon/connection/handler/anonym.py b/pyhon/connection/handler/anonym.py index 1ed0410..42cdc79 100644 --- a/pyhon/connection/handler/anonym.py +++ b/pyhon/connection/handler/anonym.py @@ -1,19 +1,24 @@ import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from typing import Callable, Dict +from typing import Dict, Any + +import aiohttp from pyhon import const from pyhon.connection.handler.base import ConnectionHandler +from pyhon.typedefs import Callback _LOGGER = logging.getLogger(__name__) class HonAnonymousConnectionHandler(ConnectionHandler): - _HEADERS: Dict = ConnectionHandler._HEADERS | {"x-api-key": const.API_KEY} + _HEADERS: Dict[str, str] = ConnectionHandler._HEADERS | {"x-api-key": const.API_KEY} @asynccontextmanager - async def _intercept(self, method: Callable, *args, **kwargs) -> AsyncIterator: + async def _intercept( + self, method: Callback, *args: Any, **kwargs: Any + ) -> AsyncIterator[aiohttp.ClientResponse]: kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS async with method(*args, **kwargs) as response: if response.status == 403: diff --git a/pyhon/connection/handler/auth.py b/pyhon/connection/handler/auth.py index ecba4cb..212395c 100644 --- a/pyhon/connection/handler/auth.py +++ b/pyhon/connection/handler/auth.py @@ -1,12 +1,13 @@ import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from typing import Optional, Callable, List, Tuple +from typing import Optional, List, Tuple, Any import aiohttp from pyhon import const from pyhon.connection.handler.base import ConnectionHandler +from pyhon.typedefs import Callback _LOGGER = logging.getLogger(__name__) @@ -28,9 +29,9 @@ class HonAuthConnectionHandler(ConnectionHandler): @asynccontextmanager async def _intercept( - self, method: Callable, *args, loop: int = 0, **kwargs - ) -> AsyncIterator: + self, method: Callback, *args: Any, **kwargs: Any + ) -> AsyncIterator[aiohttp.ClientResponse]: kwargs["headers"] = kwargs.pop("headers", {}) | self._HEADERS async with method(*args, **kwargs) as response: - self._called_urls.append((response.status, response.request_info.url)) + self._called_urls.append((response.status, str(response.request_info.url))) yield response diff --git a/pyhon/connection/handler/base.py b/pyhon/connection/handler/base.py index 7ccf5f4..eb254b5 100644 --- a/pyhon/connection/handler/base.py +++ b/pyhon/connection/handler/base.py @@ -1,18 +1,20 @@ import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from typing import Optional, Callable, Dict +from types import TracebackType +from typing import Optional, Dict, Type, Any, Protocol import aiohttp from typing_extensions import Self from pyhon import const, exceptions +from pyhon.typedefs import Callback _LOGGER = logging.getLogger(__name__) class ConnectionHandler: - _HEADERS: Dict = { + _HEADERS: Dict[str, str] = { "user-agent": const.USER_AGENT, "Content-Type": "application/json", } @@ -24,32 +26,49 @@ class ConnectionHandler: async def __aenter__(self) -> Self: return await self.create() - async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: + async def __aexit__( + self, + exc_type: Optional[Type[BaseException]], + exc: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: await self.close() + @property + def session(self) -> aiohttp.ClientSession: + if self._session is None: + raise exceptions.NoSessionException + return self._session + async def create(self) -> Self: if self._create_session: self._session = aiohttp.ClientSession() return self @asynccontextmanager - def _intercept(self, method: Callable, *args, loop: int = 0, **kwargs): + def _intercept( + self, method: Callback, *args: Any, loop: int = 0, **kwargs: Any + ) -> AsyncIterator[aiohttp.ClientResponse]: raise NotImplementedError @asynccontextmanager - async def get(self, *args, **kwargs) -> AsyncIterator[aiohttp.ClientResponse]: + async def get( + self, *args: Any, **kwargs: Any + ) -> AsyncIterator[aiohttp.ClientResponse]: if self._session is None: raise exceptions.NoSessionException() response: aiohttp.ClientResponse - async with self._intercept(self._session.get, *args, **kwargs) as response: + async with self._intercept(self._session.get, *args, **kwargs) as response: # type: ignore[arg-type] yield response @asynccontextmanager - async def post(self, *args, **kwargs) -> AsyncIterator[aiohttp.ClientResponse]: + async def post( + self, *args: Any, **kwargs: Any + ) -> AsyncIterator[aiohttp.ClientResponse]: if self._session is None: raise exceptions.NoSessionException() response: aiohttp.ClientResponse - async with self._intercept(self._session.post, *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: diff --git a/pyhon/connection/handler/hon.py b/pyhon/connection/handler/hon.py index f50c3a9..15654de 100644 --- a/pyhon/connection/handler/hon.py +++ b/pyhon/connection/handler/hon.py @@ -2,7 +2,7 @@ import json import logging from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from typing import Optional, Callable, Dict +from typing import Optional, Dict, Any import aiohttp from typing_extensions import Self @@ -11,6 +11,7 @@ from pyhon.connection.auth import HonAuth from pyhon.connection.device import HonDevice from pyhon.connection.handler.base import ConnectionHandler from pyhon.exceptions import HonAuthenticationError, NoAuthenticationException +from pyhon.typedefs import Callback _LOGGER = logging.getLogger(__name__) @@ -41,10 +42,10 @@ class HonConnectionHandler(ConnectionHandler): async def create(self) -> Self: await super().create() - self._auth = HonAuth(self._session, self._email, self._password, self._device) + self._auth = HonAuth(self.session, self._email, self._password, self._device) return self - async def _check_headers(self, headers: Dict) -> Dict: + async def _check_headers(self, headers: Dict[str, str]) -> Dict[str, str]: if not (self.auth.cognito_token and self.auth.id_token): await self.auth.authenticate() headers["cognito-token"] = self.auth.cognito_token @@ -53,18 +54,16 @@ class HonConnectionHandler(ConnectionHandler): @asynccontextmanager async def _intercept( - self, method: Callable, *args, loop: int = 0, **kwargs - ) -> AsyncIterator: + self, method: Callback, *args: Any, loop: int = 0, **kwargs: Dict[str, str] + ) -> AsyncIterator[aiohttp.ClientResponse]: kwargs["headers"] = await self._check_headers(kwargs.get("headers", {})) - async with method(*args, **kwargs) as response: + async with method(args[0], *args[1:], **kwargs) as response: if ( self.auth.token_expires_soon or response.status in [401, 403] ) and loop == 0: _LOGGER.info("Try refreshing token...") await self.auth.refresh() - async with self._intercept( - method, *args, loop=loop + 1, **kwargs - ) as result: + async with self._intercept(method, loop=loop + 1, **kwargs) as result: yield result elif ( self.auth.token_is_expired or response.status in [401, 403] @@ -76,9 +75,7 @@ class HonConnectionHandler(ConnectionHandler): await response.text(), ) await self.create() - async with self._intercept( - method, *args, loop=loop + 1, **kwargs - ) as result: + async with self._intercept(method, loop=loop + 1, **kwargs) as result: yield result elif loop >= 2: _LOGGER.error( diff --git a/pyhon/diagnose.py b/pyhon/diagnose.py index 25c934a..bffb17c 100644 --- a/pyhon/diagnose.py +++ b/pyhon/diagnose.py @@ -5,7 +5,7 @@ import shutil from pathlib import Path from typing import TYPE_CHECKING, List, Tuple -from pyhon import helper +from pyhon import printer if TYPE_CHECKING: from pyhon.appliance import HonAppliance @@ -38,7 +38,7 @@ async def load_data(appliance: "HonAppliance", topic: str) -> Tuple[str, str]: return topic, await getattr(appliance.api, f"load_{topic}")(appliance) -def write_to_json(data: str, topic: str, path: Path, anonymous: bool = False): +def write_to_json(data: str, topic: str, path: Path, anonymous: bool = False) -> Path: json_data = json.dumps(data, indent=4) if anonymous: json_data = anonymize_data(json_data) @@ -65,7 +65,9 @@ async def appliance_data( return [write_to_json(data, topic, path, anonymous) for topic, data in api_data] -async def zip_archive(appliance: "HonAppliance", path: Path, anonymous: bool = False): +async def zip_archive( + appliance: "HonAppliance", path: Path, anonymous: bool = False +) -> str: data = await appliance_data(appliance, path, anonymous) archive = data[0].parent shutil.make_archive(str(archive.parent), "zip", archive) @@ -73,7 +75,7 @@ async def zip_archive(appliance: "HonAppliance", path: Path, anonymous: bool = F return f"{archive.stem}.zip" -def yaml_export(appliance: "HonAppliance", anonymous=False) -> str: +def yaml_export(appliance: "HonAppliance", anonymous: bool = False) -> str: data = { "attributes": appliance.attributes.copy(), "appliance": appliance.info, @@ -89,10 +91,10 @@ def yaml_export(appliance: "HonAppliance", anonymous=False) -> str: data.get("appliance", {}).pop(sensible, None) data = { "data": data, - "commands": helper.create_command(appliance.commands), - "rules": helper.create_rules(appliance.commands), + "commands": printer.create_command(appliance.commands), + "rules": printer.create_rules(appliance.commands), } - result = helper.pretty_print(data) + result = printer.pretty_print(data) if anonymous: result = anonymize_data(result) return result diff --git a/pyhon/helper.py b/pyhon/helper.py index fc7555e..0cc3870 100644 --- a/pyhon/helper.py +++ b/pyhon/helper.py @@ -1,80 +1,3 @@ -def key_print(data, key="", start=True): - result = "" - if isinstance(data, list): - for i, value in enumerate(data): - result += key_print(value, key=f"{key}.{i}", start=False) - elif isinstance(data, dict): - for k, value in sorted(data.items()): - result += key_print(value, key=k if start else f"{key}.{k}", start=False) - else: - result += f"{key}: {data}\n" - return result - - -# yaml.dump() would be done the same, but needs an additional dependency... -def pretty_print(data, key="", intend=0, is_list=False, whitespace=" "): - result = "" - 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 - ) - else: - result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n" - return result - - -def create_command(commands, concat=False): - result = {} - for name, command in commands.items(): - for parameter, data in command.available_settings.items(): - if data.typology == "enum": - value = data.values - elif data.typology == "range": - value = {"min": data.min, "max": data.max, "step": data.step} - else: - continue - if not concat: - result.setdefault(name, {})[parameter] = value - else: - result[f"{name}.{parameter}"] = value - return result - - -def create_rules(commands, concat=False): - result = {} - for name, command in commands.items(): - for parameter, data in command.available_settings.items(): - value = data.triggers - if not value: - continue - if not concat: - result.setdefault(name, {})[parameter] = value - else: - result[f"{name}.{parameter}"] = value - return result - - def str_to_float(string: str | float) -> float: try: return int(string) diff --git a/pyhon/hon.py b/pyhon/hon.py index e385337..b207b87 100644 --- a/pyhon/hon.py +++ b/pyhon/hon.py @@ -70,11 +70,11 @@ class Hon: return self._appliances @appliances.setter - def appliances(self, appliances) -> None: + def appliances(self, appliances: List[HonAppliance]) -> None: self._appliances = appliances async def _create_appliance( - self, appliance_data: Dict[str, Any], api: HonAPI, zone=0 + self, appliance_data: Dict[str, Any], api: HonAPI, zone: int = 0 ) -> None: appliance = HonAppliance(api, appliance_data, zone=zone) if appliance.mac_address == "": diff --git a/pyhon/parameter/base.py b/pyhon/parameter/base.py index aa66e6b..6bf81ca 100644 --- a/pyhon/parameter/base.py +++ b/pyhon/parameter/base.py @@ -12,7 +12,9 @@ class HonParameter: self._mandatory: int = attributes.get("mandatory", 0) self._value: str | float = "" self._group: str = group - self._triggers: Dict[str, List[Tuple[Callable, "HonRule"]]] = {} + self._triggers: Dict[ + str, List[Tuple[Callable[["HonRule"], None], "HonRule"]] + ] = {} @property def key(self) -> str: @@ -51,20 +53,22 @@ class HonParameter: def group(self) -> str: return self._group - def add_trigger(self, value, func, data): + def add_trigger( + self, value: str, func: Callable[["HonRule"], None], data: "HonRule" + ) -> None: if self._value == value: func(data) self._triggers.setdefault(value, []).append((func, data)) - def check_trigger(self, value) -> None: + def check_trigger(self, value: str | float) -> None: if str(value) in self._triggers: for trigger in self._triggers[str(value)]: func, args = trigger func(args) @property - def triggers(self): - result = {} + def triggers(self) -> Dict[str, Any]: + result: Dict[str, Any] = {} for value, rules in self._triggers.items(): for _, rule in rules: if rule.extras: diff --git a/pyhon/parameter/enum.py b/pyhon/parameter/enum.py index 2dd2633..c6b1dd0 100644 --- a/pyhon/parameter/enum.py +++ b/pyhon/parameter/enum.py @@ -3,7 +3,7 @@ from typing import Dict, Any, List from pyhon.parameter.base import HonParameter -def clean_value(value): +def clean_value(value: str | float) -> str: return str(value).strip("[]").replace("|", "_").lower() @@ -24,7 +24,7 @@ class HonParameterEnum(HonParameter): return [clean_value(value) for value in self._values] @values.setter - def values(self, values) -> None: + def values(self, values: List[str]) -> None: self._values = values @property diff --git a/pyhon/parameter/program.py b/pyhon/parameter/program.py index 5b68c2e..841a633 100644 --- a/pyhon/parameter/program.py +++ b/pyhon/parameter/program.py @@ -36,7 +36,7 @@ class HonParameterProgram(HonParameterEnum): return sorted(values) @values.setter - def values(self, values) -> None: + def values(self, values: List[str]) -> None: return @property @@ -50,5 +50,5 @@ class HonParameterProgram(HonParameterEnum): } return dict(sorted(values.items())) - def set_value(self, value: str): + def set_value(self, value: str) -> None: self._value = value diff --git a/pyhon/parameter/range.py b/pyhon/parameter/range.py index 2cb0772..5928369 100644 --- a/pyhon/parameter/range.py +++ b/pyhon/parameter/range.py @@ -13,7 +13,7 @@ class HonParameterRange(HonParameter): self._default: float = str_to_float(attributes.get("defaultValue", self.min)) self._value: float = self._default - def __repr__(self): + def __repr__(self) -> str: return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])" @property diff --git a/pyhon/printer.py b/pyhon/printer.py new file mode 100644 index 0000000..ca7190e --- /dev/null +++ b/pyhon/printer.py @@ -0,0 +1,94 @@ +from typing import Dict, Any, TYPE_CHECKING, List + +from pyhon.parameter.enum import HonParameterEnum +from pyhon.parameter.range import HonParameterRange + +if TYPE_CHECKING: + from pyhon.commands import HonCommand + + +def key_print(data: Any, key: str = "", start: bool = True) -> str: + result = "" + if isinstance(data, list): + for i, value in enumerate(data): + result += key_print(value, key=f"{key}.{i}", start=False) + elif isinstance(data, dict): + for k, value in sorted(data.items()): + result += key_print(value, key=k if start else f"{key}.{k}", start=False) + else: + result += f"{key}: {data}\n" + return result + + +# yaml.dump() would be done the same, but needs an additional dependency... +def pretty_print( + data: Any, + key: str = "", + intend: int = 0, + is_list: bool = False, + whitespace: str = " ", +) -> str: + result = "" + 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 + ) + else: + result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n" + return result + + +def create_command( + commands: Dict[str, "HonCommand"], concat: bool = False +) -> Dict[str, Any]: + result: Dict[str, Any] = {} + for name, command in commands.items(): + for parameter, data in command.available_settings.items(): + if isinstance(data, HonParameterEnum): + value: List[str] | Dict[str, str | float] = data.values + elif isinstance(data, HonParameterRange): + value = {"min": data.min, "max": data.max, "step": data.step} + else: + continue + if not concat: + result.setdefault(name, {})[parameter] = value + else: + result[f"{name}.{parameter}"] = value + return result + + +def create_rules( + commands: Dict[str, "HonCommand"], concat: bool = False +) -> Dict[str, Any]: + result: Dict[str, Any] = {} + for name, command in commands.items(): + for parameter, data in command.available_settings.items(): + value = data.triggers + if not value: + continue + if not concat: + result.setdefault(name, {})[parameter] = value + else: + result[f"{name}.{parameter}"] = value + return result diff --git a/pyhon/rules.py b/pyhon/rules.py index 1bcd71c..eb580e1 100644 --- a/pyhon/rules.py +++ b/pyhon/rules.py @@ -6,6 +6,7 @@ from pyhon.parameter.range import HonParameterRange if TYPE_CHECKING: from pyhon.commands import HonCommand + from pyhon.parameter.base import HonParameter @dataclass @@ -18,18 +19,24 @@ class HonRule: class HonRuleSet: - def __init__(self, command: "HonCommand", rule): + def __init__(self, command: "HonCommand", rule: Dict[str, Any]): self._command: "HonCommand" = command self._rules: Dict[str, List[HonRule]] = {} self._parse_rule(rule) - def _parse_rule(self, rule): + 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) for trigger_key, trigger_data in params.items(): self._parse_conditions(param_key, trigger_key, trigger_data) - def _parse_conditions(self, param_key, trigger_key, trigger_data, extra=None): + def _parse_conditions( + self, + param_key: str, + trigger_key: str, + trigger_data: Dict[str, Any], + extra: Optional[Dict[str, str]] = None, + ) -> None: trigger_key = trigger_key.replace("@", "") trigger_key = self._command.appliance.options.get(trigger_key, trigger_key) for multi_trigger_value, param_data in trigger_data.items(): @@ -46,16 +53,21 @@ class HonRuleSet: self._parse_conditions(param_key, extra_key, extra_data, extra) def _create_rule( - self, param_key, trigger_key, trigger_value, param_data, extras=None - ): + self, + param_key: str, + trigger_key: str, + trigger_value: str, + param_data: Dict[str, Any], + extras: Optional[Dict[str, str]] = None, + ) -> None: if param_data.get("fixedValue") == f"@{param_key}": return self._rules.setdefault(trigger_key, []).append( HonRule(trigger_key, trigger_value, param_key, param_data, extras) ) - def _duplicate_for_extra_conditions(self): - new = {} + def _duplicate_for_extra_conditions(self) -> None: + new: Dict[str, List[HonRule]] = {} for rules in self._rules.values(): for rule in rules: if rule.extras is None: @@ -71,8 +83,8 @@ class HonRuleSet: for rule in rules: self._rules.setdefault(key, []).append(rule) - def _add_trigger(self, parameter, data): - def apply(rule: HonRule): + 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): @@ -96,10 +108,10 @@ class HonRuleSet: parameter.add_trigger(data.trigger_value, apply, data) - def patch(self): + def patch(self) -> None: self._duplicate_for_extra_conditions() for name, parameter in self._command.parameters.items(): if name not in self._rules: continue - for data in self._rules.get(name): + for data in self._rules.get(name, []): self._add_trigger(parameter, data) diff --git a/pyhon/typedefs.py b/pyhon/typedefs.py new file mode 100644 index 0000000..2a22319 --- /dev/null +++ b/pyhon/typedefs.py @@ -0,0 +1,27 @@ +from typing import Union, Any, TYPE_CHECKING, Protocol + +import aiohttp +from yarl import URL + +if TYPE_CHECKING: + from pyhon.parameter.base import HonParameter + from pyhon.parameter.enum import HonParameterEnum + from pyhon.parameter.fixed import HonParameterFixed + from pyhon.parameter.program import HonParameterProgram + from pyhon.parameter.range import HonParameterRange + + +class Callback(Protocol): + def __call__( + self, url: str | URL, *args: Any, **kwargs: Any + ) -> aiohttp.client._RequestContextManager: + ... + + +Parameter = Union[ + "HonParameter", + "HonParameterRange", + "HonParameterEnum", + "HonParameterFixed", + "HonParameterProgram", +]