pyhOn/pyhon/appliance.py

372 lines
13 KiB
Python
Raw Permalink Normal View History

2023-03-05 18:46:51 +01:00
import importlib
2023-05-11 00:43:48 +02:00
import json
2023-04-23 20:14:52 +02:00
import logging
from contextlib import suppress
2023-05-28 19:24:02 +02:00
from copy import copy
2023-05-07 00:28:24 +02:00
from datetime import datetime, timedelta
2023-05-11 00:43:48 +02:00
from pathlib import Path
2023-04-15 15:55:22 +02:00
from typing import Optional, Dict, Any
from typing import TYPE_CHECKING
2023-03-05 18:46:51 +01:00
2023-05-11 00:43:48 +02:00
from pyhon import helper
2023-02-13 03:36:09 +01:00
from pyhon.commands import HonCommand
2023-05-08 02:23:48 +02:00
from pyhon.parameter.base import HonParameter
2023-04-16 01:43:37 +02:00
from pyhon.parameter.fixed import HonParameterFixed
2023-05-28 17:34:32 +02:00
from pyhon.parameter.range import HonParameterRange
2023-02-13 01:41:38 +01:00
2023-04-15 15:55:22 +02:00
if TYPE_CHECKING:
from pyhon import HonAPI
2023-04-23 20:14:52 +02:00
_LOGGER = logging.getLogger(__name__)
2023-04-09 18:13:50 +02:00
class HonAppliance:
2023-05-07 00:28:24 +02:00
_MINIMAL_UPDATE_INTERVAL = 5 # seconds
2023-04-15 15:55:22 +02:00
def __init__(
self, api: Optional["HonAPI"], info: Dict[str, Any], zone: int = 0
) -> None:
2023-04-09 20:50:28 +02:00
if attributes := info.get("attributes"):
info["attributes"] = {v["parName"]: v["parValue"] for v in attributes}
2023-04-15 15:55:22 +02:00
self._info: Dict = info
self._api: Optional[HonAPI] = api
self._appliance_model: Dict = {}
2023-02-13 01:41:38 +01:00
2023-04-15 15:55:22 +02:00
self._commands: Dict = {}
self._statistics: Dict = {}
self._attributes: Dict = {}
2023-04-15 22:25:34 +02:00
self._zone: int = zone
2023-05-06 20:00:13 +02:00
self._additional_data: Dict[str, Any] = {}
2023-05-07 00:28:24 +02:00
self._last_update = None
2023-05-08 02:23:48 +02:00
self._default_setting = HonParameter("", {}, "")
2023-02-13 01:41:38 +01:00
2023-03-08 00:58:25 +01:00
try:
2023-04-09 20:55:36 +02:00
self._extra = importlib.import_module(
f"pyhon.appliances.{self.appliance_type.lower()}"
2023-04-23 21:42:44 +02:00
).Appliance(self)
2023-03-08 00:58:25 +01:00
except ModuleNotFoundError:
self._extra = None
def __getitem__(self, item):
2023-04-15 04:12:38 +02:00
if self._zone:
item += f"Z{self._zone}"
2023-03-08 00:58:25 +01:00
if "." in item:
result = self.data
for key in item.split("."):
2023-04-15 04:12:38 +02:00
if all(k in "0123456789" for k in key) and isinstance(result, list):
2023-03-08 00:58:25 +01:00
result = result[int(key)]
else:
result = result[key]
return result
2023-04-15 04:12:38 +02:00
if item in self.data:
return self.data[item]
if item in self.attributes["parameters"]:
return self.attributes["parameters"].get(item)
return self.info[item]
2023-02-13 01:41:38 +01:00
2023-03-08 21:53:53 +01:00
def get(self, item, default=None):
try:
return self[item]
2023-03-08 22:18:44 +01:00
except (KeyError, IndexError):
2023-03-08 21:53:53 +01:00
return default
2023-04-15 04:12:38 +02:00
def _check_name_zone(self, name: str, frontend: bool = True) -> str:
middle = " Z" if frontend else "_z"
if (attribute := self._info.get(name, "")) and self._zone:
return f"{attribute}{middle}{self._zone}"
return attribute
2023-02-13 01:41:38 +01:00
@property
2023-04-15 04:12:38 +02:00
def appliance_model_id(self) -> str:
2023-04-15 15:55:22 +02:00
return self._info.get("applianceModelId", "")
2023-02-13 01:41:38 +01:00
@property
2023-04-15 04:12:38 +02:00
def appliance_type(self) -> str:
2023-04-15 15:55:22 +02:00
return self._info.get("applianceTypeName", "")
2023-02-13 01:41:38 +01:00
@property
2023-04-15 04:12:38 +02:00
def mac_address(self) -> str:
2023-04-15 21:58:20 +02:00
return self.info.get("macAddress", "")
@property
def unique_id(self) -> str:
2023-04-15 04:12:38 +02:00
return self._check_name_zone("macAddress", frontend=False)
2023-02-13 01:41:38 +01:00
@property
2023-04-15 04:12:38 +02:00
def model_name(self) -> str:
return self._check_name_zone("modelName")
2023-02-13 01:41:38 +01:00
@property
2023-04-15 04:12:38 +02:00
def nick_name(self) -> str:
return self._check_name_zone("nickName")
2023-02-13 01:41:38 +01:00
2023-05-20 13:24:24 +02:00
@property
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]
2023-02-13 01:41:38 +01:00
@property
2023-05-21 02:25:43 +02:00
def options(self):
return self._appliance_model.get("options", {})
2023-02-13 01:41:38 +01:00
@property
def commands(self):
return self._commands
@property
def attributes(self):
2023-03-04 21:27:10 +01:00
return self._attributes
2023-02-13 01:41:38 +01:00
@property
def statistics(self):
return self._statistics
2023-03-05 18:46:51 +01:00
@property
2023-04-09 20:50:28 +02:00
def info(self):
return self._info
2023-03-05 18:46:51 +01:00
2023-05-06 16:07:28 +02:00
@property
def additional_data(self):
return self._additional_data
2023-04-15 22:25:34 +02:00
@property
def zone(self) -> int:
return self._zone
2023-05-06 20:00:13 +02:00
@property
2023-05-11 00:43:48 +02:00
def api(self) -> Optional["HonAPI"]:
2023-05-06 20:00:13 +02:00
return self._api
2023-05-06 16:07:28 +02:00
async def _recover_last_command_states(self):
2023-05-06 20:00:13 +02:00
command_history = await self.api.command_history(self)
2023-05-06 16:07:28 +02:00
for name, command in self._commands.items():
2023-04-09 20:55:36 +02:00
last = next(
(
index
for (index, d) in enumerate(command_history)
if d.get("command", {}).get("commandName") == name
),
None,
)
2023-03-11 02:31:56 +01:00
if last is None:
continue
parameters = command_history[last].get("command", {}).get("parameters", {})
2023-05-07 17:39:22 +02:00
if command.categories and (
parameters.get("program") or parameters.get("category")
):
2023-05-07 00:47:08 +02:00
if parameters.get("program"):
command.category = parameters.pop("program").split(".")[-1].lower()
else:
command.category = parameters.pop("category")
2023-03-11 02:31:56 +01:00
command = self.commands[name]
for key, data in command.settings.items():
2023-04-09 20:55:36 +02:00
if (
not isinstance(data, HonParameterFixed)
and parameters.get(key) is not None
):
with suppress(ValueError):
data.value = parameters.get(key)
2023-03-11 02:31:56 +01:00
2023-05-06 16:07:28 +02:00
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:
2023-05-15 19:29:42 +02:00
if "setParameters" in categories:
return [categories["setParameters"]]
return [list(categories.values())[0]]
2023-05-06 16:07:28 +02:00
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(
2023-04-16 00:11:50 +02:00
command,
2023-05-06 16:07:28 +02:00
data,
2023-04-16 00:11:50 +02:00
self,
2023-05-06 16:07:28 +02:00
category_name=category,
categories=categories,
2023-04-16 00:11:50 +02:00
)
2023-05-06 16:07:28 +02:00
]
else:
commands += self._get_categories(command, data)
2023-05-06 20:00:13 +02:00
elif category:
self._additional_data.setdefault(command, {})[category] = data
2023-05-06 16:07:28 +02:00
else:
self._additional_data[command] = data
return commands
2023-02-19 19:43:41 +01:00
2023-05-06 16:07:28 +02:00
async def load_commands(self):
2023-05-06 20:00:13 +02:00
raw = await self.api.load_commands(self)
2023-05-06 16:07:28 +02:00
self._appliance_model = raw.pop("applianceModel")
2023-05-08 00:03:29 +02:00
raw.pop("dictionaryId", None)
2023-05-06 16:07:28 +02:00
self._commands = self._get_commands(raw)
2023-05-28 19:24:02 +02:00
await self._add_favourites()
2023-05-06 16:07:28 +02:00
await self._recover_last_command_states()
2023-02-18 22:25:51 +01:00
2023-05-28 19:24:02 +02:00
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])
2023-05-29 18:58:50 +02:00
for data in command.values():
2023-05-28 19:24:02 +02:00
if isinstance(data, str):
continue
for key, value in data.items():
if parameter := base.parameters.get(key):
with suppress(ValueError):
parameter.value = value
2023-05-29 18:58:50 +02:00
extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom")
base.parameters.update(favourite=extra_param)
2023-05-28 19:24:02 +02:00
base.parameters["program"].set_value(name)
self._commands[command_name].categories[name] = base
2023-02-13 01:41:38 +01:00
async def load_attributes(self):
2023-05-06 20:00:13 +02:00
self._attributes = await self.api.load_attributes(self)
2023-04-24 04:33:00 +02:00
for name, values in self._attributes.pop("shadow").get("parameters").items():
2023-03-08 00:58:25 +01:00
self._attributes.setdefault("parameters", {})[name] = values["parNewVal"]
2023-02-13 01:41:38 +01:00
async def load_statistics(self):
2023-05-06 20:00:13 +02:00
self._statistics = await self.api.load_statistics(self)
2023-05-13 00:16:52 +02:00
self._statistics |= await self.api.load_maintenance(self)
2023-02-13 03:36:09 +01:00
async def update(self):
2023-05-07 00:28:24 +02:00
now = datetime.now()
if not self._last_update or self._last_update < now - timedelta(
seconds=self._MINIMAL_UPDATE_INTERVAL
):
self._last_update = now
await self.load_attributes()
2023-03-03 18:24:19 +01:00
2023-05-06 16:07:28 +02:00
@property
2023-05-06 20:00:13 +02:00
def command_parameters(self):
return {n: c.parameter_value for n, c in self._commands.items()}
2023-05-06 16:07:28 +02:00
@property
def settings(self):
result = {}
for name, command in self._commands.items():
for key in command.setting_keys:
2023-05-08 02:23:48 +02:00
setting = command.settings.get(key, self._default_setting)
2023-05-06 16:07:28 +02:00
result[f"{name}.{key}"] = setting
if self._extra:
return self._extra.settings(result)
return result
2023-05-07 00:28:24 +02:00
@property
def available_settings(self):
result = []
for name, command in self._commands.items():
for key in command.setting_keys:
result.append(f"{name}.{key}")
return result
2023-03-03 18:24:19 +01:00
@property
def data(self):
2023-04-09 20:55:36 +02:00
result = {
"attributes": self.attributes,
"appliance": self.info,
"statistics": self.statistics,
2023-05-06 16:07:28 +02:00
"additional_data": self._additional_data,
2023-05-06 20:00:13 +02:00
**self.command_parameters,
2023-04-09 20:55:36 +02:00
}
2023-03-08 00:58:25 +01:00
if self._extra:
2023-04-08 04:06:36 +02:00
return self._extra.data(result)
2023-03-08 00:58:25 +01:00
return result
2023-04-11 22:14:36 +02:00
2023-05-13 00:16:52 +02:00
def diagnose(self, whitespace=" ", command_only=False):
2023-05-06 20:00:13 +02:00
data = {
"attributes": self.attributes.copy(),
"appliance": self.info,
2023-05-13 00:16:52 +02:00
"statistics": self.statistics,
2023-05-06 20:00:13 +02:00
"additional_data": self._additional_data,
}
2023-05-29 18:58:50 +02:00
if self._extra and data.get("attributes"):
data = self._extra.data(data)
2023-05-08 00:03:29 +02:00
if command_only:
data.pop("attributes")
data.pop("appliance")
2023-05-13 00:16:52 +02:00
data.pop("statistics")
2023-05-06 20:00:13 +02:00
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}
2023-05-19 00:48:08 +02:00
for sensible in ["PK", "SK", "serialNumber", "coords", "device"]:
2023-05-08 00:03:29 +02:00
data.get("appliance", {}).pop(sensible, None)
2023-05-06 20:00:13 +02:00
result = helper.pretty_print({"data": data}, whitespace=whitespace)
2023-04-11 22:14:36 +02:00
result += helper.pretty_print(
2023-05-21 02:25:43 +02:00
{
"commands": helper.create_command(self.commands),
"rules": helper.create_rules(self.commands),
},
2023-05-06 20:00:13 +02:00
whitespace=whitespace,
2023-04-11 22:14:36 +02:00
)
2023-04-24 04:33:00 +02:00
return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx")
2023-05-11 00:43:48 +02:00
2023-05-28 06:17:43 +02:00
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):
2023-05-28 17:34:32 +02:00
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
2023-05-28 06:17:43 +02:00
parameter.value = base_value.value
2023-05-11 00:43:48 +02:00
class HonApplianceTest(HonAppliance):
def __init__(self, name):
super().__init__(None, {})
self._name = name
self.load_commands()
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"