pyhOn/pyhon/commands.py

209 lines
7.4 KiB
Python
Raw Normal View History

2023-05-21 02:25:43 +02:00
import logging
2023-05-06 20:00:13 +02:00
from typing import Optional, Dict, Any, List, TYPE_CHECKING, Union
2023-04-16 01:36:10 +02:00
2023-05-14 23:03:46 +02:00
from pyhon import exceptions
2023-06-21 19:53:36 +02:00
from pyhon.exceptions import ApiError, NoAuthenticationException
2023-04-16 01:43:37 +02:00
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
2023-05-21 02:25:43 +02:00
from pyhon.rules import HonRuleSet
2023-06-28 19:02:11 +02:00
from pyhon.typedefs import Parameter
2023-02-13 01:41:38 +01:00
2023-04-16 01:36:10 +02:00
if TYPE_CHECKING:
2023-05-14 23:03:46 +02:00
from pyhon import HonAPI
2023-04-16 01:36:10 +02:00
from pyhon.appliance import HonAppliance
2023-05-21 02:25:43 +02:00
_LOGGER = logging.getLogger(__name__)
2023-02-13 01:41:38 +01:00
class HonCommand:
2023-04-16 00:11:50 +02:00
def __init__(
2023-04-16 01:36:10 +02:00
self,
name: str,
attributes: Dict[str, Any],
appliance: "HonAppliance",
2023-05-06 16:07:28 +02:00
categories: Optional[Dict[str, "HonCommand"]] = None,
category_name: str = "",
2023-04-16 00:11:50 +02:00
):
2023-07-16 05:53:23 +02:00
self._name: str = name
2023-07-18 21:31:16 +02:00
self._api: Optional[HonAPI] = None
2023-04-16 01:36:10 +02:00
self._appliance: "HonAppliance" = appliance
2023-05-06 16:07:28 +02:00
self._categories: Optional[Dict[str, "HonCommand"]] = categories
self._category_name: str = category_name
2023-07-16 05:53:23 +02:00
self._parameters: Dict[str, Parameter] = {}
2023-05-06 20:00:13 +02:00
self._data: Dict[str, Any] = {}
2023-05-21 02:25:43 +02:00
self._rules: List[HonRuleSet] = []
2023-07-16 05:53:23 +02:00
attributes.pop("description", "")
attributes.pop("protocolType", "")
2023-05-06 16:07:28 +02:00
self._load_parameters(attributes)
2023-02-18 22:25:51 +01:00
2023-04-16 01:36:10 +02:00
def __repr__(self) -> str:
2023-02-18 22:25:51 +01:00
return f"{self._name} command"
2023-02-13 01:41:38 +01:00
2023-05-06 16:07:28 +02:00
@property
2023-06-28 19:02:11 +02:00
def name(self) -> str:
2023-05-06 16:07:28 +02:00
return self._name
2023-05-11 00:43:48 +02:00
@property
def api(self) -> "HonAPI":
2023-07-18 21:31:16 +02:00
if self._api is None and self._appliance:
self._api = self._appliance.api
2023-05-11 00:43:48 +02:00
if self._api is None:
2023-05-28 06:17:43 +02:00
raise exceptions.NoAuthenticationException("Missing hOn login")
2023-05-11 00:43:48 +02:00
return self._api
2023-05-21 02:25:43 +02:00
@property
def appliance(self) -> "HonAppliance":
return self._appliance
2023-05-06 16:07:28 +02:00
@property
2023-06-28 19:02:11 +02:00
def data(self) -> Dict[str, Any]:
2023-05-06 16:07:28 +02:00
return self._data
2023-02-13 01:41:38 +01:00
@property
2023-04-16 01:36:10 +02:00
def parameters(self) -> Dict[str, HonParameter]:
2023-03-06 19:45:46 +01:00
return self._parameters
2023-02-13 01:41:38 +01:00
2023-05-07 00:28:24 +02:00
@property
def settings(self) -> Dict[str, HonParameter]:
return self._parameters
2023-05-06 20:00:13 +02:00
@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():
2023-05-22 01:07:55 +02:00
result.setdefault(parameter.group, {})[name] = parameter.intern_value
2023-05-06 20:00:13 +02:00
return result
@property
def mandatory_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():
if parameter.mandatory:
result.setdefault(parameter.group, {})[name] = parameter.intern_value
return result
2023-05-06 20:00:13 +02:00
@property
def parameter_value(self) -> Dict[str, Union[str, float]]:
return {n: p.value for n, p in self._parameters.items()}
2023-06-28 19:02:11 +02:00
def _load_parameters(self, attributes: Dict[str, Dict[str, Any]]) -> None:
2023-05-06 16:07:28 +02:00
for key, items in attributes.items():
if not isinstance(items, dict):
_LOGGER.info("Loading Attributes - Skipping %s", str(items))
continue
2023-05-06 16:07:28 +02:00
for name, data in items.items():
self._create_parameters(data, name, key)
2023-05-21 02:25:43 +02:00
for rule in self._rules:
rule.patch()
2023-05-06 16:07:28 +02:00
2023-06-28 19:02:11 +02:00
def _create_parameters(
self, data: Dict[str, Any], name: str, parameter: str
) -> None:
2023-05-06 16:07:28 +02:00
if name == "zoneMap" and self._appliance.zone:
data["default"] = self._appliance.zone
2023-05-21 02:25:43 +02:00
if data.get("category") == "rule":
2023-10-02 01:38:40 +02:00
if "fixedValue" in data:
2023-05-21 02:25:43 +02:00
self._rules.append(HonRuleSet(self, data["fixedValue"]))
2023-10-02 01:38:40 +02:00
elif "enumValues" in data:
self._rules.append(HonRuleSet(self, data["enumValues"]))
else:
_LOGGER.warning("Rule not supported: %s", data)
2023-05-06 16:07:28 +02:00
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:
2023-05-07 00:28:24 +02:00
name = "program" if "PROGRAM" in self._category_name else "category"
self._parameters[name] = HonParameterProgram(name, self, "custom")
2023-02-13 01:41:38 +01:00
async def send(self, only_mandatory: bool = False) -> bool:
grouped_params = (
self.mandatory_parameter_groups if only_mandatory else self.parameter_groups
)
params = grouped_params.get("parameters", {})
return await self.send_parameters(params)
async def send_specific(self, param_names: List[str]) -> bool:
params: Dict[str, str | float] = {}
for key, parameter in self._parameters.items():
if key in param_names or parameter.mandatory:
params[key] = parameter.value
return await self.send_parameters(params)
async def send_parameters(self, params: Dict[str, str | float]) -> bool:
2023-05-07 01:17:02 +02:00
ancillary_params = self.parameter_groups.get("ancillaryParameters", {})
2023-06-10 06:40:55 +02:00
ancillary_params.pop("programRules", None)
2023-10-06 01:28:47 +02:00
if "prStr" in params:
params["prStr"] = self._category_name.upper()
2023-07-01 16:04:34 +02:00
self.appliance.sync_command_to_params(self.name)
2023-06-21 19:53:36 +02:00
try:
result = await self.api.send_command(
2023-11-21 02:23:38 +01:00
self._appliance,
self._name,
params,
ancillary_params,
self._category_name,
2023-06-21 19:53:36 +02:00
)
if not result:
_LOGGER.error(result)
raise ApiError("Can't send command")
except NoAuthenticationException:
_LOGGER.error("No Authentication")
return False
2023-06-09 02:09:20 +02:00
return result
2023-02-13 01:41:38 +01:00
2023-04-15 23:02:37 +02:00
@property
2023-05-06 16:07:28 +02:00
def categories(self) -> Dict[str, "HonCommand"]:
if self._categories is None:
return {"_": self}
return self._categories
2023-04-15 23:02:37 +02:00
@property
2023-05-06 16:07:28 +02:00
def category(self) -> str:
return self._category_name
@category.setter
def category(self, category: str) -> None:
2023-05-08 00:03:29 +02:00
if category in self.categories:
self._appliance.commands[self._name] = self.categories[category]
2023-02-18 22:25:51 +01:00
@property
2023-04-16 01:36:10 +02:00
def setting_keys(self) -> List[str]:
2023-05-06 16:07:28 +02:00
return list(
{param for cmd in self.categories.values() for param in cmd.parameters}
)
2023-02-18 22:25:51 +01:00
2023-05-06 20:00:13 +02:00
@staticmethod
2023-06-28 19:02:11 +02:00
def _more_options(first: Parameter, second: Parameter) -> Parameter:
2023-05-06 20:00:13 +02:00
if isinstance(first, HonParameterFixed) and not isinstance(
second, HonParameterFixed
):
return second
if len(second.values) > len(first.values):
return second
return first
2023-02-18 22:25:51 +01:00
@property
2023-06-28 19:02:11 +02:00
def available_settings(self) -> Dict[str, Parameter]:
result: Dict[str, Parameter] = {}
2023-05-06 16:07:28 +02:00
for command in self.categories.values():
for name, parameter in command.parameters.items():
if name in result:
2023-05-06 20:00:13 +02:00
result[name] = self._more_options(result[name], parameter)
2023-05-11 00:43:48 +02:00
else:
result[name] = parameter
2023-05-06 16:07:28 +02:00
return result
2023-07-12 00:05:27 +02:00
2023-07-12 19:36:32 +02:00
def reset(self) -> None:
2023-07-12 00:05:27 +02:00
for parameter in self._parameters.values():
parameter.reset()