Compare commits
	
		
			5 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| eea79e28b9 | |||
| ecbf438889 | |||
| 9cd12e3234 | |||
| c2765fe953 | |||
| 9971fe95e2 | 
| @ -99,8 +99,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): | ||||
| @ -280,7 +287,10 @@ class HonAppliance: | ||||
|             data.get("appliance", {}).pop(sensible, None) | ||||
|         result = helper.pretty_print({"data": data}, whitespace=whitespace) | ||||
|         result += helper.pretty_print( | ||||
|             {"commands": helper.create_command(self.commands)}, | ||||
|             { | ||||
|                 "commands": helper.create_command(self.commands), | ||||
|                 "rules": helper.create_rules(self.commands), | ||||
|             }, | ||||
|             whitespace=whitespace, | ||||
|         ) | ||||
|         return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx") | ||||
|  | ||||
| @ -1,3 +1,4 @@ | ||||
| import logging | ||||
| from typing import Optional, Dict, Any, List, TYPE_CHECKING, Union | ||||
|  | ||||
| from pyhon import exceptions | ||||
| @ -6,11 +7,14 @@ 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__( | ||||
| @ -31,6 +35,7 @@ class HonCommand: | ||||
|         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: | ||||
| @ -46,6 +51,10 @@ class HonCommand: | ||||
|             raise exceptions.NoAuthenticationException | ||||
|         return self._api | ||||
|  | ||||
|     @property | ||||
|     def appliance(self) -> "HonAppliance": | ||||
|         return self._appliance | ||||
|  | ||||
|     @property | ||||
|     def data(self): | ||||
|         return self._data | ||||
| @ -62,7 +71,7 @@ class HonCommand: | ||||
|     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.value | ||||
|             result.setdefault(parameter.group, {})[name] = parameter.intern_value | ||||
|         return result | ||||
|  | ||||
|     @property | ||||
| @ -73,10 +82,17 @@ class HonCommand: | ||||
|         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) | ||||
|  | ||||
| @ -76,18 +76,19 @@ class HonAPI: | ||||
|             "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 code := appliance.info.get("code"): | ||||
|             params["code"] = code | ||||
|         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 | ||||
|  | ||||
| @ -170,7 +171,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", | ||||
|  | ||||
| @ -47,8 +47,6 @@ 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.available_settings.items(): | ||||
|             if data.typology == "enum": | ||||
|                 value = data.values | ||||
| @ -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 | ||||
|  | ||||
| @ -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: | ||||
| @ -9,6 +12,7 @@ class HonParameter: | ||||
|         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: | ||||
| @ -18,6 +22,10 @@ class HonParameter: | ||||
|     def value(self) -> str | float: | ||||
|         return self._value if self._value is not None else "0" | ||||
|  | ||||
|     @property | ||||
|     def intern_value(self) -> str | float: | ||||
|         return str(self._value) if self._value is not None else "" | ||||
|  | ||||
|     @property | ||||
|     def values(self) -> List[str]: | ||||
|         return [str(self.value)] | ||||
| @ -37,3 +45,21 @@ class HonParameter: | ||||
|     @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(): | ||||
|             result[value] = {rule.param_key: rule.param_value for _, rule in rules} | ||||
|         return result | ||||
|  | ||||
| @ -3,13 +3,17 @@ 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], 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 str(self._default) not in self.values: | ||||
|         if self._default and clean_value(self._default.strip("[]")) not in self.values: | ||||
|             self._values.append(self._default) | ||||
|  | ||||
|     def __repr__(self) -> str: | ||||
| @ -17,15 +21,20 @@ class HonParameterEnum(HonParameter): | ||||
|  | ||||
|     @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 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}") | ||||
|  | ||||
| @ -19,3 +19,4 @@ class HonParameterFixed(HonParameter): | ||||
|     def value(self, value: str | float) -> None: | ||||
|         # Fixed values seems being not so fixed as thought | ||||
|         self._value = value | ||||
|         self.check_trigger(value) | ||||
|  | ||||
| @ -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)] | ||||
|         return sorted(values) | ||||
|  | ||||
|     @values.setter | ||||
|     def values(self, values) -> None: | ||||
|         return | ||||
|  | ||||
|     @property | ||||
|     def ids(self): | ||||
|     def ids(self) -> Dict[int, str]: | ||||
|         values = { | ||||
|             int(p.parameters["prCode"].value): n | ||||
|             for i, (n, p) in enumerate(self._programs.items()) | ||||
|  | ||||
| @ -45,6 +45,7 @@ class HonParameterRange(HonParameter): | ||||
|         value = str_to_float(value) | ||||
|         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}" | ||||
|  | ||||
							
								
								
									
										62
									
								
								pyhon/rules.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										62
									
								
								pyhon/rules.py
									
									
									
									
									
										Normal file
									
								
							| @ -0,0 +1,62 @@ | ||||
| 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 = trigger_key.replace("@", "") | ||||
|                 trigger_key = self._command.appliance.options.get( | ||||
|                     trigger_key, trigger_key | ||||
|                 ) | ||||
|                 for trigger_value, entity_value in values.items(): | ||||
|                     if entity_value.get("fixedValue") == f"@{entity_key}": | ||||
|                         continue | ||||
|                     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) | ||||
		Reference in New Issue
	
	Block a user
	