diff --git a/pyhon/appliance.py b/pyhon/appliance.py index 76514ff..becfbba 100644 --- a/pyhon/appliance.py +++ b/pyhon/appliance.py @@ -1,18 +1,15 @@ import importlib import json import logging -from contextlib import suppress -from copy import copy from datetime import datetime, timedelta from pathlib import Path -from typing import Optional, Dict, Any -from typing import TYPE_CHECKING +from typing import Optional, Dict, Any, TYPE_CHECKING from pyhon import helper from pyhon.attributes import HonAttribute +from pyhon.command_loader import HonCommandLoader from pyhon.commands import HonCommand from pyhon.parameter.base import HonParameter -from pyhon.parameter.fixed import HonParameterFixed from pyhon.parameter.range import HonParameterRange if TYPE_CHECKING: @@ -140,104 +137,12 @@ class HonAppliance: def api(self) -> Optional["HonAPI"]: return self._api - async def _recover_last_command_states(self): - command_history = await self.api.command_history(self) - for name, command in self._commands.items(): - last = next( - ( - index - for (index, d) in enumerate(command_history) - if d.get("command", {}).get("commandName") == name - ), - None, - ) - if last is None: - continue - parameters = command_history[last].get("command", {}).get("parameters", {}) - if command.categories and ( - parameters.get("program") or parameters.get("category") - ): - if parameters.get("program"): - command.category = parameters.pop("program").split(".")[-1].lower() - else: - command.category = parameters.pop("category") - command = self.commands[name] - for key, data in command.settings.items(): - if ( - not isinstance(data, HonParameterFixed) - and parameters.get(key) is not None - ): - with suppress(ValueError): - data.value = parameters.get(key) - - 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: - if "setParameters" in categories: - return [categories["setParameters"]] - return [list(categories.values())[0]] - 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( - command, - data, - self, - category_name=category, - categories=categories, - ) - ] - else: - commands += self._get_categories(command, data) - elif category: - self._additional_data.setdefault(command, {})[category] = data - else: - self._additional_data[command] = data - return commands - - async def load_commands(self): - raw = await self.api.load_commands(self) - self._appliance_model = raw.pop("applianceModel") - raw.pop("dictionaryId", None) - self._commands = self._get_commands(raw) - await self._add_favourites() - await self._recover_last_command_states() - - 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]) - for data in command.values(): - if isinstance(data, str): - continue - for key, value in data.items(): - if parameter := base.parameters.get(key): - with suppress(ValueError): - parameter.value = value - extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") - base.parameters.update(favourite=extra_param) - base.parameters["program"].set_value(name) - self._commands[command_name].categories[name] = base + async def load_commands(self, data=None): + command_loader = HonCommandLoader(self.api, self) + await command_loader.load_commands(data) + self._commands = command_loader.commands + self._additional_data = command_loader.additional_data + self._appliance_model = command_loader.appliance_data async def load_attributes(self): self._attributes = await self.api.load_attributes(self) @@ -245,7 +150,9 @@ class HonAppliance: if name in self._attributes.get("parameters", {}): self._attributes["parameters"][name].update(values) else: - self._attributes.setdefault("parameters", {})[name] = HonAttribute(values) + self._attributes.setdefault("parameters", {})[name] = HonAttribute( + values + ) if self._extra: self._attributes = self._extra.attributes(self._attributes) @@ -330,7 +237,9 @@ class HonAppliance: command: HonCommand = self.commands.get(command_name) for key, value in self.attributes.get("parameters", {}).items(): if isinstance(value, str) and (new := command.parameters.get(key)): - self.attributes["parameters"][key].update(str(new.intern_value), shield=True) + self.attributes["parameters"][key].update( + str(new.intern_value), shield=True + ) def sync_command(self, main, target=None) -> None: base: HonCommand = self.commands.get(main) diff --git a/pyhon/command_loader.py b/pyhon/command_loader.py new file mode 100644 index 0000000..7ccb3c5 --- /dev/null +++ b/pyhon/command_loader.py @@ -0,0 +1,197 @@ +import asyncio +import json +from contextlib import suppress +from copy import copy +from typing import Dict, Any, Optional, TYPE_CHECKING, List + +from pyhon.commands import HonCommand +from pyhon.parameter.fixed import HonParameterFixed +from pyhon.parameter.program import HonParameterProgram + +if TYPE_CHECKING: + from pyhon import HonAPI, exceptions + from pyhon.appliance import HonAppliance + + +class HonCommandLoader: + """Loads and parses hOn command data""" + + def __init__(self, api, appliance): + self._api_commands: Dict[str, Any] = {} + self._favourites: List[Dict[str, Any]] = [] + self._command_history: List[Dict[str, Any]] = [] + self._commands: Dict[str, HonCommand] = {} + self._api: "HonAPI" = api + self._appliance: "HonAppliance" = appliance + self._appliance_data: Dict[str, Any] = {} + self._additional_data: Dict[str, Any] = {} + + @property + def api(self) -> "HonAPI": + """api connection object""" + if self._api is None: + raise exceptions.NoAuthenticationException("Missing hOn login") + return self._api + + @property + def appliance(self) -> "HonAppliance": + """appliance object""" + return self._appliance + + @property + def commands(self) -> Dict[str, HonCommand]: + """Get list of hon commands""" + return self._commands + + @property + def appliance_data(self) -> Dict[str, Any]: + """Get command appliance data""" + return self._appliance_data + + @property + def additional_data(self) -> Dict[str, Any]: + """Get command additional data""" + return self._additional_data + + async def load_commands(self, data=None): + """Trigger loading of command data""" + if data: + self._api_commands = data + else: + await self._load_data() + self._appliance_data = self._api_commands.pop("applianceModel") + self._get_commands() + self._add_favourites() + self._recover_last_command_states() + + async def _load_commands(self): + self._api_commands = await self._api.load_commands(self._appliance) + + async def _load_favourites(self): + self._favourites = await self._api.command_favourites(self._appliance) + + async def _load_command_history(self): + self._command_history = await self._api.command_history(self._appliance) + + async def _load_data(self): + """Request parallel all relevant data""" + await asyncio.gather( + *[ + self._load_commands(), + self._load_favourites(), + self._load_command_history(), + ] + ) + + @staticmethod + def _is_command(data: Dict[str, Any]) -> bool: + """Check if dict can be parsed as command""" + return ( + data.get("description") is not None and data.get("protocolType") is not None + ) + + @staticmethod + def _clean_name(category: str) -> str: + """Clean up category name""" + if "PROGRAM" in category: + return category.split(".")[-1].lower() + return category + + def _get_commands(self) -> None: + """Generates HonCommand dict from api data""" + commands = [] + for name, data in self._api_commands.items(): + if command := self._parse_command(data, name): + commands.append(command) + self._commands = {c.name: c for c in commands} + + def _parse_command( + self, data: Dict[str, Any] | str, command_name: str, **kwargs + ) -> Optional[HonCommand]: + """Try to crate HonCommand object""" + if not isinstance(data, dict): + self._additional_data[command_name] = data + return None + if self._is_command(data): + return HonCommand(command_name, data, self._appliance, **kwargs) + if category := self._parse_categories(data, command_name): + return category + return None + + def _parse_categories( + self, data: Dict[str, Any], command_name: str + ) -> Optional[HonCommand]: + """Parse categories and create reference to other""" + categories: Dict[str, HonCommand] = {} + for category, value in data.items(): + kwargs = {"category_name": category, "categories": categories} + if command := self._parse_command(value, command_name, **kwargs): + categories[self._clean_name(category)] = command + if categories: + # setParameters should be at first place + if "setParameters" in categories: + return categories["setParameters"] + return list(categories.values())[0] + return None + + def _get_last_command_index(self, name: str) -> Optional[int]: + """Get index of last command execution""" + return next( + ( + index + for (index, d) in enumerate(self._command_history) + if d.get("command", {}).get("commandName") == name + ), + None, + ) + + def _set_last_category( + self, command: HonCommand, name: str, parameters: Dict[str, Any] + ) -> HonCommand: + """Set category to last state""" + if command.categories: + if program := parameters.pop("program", None): + command.category = self._clean_name(program) + elif category := parameters.pop("category", None): + command.category = category + else: + return command + return self.commands[name] + return command + + def _recover_last_command_states(self) -> None: + """Set commands to last state""" + for name, command in self.commands.items(): + if (last_index := self._get_last_command_index(name)) is None: + continue + last_command = self._command_history[last_index] + parameters = last_command.get("command", {}).get("parameters", {}) + command = self._set_last_category(command, name, parameters) + for key, data in command.settings.items(): + if parameters.get(key) is None: + continue + with suppress(ValueError): + data.value = parameters.get(key) + + def _add_favourites(self) -> None: + """Patch program categories with favourites""" + for favourite in self._favourites: + name = favourite.get("favouriteName", {}) + command = favourite.get("command", {}) + command_name = command.get("commandName", "") + program_name = self._clean_name(command.get("programName", "")) + base: HonCommand = copy( + self.commands[command_name].categories[program_name] + ) + for data in command.values(): + if isinstance(data, str): + continue + for key, value in data.items(): + if parameter := base.parameters.get(key): + with suppress(ValueError): + parameter.value = value + extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") + base.parameters.update(favourite=extra_param) + if isinstance(program := base.parameters["program"], HonParameterProgram): + program.set_value(name) + self.commands[command_name].categories[name] = base