Compare commits
4 Commits
Author | SHA1 | Date | |
---|---|---|---|
5db13a90e7 | |||
9ee5dbc956 | |||
d4c6ccdce3 | |||
9594b9ebd8 |
0
pyhon/__main__.py
Normal file → Executable file
0
pyhon/__main__.py
Normal file → Executable 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")
|
||||
|
@ -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"
|
||||
|
@ -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"
|
||||
|
@ -2,15 +2,21 @@ 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["startProgram.dryLevel"]
|
||||
dry_level = settings.get("startProgram.dryLevel")
|
||||
if isinstance(dry_level, HonParameterFixed) and dry_level.value == "11":
|
||||
settings.pop("startProgram.dryLevel", None)
|
||||
return settings
|
||||
|
@ -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"
|
||||
|
@ -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"
|
||||
|
@ -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(
|
||||
|
@ -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:
|
||||
@ -75,13 +75,14 @@ class HonAPI:
|
||||
"code": appliance.info["code"],
|
||||
"applianceModelId": appliance.appliance_model_id,
|
||||
"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", {})
|
||||
|
@ -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":
|
||||
|
@ -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(
|
||||
*[
|
||||
|
@ -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
|
||||
|
@ -1,4 +1,4 @@
|
||||
from typing import Dict, Any, List
|
||||
from typing import Dict, Any
|
||||
|
||||
from pyhon.parameter.base import HonParameter
|
||||
|
||||
@ -19,7 +19,3 @@ class HonParameterFixed(HonParameter):
|
||||
def value(self, value: str | float) -> None:
|
||||
# Fixed values seems being not so fixed as thought
|
||||
self._value = value
|
||||
|
||||
@property
|
||||
def values(self) -> List[str]:
|
||||
return list(str(self.value))
|
||||
|
@ -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()))
|
||||
|
Reference in New Issue
Block a user