Cool_Tools/pyhOn
1
0
Fork 0
mirror of https://github.com/Andre0512/pyhOn.git synced 2024-11-28 00:10:50 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add more type hints

Browse Source
This commit is contained in:
Andre Basche 2023-06-28 19:02:11 +02:00
parent ad0d065b03
commit 9eb99f283b
30 changed files with 392 additions and 243 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
mypy.ini Normal file
View File

@ -0,0 +1,4 @@
[mypy]
check_untyped_defs = True
disallow_any_generics = True
disallow_untyped_defs = True

33
pyhon/__main__.py
View File

@ -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:

57
pyhon/appliance.py
View File

@ -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

18
pyhon/appliances/base.py
View File

@ -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

4
pyhon/appliances/dw.py
View File

@ -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"

11
pyhon/appliances/ov.py
View File

@ -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

4
pyhon/appliances/ref.py
View File

@ -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"

6
pyhon/appliances/td.py
View File

@ -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)

6
pyhon/appliances/wd.py
View File

@ -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

6
pyhon/appliances/wm.py
View File

@ -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

4
pyhon/attributes.py
View File

@ -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

35
pyhon/command_loader.py
View File

@ -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

17
pyhon/commands.py
View File

@ -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:

54
pyhon/connection/api.py
View File

@ -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

16
pyhon/connection/auth.py
View File

@ -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"

8
pyhon/connection/device.py
View File

@ -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

11
pyhon/connection/handler/anonym.py
View File

@ -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:

9
pyhon/connection/handler/auth.py
View File

@ -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

35
pyhon/connection/handler/base.py
View File

@ -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:

21
pyhon/connection/handler/hon.py
View File

@ -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(

16
pyhon/diagnose.py
View File

@ -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

77
pyhon/helper.py
View File

@ -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)

4
pyhon/hon.py
View File

@ -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 == "":

14
pyhon/parameter/base.py
View File

@ -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:

4
pyhon/parameter/enum.py
View File

@ -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

4
pyhon/parameter/program.py
View File

@ -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

2
pyhon/parameter/range.py
View File

@ -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

94
pyhon/printer.py Normal file
View File

@ -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

34
pyhon/rules.py
View File

@ -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)

27
pyhon/typedefs.py Normal file
View File

@ -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",
]