Compare commits

...

37 Commits

Author SHA1 Message Date
310d1bafd7 Improve rule parsing 2023-06-10 06:47:37 +02:00
9e35dcf9cf Don't send optional program rules 2023-06-10 06:40:55 +02:00
f9d0fa4ae8 Add wine cellar 2023-06-09 22:52:33 +02:00
11988c73a6 Fix command parameter issue hon#63 2023-06-09 02:09:20 +02:00
7b51caecca Improve update performance 2023-06-08 19:52:08 +02:00
38d09e2ef5 Fix step is 0 hon#60 2023-06-08 14:20:08 +02:00
3c7ad3f395 Fix changed hOn login 2023-06-07 02:27:02 +02:00
31c03faca8 Get program name 2023-05-29 19:05:37 +02:00
a081ef1f97 Add favourites to progams hon#47 2023-05-28 19:24:02 +02:00
4888f2b1d0 Add oven climate support 2023-05-28 17:41:20 +02:00
7c6ac15901 Add and improve fridge 2023-05-28 07:37:38 +02:00
eea79e28b9 Fix for fridge program names 2023-05-22 01:07:55 +02:00
ecbf438889 Bump version 2023-05-21 20:48:27 +02:00
9cd12e3234 Fixes for fridge 2023-05-21 20:33:08 +02:00
c2765fe953 Add rule handling 2023-05-21 02:25:43 +02:00
9971fe95e2 Get appliance code from serial number 2023-05-20 13:24:24 +02:00
9f130e2e85 Make 'code' attribute really optional 2023-05-19 00:48:08 +02:00
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
dd61b24eed Improve output 2023-05-06 20:06:51 +02:00
ea8f481b01 More general parsing 2023-05-06 16:08:36 +02:00
7dcb34559b Raise error if missing token 2023-04-24 21:50:25 +02:00
5db13a90e7 Fix some stuff for hoover appliances 2023-04-24 04:33:00 +02:00
9ee5dbc956 Get program name by prcode 2023-04-23 21:42:44 +02:00
d4c6ccdce3 Use parameter optional 2023-04-23 20:15:07 +02:00
24 changed files with 783 additions and 236 deletions

7
pyhon/__main__.py Normal file → Executable file
View File

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

View File

