Compare commits

...

3 Commits

Author SHA1 Message Date
ccff32e6c1 Fix missing program 2023-05-07 00:47:08 +02:00
22cbd7474a Fix performance issues 2023-05-07 00:28:24 +02:00
dd61b24eed Improve output 2023-05-06 20:06:51 +02:00
6 changed files with 97 additions and 62 deletions

View File

@ -78,12 +78,7 @@ async def main():
) )
) )
else: else:
print(helper.pretty_print({"data": device.data})) print(device.diagnose(" "))
print(
helper.pretty_print(
{"settings": helper.create_command(device.commands)}
)
)
def start(): def start():

View File

@ -1,12 +1,12 @@
import importlib import importlib
import logging import logging
from contextlib import suppress from contextlib import suppress
from datetime import datetime, timedelta
from typing import Optional, Dict, Any from typing import Optional, Dict, Any
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from pyhon import helper from pyhon import helper, exceptions
from pyhon.commands import HonCommand from pyhon.commands import HonCommand
from pyhon.parameter.base import HonParameter
from pyhon.parameter.fixed import HonParameterFixed from pyhon.parameter.fixed import HonParameterFixed
if TYPE_CHECKING: if TYPE_CHECKING:
@ -16,6 +16,8 @@ _LOGGER = logging.getLogger(__name__)
class HonAppliance: class HonAppliance:
_MINIMAL_UPDATE_INTERVAL = 5 # seconds
def __init__( def __init__(
self, api: Optional["HonAPI"], info: Dict[str, Any], zone: int = 0 self, api: Optional["HonAPI"], info: Dict[str, Any], zone: int = 0
) -> None: ) -> None:
@ -29,7 +31,8 @@ class HonAppliance:
self._statistics: Dict = {} self._statistics: Dict = {}
self._attributes: Dict = {} self._attributes: Dict = {}
self._zone: int = zone self._zone: int = zone
self._additional_data = {} self._additional_data: Dict[str, Any] = {}
self._last_update = None
try: try:
self._extra = importlib.import_module( self._extra = importlib.import_module(
@ -119,8 +122,14 @@ class HonAppliance:
def zone(self) -> int: def zone(self) -> int:
return self._zone return self._zone
@property
def api(self) -> "HonAPI":
if self._api is None:
raise exceptions.NoAuthenticationException
return self._api
async def _recover_last_command_states(self): async def _recover_last_command_states(self):
command_history = await self._api.command_history(self) command_history = await self.api.command_history(self)
for name, command in self._commands.items(): for name, command in self._commands.items():
last = next( last = next(
( (
@ -133,8 +142,11 @@ class HonAppliance:
if last is None: if last is None:
continue continue
parameters = command_history[last].get("command", {}).get("parameters", {}) parameters = command_history[last].get("command", {}).get("parameters", {})
if command.categories and parameters.get("category"): if command.categories:
command.category = parameters.pop("category").split(".")[-1].lower() if parameters.get("program"):
command.category = parameters.pop("program").split(".")[-1].lower()
else:
command.category = parameters.pop("category")
command = self.commands[name] command = self.commands[name]
for key, data in command.settings.items(): for key, data in command.settings.items():
if ( if (
@ -170,7 +182,6 @@ class HonAppliance:
HonCommand( HonCommand(
command, command,
data, data,
self._api,
self, self,
category_name=category, category_name=category,
categories=categories, categories=categories,
@ -178,47 +189,58 @@ class HonAppliance:
] ]
else: else:
commands += self._get_categories(command, data) commands += self._get_categories(command, data)
elif category:
self._additional_data.setdefault(command, {})[category] = data
else: else:
self._additional_data[command] = data self._additional_data[command] = data
return commands return commands
async def load_commands(self): async def load_commands(self):
raw = await self._api.load_commands(self) raw = await self.api.load_commands(self)
self._appliance_model = raw.pop("applianceModel") self._appliance_model = raw.pop("applianceModel")
raw.pop("dictionaryId") raw.pop("dictionaryId")
self._commands = self._get_commands(raw) self._commands = self._get_commands(raw)
await self._recover_last_command_states() await self._recover_last_command_states()
async def load_attributes(self): async def load_attributes(self):
self._attributes = await self._api.load_attributes(self) self._attributes = await self.api.load_attributes(self)
for name, values in self._attributes.pop("shadow").get("parameters").items(): for name, values in self._attributes.pop("shadow").get("parameters").items():
self._attributes.setdefault("parameters", {})[name] = values["parNewVal"] self._attributes.setdefault("parameters", {})[name] = values["parNewVal"]
async def load_statistics(self): async def load_statistics(self):
self._statistics = await self._api.load_statistics(self) self._statistics = await self.api.load_statistics(self)
async def update(self): async def update(self):
await self.load_attributes() now = datetime.now()
if not self._last_update or self._last_update < now - timedelta(
seconds=self._MINIMAL_UPDATE_INTERVAL
):
self._last_update = now
await self.load_attributes()
@property @property
def parameters(self): def command_parameters(self):
result = {} return {n: c.parameter_value for n, c in self._commands.items()}
for name, command in self._commands.items():
for key, parameter in command.parameters.items():
result.setdefault(name, {})[key] = parameter.value
return result
@property @property
def settings(self): def settings(self):
result = {} result = {}
for name, command in self._commands.items(): for name, command in self._commands.items():
for key in command.setting_keys: for key in command.setting_keys:
setting = command.settings.get(key, HonParameter(key, {}, name)) setting = command.settings.get(key)
result[f"{name}.{key}"] = setting result[f"{name}.{key}"] = setting
if self._extra: if self._extra:
return self._extra.settings(result) return self._extra.settings(result)
return result return result
@property
def available_settings(self):
result = []
for name, command in self._commands.items():
for key in command.setting_keys:
result.append(f"{name}.{key}")
return result
@property @property
def data(self): def data(self):
result = { result = {
@ -226,20 +248,27 @@ class HonAppliance:
"appliance": self.info, "appliance": self.info,
"statistics": self.statistics, "statistics": self.statistics,
"additional_data": self._additional_data, "additional_data": self._additional_data,
**self.parameters, **self.command_parameters,
} }
if self._extra: if self._extra:
return self._extra.data(result) return self._extra.data(result)
return result return result
@property def diagnose(self, whitespace="\u200B \u200B "):
def diagnose(self): data = {
data = self.data.copy() "attributes": self.attributes.copy(),
"appliance": self.info,
"additional_data": self._additional_data,
}
data |= {n: c.parameter_groups for n, c in self._commands.items()}
extra = {n: c.data for n, c in self._commands.items() if c.data}
if extra:
data |= {"extra_command_data": extra}
for sensible in ["PK", "SK", "serialNumber", "code", "coords"]: for sensible in ["PK", "SK", "serialNumber", "code", "coords"]:
data["appliance"].pop(sensible, None) data["appliance"].pop(sensible, None)
result = helper.pretty_print({"data": self.data}, whitespace="\u200B \u200B ") result = helper.pretty_print({"data": data}, whitespace=whitespace)
result += helper.pretty_print( result += helper.pretty_print(
{"commands": helper.create_command(self.commands)}, {"commands": helper.create_command(self.commands)},
whitespace="\u200B \u200B ", whitespace=whitespace,
) )
return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx") return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx")

View File

@ -1,4 +1,4 @@
from typing import Optional, Dict, Any, List, TYPE_CHECKING from typing import Optional, Dict, Any, List, TYPE_CHECKING, Union
from pyhon.parameter.base import HonParameter from pyhon.parameter.base import HonParameter
from pyhon.parameter.enum import HonParameterEnum from pyhon.parameter.enum import HonParameterEnum
@ -16,12 +16,11 @@ class HonCommand:
self, self,
name: str, name: str,
attributes: Dict[str, Any], attributes: Dict[str, Any],
api: "HonAPI",
appliance: "HonAppliance", appliance: "HonAppliance",
categories: Optional[Dict[str, "HonCommand"]] = None, categories: Optional[Dict[str, "HonCommand"]] = None,
category_name: str = "", category_name: str = "",
): ):
self._api: HonAPI = api self._api: HonAPI = appliance.api
self._appliance: "HonAppliance" = appliance self._appliance: "HonAppliance" = appliance
self._name: str = name self._name: str = name
self._categories: Optional[Dict[str, "HonCommand"]] = categories self._categories: Optional[Dict[str, "HonCommand"]] = categories
@ -29,7 +28,8 @@ class HonCommand:
self._description: str = attributes.pop("description", "") self._description: str = attributes.pop("description", "")
self._protocol_type: str = attributes.pop("protocolType", "") self._protocol_type: str = attributes.pop("protocolType", "")
self._parameters: Dict[str, HonParameter] = {} self._parameters: Dict[str, HonParameter] = {}
self._data = {} self._data: Dict[str, Any] = {}
self._available_settings: Dict[str, HonParameter] = {}
self._load_parameters(attributes) self._load_parameters(attributes)
def __repr__(self) -> str: def __repr__(self) -> str:
@ -47,6 +47,21 @@ class HonCommand:
def parameters(self) -> Dict[str, HonParameter]: def parameters(self) -> Dict[str, HonParameter]:
return self._parameters return self._parameters
@property
def settings(self) -> Dict[str, HonParameter]:
return self._parameters
@property
def parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]:
result: Dict[str, Dict[str, Union[str, float]]] = {}
for name, parameter in self._parameters.items():
result.setdefault(parameter.group, {})[name] = parameter.value
return result
@property
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):
for key, items in attributes.items(): for key, items in attributes.items():
for name, data in items.items(): for name, data in items.items():
@ -66,17 +81,12 @@ class HonCommand:
self._data[name] = data self._data[name] = data
return return
if self._category_name: if self._category_name:
if not self._categories: name = "program" if "PROGRAM" in self._category_name else "category"
self._parameters["program"] = HonParameterProgram("program", self, name) self._parameters[name] = HonParameterProgram(name, self, "custom")
def _parameters_by_group(self, group):
return {
name: v.value for name, v in self._parameters.items() if v.group == group
}
async def send(self) -> bool: async def send(self) -> bool:
params = self._parameters_by_group("parameters") params = self.parameter_groups["parameters"]
ancillary_params = self._parameters_by_group("ancillary_parameters") ancillary_params = self.parameter_groups["ancillary_parameters"]
return await self._api.send_command( return await self._api.send_command(
self._appliance, self._name, params, ancillary_params self._appliance, self._name, params, ancillary_params
) )
@ -101,12 +111,22 @@ class HonCommand:
{param for cmd in self.categories.values() for param in cmd.parameters} {param for cmd in self.categories.values() for param in cmd.parameters}
) )
@staticmethod
def _more_options(first: HonParameter, second: HonParameter):
if isinstance(first, HonParameterFixed) and not isinstance(
second, HonParameterFixed
):
return second
if len(second.values) > len(first.values):
return second
return first
@property @property
def settings(self) -> Dict[str, HonParameter]: def available_settings(self) -> Dict[str, HonParameter]:
result = {} result: Dict[str, HonParameter] = {}
for command in self.categories.values(): for command in self.categories.values():
for name, parameter in command.parameters.items(): for name, parameter in command.parameters.items():
if name in result: if name in result:
continue result[name] = self._more_options(result[name], parameter)
result[name] = parameter result[name] = parameter
return result return result

View File

@ -1,6 +1,3 @@
from pyhon.parameter.base import HonParameter
def key_print(data, key="", start=True): def key_print(data, key="", start=True):
result = "" result = ""
if isinstance(data, list): if isinstance(data, list):
@ -47,21 +44,12 @@ def pretty_print(data, key="", intend=0, is_list=False, whitespace=" "):
return result return result
def get_parameter(command, parameter):
if programs := command.categories:
for program in programs.values():
if data := program.settings.get(parameter):
return data
return command.settings.get(parameter)
def create_command(commands, concat=False): def create_command(commands, concat=False):
result = {} result = {}
for name, command in commands.items(): for name, command in commands.items():
if not concat: if not concat:
result[name] = {} result[name] = {}
for parameter in command.setting_keys: for parameter, data in command.available_settings.items():
data = get_parameter(command, parameter)
if data.typology == "enum": if data.typology == "enum":
value = data.values value = data.values
elif data.typology == "range": elif data.typology == "range":

View File

@ -12,7 +12,10 @@ class HonParameterProgram(HonParameterEnum):
def __init__(self, key: str, command: "HonCommand", group: str) -> None: def __init__(self, key: str, command: "HonCommand", group: str) -> None:
super().__init__(key, {}, group) super().__init__(key, {}, group)
self._command = command self._command = command
self._value: str = command.category if "PROGRAM" in command.category:
self._value: str = command.category.split(".")[-1].lower()
else:
self._value: str = command.category
self._programs: Dict[str, "HonCommand"] = command.categories self._programs: Dict[str, "HonCommand"] = command.categories
self._typology: str = "enum" self._typology: str = "enum"
@ -23,7 +26,7 @@ class HonParameterProgram(HonParameterEnum):
@value.setter @value.setter
def value(self, value: str) -> None: def value(self, value: str) -> None:
if value in self.values: if value in self.values:
self._command.program = value self._command.category = value
else: else:
raise ValueError(f"Allowed values {self.values}") raise ValueError(f"Allowed values {self.values}")

View File

@ -7,7 +7,7 @@ with open("README.md", "r") as f:
setup( setup(
name="pyhOn", name="pyhOn",
version="0.10.0b0", version="0.10.2",
author="Andre Basche", author="Andre Basche",
description="Control hOn devices with python", description="Control hOn devices with python",
long_description=long_description, long_description=long_description,