Compare commits

...

14 Commits

Author SHA1 Message Date
dfbc24452b Fix value check 2023-05-16 21:26:24 +02:00
af4fbdd8cd Make 'code' attribute optional, fix hon#51 2023-05-16 20:46:46 +02:00
b5af81b744 Improve set parameters 2023-05-15 19:29:42 +02:00
22a98e1781 Add favorites request 2023-05-14 23:04:36 +02:00
2feb3295e1 Fix some dw bugs 2023-05-14 23:03:46 +02:00
d350d639cc Add maintenance cycle 2023-05-13 00:16:52 +02:00
81c202d730 Use test devies 2023-05-11 00:44:18 +02:00
022da71800 Fix none-type attributes 2023-05-08 02:23:48 +02:00
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
9 changed files with 165 additions and 50 deletions

View File

@ -1,10 +1,13 @@
import importlib import importlib
import json
import logging import logging
from contextlib import suppress from contextlib import suppress
from datetime import datetime, timedelta
from pathlib import Path
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
from pyhon.commands import HonCommand from pyhon.commands import HonCommand
from pyhon.parameter.base import HonParameter from pyhon.parameter.base import HonParameter
from pyhon.parameter.fixed import HonParameterFixed from pyhon.parameter.fixed import HonParameterFixed
@ -16,6 +19,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 +35,8 @@ 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
self._default_setting = HonParameter("", {}, "")
try: try:
self._extra = importlib.import_module( self._extra = importlib.import_module(
@ -120,9 +127,7 @@ class HonAppliance:
return self._zone return self._zone
@property @property
def api(self) -> "HonAPI": def api(self) -> Optional["HonAPI"]:
if self._api is None:
raise exceptions.NoAuthenticationException
return self._api return self._api
async def _recover_last_command_states(self): async def _recover_last_command_states(self):
@ -139,8 +144,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 (
@ -159,6 +169,8 @@ class HonAppliance:
category = category.split(".")[-1].lower() category = category.split(".")[-1].lower()
categories[category] = result[0] categories[category] = result[0]
if categories: if categories:
if "setParameters" in categories:
return [categories["setParameters"]]
return [list(categories.values())[0]] return [list(categories.values())[0]]
return [] return []
@ -192,7 +204,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()
@ -203,9 +215,15 @@ class HonAppliance:
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)
self._statistics |= await self.api.load_maintenance(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 +234,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, self._default_setting)
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,21 +261,57 @@ 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=" ", command_only=False):
data = { data = {
"attributes": self.attributes.copy(), "attributes": self.attributes.copy(),
"appliance": self.info, "appliance": self.info,
"statistics": self.statistics,
"additional_data": self._additional_data, "additional_data": self._additional_data,
} }
if command_only:
data.pop("attributes")
data.pop("appliance")
data.pop("statistics")
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", "device"]:
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)},
whitespace=whitespace, 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")
class HonApplianceTest(HonAppliance):
def __init__(self, name):
super().__init__(None, {})
self._name = name
self.load_commands()
self._info = self._appliance_model
def load_commands(self):
device = Path(__file__).parent / "test_data" / f"{self._name}.json"
with open(str(device)) as f:
raw = json.loads(f.read())
self._appliance_model = raw.pop("applianceModel")
raw.pop("dictionaryId", None)
self._commands = self._get_commands(raw)
async def update(self):
return
@property
def nick_name(self) -> str:
return self._name
@property
def unique_id(self) -> str:
return self._name
@property
def mac_address(self) -> str:
return "xx-xx-xx-xx-xx-xx"

View File

