Compare commits

...

11 Commits

15 changed files with 82 additions and 18 deletions

View File

@ -1,16 +1,21 @@
import importlib
import logging
from contextlib import suppress
from typing import Optional, Dict, Any
from typing import TYPE_CHECKING
from pyhon import helper
from pyhon.commands import HonCommand
from pyhon.parameter.base import HonParameter
from pyhon.parameter.fixed import HonParameterFixed
if TYPE_CHECKING:
from pyhon import HonAPI
_LOGGER = logging.getLogger(__name__)
class HonAppliance:
def __init__(
self, api: Optional["HonAPI"], info: Dict[str, Any], zone: int = 0
@ -29,7 +34,7 @@ class HonAppliance:
try:
self._extra = importlib.import_module(
f"pyhon.appliances.{self.appliance_type.lower()}"
).Appliance()
).Appliance(self)
except ModuleNotFoundError:
self._extra = None
@ -165,7 +170,8 @@ class HonAppliance:
def settings(self):
result = {}
for name, command in self._commands.items():
for key, setting in command.settings.items():
for key in command.setting_keys:
setting = command.settings.get(key, HonParameter(key, {}))
result[f"{name}.{key}"] = setting
if self._extra:
return self._extra.settings(result)
@ -214,4 +220,4 @@ class HonAppliance:
{"commands": helper.create_command(self.commands)},
whitespace="\u200B \u200B ",
)
return result.replace(self.mac_address, "12-34-56-78-90-ab")
return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx")

View File

@ -1,4 +1,7 @@
class Appliance:
def __init__(self, appliance):
self.parent = appliance
def data(self, data):
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
data["attributes"]["parameters"]["machMode"] = "0"

View File

@ -1,4 +1,7 @@
class Appliance:
def __init__(self, appliance):
self.parent = appliance
def data(self, data):
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
data["attributes"]["parameters"]["temp"] = "0"

View File

@ -1,10 +1,22 @@
from pyhon.parameter.fixed import HonParameterFixed
class Appliance:
def __init__(self, appliance):
self.parent = appliance
def data(self, data):
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
data["attributes"]["parameters"]["machMode"] = "0"
data["active"] = bool(data.get("attributes", {}).get("activity"))
data["pause"] = data["attributes"]["parameters"]["machMode"] == "3"
if program := int(data["attributes"]["parameters"]["prCode"]):
ids = self.parent.settings["startProgram.program"].ids
data["programName"] = ids.get(program, "")
return data
def settings(self, settings):
dry_level = settings.get("startProgram.dryLevel")
if isinstance(dry_level, HonParameterFixed) and dry_level.value == "11":
settings.pop("startProgram.dryLevel", None)
return settings

View File

@ -1,4 +1,7 @@
class Appliance:
def __init__(self, appliance):
self.parent = appliance
def data(self, data):
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
data["attributes"]["parameters"]["machMode"] = "0"

View File

@ -1,4 +1,7 @@
class Appliance:
def __init__(self, appliance):
self.parent = appliance
def data(self, data):
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
data["attributes"]["parameters"]["machMode"] = "0"

View File

