From f6139db0b5955e865bb17da3155193715ad28024 Mon Sep 17 00:00:00 2001 From: Andre Basche Date: Tue, 13 Jun 2023 00:12:29 +0200 Subject: [PATCH 1/3] Use class for attributes --- pyhon/appliance.py | 10 +++++++--- pyhon/appliances/base.py | 2 +- pyhon/appliances/dw.py | 2 +- pyhon/appliances/ov.py | 8 ++++---- pyhon/appliances/td.py | 2 +- pyhon/appliances/wd.py | 2 +- pyhon/appliances/wm.py | 2 +- pyhon/attributes.py | 37 +++++++++++++++++++++++++++++++++++++ pyhon/helper.py | 7 +++++++ pyhon/parameter/range.py | 8 +------- 10 files changed, 61 insertions(+), 19 deletions(-) create mode 100644 pyhon/attributes.py diff --git a/pyhon/appliance.py b/pyhon/appliance.py index eb13e12..bfb16b2 100644 --- a/pyhon/appliance.py +++ b/pyhon/appliance.py @@ -9,6 +9,7 @@ from typing import Optional, Dict, Any from typing import TYPE_CHECKING from pyhon import helper +from pyhon.attributes import HonAttribute from pyhon.commands import HonCommand from pyhon.parameter.base import HonParameter from pyhon.parameter.fixed import HonParameterFixed @@ -61,7 +62,7 @@ class HonAppliance: if item in self.data: return self.data[item] if item in self.attributes["parameters"]: - return self.attributes["parameters"].get(item) + return self.attributes["parameters"][item].value return self.info[item] def get(self, item, default=None): @@ -241,7 +242,10 @@ class HonAppliance: async def load_attributes(self): self._attributes = await self.api.load_attributes(self) for name, values in self._attributes.pop("shadow").get("parameters").items(): - self._attributes.setdefault("parameters", {})[name] = values["parNewVal"] + if name in self._attributes.get("parameters", {}): + self._attributes["parameters"][name].update(values) + else: + self._attributes.setdefault("parameters", {})[name] = HonAttribute(values) if self._extra: self._attributes = self._extra.attributes(self._attributes) @@ -326,7 +330,7 @@ class HonAppliance: command: HonCommand = self.commands.get(command_name) for key, value in self.attributes.get("parameters", {}).items(): if isinstance(value, str) and (new := command.parameters.get(key)): - self.attributes["parameters"][key] = str(new.intern_value) + self.attributes["parameters"][key].value = str(new.intern_value) def sync_command(self, main, target=None) -> None: base: HonCommand = self.commands.get(main) diff --git a/pyhon/appliances/base.py b/pyhon/appliances/base.py index 0e24b63..dfb4c51 100644 --- a/pyhon/appliances/base.py +++ b/pyhon/appliances/base.py @@ -4,7 +4,7 @@ class ApplianceBase: def attributes(self, data): program_name = "No Program" - if program := int(data["parameters"].get("prCode", "0")): + if program := int(str(data.get("parameters", {}).get("prCode", "0"))): if start_cmd := self.parent.settings.get("startProgram.program"): if ids := start_cmd.ids: program_name = ids.get(program, program_name) diff --git a/pyhon/appliances/dw.py b/pyhon/appliances/dw.py index 6e980b7..af3da74 100644 --- a/pyhon/appliances/dw.py +++ b/pyhon/appliances/dw.py @@ -5,6 +5,6 @@ class Appliance(ApplianceBase): def attributes(self, data): data = super().attributes(data) if data["lastConnEvent"]["category"] == "DISCONNECTED": - data["parameters"]["machMode"] = "0" + data["parameters"]["machMode"].value = "0" data["active"] = bool(data.get("activity")) return data diff --git a/pyhon/appliances/ov.py b/pyhon/appliances/ov.py index 5558690..74afeac 100644 --- a/pyhon/appliances/ov.py +++ b/pyhon/appliances/ov.py @@ -5,10 +5,10 @@ class Appliance(ApplianceBase): def attributes(self, data): data = super().attributes(data) if data["lastConnEvent"]["category"] == "DISCONNECTED": - data["parameters"]["temp"] = "0" - data["parameters"]["onOffStatus"] = "0" - data["parameters"]["remoteCtrValid"] = "0" - data["parameters"]["remainingTimeMM"] = "0" + data["parameters"]["temp"].value = "0" + data["parameters"]["onOffStatus"].value = "0" + data["parameters"]["remoteCtrValid"].value = "0" + data["parameters"]["remainingTimeMM"].value = "0" data["active"] = data["parameters"]["onOffStatus"] == "1" diff --git a/pyhon/appliances/td.py b/pyhon/appliances/td.py index ee21748..6b9894e 100644 --- a/pyhon/appliances/td.py +++ b/pyhon/appliances/td.py @@ -6,7 +6,7 @@ class Appliance(ApplianceBase): def attributes(self, data): data = super().attributes(data) if data["lastConnEvent"]["category"] == "DISCONNECTED": - data["parameters"]["machMode"] = "0" + data["parameters"]["machMode"].value = "0" data["active"] = bool(data.get("activity")) data["pause"] = data["parameters"]["machMode"] == "3" return data diff --git a/pyhon/appliances/wd.py b/pyhon/appliances/wd.py index ee4ccac..3f847d0 100644 --- a/pyhon/appliances/wd.py +++ b/pyhon/appliances/wd.py @@ -5,7 +5,7 @@ class Appliance(ApplianceBase): def attributes(self, data): data = super().attributes(data) if data["lastConnEvent"]["category"] == "DISCONNECTED": - data["parameters"]["machMode"] = "0" + data["parameters"]["machMode"].value = "0" data["active"] = bool(data.get("activity")) data["pause"] = data["parameters"]["machMode"] == "3" return data diff --git a/pyhon/appliances/wm.py b/pyhon/appliances/wm.py index ee4ccac..3f847d0 100644 --- a/pyhon/appliances/wm.py +++ b/pyhon/appliances/wm.py @@ -5,7 +5,7 @@ class Appliance(ApplianceBase): def attributes(self, data): data = super().attributes(data) if data["lastConnEvent"]["category"] == "DISCONNECTED": - data["parameters"]["machMode"] = "0" + data["parameters"]["machMode"].value = "0" data["active"] = bool(data.get("activity")) data["pause"] = data["parameters"]["machMode"] == "3" return data diff --git a/pyhon/attributes.py b/pyhon/attributes.py new file mode 100644 index 0000000..3a45c29 --- /dev/null +++ b/pyhon/attributes.py @@ -0,0 +1,37 @@ +from datetime import datetime +from typing import Optional + +from pyhon.helper import str_to_float + + +class HonAttribute: + def __init__(self, data): + self._value: str = "" + self._last_update: Optional[datetime] = None + self.update(data) + + @property + def value(self) -> float | str: + try: + return str_to_float(self._value) + except ValueError: + return self._value + + @value.setter + def value(self, value) -> None: + self._value = value + + @property + def last_update(self) -> Optional[datetime]: + return self._last_update + + def update(self, data): + self._value = data.get("parNewVal", "") + if last_update := data.get("lastUpdate"): + try: + self._last_update = datetime.fromisoformat(last_update) + except ValueError: + self._last_update = None + + def __str__(self) -> str: + return self._value diff --git a/pyhon/helper.py b/pyhon/helper.py index c37ffb2..fc7555e 100644 --- a/pyhon/helper.py +++ b/pyhon/helper.py @@ -73,3 +73,10 @@ def create_rules(commands, concat=False): else: result[f"{name}.{parameter}"] = value return result + + +def str_to_float(string: str | float) -> float: + try: + return int(string) + except ValueError: + return float(str(string).replace(",", ".")) diff --git a/pyhon/parameter/range.py b/pyhon/parameter/range.py index e43f22f..2cb0772 100644 --- a/pyhon/parameter/range.py +++ b/pyhon/parameter/range.py @@ -1,15 +1,9 @@ from typing import Dict, Any, List +from pyhon.helper import str_to_float from pyhon.parameter.base import HonParameter -def str_to_float(string: str | float) -> float: - try: - return int(string) - except ValueError: - return float(str(string).replace(",", ".")) - - class HonParameterRange(HonParameter): def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: super().__init__(key, attributes, group) From 1ca89995a2e35d2f537225ccc210311f0d37c2ed Mon Sep 17 00:00:00 2001 From: Andre Basche Date: Tue, 13 Jun 2023 00:39:18 +0200 Subject: [PATCH 2/3] Lock attributes --- pyhon/appliance.py | 2 +- pyhon/attributes.py | 29 +++++++++++++++++++++++++---- 2 files changed, 26 insertions(+), 5 deletions(-) diff --git a/pyhon/appliance.py b/pyhon/appliance.py index bfb16b2..76514ff 100644 --- a/pyhon/appliance.py +++ b/pyhon/appliance.py @@ -330,7 +330,7 @@ class HonAppliance: command: HonCommand = self.commands.get(command_name) for key, value in self.attributes.get("parameters", {}).items(): if isinstance(value, str) and (new := command.parameters.get(key)): - self.attributes["parameters"][key].value = str(new.intern_value) + self.attributes["parameters"][key].update(str(new.intern_value), shield=True) def sync_command(self, main, target=None) -> None: base: HonCommand = self.commands.get(main) diff --git a/pyhon/attributes.py b/pyhon/attributes.py index 3a45c29..55ec0b7 100644 --- a/pyhon/attributes.py +++ b/pyhon/attributes.py @@ -1,17 +1,21 @@ -from datetime import datetime -from typing import Optional +from datetime import datetime, timedelta +from typing import Optional, Final, Dict from pyhon.helper import str_to_float class HonAttribute: + _LOCK_TIMEOUT: Final = 10 + def __init__(self, data): self._value: str = "" self._last_update: Optional[datetime] = None + self._lock_timestamp: Optional[datetime] = None self.update(data) @property def value(self) -> float | str: + """Attribute value""" try: return str_to_float(self._value) except ValueError: @@ -23,15 +27,32 @@ class HonAttribute: @property def last_update(self) -> Optional[datetime]: + """Timestamp of last api update""" return self._last_update - def update(self, data): - self._value = data.get("parNewVal", "") + @property + def lock(self) -> bool: + """Shows if value changes are forbidden""" + if not self._lock_timestamp: + return False + lock_until = self._lock_timestamp + timedelta(seconds=self._LOCK_TIMEOUT) + return lock_until >= datetime.utcnow() + + def update(self, data: Dict[str, str] | str, shield: bool = False) -> bool: + if self.lock and not shield: + return False + if shield: + self._lock_timestamp = datetime.utcnow() + if isinstance(data, str): + self.value = data + return True + self.value = data.get("parNewVal", "") if last_update := data.get("lastUpdate"): try: self._last_update = datetime.fromisoformat(last_update) except ValueError: self._last_update = None + return True def __str__(self) -> str: return self._value From 8c65a37f293212b56b8b0dbc6f619b0af5d701d6 Mon Sep 17 00:00:00 2001 From: Andre Basche Date: Thu, 15 Jun 2023 02:16:03 +0200 Subject: [PATCH 3/3] Add command loader class --- pyhon/appliance.py | 119 +++--------------------- pyhon/command_loader.py | 197 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 211 insertions(+), 105 deletions(-) create mode 100644 pyhon/command_loader.py diff --git a/pyhon/appliance.py b/pyhon/appliance.py index 76514ff..becfbba 100644 --- a/pyhon/appliance.py +++ b/pyhon/appliance.py @@ -1,18 +1,15 @@ import importlib import json import logging -from contextlib import suppress -from copy import copy from datetime import datetime, timedelta from pathlib import Path -from typing import Optional, Dict, Any -from typing import TYPE_CHECKING +from typing import Optional, Dict, Any, TYPE_CHECKING from pyhon import helper 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.fixed import HonParameterFixed from pyhon.parameter.range import HonParameterRange if TYPE_CHECKING: @@ -140,104 +137,12 @@ class HonAppliance: def api(self) -> Optional["HonAPI"]: return self._api - async def _recover_last_command_states(self): - command_history = await self.api.command_history(self) - for name, command in self._commands.items(): - last = next( - ( - index - for (index, d) in enumerate(command_history) - if d.get("command", {}).get("commandName") == name - ), - None, - ) - if last is None: - continue - parameters = command_history[last].get("command", {}).get("parameters", {}) - if command.categories and ( - parameters.get("program") or parameters.get("category") - ): - if parameters.get("program"): - command.category = parameters.pop("program").split(".")[-1].lower() - else: - command.category = parameters.pop("category") - command = self.commands[name] - for key, data in command.settings.items(): - if ( - not isinstance(data, HonParameterFixed) - and parameters.get(key) is not None - ): - with suppress(ValueError): - data.value = parameters.get(key) - - def _get_categories(self, command, data): - categories = {} - for category, value in data.items(): - result = self._get_command(value, command, category, categories) - if result: - if "PROGRAM" in category: - category = category.split(".")[-1].lower() - categories[category] = result[0] - if categories: - if "setParameters" in categories: - return [categories["setParameters"]] - return [list(categories.values())[0]] - return [] - - def _get_commands(self, data): - commands = [] - for command, value in data.items(): - commands += self._get_command(value, command, "") - return {c.name: c for c in commands} - - def _get_command(self, data, command="", category="", categories=None): - commands = [] - if isinstance(data, dict): - if data.get("description") and data.get("protocolType", None): - commands += [ - HonCommand( - command, - data, - self, - category_name=category, - categories=categories, - ) - ] - else: - commands += self._get_categories(command, data) - elif category: - self._additional_data.setdefault(command, {})[category] = data - else: - self._additional_data[command] = data - return commands - - async def load_commands(self): - raw = await self.api.load_commands(self) - self._appliance_model = raw.pop("applianceModel") - raw.pop("dictionaryId", None) - self._commands = self._get_commands(raw) - await self._add_favourites() - await self._recover_last_command_states() - - async def _add_favourites(self): - favourites = await self._api.command_favourites(self) - for favourite in favourites: - name = favourite.get("favouriteName") - command = favourite.get("command") - command_name = command.get("commandName") - program_name = command.get("programName", "").split(".")[-1].lower() - base = copy(self._commands[command_name].categories[program_name]) - for data in command.values(): - if isinstance(data, str): - continue - for key, value in data.items(): - if parameter := base.parameters.get(key): - with suppress(ValueError): - parameter.value = value - extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") - base.parameters.update(favourite=extra_param) - base.parameters["program"].set_value(name) - self._commands[command_name].categories[name] = base + async def load_commands(self, data=None): + command_loader = HonCommandLoader(self.api, self) + await command_loader.load_commands(data) + self._commands = command_loader.commands + self._additional_data = command_loader.additional_data + self._appliance_model = command_loader.appliance_data async def load_attributes(self): self._attributes = await self.api.load_attributes(self) @@ -245,7 +150,9 @@ class HonAppliance: if name in self._attributes.get("parameters", {}): self._attributes["parameters"][name].update(values) else: - self._attributes.setdefault("parameters", {})[name] = HonAttribute(values) + self._attributes.setdefault("parameters", {})[name] = HonAttribute( + values + ) if self._extra: self._attributes = self._extra.attributes(self._attributes) @@ -330,7 +237,9 @@ class HonAppliance: command: HonCommand = self.commands.get(command_name) 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) + self.attributes["parameters"][key].update( + str(new.intern_value), shield=True + ) def sync_command(self, main, target=None) -> None: base: HonCommand = self.commands.get(main) diff --git a/pyhon/command_loader.py b/pyhon/command_loader.py new file mode 100644 index 0000000..7ccb3c5 --- /dev/null +++ b/pyhon/command_loader.py @@ -0,0 +1,197 @@ +import asyncio +import json +from contextlib import suppress +from copy import copy +from typing import Dict, Any, Optional, TYPE_CHECKING, List + +from pyhon.commands import HonCommand +from pyhon.parameter.fixed import HonParameterFixed +from pyhon.parameter.program import HonParameterProgram + +if TYPE_CHECKING: + from pyhon import HonAPI, exceptions + from pyhon.appliance import HonAppliance + + +class HonCommandLoader: + """Loads and parses hOn command data""" + + def __init__(self, api, appliance): + self._api_commands: Dict[str, Any] = {} + self._favourites: List[Dict[str, Any]] = [] + self._command_history: List[Dict[str, Any]] = [] + self._commands: Dict[str, HonCommand] = {} + self._api: "HonAPI" = api + self._appliance: "HonAppliance" = appliance + self._appliance_data: Dict[str, Any] = {} + self._additional_data: Dict[str, Any] = {} + + @property + def api(self) -> "HonAPI": + """api connection object""" + if self._api is None: + raise exceptions.NoAuthenticationException("Missing hOn login") + return self._api + + @property + def appliance(self) -> "HonAppliance": + """appliance object""" + return self._appliance + + @property + def commands(self) -> Dict[str, HonCommand]: + """Get list of hon commands""" + return self._commands + + @property + def appliance_data(self) -> Dict[str, Any]: + """Get command appliance data""" + return self._appliance_data + + @property + def additional_data(self) -> Dict[str, Any]: + """Get command additional data""" + return self._additional_data + + async def load_commands(self, data=None): + """Trigger loading of command data""" + if data: + self._api_commands = data + else: + await self._load_data() + self._appliance_data = self._api_commands.pop("applianceModel") + self._get_commands() + self._add_favourites() + self._recover_last_command_states() + + async def _load_commands(self): + self._api_commands = await self._api.load_commands(self._appliance) + + async def _load_favourites(self): + self._favourites = await self._api.command_favourites(self._appliance) + + async def _load_command_history(self): + self._command_history = await self._api.command_history(self._appliance) + + async def _load_data(self): + """Request parallel all relevant data""" + await asyncio.gather( + *[ + self._load_commands(), + self._load_favourites(), + self._load_command_history(), + ] + ) + + @staticmethod + def _is_command(data: Dict[str, Any]) -> bool: + """Check if dict can be parsed as command""" + return ( + data.get("description") is not None and data.get("protocolType") is not None + ) + + @staticmethod + def _clean_name(category: str) -> str: + """Clean up category name""" + if "PROGRAM" in category: + return category.split(".")[-1].lower() + return category + + def _get_commands(self) -> None: + """Generates HonCommand dict from api data""" + commands = [] + for name, data in self._api_commands.items(): + if command := self._parse_command(data, name): + commands.append(command) + self._commands = {c.name: c for c in commands} + + def _parse_command( + self, data: Dict[str, Any] | str, command_name: str, **kwargs + ) -> 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) + if category := self._parse_categories(data, command_name): + return category + return None + + def _parse_categories( + self, data: Dict[str, Any], command_name: str + ) -> Optional[HonCommand]: + """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): + categories[self._clean_name(category)] = command + if categories: + # setParameters should be at first place + if "setParameters" in categories: + return categories["setParameters"] + return list(categories.values())[0] + return None + + def _get_last_command_index(self, name: str) -> Optional[int]: + """Get index of last command execution""" + return next( + ( + index + for (index, d) in enumerate(self._command_history) + if d.get("command", {}).get("commandName") == name + ), + None, + ) + + def _set_last_category( + self, command: HonCommand, name: str, parameters: Dict[str, Any] + ) -> HonCommand: + """Set category to last state""" + if command.categories: + if program := parameters.pop("program", None): + command.category = self._clean_name(program) + elif category := parameters.pop("category", None): + command.category = category + else: + return command + return self.commands[name] + return command + + def _recover_last_command_states(self) -> None: + """Set commands to last state""" + for name, command in self.commands.items(): + if (last_index := self._get_last_command_index(name)) is None: + continue + last_command = self._command_history[last_index] + parameters = last_command.get("command", {}).get("parameters", {}) + command = self._set_last_category(command, name, parameters) + for key, data in command.settings.items(): + if parameters.get(key) is None: + continue + with suppress(ValueError): + data.value = parameters.get(key) + + def _add_favourites(self) -> None: + """Patch program categories with favourites""" + for favourite in self._favourites: + name = favourite.get("favouriteName", {}) + command = favourite.get("command", {}) + command_name = command.get("commandName", "") + program_name = self._clean_name(command.get("programName", "")) + base: HonCommand = copy( + self.commands[command_name].categories[program_name] + ) + for data in command.values(): + if isinstance(data, str): + continue + for key, value in data.items(): + if parameter := base.parameters.get(key): + with suppress(ValueError): + parameter.value = value + extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") + base.parameters.update(favourite=extra_param) + if isinstance(program := base.parameters["program"], HonParameterProgram): + program.set_value(name) + self.commands[command_name].categories[name] = base