Compare commits

...

12 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
20 changed files with 307 additions and 115 deletions

View File

@ -2,6 +2,7 @@ import importlib
import json import json
import logging import logging
from contextlib import suppress from contextlib import suppress
from copy import copy
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pathlib import Path from pathlib import Path
from typing import Optional, Dict, Any from typing import Optional, Dict, Any
@ -11,6 +12,7 @@ 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
from pyhon.parameter.range import HonParameterRange
if TYPE_CHECKING: if TYPE_CHECKING:
from pyhon import HonAPI from pyhon import HonAPI
@ -213,21 +215,47 @@ class HonAppliance:
self._appliance_model = raw.pop("applianceModel") self._appliance_model = raw.pop("applianceModel")
raw.pop("dictionaryId", None) raw.pop("dictionaryId", None)
self._commands = self._get_commands(raw) self._commands = self._get_commands(raw)
await self._add_favourites()
await self._recover_last_command_states() 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): async def load_attributes(self):
self._attributes = await self.api.load_attributes(self) self._attributes = await self.api.load_attributes(self)
for name, values in self._attributes.pop("shadow").get("parameters").items(): for name, values in self._attributes.pop("shadow").get("parameters").items():
self._attributes.setdefault("parameters", {})[name] = values["parNewVal"] self._attributes.setdefault("parameters", {})[name] = values["parNewVal"]
if self._extra:
self._attributes = self._extra.attributes(self._attributes)
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) self._statistics |= await self.api.load_maintenance(self)
async def update(self): async def update(self, force=False):
now = datetime.now() now = datetime.now()
if not self._last_update or self._last_update < now - timedelta( if (
seconds=self._MINIMAL_UPDATE_INTERVAL force
or not self._last_update
or self._last_update
< now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL)
): ):
self._last_update = now self._last_update = now
await self.load_attributes() await self.load_attributes()
@ -263,9 +291,8 @@ class HonAppliance:
"statistics": self.statistics, "statistics": self.statistics,
"additional_data": self._additional_data, "additional_data": self._additional_data,
**self.command_parameters, **self.command_parameters,
**self.attributes,
} }
if self._extra:
return self._extra.data(result)
return result return result
def diagnose(self, whitespace=" ", command_only=False): def diagnose(self, whitespace=" ", command_only=False):
@ -295,12 +322,38 @@ class HonAppliance:
) )
return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx") 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): class HonApplianceTest(HonAppliance):
def __init__(self, name): def __init__(self, name):
super().__init__(None, {}) super().__init__(None, {})
self._name = name self._name = name
self.load_commands() self.load_commands()
self.load_attributes()
self._info = self._appliance_model self._info = self._appliance_model
def load_commands(self): def load_commands(self):

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

View File

@ -1,17 +1,19 @@
class Appliance: from pyhon.appliances.base import ApplianceBase
def __init__(self, appliance):
self.parent = 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"
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 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,18 +1,14 @@
from pyhon.appliances.base import ApplianceBase
from pyhon.parameter.fixed import HonParameterFixed from pyhon.parameter.fixed import HonParameterFixed
class Appliance: class Appliance(ApplianceBase):
def __init__(self, appliance): def attributes(self, data):
self.parent = appliance data = super().attributes(data)
if data["lastConnEvent"]["category"] == "DISCONNECTED":
def data(self, data): data["parameters"]["machMode"] = "0"
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": data["active"] = bool(data.get("activity"))
data["attributes"]["parameters"]["machMode"] = "0" data["pause"] = data["parameters"]["machMode"] == "3"
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 return data
def settings(self, settings): def settings(self, 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,12 +1,13 @@
class Appliance: from pyhon.appliances.base import ApplianceBase
def __init__(self, appliance):
self.parent = appliance
def data(self, data):
if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": class Appliance(ApplianceBase):
data["attributes"]["parameters"]["machMode"] = "0" def attributes(self, data):
data["active"] = bool(data.get("attributes", {}).get("activity")) data = super().attributes(data)
data["pause"] = data["attributes"]["parameters"]["machMode"] == "3" if data["lastConnEvent"]["category"] == "DISCONNECTED":
data["parameters"]["machMode"] = "0"
data["active"] = bool(data.get("activity"))
data["pause"] = data["parameters"]["machMode"] == "3"
return data return data
def settings(self, settings): def settings(self, settings):

View File

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

View File

@ -2,6 +2,7 @@ import logging
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 import exceptions
from pyhon.exceptions import ApiError
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
@ -48,7 +49,7 @@ class HonCommand:
@property @property
def api(self) -> "HonAPI": def api(self) -> "HonAPI":
if self._api is None: if self._api is None:
raise exceptions.NoAuthenticationException raise exceptions.NoAuthenticationException("Missing hOn login")
return self._api return self._api
@property @property
@ -71,7 +72,7 @@ class HonCommand:
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]]] = {}
for name, parameter in self._parameters.items(): for name, parameter in self._parameters.items():
result.setdefault(parameter.group, {})[name] = parameter.value result.setdefault(parameter.group, {})[name] = parameter.intern_value
return result return result
@property @property
@ -110,9 +111,14 @@ class HonCommand:
async def send(self) -> bool: async def send(self) -> bool:
params = self.parameter_groups.get("parameters", {}) params = self.parameter_groups.get("parameters", {})
ancillary_params = self.parameter_groups.get("ancillaryParameters", {}) ancillary_params = self.parameter_groups.get("ancillaryParameters", {})
return await self.api.send_command( 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 self._appliance, self._name, params, ancillary_params
) )
if not result:
raise ApiError("Can't send command")
return result
@property @property
def categories(self) -> Dict[str, "HonCommand"]: def categories(self) -> Dict[str, "HonCommand"]:

