Merge branch 'refactor2'

This commit is contained in:
Andre Basche 2023-06-22 00:03:07 +02:00
commit 66cb7bcc24
11 changed files with 291 additions and 122 deletions

View file

@ -1,17 +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:
@ -61,7 +59,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):
@ -139,109 +137,22 @@ 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)
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 +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] = str(new.intern_value)
self.attributes["parameters"][key].update(
str(new.intern_value), shield=True
)
def sync_command(self, main, target=None) -> None:
base: Optional[HonCommand] = self.commands.get(main)

View file

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

View file

@ -5,6 +5,6 @@ class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["machMode"] = "0"
data["parameters"]["machMode"].value = "0"
data["active"] = bool(data.get("activity"))
return data

View file

@ -5,10 +5,10 @@ class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data.get("lastConnEvent", {}).get("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"

View file

@ -6,7 +6,7 @@ class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data.get("lastConnEvent", {}).get("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

View file

@ -5,7 +5,7 @@ class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data.get("lastConnEvent", {}).get("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

View file

@ -5,7 +5,7 @@ class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data.get("lastConnEvent", {}).get("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

58
pyhon/attributes.py Normal file
View file

@ -0,0 +1,58 @@
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:
return self._value
@value.setter
def value(self, value) -> None:
self._value = value
@property
def last_update(self) -> Optional[datetime]:
"""Timestamp of last api update"""
return self._last_update
@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

197
pyhon/command_loader.py Normal file
View file

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

View file

@ -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(",", "."))

View file

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