@ -24,7 +24,7 @@ class HonCommand:
self._api: HonAPI = api
self._appliance: "HonAppliance" = appliance
self._name: str = name
self._programs: Optional[Dict[str, "HonCommand"]] = programs or {}
self._programs: Optional[Dict[str, "HonCommand"]] = programs
self._program_name: str = program_name
self._description: str = attributes.get("description", "")
self._parameters: Dict[str, HonParameter] = self._create_parameters(
@ -89,8 +89,6 @@ class HonCommand:
for key, parameter in (
command._parameters | command._ancillary_parameters
).items():
if isinstance(parameter, HonParameterFixed):
continue
if key not in keys:
keys.append(key)
return keys
@ -108,7 +106,6 @@ class HonCommand:
@property
def settings(self) -> Dict[str, HonParameter]:
"""Parameters with typology enum and range"""
return {
s: param
for s in self.setting_keys

View File

@ -12,7 +12,7 @@ from pyhon.connection.auth import HonAuth
from pyhon.connection.handler.anonym import HonAnonymousConnectionHandler
from pyhon.connection.handler.hon import HonConnectionHandler
_LOGGER = logging.getLogger()
_LOGGER = logging.getLogger(__name__)
class HonAPI:
@ -74,13 +74,15 @@ class HonAPI:
"applianceType": appliance.appliance_type,
"code": appliance.info["code"],
"applianceModelId": appliance.appliance_model_id,
"firmwareId": appliance.info["eepromId"],
"macAddress": appliance.mac_address,
"fwVersion": appliance.info["fwVersion"],
"os": const.OS,
"appVersion": const.APP_VERSION,
"series": appliance.info["series"],
}
if firmware_id := appliance.info.get("eepromId"):
params["firmwareId"] = firmware_id
if firmware_version := appliance.info.get("fwVersion"):
params["fwVersion"] = firmware_version
url: str = f"{const.API_URL}/commands/v1/retrieve"
async with self._hon.get(url, params=params) as response:
result: Dict = (await response.json()).get("payload", {})

View File

@ -182,13 +182,14 @@ class HonAuth:
await self._error_logger(response)
return ""
def _parse_token_data(self, text: str) -> None:
def _parse_token_data(self, text: str) -> bool:
if access_token := re.findall("access_token=(.*?)&", text):
self._access_token = access_token[0]
if refresh_token := re.findall("refresh_token=(.*?)&", text):
self._refresh_token = refresh_token[0]
if id_token := re.findall("id_token=(.*?)&", text):
self._id_token = id_token[0]
return True if access_token and refresh_token and id_token else False
async def _get_token(self, url: str) -> bool:
async with self._request.get(url) as response:
@ -214,7 +215,9 @@ class HonAuth:
if response.status != 200:
await self._error_logger(response)
return False
self._parse_token_data(await response.text())
if not self._parse_token_data(await response.text()):
await self._error_logger(response)
return False
return True
async def _api_auth(self) -> bool:
@ -228,7 +231,10 @@ class HonAuth:
except json.JSONDecodeError:
await self._error_logger(response)
return False
self._cognito_token = json_data["cognitoUser"]["Token"]
self._cognito_token = json_data.get("cognitoUser", {}).get("Token", "")
if not self._cognito_token:
_LOGGER.error(json_data)
raise exceptions.HonAuthenticationError()
return True
async def authenticate(self) -> None:

View File

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

View File

@ -47,7 +47,7 @@ class Hon:
async def _create_appliance(self, appliance_data: Dict[str, Any], zone=0) -> None:
appliance = HonAppliance(self._api, appliance_data, zone=zone)
if appliance.mac_address is None:
if appliance.mac_address == "":
return
await asyncio.gather(
*[

View File

@ -1,4 +1,4 @@
from typing import Dict, Any
from typing import Dict, Any, List
class HonParameter:
@ -17,6 +17,10 @@ class HonParameter:
def value(self) -> str | float:
return self._value if self._value is not None else "0"
@property
def values(self) -> List[str]:
return list(str(self.value))
@property
def category(self) -> str:
return self._category

View File

@ -31,3 +31,12 @@ class HonParameterProgram(HonParameterEnum):
def values(self) -> List[str]:
values = [v for v in self._programs if all(f not in v for f in self._FILTER)]
return sorted(values)
@property
def ids(self):
values = {
int(p.parameters["prCode"].value): n
for i, (n, p) in enumerate(self._programs.items())
if "iot_" not in n and p.parameters.get("prCode")
}
return dict(sorted(values.items()))

View File

@ -1,4 +1,4 @@
from typing import Dict, Any
from typing import Dict, Any, List
from pyhon.parameter.base import HonParameter
@ -47,3 +47,7 @@ class HonParameterRange(HonParameter):
raise ValueError(
f"Allowed: min {self._min} max {self._max} step {self._step}"
)
@property
def values(self) -> List[str]:
return [str(i) for i in range(int(self.min), int(self.max) + 1, int(self.step))]

View File

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