View File

@ -1,6 +1,7 @@
import json import json
import logging import logging
from datetime import datetime from datetime import datetime
from pprint import pformat
from typing import Dict, Optional from typing import Dict, Optional
from aiohttp import ClientSession from aiohttp import ClientSession
@ -188,6 +189,7 @@ class HonAPI:
if json_data.get("payload", {}).get("resultCode") == "0": if json_data.get("payload", {}).get("resultCode") == "0":
return True return True
_LOGGER.error(await response.text()) _LOGGER.error(await response.text())
_LOGGER.error("%s - Payload:\n%s", url, pformat(data))
return False return False
async def appliance_configuration(self) -> Dict: async def appliance_configuration(self) -> Dict:

View File

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

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_URL = "https://api-iot.he.services"
API_KEY = "GRCqFhC6Gk@ikWXm1RmnSmX1cm,MxY-configuration" API_KEY = "GRCqFhC6Gk@ikWXm1RmnSmX1cm,MxY-configuration"
APP = "hon" APP = "hon"
# All seen id's (different accounts, different devices) are the same, so I guess this hash is static # 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" CLIENT_ID = "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9.HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
APP_VERSION = "1.53.7" APP_VERSION = "2.0.10"
OS_VERSION = 31 OS_VERSION = 31
OS = "android" OS = "android"
DEVICE_MODEL = "exynos9820" DEVICE_MODEL = "exynos9820"

View File

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

View File

@ -22,6 +22,15 @@ class HonParameter:
def value(self) -> str | float: def value(self) -> str | float:
return self._value if self._value is not None else "0" 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 @property
def values(self) -> List[str]: def values(self) -> List[str]:
return [str(self.value)] return [str(self.value)]
@ -57,5 +66,18 @@ class HonParameter:
def triggers(self): def triggers(self):
result = {} result = {}
for value, rules in self._triggers.items(): for value, rules in self._triggers.items():
result[value] = {rule.param_key: rule.param_value for _, rule in rules} 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 return result

View File

@ -3,13 +3,17 @@ from typing import Dict, Any, List
from pyhon.parameter.base import HonParameter from pyhon.parameter.base import HonParameter
def clean_value(value):
return str(value).strip("[]").replace("|", "_").lower()
class HonParameterEnum(HonParameter): class HonParameterEnum(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None: def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
super().__init__(key, attributes, group) super().__init__(key, attributes, group)
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.strip("[]")) not in self.values: if self._default and clean_value(self._default.strip("[]")) not in self.values:
self._values.append(self._default) self._values.append(self._default)
def __repr__(self) -> str: def __repr__(self) -> str:
@ -17,15 +21,19 @@ class HonParameterEnum(HonParameter):
@property @property
def values(self) -> List[str]: def values(self) -> List[str]:
return [str(value) for value in self._values] return [clean_value(value) for value in self._values]
@values.setter @values.setter
def values(self, values) -> None: def values(self, values) -> None:
self._values = values 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 @property
def value(self) -> str | float: 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 @value.setter
def value(self, value: str) -> None: def value(self, value: str) -> None:

View File

@ -44,6 +44,11 @@ class HonParameterProgram(HonParameterEnum):
values = { values = {
int(p.parameters["prCode"].value): n int(p.parameters["prCode"].value): n
for i, (n, p) in enumerate(self._programs.items()) for i, (n, p) in enumerate(self._programs.items())
if "iot_" not in n and p.parameters.get("prCode") 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())) return dict(sorted(values.items()))
def set_value(self, value: str):
self._value = value

View File