@ -1,17 +1,28 @@
import importlib
import json
import logging
from contextlib import suppress
from copy import copy
from datetime import datetime, timedelta
from pathlib import Path
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
from pyhon.parameter.range import HonParameterRange
if TYPE_CHECKING:
from pyhon import HonAPI
_LOGGER = logging.getLogger(__name__)
class HonAppliance:
_MINIMAL_UPDATE_INTERVAL = 5 # seconds
def __init__(
self, api: Optional["HonAPI"], info: Dict[str, Any], zone: int = 0
) -> None:
@ -25,11 +36,14 @@ class HonAppliance:
self._statistics: Dict = {}
self._attributes: Dict = {}
self._zone: int = zone
self._additional_data: Dict[str, Any] = {}
self._last_update = None
self._default_setting = HonParameter("", {}, "")
try:
self._extra = importlib.import_module(
f"pyhon.appliances.{self.appliance_type.lower()}"
).Appliance()
).Appliance(self)
except ModuleNotFoundError:
self._extra = None
@ -87,8 +101,15 @@ class HonAppliance:
return self._check_name_zone("nickName")
@property
def commands_options(self):
return self._appliance_model.get("options")
def code(self) -> str:
if code := self.info.get("code"):
return code
serial_number = self.info.get("serialNumber", "")
return serial_number[:8] if len(serial_number) < 18 else serial_number[:11]
@property
def options(self):
return self._appliance_model.get("options", {})
@property
def commands(self):
@ -106,13 +127,21 @@ class HonAppliance:
def info(self):
return self._info
@property
def additional_data(self):
return self._additional_data
@property
def zone(self) -> int:
return self._zone
async def _recover_last_command_states(self, commands):
command_history = await self._api.command_history(self)
for name, command in commands.items():
@property
def api(self) -> Optional["HonAPI"]:
return self._api
async def _recover_last_command_states(self):
command_history = await self.api.command_history(self)
for name, command in self._commands.items():
last = next(
(
index
@ -124,8 +153,13 @@ class HonAppliance:
if last is None:
continue
parameters = command_history[last].get("command", {}).get("parameters", {})
if command.programs and parameters.get("program"):
command.program = parameters.pop("program").split(".")[-1].lower()
if command.categories and (
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]
for key, data in command.settings.items():
if (
@ -135,83 +169,212 @@ class HonAppliance:
with suppress(ValueError):
data.value = parameters.get(key)
async def load_commands(self):
raw = await self._api.load_commands(self)
self._appliance_model = raw.pop("applianceModel")
for item in ["settings", "options", "dictionaryId"]:
raw.pop(item)
commands = {}
for command, attr in raw.items():
if "parameters" in attr:
commands[command] = HonCommand(command, attr, self._api, self)
elif "parameters" in attr[list(attr)[0]]:
multi = {}
for program, attr2 in attr.items():
program = program.split(".")[-1].lower()
cmd = HonCommand(
def _get_categories(self, command, data):
categories = {}
for category, value in data.items():
result = self._get_command(value, command, category, categories)
if result:
if "PROGRAM" in category:
category = category.split(".")[-1].lower()
categories[category] = result[0]
if categories:
if "setParameters" in categories:
return [categories["setParameters"]]
return [list(categories.values())[0]]
return []
def _get_commands(self, data):
commands = []
for command, value in data.items():
commands += self._get_command(value, command, "")
return {c.name: c for c in commands}
def _get_command(self, data, command="", category="", categories=None):
commands = []
if isinstance(data, dict):
if data.get("description") and data.get("protocolType", None):
commands += [
HonCommand(
command,
attr2,
self._api,
data,
self,
programs=multi,
program_name=program,
category_name=category,
categories=categories,
)
multi[program] = cmd
commands[command] = cmd
self._commands = commands
await self._recover_last_command_states(commands)
]
else:
commands += self._get_categories(command, data)
elif category:
self._additional_data.setdefault(command, {})[category] = data
else:
self._additional_data[command] = data
return commands
async def load_commands(self):
raw = await self.api.load_commands(self)
self._appliance_model = raw.pop("applianceModel")
raw.pop("dictionaryId", None)
self._commands = self._get_commands(raw)
await self._add_favourites()
await self._recover_last_command_states()
async def _add_favourites(self):
favourites = await self._api.command_favourites(self)
for favourite in favourites:
name = favourite.get("favouriteName")
command = favourite.get("command")
command_name = command.get("commandName")
program_name = command.get("programName", "").split(".")[-1].lower()
base = copy(self._commands[command_name].categories[program_name])
for data in command.values():
if isinstance(data, str):
continue
for key, value in data.items():
if parameter := base.parameters.get(key):
with suppress(ValueError):
parameter.value = value
extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom")
base.parameters.update(favourite=extra_param)
base.parameters["program"].set_value(name)
self._commands[command_name].categories[name] = base
async def load_attributes(self):
self._attributes = await self.api.load_attributes(self)
for name, values in self._attributes.pop("shadow").get("parameters").items():
self._attributes.setdefault("parameters", {})[name] = values["parNewVal"]
if self._extra:
self._attributes = self._extra.attributes(self._attributes)
async def load_statistics(self):
self._statistics = await self.api.load_statistics(self)
self._statistics |= await self.api.load_maintenance(self)
async def update(self, force=False):
now = datetime.now()
if (
force
or not self._last_update
or self._last_update
< now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL)
):
self._last_update = now
await self.load_attributes()
@property
def command_parameters(self):
return {n: c.parameter_value for n, c in self._commands.items()}
@property
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, self._default_setting)
result[f"{name}.{key}"] = setting
if self._extra:
return self._extra.settings(result)
return result
@property
def parameters(self):
result = {}
def available_settings(self):
result = []
for name, command in self._commands.items():
for key, parameter in (
command.parameters | command.ancillary_parameters
).items():
result.setdefault(name, {})[key] = parameter.value
for key in command.setting_keys:
result.append(f"{name}.{key}")
return result
async def load_attributes(self):
self._attributes = await self._api.load_attributes(self)
for name, values in self._attributes.pop("shadow").get("parameters").items():
self._attributes.setdefault("parameters", {})[name] = values["parNewVal"]
async def load_statistics(self):
self._statistics = await self._api.load_statistics(self)
async def update(self):
await self.load_attributes()
@property
def data(self):
result = {
"attributes": self.attributes,
"appliance": self.info,
"statistics": self.statistics,
**self.parameters,
"additional_data": self._additional_data,
**self.command_parameters,
**self.attributes,
}
if self._extra:
return self._extra.data(result)
return result
@property
def diagnose(self):
data = self.data.copy()
for sensible in ["PK", "SK", "serialNumber", "code", "coords"]:
data["appliance"].pop(sensible, None)
result = helper.pretty_print({"data": self.data}, whitespace="\u200B \u200B ")
def diagnose(self, whitespace=" ", command_only=False):
data = {
"attributes": self.attributes.copy(),
"appliance": self.info,
"statistics": self.statistics,
"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()}
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", "coords", "device"]:
data.get("appliance", {}).pop(sensible, None)
result = helper.pretty_print({"data": data}, whitespace=whitespace)
result += helper.pretty_print(
{"commands": helper.create_command(self.commands)},
whitespace="\u200B \u200B ",
{
"commands": helper.create_command(self.commands),
"rules": helper.create_rules(self.commands),
},
whitespace=whitespace,
)
return result.replace(self.mac_address, "12-34-56-78-90-ab")
return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx")
def sync_to_params(self, command_name):
command: HonCommand = self.commands.get(command_name)
for key, value in self.attributes.get("parameters", {}).items():
if isinstance(value, str) and (new := command.parameters.get(key)):
self.attributes["parameters"][key] = str(new.intern_value)
def sync_command(self, main, target=None) -> None:
base: HonCommand = self.commands.get(main)
for command, data in self.commands.items():
if command == main or target and command not in target:
continue
for name, parameter in data.parameters.items():
if base_value := base.parameters.get(name):
if isinstance(base_value, HonParameterRange) and isinstance(
parameter, HonParameterRange
):
parameter.max = base_value.max
parameter.min = base_value.min
parameter.step = base_value.step
elif isinstance(parameter, HonParameterRange):
parameter.max = int(base_value.value)
parameter.min = int(base_value.value)
parameter.step = 1
parameter.value = base_value.value
class HonApplianceTest(HonAppliance):
def __init__(self, name):
super().__init__(None, {})
self._name = name
self.load_commands()
self.load_attributes()
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"

15
pyhon/appliances/base.py Normal file
View File

@ -0,0 +1,15 @@
class ApplianceBase:
def __init__(self, appliance):
self.parent = appliance
def attributes(self, data):
program_name = "No Program"
if program := int(data["parameters"].get("prCode", "0")):
if start_cmd := self.parent.settings.get("startProgram.program"):
if ids := start_cmd.ids:
program_name = ids.get(program, program_name)
data["programName"] = program_name
return data
def settings(self, settings):
return settings

View File

@ -1,9 +1,10 @@
class Appliance:
def data(self, data):
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
data["attributes"]["parameters"]["machMode"] = "0"
data["active"] = bool(data.get("attributes", {}).get("activity"))
return data
from pyhon.appliances.base import ApplianceBase
def settings(self, settings):
return settings
class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data["lastConnEvent"]["category"] == "DISCONNECTED":
data["parameters"]["machMode"] = "0"
data["active"] = bool(data.get("activity"))
return data

View File

@ -1,14 +1,19 @@
class Appliance:
def data(self, data):
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED":
data["attributes"]["parameters"]["temp"] = "0"
data["attributes"]["parameters"]["onOffStatus"] = "0"
data["attributes"]["parameters"]["remoteCtrValid"] = "0"
data["attributes"]["parameters"]["remainingTimeMM"] = "0"
from pyhon.appliances.base import ApplianceBase
data["active"] = data["attributes"]["parameters"]["onOffStatus"] == "1"
class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data["lastConnEvent"]["category"] == "DISCONNECTED":
data["parameters"]["temp"] = "0"
data["parameters"]["onOffStatus"] = "0"
data["parameters"]["remoteCtrValid"] = "0"
data["parameters"]["remainingTimeMM"] = "0"
data["active"] = data["parameters"]["onOffStatus"] == "1"
if program := int(data["parameters"]["prCode"]):
ids = self.parent.settings["startProgram.program"].ids
data["programName"] = ids.get(program, "")
return data
def settings(self, settings):
return settings

23
pyhon/appliances/ref.py Normal file
View File

@ -0,0 +1,23 @@
from pyhon.appliances.base import ApplianceBase
class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data["parameters"]["holidayMode"] == "1":
data["modeZ1"] = "holiday"
elif data["parameters"]["intelligenceMode"] == "1":
data["modeZ1"] = "auto_set"
elif data["parameters"]["quickModeZ1"] == "1":
data["modeZ1"] = "super_cool"
else:
data["modeZ1"] = "no_mode"
if data["parameters"]["quickModeZ2"] == "1":
data["modeZ2"] = "super_freeze"
elif data["parameters"]["intelligenceMode"] == "1":
data["modeZ2"] = "auto_set"
else:
data["modeZ2"] = "no_mode"
return data

View File

@ -1,16 +1,18 @@
from pyhon.appliances.base import ApplianceBase
from pyhon.parameter.fixed import HonParameterFixed
class 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"
class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data["lastConnEvent"]["category"] == "DISCONNECTED":
data["parameters"]["machMode"] = "0"
data["active"] = bool(data.get("activity"))
data["pause"] = data["parameters"]["machMode"] == "3"
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

5
pyhon/appliances/wc.py Normal file
View File

@ -0,0 +1,5 @@
from pyhon.appliances.base import ApplianceBase
class Appliance(ApplianceBase):
pass

View File

@ -1,9 +1,13 @@
class 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"
from pyhon.appliances.base import ApplianceBase
class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data["lastConnEvent"]["category"] == "DISCONNECTED":
data["parameters"]["machMode"] = "0"
data["active"] = bool(data.get("activity"))
data["pause"] = data["parameters"]["machMode"] == "3"
return data
def settings(self, settings):

View File

@ -1,9 +1,13 @@
class 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"
from pyhon.appliances.base import ApplianceBase
class Appliance(ApplianceBase):
def attributes(self, data):
data = super().attributes(data)
if data["lastConnEvent"]["category"] == "DISCONNECTED":
data["parameters"]["machMode"] = "0"
data["active"] = bool(data.get("activity"))
data["pause"] = data["parameters"]["machMode"] == "3"
return data
def settings(self, settings):

View File

@ -1,114 +1,163 @@
from typing import Optional, Dict, Any, List, TYPE_CHECKING
import logging
from typing import Optional, Dict, Any, List, TYPE_CHECKING, Union
from pyhon import exceptions
from pyhon.exceptions import ApiError
from pyhon.parameter.base import HonParameter
from pyhon.parameter.enum import HonParameterEnum
from pyhon.parameter.fixed import HonParameterFixed
from pyhon.parameter.program import HonParameterProgram
from pyhon.parameter.range import HonParameterRange
from pyhon.rules import HonRuleSet
if TYPE_CHECKING:
from pyhon import HonAPI
from pyhon.appliance import HonAppliance
_LOGGER = logging.getLogger(__name__)
class HonCommand:
def __init__(
self,
name: str,
attributes: Dict[str, Any],
api: "HonAPI",
appliance: "HonAppliance",
programs: Optional[Dict[str, "HonCommand"]] = None,
program_name: str = "",
categories: Optional[Dict[str, "HonCommand"]] = None,
category_name: str = "",
):
self._api: HonAPI = api
self._api: Optional[HonAPI] = appliance.api
self._appliance: "HonAppliance" = appliance
self._name: str = name
self._programs: Optional[Dict[str, "HonCommand"]] = programs or {}
self._program_name: str = program_name
self._description: str = attributes.get("description", "")
self._parameters: Dict[str, HonParameter] = self._create_parameters(
attributes.get("parameters", {})
)
self._ancillary_parameters: Dict[str, HonParameter] = self._create_parameters(
attributes.get("ancillaryParameters", {})
)
self._categories: Optional[Dict[str, "HonCommand"]] = categories
self._category_name: str = category_name
self._description: str = attributes.pop("description", "")
self._protocol_type: str = attributes.pop("protocolType", "")
self._parameters: Dict[str, HonParameter] = {}
self._data: Dict[str, Any] = {}
self._available_settings: Dict[str, HonParameter] = {}
self._rules: List[HonRuleSet] = []
self._load_parameters(attributes)
def __repr__(self) -> str:
return f"{self._name} command"
def _create_parameters(self, parameters: Dict) -> Dict[str, HonParameter]:
result: Dict[str, HonParameter] = {}
for parameter, attributes in parameters.items():
if parameter == "zoneMap" and self._appliance.zone:
attributes["default"] = self._appliance.zone
match attributes.get("typology"):
case "range":
result[parameter] = HonParameterRange(parameter, attributes)
case "enum":
result[parameter] = HonParameterEnum(parameter, attributes)
case "fixed":
result[parameter] = HonParameterFixed(parameter, attributes)
if self._programs:
result["program"] = HonParameterProgram("program", self)
return result
@property
def name(self):
return self._name
@property
def api(self) -> "HonAPI":
if self._api is None:
raise exceptions.NoAuthenticationException("Missing hOn login")
return self._api
@property
def appliance(self) -> "HonAppliance":
return self._appliance
@property
def data(self):
return self._data
@property
def parameters(self) -> Dict[str, HonParameter]:
return self._parameters
@property
def ancillary_parameters(self) -> Dict[str, HonParameter]:
return self._ancillary_parameters
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.intern_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):
for key, items in attributes.items():
for name, data in items.items():
self._create_parameters(data, name, key)
for rule in self._rules:
rule.patch()
def _create_parameters(self, data: Dict, name: str, parameter: str) -> None:
if name == "zoneMap" and self._appliance.zone:
data["default"] = self._appliance.zone
if data.get("category") == "rule":
if "fixedValue" not in data:
_LOGGER.error("Rule not supported: %s", data)
else:
self._rules.append(HonRuleSet(self, data["fixedValue"]))
match data.get("typology"):
case "range":
self._parameters[name] = HonParameterRange(name, data, parameter)
case "enum":
self._parameters[name] = HonParameterEnum(name, data, parameter)
case "fixed":
self._parameters[name] = HonParameterFixed(name, data, parameter)
case _:
self._data[name] = data
return
if self._category_name:
name = "program" if "PROGRAM" in self._category_name else "category"
self._parameters[name] = HonParameterProgram(name, self, "custom")
async def send(self) -> bool:
params = {k: v.value for k, v in self._parameters.items()}
ancillary_params = {k: v.value for k, v in self._ancillary_parameters.items()}
return await self._api.send_command(
params = self.parameter_groups.get("parameters", {})
ancillary_params = self.parameter_groups.get("ancillaryParameters", {})
ancillary_params.pop("programRules", None)
self.appliance.sync_to_params(self.name)
result = await self.api.send_command(
self._appliance, self._name, params, ancillary_params
)
if not result:
raise ApiError("Can't send command")
return result
@property
def programs(self) -> Dict[str, "HonCommand"]:
if self._programs is None:
return {}
return self._programs
def categories(self) -> Dict[str, "HonCommand"]:
if self._categories is None:
return {"_": self}
return self._categories
@property
def program(self) -> str:
return self._program_name
def category(self) -> str:
return self._category_name
@program.setter
def program(self, program: str) -> None:
self._appliance.commands[self._name] = self.programs[program]
def _get_settings_keys(self, command: Optional["HonCommand"] = None) -> List[str]:
if command is None:
command = self
keys = []
for key, parameter in (
command._parameters | command._ancillary_parameters
).items():
if key not in keys:
keys.append(key)
return keys
@category.setter
def category(self, category: str) -> None:
if category in self.categories:
self._appliance.commands[self._name] = self.categories[category]
@property
def setting_keys(self) -> List[str]:
if not self._programs:
return self._get_settings_keys()
result = [
key
for cmd in self._programs.values()
for key in self._get_settings_keys(cmd)
]
return list(set(result + ["program"]))
return list(
{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
def settings(self) -> Dict[str, HonParameter]:
return {
s: param
for s in self.setting_keys
if (param := self._parameters.get(s)) is not None
or (param := self._ancillary_parameters.get(s)) is not None
}
def available_settings(self) -> Dict[str, HonParameter]:
result: Dict[str, HonParameter] = {}
for command in self.categories.values():
for name, parameter in command.parameters.items():
if name in result:
result[name] = self._more_options(result[name], parameter)
else:
result[name] = parameter
return result

View File

@ -1,6 +1,7 @@
import json
import logging
from datetime import datetime
from pprint import pformat
from typing import Dict, Optional
from aiohttp import ClientSession
@ -12,7 +13,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:
@ -72,21 +73,23 @@ class HonAPI:
async def load_commands(self, appliance: HonAppliance) -> Dict:
params: Dict = {
"applianceType": appliance.appliance_type,
"code": appliance.info["code"],
"applianceModelId": appliance.appliance_model_id,
"macAddress": appliance.mac_address,
"os": const.OS,
"appVersion": const.APP_VERSION,
"series": appliance.info["series"],
"code": appliance.code,
}
if firmware_id := appliance.info.get("eepromId"):
params["firmwareId"] = firmware_id
if firmware_version := appliance.info.get("fwVersion"):
params["fwVersion"] = firmware_version
if series := appliance.info.get("series"):
params["series"] = series
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", {})
if not result or result.pop("resultCode") != "0":
_LOGGER.error(await response.json())
return {}
return result
@ -100,6 +103,16 @@ class HonAPI:
return {}
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:
url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity"
params: Dict = {"macAddress": appliance.mac_address}
@ -109,6 +122,18 @@ class HonAPI:
return activity
return {}
async def appliance_model(self, appliance: HonAppliance) -> Dict:
url: str = f"{const.API_URL}/commands/v1/appliance-model"
params: Dict = {
"code": appliance.info["code"],
"macAddress": appliance.mac_address,
}
async with self._hon.get(url, params=params) as response:
result: Dict = await response.json()
if result and (activity := result.get("attributes")):
return activity
return {}
async def load_attributes(self, appliance: HonAppliance) -> Dict:
params: Dict = {
"macAddress": appliance.mac_address,
@ -128,6 +153,12 @@ class HonAPI:
async with self._hon.get(url, params=params) as response:
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(
self,
appliance: HonAppliance,
@ -141,7 +172,7 @@ class HonAPI:
"timestamp": f"{now[:-3]}Z",
"commandName": command,
"transactionId": f"{appliance.mac_address}_{now[:-3]}Z",
"applianceOptions": appliance.commands_options,
"applianceOptions": appliance.options,
"device": self._hon.device.get(mobile=True),
"attributes": {
"channel": "mobileApp",
@ -158,10 +189,11 @@ class HonAPI:
if json_data.get("payload", {}).get("resultCode") == "0":
return True
_LOGGER.error(await response.text())
_LOGGER.error("%s - Payload:\n%s", url, pformat(data))
return False
async def appliance_configuration(self) -> Dict:
url: str = f"{const.API_URL}/config/v1/appliance-configuration"
url: str = f"{const.API_URL}/config/v1/program-list-rules"
async with self._hon_anonymous.get(url) as response:
result: Dict = await response.json()
if result and (data := result.get("payload")):

View File

@ -134,9 +134,7 @@ class HonAuth:
fw_uid, loaded_str = context[0]
self._login_data.fw_uid = fw_uid
self._login_data.loaded = json.loads(loaded_str)
self._login_data.url = login_url.replace(
"/".join(const.AUTH_API.split("/")[:-1]), ""
)
self._login_data.url = login_url.replace(const.AUTH_API, "")
return True
await self._error_logger(response)
return False
@ -149,8 +147,8 @@ class HonAuth:
"descriptor": "apex://LightningLoginCustomController/ACTION$login",
"callingDescriptor": "markup://c:loginForm",
"params": {
"username": quote(self._login_data.email),
"password": quote(self._login_data.password),
"username": self._login_data.email,
"password": self._login_data.password,
"startUrl": start_url,
},
}
@ -172,7 +170,7 @@ class HonAuth:
async with self._request.post(
const.AUTH_API + "/s/sfsites/aura",
headers={"Content-Type": "application/x-www-form-urlencoded"},
data="&".join(f"{k}={json.dumps(v)}" for k, v in data.items()),
data="&".join(f"{k}={quote(json.dumps(v))}" for k, v in data.items()),
params=params,
) as response:
if response.status == 200:
@ -182,13 +180,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:
@ -209,12 +208,14 @@ class HonAuth:
url_search = re.findall(
"href\\s*=\\s*[\"'](.*?)[\"']", await response.text()
)
url = "/".join(const.AUTH_API.split("/")[:-1]) + url_search[0]
url = const.AUTH_API + url_search[0]
async with self._request.get(url) as response:
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:

View File

@ -1,10 +1,10 @@
AUTH_API = "https://he-accounts.force.com/SmartHome"
AUTH_API = "https://account2.hon-smarthome.com"
API_URL = "https://api-iot.he.services"
API_KEY = "GRCqFhC6Gk@ikWXm1RmnSmX1cm,MxY-configuration"
APP = "hon"
# All seen id's (different accounts, different devices) are the same, so I guess this hash is static
CLIENT_ID = "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9.HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
APP_VERSION = "1.53.7"
APP_VERSION = "2.0.10"
OS_VERSION = 31
OS = "android"
DEVICE_MODEL = "exynos9820"

View File

@ -12,3 +12,7 @@ class NoSessionException(Exception):
class NoAuthenticationException(Exception):
pass
class ApiError(Exception):
pass

View File

@ -47,9 +47,7 @@ def pretty_print(data, key="", intend=0, is_list=False, whitespace=" "):
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, data in command.available_settings.items():
if data.typology == "enum":
value = data.values
elif data.typology == "range":
@ -57,7 +55,21 @@ def create_command(commands, concat=False):
else:
continue
if not concat:
result[name][parameter] = value
result.setdefault(name, {})[parameter] = value
else:
result[f"{name}.{parameter}"] = value
return result
def create_rules(commands, concat=False):
result = {}
for name, command in commands.items():
for parameter, data in command.available_settings.items():
value = data.triggers
if not value:
continue
if not concat:
result.setdefault(name, {})[parameter] = value
else:
result[f"{name}.{parameter}"] = value
return result

View File

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

View File

@ -1,13 +1,18 @@
from typing import Dict, Any
from typing import Dict, Any, List, Tuple, Callable, TYPE_CHECKING
if TYPE_CHECKING:
from pyhon.rules import HonRule
class HonParameter:
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
self._key = key
self._category: str = attributes.get("category", "")
self._typology: str = attributes.get("typology", "")
self._mandatory: int = attributes.get("mandatory", 0)
self._value: str | float = ""
self._group: str = group
self._triggers: Dict[str, List[Tuple[Callable, "HonRule"]]] = {}
@property
def key(self) -> str:
@ -17,6 +22,19 @@ class HonParameter:
def value(self) -> str | float:
return self._value if self._value is not None else "0"
@value.setter
def value(self, value: str | float) -> None:
self._value = value
self.check_trigger(value)
@property
def intern_value(self) -> str:
return str(self.value)
@property
def values(self) -> List[str]:
return [str(self.value)]
@property
def category(self) -> str:
return self._category
@ -28,3 +46,38 @@ class HonParameter:
@property
def mandatory(self) -> int:
return self._mandatory
@property
def group(self) -> str:
return self._group
def add_trigger(self, value, func, data):
if self._value == value:
func(data)
self._triggers.setdefault(value, []).append((func, data))
def check_trigger(self, value) -> None:
if str(value) in self._triggers:
for trigger in self._triggers[str(value)]:
func, args = trigger
func(args)
@property
def triggers(self):
result = {}
for value, rules in self._triggers.items():
for _, rule in rules:
if rule.extras:
param = result.setdefault(value, {})
for extra_key, extra_value in rule.extras.items():
param = param.setdefault(extra_key, {}).setdefault(
extra_value, {}
)
else:
param = result.setdefault(value, {})
if fixed_value := rule.param_data.get("fixedValue"):
param[rule.param_key] = fixed_value
else:
param[rule.param_key] = rule.param_data.get("defaultValue", "")
return result

View File

@ -3,27 +3,42 @@ from typing import Dict, Any, List
from pyhon.parameter.base import HonParameter
def clean_value(value):
return str(value).strip("[]").replace("|", "_").lower()
class HonParameterEnum(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
super().__init__(key, attributes)
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
super().__init__(key, attributes, group)
self._default = attributes.get("defaultValue")
self._value = self._default or "0"
self._values: List[str] = attributes.get("enumValues", [])
if self._default and clean_value(self._default.strip("[]")) not in self.values:
self._values.append(self._default)
def __repr__(self) -> str:
return f"{self.__class__} (<{self.key}> {self.values})"
@property
def values(self) -> List[str]:
return [str(value) for value in self._values]
return [clean_value(value) for value in self._values]
@values.setter
def values(self, values) -> None:
self._values = values
@property
def intern_value(self) -> str:
return str(self._value) if self._value is not None else str(self.values[0])
@property
def value(self) -> str | float:
return self._value if self._value is not None else self.values[0]
return clean_value(self._value) if self._value is not None else self.values[0]
@value.setter
def value(self, value: str) -> None:
if value in self.values:
self._value = value
self.check_trigger(value)
else:
raise ValueError(f"Allowed values {self._value}")
raise ValueError(f"Allowed values {self._values}")

View File

@ -1,11 +1,11 @@
from typing import Dict, Any, List
from typing import Dict, Any
from pyhon.parameter.base import HonParameter
class HonParameterFixed(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
super().__init__(key, attributes)
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
super().__init__(key, attributes, group)
self._value = attributes.get("fixedValue", None)
def __repr__(self) -> str:
@ -19,7 +19,4 @@ 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))
self.check_trigger(value)

View File

@ -9,11 +9,14 @@ if TYPE_CHECKING:
class HonParameterProgram(HonParameterEnum):
_FILTER = ["iot_recipe", "iot_guided"]
def __init__(self, key: str, command: "HonCommand") -> None:
super().__init__(key, {})
def __init__(self, key: str, command: "HonCommand", group: str) -> None:
super().__init__(key, {}, group)
self._command = command
self._value: str = command.program
self._programs: Dict[str, "HonCommand"] = command.programs
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._typology: str = "enum"
@property
@ -23,7 +26,7 @@ class HonParameterProgram(HonParameterEnum):
@value.setter
def value(self, value: str) -> None:
if value in self.values:
self._command.program = value
self._command.category = value
else:
raise ValueError(f"Allowed values {self.values}")
@ -31,3 +34,21 @@ 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)
@values.setter
def values(self, values) -> None:
return
@property
def ids(self) -> Dict[int, str]:
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")
and not ((fav := p.parameters.get("favourite")) and fav.value == "1")
}
return dict(sorted(values.items()))
def set_value(self, value: str):
self._value = value

View File

@ -11,42 +11,55 @@ def str_to_float(string: str | float) -> float:
class HonParameterRange(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any]) -> None:
super().__init__(key, attributes)
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
super().__init__(key, attributes, group)
self._min: float = str_to_float(attributes["minimumValue"])
self._max: float = str_to_float(attributes["maximumValue"])
self._step: float = str_to_float(attributes["incrementValue"])
self._default: float = str_to_float(attributes.get("defaultValue", self._min))
self._default: float = str_to_float(attributes.get("defaultValue", self.min))
self._value: float = self._default
def __repr__(self):
return f"{self.__class__} (<{self.key}> [{self._min} - {self._max}])"
return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])"
@property
def min(self) -> float:
return self._min
@min.setter
def min(self, mini: float) -> None:
self._min = mini
@property
def max(self) -> float:
return self._max
@max.setter
def max(self, maxi: float) -> None:
self._max = maxi
@property
def step(self) -> float:
if not self._step:
return 1
return self._step
@step.setter
def step(self, step: float) -> None:
self._step = step
@property
def value(self) -> float:
return self._value if self._value is not None else self._min
def value(self) -> str | float:
return self._value if self._value is not None else self.min
@value.setter
def value(self, value: float) -> None:
def value(self, value: str | float) -> None:
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.check_trigger(value)
else:
raise ValueError(
f"Allowed: min {self._min} max {self._max} step {self._step}"
)
raise ValueError(f"Allowed: min {self.min} max {self.max} step {self.step}")
@property
def values(self) -> List[str]:

105
pyhon/rules.py Normal file
View File

@ -0,0 +1,105 @@
from dataclasses import dataclass
from typing import List, Dict, TYPE_CHECKING, Any, Optional
from pyhon.parameter.enum import HonParameterEnum
from pyhon.parameter.range import HonParameterRange
if TYPE_CHECKING:
from pyhon.commands import HonCommand
@dataclass
class HonRule:
trigger_key: str
trigger_value: str
param_key: str
param_data: Dict[str, Any]
extras: Optional[Dict[str, str]] = None
class HonRuleSet:
def __init__(self, command: "HonCommand", rule):
self._command: "HonCommand" = command
self._rules: Dict[str, List[HonRule]] = {}
self._parse_rule(rule)
def _parse_rule(self, rule):
for param_key, params in rule.items():
param_key = self._command.appliance.options.get(param_key, param_key)
for trigger_key, trigger_data in params.items():
self._parse_conditions(param_key, trigger_key, trigger_data)
def _parse_conditions(self, param_key, trigger_key, trigger_data, extra=None):
trigger_key = trigger_key.replace("@", "")
trigger_key = self._command.appliance.options.get(trigger_key, trigger_key)
for multi_trigger_value, param_data in trigger_data.items():
for trigger_value in multi_trigger_value.split("|"):
if isinstance(param_data, dict) and "typology" in param_data:
self._create_rule(
param_key, trigger_key, trigger_value, param_data, extra
)
elif isinstance(param_data, dict):
if extra is None:
extra = {}
extra[trigger_key] = trigger_value
for extra_key, extra_data in param_data.items():
self._parse_conditions(param_key, extra_key, extra_data, extra)
def _create_rule(
self, param_key, trigger_key, trigger_value, param_data, extras=None
):
if param_data.get("fixedValue") == f"@{param_key}":
return
self._rules.setdefault(trigger_key, []).append(
HonRule(trigger_key, trigger_value, param_key, param_data, extras)
)
def _duplicate_for_extra_conditions(self):
new = {}
for rules in self._rules.values():
for rule in rules:
if rule.extras is None:
continue
for key, value in rule.extras.items():
extras = rule.extras.copy()
extras.pop(key)
extras[rule.trigger_key] = rule.trigger_value
new.setdefault(key, []).append(
HonRule(key, value, rule.param_key, rule.param_data, extras)
)
for key, rules in new.items():
for rule in rules:
self._rules.setdefault(key, []).append(rule)
def _add_trigger(self, parameter, data):
def apply(rule: HonRule):
if rule.extras is not None:
for key, value in rule.extras.items():
if str(self._command.parameters.get(key)) != str(value):
return
if param := self._command.parameters.get(rule.param_key):
if value := rule.param_data.get("fixedValue", ""):
if isinstance(param, HonParameterEnum) and set(param.values) != {
str(value)
}:
param.values = [str(value)]
elif isinstance(param, HonParameterRange):
param.value = float(value)
return
param.value = str(value)
elif rule.param_data.get("typology") == "enum":
if isinstance(param, HonParameterEnum):
if enum_values := rule.param_data.get("enumValues"):
param.values = enum_values.split("|")
if default_value := rule.param_data.get("defaultValue"):
param.value = default_value
parameter.add_trigger(data.trigger_value, apply, data)
def patch(self):
self._duplicate_for_extra_conditions()
for name, parameter in self._command.parameters.items():
if name not in self._rules:
continue
for data in self._rules.get(name):
self._add_trigger(parameter, data)

View File

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