Compare commits

..

6 Commits

Author SHA1 Message Date
8e16b4a215 Fix some errors 2023-05-08 00:03:53 +02:00
7bd3aac7c5 Fix error for no category 2023-05-07 17:39:22 +02:00
365a37b42d Fix send command error 2023-05-07 01:17:02 +02:00
2bde6bb61c Fix mypy error 2023-05-07 00:48:42 +02:00
ccff32e6c1 Fix missing program 2023-05-07 00:47:08 +02:00
22cbd7474a Fix performance issues 2023-05-07 00:28:24 +02:00
6 changed files with 52 additions and 31 deletions

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, exceptions 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:
@ -30,6 +32,7 @@ class HonAppliance:
self._attributes: Dict = {} self._attributes: Dict = {}
self._zone: int = zone self._zone: int = zone
self._additional_data: Dict[str, Any] = {} self._additional_data: Dict[str, Any] = {}
self._last_update = None
try: try:
self._extra = importlib.import_module( self._extra = importlib.import_module(
@ -139,8 +142,13 @@ 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 and (
command.category = parameters.pop("category").split(".")[-1].lower() 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] command = self.commands[name]
for key, data in command.settings.items(): for key, data in command.settings.items():
if ( if (
@ -192,7 +200,7 @@ class HonAppliance:
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", None)
self._commands = self._get_commands(raw) self._commands = self._get_commands(raw)
await self._recover_last_command_states() await self._recover_last_command_states()
@ -205,7 +213,12 @@ class HonAppliance:
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 command_parameters(self): def command_parameters(self):
@ -216,12 +229,20 @@ class HonAppliance:
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 = {
@ -235,18 +256,21 @@ class HonAppliance:
return self._extra.data(result) return self._extra.data(result)
return result return result
def diagnose(self, whitespace="\u200B \u200B "): def diagnose(self, whitespace="\u200B \u200B ", command_only=False):
data = { data = {
"attributes": self.attributes.copy(), "attributes": self.attributes.copy(),
"appliance": self.info, "appliance": self.info,
"additional_data": self._additional_data, "additional_data": self._additional_data,
} }
if command_only:
data.pop("attributes")
data.pop("appliance")
data |= {n: c.parameter_groups for n, c in self._commands.items()} 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} extra = {n: c.data for n, c in self._commands.items() if c.data}
if extra: if extra:
data |= {"extra_command_data": 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.get("appliance", {}).pop(sensible, None)
result = helper.pretty_print({"data": data}, whitespace=whitespace) 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)},

View File

@ -29,6 +29,7 @@ class HonCommand:
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: Dict[str, Any] = {} 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:
@ -46,6 +47,10 @@ 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 @property
def parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]: def parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]:
result: Dict[str, Dict[str, Union[str, float]]] = {} result: Dict[str, Dict[str, Union[str, float]]] = {}
@ -76,14 +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( self._parameters[name] = HonParameterProgram(name, self, "custom")
"program", self, "custom"
)
async def send(self) -> bool: async def send(self) -> bool:
params = self.parameter_groups["parameters"] params = self.parameter_groups.get("parameters", {})
ancillary_params = self.parameter_groups["ancillary_parameters"] ancillary_params = self.parameter_groups.get("ancillaryParameters", {})
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
) )
@ -100,7 +103,8 @@ class HonCommand:
@category.setter @category.setter
def category(self, category: str) -> None: def category(self, category: str) -> None:
self._appliance.commands[self._name] = self.categories[category] if category in self.categories:
self._appliance.commands[self._name] = self.categories[category]
@property @property
def setting_keys(self) -> List[str]: def setting_keys(self) -> List[str]:
@ -119,7 +123,7 @@ class HonCommand:
return first return first
@property @property
def settings(self) -> Dict[str, HonParameter]: def available_settings(self) -> Dict[str, HonParameter]:
result: Dict[str, HonParameter] = {} 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():

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 = command.category.split(".")[-1].lower()
else:
self._value = command.category
self._programs: Dict[str, "HonCommand"] = command.categories self._programs: Dict[str, "HonCommand"] = command.categories
self._typology: str = "enum" self._typology: str = "enum"

View File

@ -32,6 +32,8 @@ class HonParameterRange(HonParameter):
@property @property
def step(self) -> float: def step(self) -> float:
if not self._step:
return 1
return self._step return self._step
@property @property

View File

@ -7,7 +7,7 @@ with open("README.md", "r") as f:
setup( setup(
name="pyhOn", name="pyhOn",
version="0.10.0", version="0.10.5",
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,