@ -1,5 +1,6 @@
from typing import Optional, Dict, Any, List, TYPE_CHECKING, Union from typing import Optional, Dict, Any, List, TYPE_CHECKING, Union
from pyhon import exceptions
from pyhon.parameter.base import HonParameter from pyhon.parameter.base import HonParameter
from pyhon.parameter.enum import HonParameterEnum from pyhon.parameter.enum import HonParameterEnum
from pyhon.parameter.fixed import HonParameterFixed from pyhon.parameter.fixed import HonParameterFixed
@ -20,7 +21,7 @@ class HonCommand:
categories: Optional[Dict[str, "HonCommand"]] = None, categories: Optional[Dict[str, "HonCommand"]] = None,
category_name: str = "", category_name: str = "",
): ):
self._api: HonAPI = appliance.api self._api: Optional[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,6 +30,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:
@ -38,6 +40,12 @@ class HonCommand:
def name(self): def name(self):
return self._name return self._name
@property
def api(self) -> "HonAPI":
if self._api is None:
raise exceptions.NoAuthenticationException
return self._api
@property @property
def data(self): def data(self):
return self._data return self._data
@ -46,6 +54,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,15 +88,13 @@ 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 +110,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,11 +130,12 @@ 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():
if name in result: if name in result:
result[name] = self._more_options(result[name], parameter) result[name] = self._more_options(result[name], parameter)
result[name] = parameter else:
result[name] = parameter
return result return result

View File

@ -83,6 +83,8 @@ class HonAPI:
params["firmwareId"] = firmware_id params["firmwareId"] = firmware_id
if firmware_version := appliance.info.get("fwVersion"): if firmware_version := appliance.info.get("fwVersion"):
params["fwVersion"] = firmware_version params["fwVersion"] = firmware_version
if code := appliance.info.get("code"):
params["code"] = code
url: str = f"{const.API_URL}/commands/v1/retrieve" url: str = f"{const.API_URL}/commands/v1/retrieve"
async with self._hon.get(url, params=params) as response: async with self._hon.get(url, params=params) as response:
result: Dict = (await response.json()).get("payload", {}) result: Dict = (await response.json()).get("payload", {})
@ -100,6 +102,16 @@ class HonAPI:
return {} return {}
return result["payload"]["history"] return result["payload"]["history"]
async def command_favourites(self, appliance: HonAppliance) -> Dict:
url: str = (
f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/favourite"
)
async with self._hon.get(url) as response:
result: Dict = await response.json()
if not result or not result.get("payload"):
return {}
return result["payload"]["favourites"]
async def last_activity(self, appliance: HonAppliance) -> Dict: async def last_activity(self, appliance: HonAppliance) -> Dict:
url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity" url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity"
params: Dict = {"macAddress": appliance.mac_address} params: Dict = {"macAddress": appliance.mac_address}
@ -140,6 +152,12 @@ class HonAPI:
async with self._hon.get(url, params=params) as response: async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {}) return (await response.json()).get("payload", {})
async def load_maintenance(self, appliance: HonAppliance):
url = f"{const.API_URL}/commands/v1/maintenance-cycle"
params = {"macAddress": appliance.mac_address}
async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {})
async def send_command( async def send_command(
self, self,
appliance: HonAppliance, appliance: HonAppliance,

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

@ -1,4 +1,5 @@
import asyncio import asyncio
import logging
from types import TracebackType from types import TracebackType
from typing import List, Optional, Dict, Any, Type from typing import List, Optional, Dict, Any, Type
@ -8,11 +9,18 @@ from typing_extensions import Self
from pyhon import HonAPI, exceptions from pyhon import HonAPI, exceptions
from pyhon.appliance import HonAppliance from pyhon.appliance import HonAppliance
_LOGGER = logging.getLogger(__name__)
class Hon: class Hon:
def __init__(self, email: str, password: str, session: ClientSession | None = None): def __init__(
self._email: str = email self,
self._password: str = password email: Optional[str] = "",
password: Optional[str] = "",
session: Optional[ClientSession] = None,
):
self._email: Optional[str] = email
self._password: Optional[str] = password
self._session: ClientSession | None = session self._session: ClientSession | None = session
self._appliances: List[HonAppliance] = [] self._appliances: List[HonAppliance] = []
self._api: Optional[HonAPI] = None self._api: Optional[HonAPI] = None
@ -34,9 +42,21 @@ class Hon:
raise exceptions.NoAuthenticationException raise exceptions.NoAuthenticationException
return self._api return self._api
@property
def email(self) -> str:
if not self._email:
raise ValueError("Missing email")
return self._email
@property
def password(self) -> str:
if not self._password:
raise ValueError("Missing password")
return self._password
async def create(self) -> Self: async def create(self) -> Self:
self._api = await HonAPI( self._api = await HonAPI(
self._email, self._password, session=self._session self.email, self.password, session=self._session
).create() ).create()
await self.setup() await self.setup()
return self return self
@ -45,17 +65,25 @@ class Hon:
def appliances(self) -> List[HonAppliance]: def appliances(self) -> List[HonAppliance]:
return self._appliances return self._appliances
@appliances.setter
def appliances(self, appliances) -> None:
self._appliances = appliances
async def _create_appliance(self, appliance_data: Dict[str, Any], zone=0) -> None: async def _create_appliance(self, appliance_data: Dict[str, Any], zone=0) -> None:
appliance = HonAppliance(self._api, appliance_data, zone=zone) appliance = HonAppliance(self._api, appliance_data, zone=zone)
if appliance.mac_address == "": if appliance.mac_address == "":
return return
await asyncio.gather( try:
*[ await asyncio.gather(
appliance.load_attributes(), *[
appliance.load_commands(), appliance.load_attributes(),
appliance.load_statistics(), appliance.load_commands(),
] appliance.load_statistics(),
) ]
)
except (KeyError, ValueError, IndexError) as error:
_LOGGER.exception(error)
_LOGGER.error(f"Device data - %s", appliance_data)
self._appliances.append(appliance) self._appliances.append(appliance)
async def setup(self) -> None: async def setup(self) -> None:

View File

@ -9,6 +9,8 @@ class HonParameterEnum(HonParameter):
self._default = attributes.get("defaultValue") self._default = attributes.get("defaultValue")
self._value = self._default or "0" self._value = self._default or "0"
self._values: List[str] = attributes.get("enumValues", []) self._values: List[str] = attributes.get("enumValues", [])
if self._default and str(self._default) not in self.values:
self._values.append(self._default)
def __repr__(self) -> str: def __repr__(self) -> str:
return f"{self.__class__} (<{self.key}> {self.values})" return f"{self.__class__} (<{self.key}> {self.values})"

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
@ -41,7 +43,7 @@ class HonParameterRange(HonParameter):
@value.setter @value.setter
def value(self, value: float) -> None: def value(self, value: float) -> None:
value = str_to_float(value) value = str_to_float(value)
if self._min <= value <= self._max and not value % self._step: if self._min <= value <= self._max and not (value - self._min) % self._step:
self._value = value self._value = value
else: else:
raise ValueError( raise ValueError(

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.8",
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,