@ -16,40 +16,50 @@ class HonParameterRange(HonParameter):
self._min: float = str_to_float(attributes["minimumValue"]) self._min: float = str_to_float(attributes["minimumValue"])
self._max: float = str_to_float(attributes["maximumValue"]) self._max: float = str_to_float(attributes["maximumValue"])
self._step: float = str_to_float(attributes["incrementValue"]) 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 self._value: float = self._default
def __repr__(self): def __repr__(self):
return f"{self.__class__} (<{self.key}> [{self._min} - {self._max}])" return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])"
@property @property
def min(self) -> float: def min(self) -> float:
return self._min return self._min
@min.setter
def min(self, mini: float) -> None:
self._min = mini
@property @property
def max(self) -> float: def max(self) -> float:
return self._max return self._max
@max.setter
def max(self, maxi: float) -> None:
self._max = maxi
@property @property
def step(self) -> float: def step(self) -> float:
if not self._step: if not self._step:
return 1 return 1
return self._step return self._step
@step.setter
def step(self, step: float) -> None:
self._step = step
@property @property
def value(self) -> float: def value(self) -> str | float:
return self._value if self._value is not None else self._min return self._value if self._value is not None else self.min
@value.setter @value.setter
def value(self, value: float) -> None: def value(self, value: str | float) -> None:
value = str_to_float(value) value = str_to_float(value)
if self._min <= value <= self._max and not (value - self._min) % self._step: if self.min <= value <= self.max and not (value - self.min) % self.step:
self._value = value self._value = value
self.check_trigger(value) self.check_trigger(value)
else: else:
raise ValueError( raise ValueError(f"Allowed: min {self.min} max {self.max} step {self.step}")
f"Allowed: min {self._min} max {self._max} step {self._step}"
)
@property @property
def values(self) -> List[str]: def values(self) -> List[str]:

View File

@ -1,5 +1,5 @@
from dataclasses import dataclass from dataclasses import dataclass
from typing import List, Dict, TYPE_CHECKING from typing import List, Dict, TYPE_CHECKING, Any, Optional
from pyhon.parameter.enum import HonParameterEnum from pyhon.parameter.enum import HonParameterEnum
from pyhon.parameter.range import HonParameterRange from pyhon.parameter.range import HonParameterRange
@ -13,7 +13,8 @@ class HonRule:
trigger_key: str trigger_key: str
trigger_value: str trigger_value: str
param_key: str param_key: str
param_value: str param_data: Dict[str, Any]
extras: Optional[Dict[str, str]] = None
class HonRuleSet: class HonRuleSet:
@ -23,40 +24,82 @@ class HonRuleSet:
self._parse_rule(rule) self._parse_rule(rule)
def _parse_rule(self, rule): def _parse_rule(self, rule):
for entity_key, params in rule.items(): for param_key, params in rule.items():
entity_key = self._command.appliance.options.get(entity_key, entity_key) param_key = self._command.appliance.options.get(param_key, param_key)
for trigger_key, values in params.items(): for trigger_key, trigger_data in params.items():
trigger_key = trigger_key.replace("@", "") self._parse_conditions(param_key, trigger_key, trigger_data)
trigger_key = self._command.appliance.options.get(
trigger_key, trigger_key def _parse_conditions(self, param_key, trigger_key, trigger_data, extra=None):
) trigger_key = trigger_key.replace("@", "")
for trigger_value, entity_value in values.items(): trigger_key = self._command.appliance.options.get(trigger_key, trigger_key)
if entity_value.get("fixedValue") == f"@{entity_key}": for multi_trigger_value, param_data in trigger_data.items():
continue for trigger_value in multi_trigger_value.split("|"):
self._rules.setdefault(trigger_key, []).append( if isinstance(param_data, dict) and "typology" in param_data:
HonRule( self._create_rule(
trigger_key, param_key, trigger_key, trigger_value, param_data, extra
trigger_value,
entity_key,
entity_value.get("fixedValue"),
)
) )
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): def patch(self):
self._duplicate_for_extra_conditions()
for name, parameter in self._command.parameters.items(): for name, parameter in self._command.parameters.items():
if name not in self._rules: if name not in self._rules:
continue continue
for data in self._rules.get(name): for data in self._rules.get(name):
self._add_trigger(parameter, data)
def apply(rule):
if param := self._command.parameters.get(rule.param_key):
if isinstance(param, HonParameterEnum) and set(
param.values
) != {str(rule.param_value)}:
param.values = [str(rule.param_value)]
elif isinstance(param, HonParameterRange):
param.value = float(rule.param_value)
return
param.value = str(rule.param_value)
parameter.add_trigger(data.trigger_value, apply, data)

View File

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