pyhOn/pyhon/appliance.py

264 lines
8.9 KiB
Python
Raw Permalink Normal View History

2023-03-05 18:46:51 +01:00
import importlib
2023-04-23 20:14:52 +02:00
import logging
2023-06-28 19:42:21 +02:00
import re
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-06-28 19:02:11 +02:00
from typing import Optional, Dict, Any, TYPE_CHECKING, List
2023-03-05 18:46:51 +01:00
2023-06-28 19:02:11 +02:00
from pyhon import diagnose, exceptions
2023-06-13 00:12:29 +02:00
from pyhon.attributes import HonAttribute
2023-06-15 02:16:03 +02:00
from pyhon.command_loader import HonCommandLoader
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-05-28 17:34:32 +02:00
from pyhon.parameter.range import HonParameterRange
2023-06-28 19:02:11 +02:00
from pyhon.typedefs import Parameter
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-06-28 19:02:11 +02:00
self._info: Dict[str, Any] = info
2023-04-15 15:55:22 +02:00
self._api: Optional[HonAPI] = api
2023-06-28 19:02:11 +02:00
self._appliance_model: Dict[str, Any] = {}
2023-02-13 01:41:38 +01:00
2023-06-21 19:53:36 +02:00
self._commands: Dict[str, HonCommand] = {}
2023-06-28 19:02:11 +02:00
self._statistics: Dict[str, Any] = {}
self._attributes: Dict[str, Any] = {}
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-06-28 19:02:11 +02:00
self._last_update: Optional[datetime] = 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
2023-06-28 19:02:11 +02:00
def __getitem__(self, item: str) -> Any:
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"]:
2023-06-13 00:12:29 +02:00
return self.attributes["parameters"][item].value
2023-04-15 04:12:38 +02:00
return self.info[item]
2023-02-13 01:41:38 +01:00
2023-06-28 19:02:11 +02:00
def get(self, item: str, default: Any = None) -> Any:
2023-03-08 21:53:53 +01:00
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:
2023-06-28 19:42:21 +02:00
zone = " Z" if frontend else "_z"
2023-04-15 04:12:38 +02:00
if (attribute := self._info.get(name, "")) and self._zone:
2023-06-28 19:42:21 +02:00
return f"{attribute}{zone}{self._zone}"
2023-04-15 04:12:38 +02:00
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-06-28 19:42:21 +02:00
default_mac = "xx-xx-xx-xx-xx-xx"
import_name = f"{self.appliance_type.lower()}_{self.appliance_model_id}"
result = self._check_name_zone("macAddress", frontend=False)
result = result.replace(default_mac, import_name)
return result
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
2023-06-25 17:29:04 +02:00
@property
def brand(self) -> str:
return self._check_name_zone("brand")
2023-02-13 01:41:38 +01:00
@property
2023-04-15 04:12:38 +02:00
def nick_name(self) -> str:
2023-06-28 19:42:21 +02:00
result = self._check_name_zone("nickName")
if not result or re.findall("^[xX\s]+$", result):
return self.model_name
return result
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-06-25 17:29:04 +02:00
@property
def model_id(self) -> int:
return self._info.get("applianceModelId", 0)
2023-02-13 01:41:38 +01:00
@property
2023-06-28 19:02:11 +02:00
def options(self) -> Dict[str, Any]:
2023-05-21 02:25:43 +02:00
return self._appliance_model.get("options", {})
2023-02-13 01:41:38 +01:00
@property
2023-06-21 19:53:36 +02:00
def commands(self) -> Dict[str, HonCommand]:
2023-02-13 01:41:38 +01:00
return self._commands
@property
2023-06-28 19:02:11 +02:00
def attributes(self) -> Dict[str, Any]:
2023-03-04 21:27:10 +01:00
return self._attributes
2023-02-13 01:41:38 +01:00
@property
2023-06-28 19:02:11 +02:00
def statistics(self) -> Dict[str, Any]:
2023-02-13 01:41:38 +01:00
return self._statistics
2023-03-05 18:46:51 +01:00
@property
2023-06-28 19:02:11 +02:00
def info(self) -> Dict[str, Any]:
2023-04-09 20:50:28 +02:00
return self._info
2023-03-05 18:46:51 +01:00
2023-05-06 16:07:28 +02:00
@property
2023-06-28 19:02:11 +02:00
def additional_data(self) -> Dict[str, Any]:
2023-05-06 16:07:28 +02:00
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-06-28 19:02:11 +02:00
def api(self) -> "HonAPI":
"""api connection object"""
if self._api is None:
raise exceptions.NoAuthenticationException("Missing hOn login")
2023-05-06 20:00:13 +02:00
return self._api
2023-06-28 19:02:11 +02:00
async def load_commands(self) -> None:
2023-06-15 02:16:03 +02:00
command_loader = HonCommandLoader(self.api, self)
2023-06-25 17:29:04 +02:00
await command_loader.load_commands()
2023-06-15 02:16:03 +02:00
self._commands = command_loader.commands
self._additional_data = command_loader.additional_data
self._appliance_model = command_loader.appliance_data
2023-05-28 19:24:02 +02:00
2023-06-28 19:02:11 +02:00
async def load_attributes(self) -> None:
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-06-13 00:12:29 +02:00
if name in self._attributes.get("parameters", {}):
self._attributes["parameters"][name].update(values)
else:
2023-06-15 02:16:03 +02:00
self._attributes.setdefault("parameters", {})[name] = HonAttribute(
values
)
2023-06-08 19:50:56 +02:00
if self._extra:
self._attributes = self._extra.attributes(self._attributes)
2023-02-13 01:41:38 +01:00
2023-06-28 19:02:11 +02:00
async def load_statistics(self) -> None:
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
2023-06-28 19:02:11 +02:00
async def update(self, force: bool = False) -> None:
2023-05-07 00:28:24 +02:00
now = datetime.now()
2023-06-08 19:50:56 +02:00
if (
force
or not self._last_update
or self._last_update
< now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL)
2023-05-07 00:28:24 +02:00
):
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-06-28 19:02:11 +02:00
def command_parameters(self) -> Dict[str, Dict[str, str | float]]:
2023-05-06 20:00:13 +02:00
return {n: c.parameter_value for n, c in self._commands.items()}
2023-05-06 16:07:28 +02:00
@property
2023-06-28 19:02:11 +02:00
def settings(self) -> Dict[str, Parameter]:
2023-05-06 16:07:28 +02:00
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
2023-06-28 19:02:11 +02:00
def available_settings(self) -> List[str]:
2023-05-07 00:28:24 +02:00
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
2023-06-28 19:02:11 +02:00
def data(self) -> Dict[str, Any]:
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-06-08 19:50:56 +02:00
**self.attributes,
2023-04-09 20:55:36 +02:00
}
2023-03-08 00:58:25 +01:00
return result
2023-04-11 22:14:36 +02:00
2023-06-25 17:29:04 +02:00
@property
def diagnose(self) -> str:
return diagnose.yaml_export(self, anonymous=True)
async def data_archive(self, path: Path) -> str:
return await diagnose.zip_archive(self, path, anonymous=True)
2023-05-11 00:43:48 +02:00
2023-06-28 19:02:11 +02:00
def sync_to_params(self, command_name: str) -> None:
if not (command := self.commands.get(command_name)):
return
2023-06-08 19:50:56 +02:00
for key, value in self.attributes.get("parameters", {}).items():
if isinstance(value, str) and (new := command.parameters.get(key)):
2023-06-15 02:16:03 +02:00
self.attributes["parameters"][key].update(
str(new.intern_value), shield=True
)
2023-06-08 19:50:56 +02:00
2023-06-28 19:02:11 +02:00
def sync_command(self, main: str, target: Optional[List[str]] = None) -> None:
2023-06-21 19:53:36 +02:00
base: Optional[HonCommand] = self.commands.get(main)
if not base:
return
2023-05-28 06:17:43 +02:00
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