Add rule handling

This commit is contained in:
Andre Basche 2023-05-21 02:25:43 +02:00
parent 9971fe95e2
commit c2765fe953
10 changed files with 133 additions and 10 deletions

View file

@ -106,8 +106,8 @@ class HonAppliance:
return serial_number[:8] if len(serial_number) < 18 else serial_number[:11] return serial_number[:8] if len(serial_number) < 18 else serial_number[:11]
@property @property
def commands_options(self): def options(self):
return self._appliance_model.get("options") return self._appliance_model.get("options", {})
@property @property
def commands(self): def commands(self):
@ -287,7 +287,10 @@ class HonAppliance:
data.get("appliance", {}).pop(sensible, None) data.get("appliance", {}).pop(sensible, None)
result = helper.pretty_print({"data": data}, whitespace=whitespace) result = helper.pretty_print({"data": data}, whitespace=whitespace)
result += helper.pretty_print( result += helper.pretty_print(
{"commands": helper.create_command(self.commands)}, {
"commands": helper.create_command(self.commands),
"rules": helper.create_rules(self.commands),
},
whitespace=whitespace, whitespace=whitespace,
) )
return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx") return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx")

View file

@ -1,3 +1,4 @@
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
@ -6,11 +7,14 @@ from pyhon.parameter.enum import HonParameterEnum
from pyhon.parameter.fixed import HonParameterFixed from pyhon.parameter.fixed import HonParameterFixed
from pyhon.parameter.program import HonParameterProgram from pyhon.parameter.program import HonParameterProgram
from pyhon.parameter.range import HonParameterRange from pyhon.parameter.range import HonParameterRange
from pyhon.rules import HonRuleSet
if TYPE_CHECKING: if TYPE_CHECKING:
from pyhon import HonAPI from pyhon import HonAPI
from pyhon.appliance import HonAppliance from pyhon.appliance import HonAppliance
_LOGGER = logging.getLogger(__name__)
class HonCommand: class HonCommand:
def __init__( def __init__(
@ -31,6 +35,7 @@ class HonCommand:
self._parameters: Dict[str, HonParameter] = {} self._parameters: Dict[str, HonParameter] = {}
self._data: Dict[str, Any] = {} self._data: Dict[str, Any] = {}
self._available_settings: Dict[str, HonParameter] = {} self._available_settings: Dict[str, HonParameter] = {}
self._rules: List[HonRuleSet] = []
self._load_parameters(attributes) self._load_parameters(attributes)
def __repr__(self) -> str: def __repr__(self) -> str:
@ -46,6 +51,10 @@ class HonCommand:
raise exceptions.NoAuthenticationException raise exceptions.NoAuthenticationException
return self._api return self._api
@property
def appliance(self) -> "HonAppliance":
return self._appliance
@property @property
def data(self): def data(self):
return self._data return self._data
@ -73,10 +82,17 @@ class HonCommand:
for key, items in attributes.items(): for key, items in attributes.items():
for name, data in items.items(): for name, data in items.items():
self._create_parameters(data, name, key) self._create_parameters(data, name, key)
for rule in self._rules:
rule.patch()
def _create_parameters(self, data: Dict, name: str, parameter: str) -> None: def _create_parameters(self, data: Dict, name: str, parameter: str) -> None:
if name == "zoneMap" and self._appliance.zone: if name == "zoneMap" and self._appliance.zone:
data["default"] = 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"): match data.get("typology"):
case "range": case "range":
self._parameters[name] = HonParameterRange(name, data, parameter) self._parameters[name] = HonParameterRange(name, data, parameter)

View file

@ -171,7 +171,7 @@ class HonAPI:
"timestamp": f"{now[:-3]}Z", "timestamp": f"{now[:-3]}Z",
"commandName": command, "commandName": command,
"transactionId": f"{appliance.mac_address}_{now[:-3]}Z", "transactionId": f"{appliance.mac_address}_{now[:-3]}Z",
"applianceOptions": appliance.commands_options, "applianceOptions": appliance.options,
"device": self._hon.device.get(mobile=True), "device": self._hon.device.get(mobile=True),
"attributes": { "attributes": {
"channel": "mobileApp", "channel": "mobileApp",

View file

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

View file

@ -1,4 +1,7 @@
from typing import Dict, Any, List from typing import Dict, Any, List, Tuple, Callable, TYPE_CHECKING
if TYPE_CHECKING:
from pyhon.rules import HonRule
class HonParameter: class HonParameter:
@ -9,6 +12,7 @@ class HonParameter:
self._mandatory: int = attributes.get("mandatory", 0) self._mandatory: int = attributes.get("mandatory", 0)
self._value: str | float = "" self._value: str | float = ""
self._group: str = group self._group: str = group
self._triggers: Dict[str, List[Tuple[Callable, "HonRule"]]] = {}
@property @property
def key(self) -> str: def key(self) -> str:
@ -37,3 +41,21 @@ class HonParameter:
@property @property
def group(self) -> str: def group(self) -> str:
return self._group 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():
result[value] = {rule.param_key: rule.param_value for _, rule in rules}
return result

View file

@ -19,6 +19,10 @@ class HonParameterEnum(HonParameter):
def values(self) -> List[str]: def values(self) -> List[str]:
return [str(value) for value in self._values] return [str(value) for value in self._values]
@values.setter
def values(self, values) -> None:
self._values = values
@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 self._value if self._value is not None else self.values[0]
@ -27,5 +31,6 @@ class HonParameterEnum(HonParameter):
def value(self, value: str) -> None: def value(self, value: str) -> None:
if value in self.values: if value in self.values:
self._value = value self._value = value
self.check_trigger(value)
else: else:
raise ValueError(f"Allowed values {self._value}") raise ValueError(f"Allowed values {self._values}")

View file

@ -19,3 +19,4 @@ class HonParameterFixed(HonParameter):
def value(self, value: str | float) -> None: def value(self, value: str | float) -> None:
# Fixed values seems being not so fixed as thought # Fixed values seems being not so fixed as thought
self._value = value self._value = value
self.check_trigger(value)

View file

@ -35,8 +35,12 @@ class HonParameterProgram(HonParameterEnum):
values = [v for v in self._programs if all(f not in v for f in self._FILTER)] values = [v for v in self._programs if all(f not in v for f in self._FILTER)]
return sorted(values) return sorted(values)
@values.setter
def values(self, values) -> None:
return
@property @property
def ids(self): def ids(self) -> Dict[int, str]:
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())

View file

@ -45,6 +45,7 @@ class HonParameterRange(HonParameter):
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)
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}"

59
pyhon/rules.py Normal file
View file

@ -0,0 +1,59 @@
from dataclasses import dataclass
from typing import List, Dict, TYPE_CHECKING
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_value: str
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 entity_key, params in rule.items():
entity_key = self._command.appliance.options.get(entity_key, entity_key)
for trigger_key, values in params.items():
trigger_key = self._command.appliance.options.get(
trigger_key, trigger_key
)
for trigger_value, entity_value in values.items():
self._rules.setdefault(trigger_key, []).append(
HonRule(
trigger_key,
trigger_value,
entity_key,
entity_value.get("fixedValue"),
)
)
def patch(self):
for name, parameter in self._command.parameters.items():
if name not in self._rules:
continue
for data in self._rules.get(name):
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)