Compare commits
	
		
			10 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| 9f130e2e85 | |||
| dfbc24452b | |||
| af4fbdd8cd | |||
| b5af81b744 | |||
| 22a98e1781 | |||
| 2feb3295e1 | |||
| d350d639cc | |||
| 81c202d730 | |||
| 022da71800 | |||
| 8e16b4a215 | 
| @ -1,12 +1,15 @@ | ||||
| import importlib | ||||
| import json | ||||
| import logging | ||||
| from contextlib import suppress | ||||
| from datetime import datetime, timedelta | ||||
| from pathlib import Path | ||||
| from typing import Optional, Dict, Any | ||||
| from typing import TYPE_CHECKING | ||||
|  | ||||
| from pyhon import helper, exceptions | ||||
| from pyhon import helper | ||||
| from pyhon.commands import HonCommand | ||||
| from pyhon.parameter.base import HonParameter | ||||
| from pyhon.parameter.fixed import HonParameterFixed | ||||
|  | ||||
| if TYPE_CHECKING: | ||||
| @ -33,6 +36,7 @@ class HonAppliance: | ||||
|         self._zone: int = zone | ||||
|         self._additional_data: Dict[str, Any] = {} | ||||
|         self._last_update = None | ||||
|         self._default_setting = HonParameter("", {}, "") | ||||
|  | ||||
|         try: | ||||
|             self._extra = importlib.import_module( | ||||
| @ -123,9 +127,7 @@ class HonAppliance: | ||||
|         return self._zone | ||||
|  | ||||
|     @property | ||||
|     def api(self) -> "HonAPI": | ||||
|         if self._api is None: | ||||
|             raise exceptions.NoAuthenticationException | ||||
|     def api(self) -> Optional["HonAPI"]: | ||||
|         return self._api | ||||
|  | ||||
|     async def _recover_last_command_states(self): | ||||
| @ -167,6 +169,8 @@ class HonAppliance: | ||||
|                     category = category.split(".")[-1].lower() | ||||
|                 categories[category] = result[0] | ||||
|         if categories: | ||||
|             if "setParameters" in categories: | ||||
|                 return [categories["setParameters"]] | ||||
|             return [list(categories.values())[0]] | ||||
|         return [] | ||||
|  | ||||
| @ -200,7 +204,7 @@ class HonAppliance: | ||||
|     async def load_commands(self): | ||||
|         raw = await self.api.load_commands(self) | ||||
|         self._appliance_model = raw.pop("applianceModel") | ||||
|         raw.pop("dictionaryId") | ||||
|         raw.pop("dictionaryId", None) | ||||
|         self._commands = self._get_commands(raw) | ||||
|         await self._recover_last_command_states() | ||||
|  | ||||
| @ -211,6 +215,7 @@ class HonAppliance: | ||||
|  | ||||
|     async def load_statistics(self): | ||||
|         self._statistics = await self.api.load_statistics(self) | ||||
|         self._statistics |= await self.api.load_maintenance(self) | ||||
|  | ||||
|     async def update(self): | ||||
|         now = datetime.now() | ||||
| @ -229,7 +234,7 @@ class HonAppliance: | ||||
|         result = {} | ||||
|         for name, command in self._commands.items(): | ||||
|             for key in command.setting_keys: | ||||
|                 setting = command.settings.get(key) | ||||
|                 setting = command.settings.get(key, self._default_setting) | ||||
|                 result[f"{name}.{key}"] = setting | ||||
|         if self._extra: | ||||
|             return self._extra.settings(result) | ||||
| @ -256,21 +261,57 @@ class HonAppliance: | ||||
|             return self._extra.data(result) | ||||
|         return result | ||||
|  | ||||
|     def diagnose(self, whitespace="\u200B \u200B "): | ||||
|     def diagnose(self, whitespace="  ", command_only=False): | ||||
|         data = { | ||||
|             "attributes": self.attributes.copy(), | ||||
|             "appliance": self.info, | ||||
|             "statistics": self.statistics, | ||||
|             "additional_data": self._additional_data, | ||||
|         } | ||||
|         if command_only: | ||||
|             data.pop("attributes") | ||||
|             data.pop("appliance") | ||||
|             data.pop("statistics") | ||||
|         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} | ||||
|         for sensible in ["PK", "SK", "serialNumber", "code", "coords"]: | ||||
|             data["appliance"].pop(sensible, None) | ||||
|         for sensible in ["PK", "SK", "serialNumber", "coords", "device"]: | ||||
|             data.get("appliance", {}).pop(sensible, None) | ||||
|         result = helper.pretty_print({"data": data}, whitespace=whitespace) | ||||
|         result += helper.pretty_print( | ||||
|             {"commands": helper.create_command(self.commands)}, | ||||
|             whitespace=whitespace, | ||||
|         ) | ||||
|         return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx") | ||||
|  | ||||
|  | ||||
| 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" | ||||
|  | ||||
| @ -1,5 +1,6 @@ | ||||
| from typing import Optional, Dict, Any, List, TYPE_CHECKING, Union | ||||
|  | ||||
| from pyhon import exceptions | ||||
| from pyhon.parameter.base import HonParameter | ||||
| from pyhon.parameter.enum import HonParameterEnum | ||||
| from pyhon.parameter.fixed import HonParameterFixed | ||||
| @ -20,7 +21,7 @@ class HonCommand: | ||||
|         categories: Optional[Dict[str, "HonCommand"]] = None, | ||||
|         category_name: str = "", | ||||
|     ): | ||||
|         self._api: HonAPI = appliance.api | ||||
|         self._api: Optional[HonAPI] = appliance.api | ||||
|         self._appliance: "HonAppliance" = appliance | ||||
|         self._name: str = name | ||||
|         self._categories: Optional[Dict[str, "HonCommand"]] = categories | ||||
| @ -39,6 +40,12 @@ class HonCommand: | ||||
|     def name(self): | ||||
|         return self._name | ||||
|  | ||||
|     @property | ||||
|     def api(self) -> "HonAPI": | ||||
|         if self._api is None: | ||||
|             raise exceptions.NoAuthenticationException | ||||
|         return self._api | ||||
|  | ||||
|     @property | ||||
|     def data(self): | ||||
|         return self._data | ||||
| @ -87,7 +94,7 @@ class HonCommand: | ||||
|     async def send(self) -> bool: | ||||
|         params = self.parameter_groups.get("parameters", {}) | ||||
|         ancillary_params = self.parameter_groups.get("ancillaryParameters", {}) | ||||
|         return await self._api.send_command( | ||||
|         return await self.api.send_command( | ||||
|             self._appliance, self._name, params, ancillary_params | ||||
|         ) | ||||
|  | ||||
| @ -103,7 +110,8 @@ class HonCommand: | ||||
|  | ||||
|     @category.setter | ||||
|     def category(self, category: str) -> None: | ||||
|         self._appliance.commands[self._name] = self.categories[category] | ||||
|         if category in self.categories: | ||||
|             self._appliance.commands[self._name] = self.categories[category] | ||||
|  | ||||
|     @property | ||||
|     def setting_keys(self) -> List[str]: | ||||
| @ -128,5 +136,6 @@ class HonCommand: | ||||
|             for name, parameter in command.parameters.items(): | ||||
|                 if name in result: | ||||
|                     result[name] = self._more_options(result[name], parameter) | ||||
|                 result[name] = parameter | ||||
|                 else: | ||||
|                     result[name] = parameter | ||||
|         return result | ||||
|  | ||||
| @ -72,7 +72,6 @@ class HonAPI: | ||||
|     async def load_commands(self, appliance: HonAppliance) -> Dict: | ||||
|         params: Dict = { | ||||
|             "applianceType": appliance.appliance_type, | ||||
|             "code": appliance.info["code"], | ||||
|             "applianceModelId": appliance.appliance_model_id, | ||||
|             "macAddress": appliance.mac_address, | ||||
|             "os": const.OS, | ||||
| @ -83,6 +82,8 @@ class HonAPI: | ||||
|             params["firmwareId"] = firmware_id | ||||
|         if firmware_version := appliance.info.get("fwVersion"): | ||||
|             params["fwVersion"] = firmware_version | ||||
|         if code := appliance.info.get("code"): | ||||
|             params["code"] = code | ||||
|         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", {}) | ||||
| @ -100,6 +101,16 @@ class HonAPI: | ||||
|                 return {} | ||||
|             return result["payload"]["history"] | ||||
|  | ||||
|     async def command_favourites(self, appliance: HonAppliance) -> Dict: | ||||
|         url: str = ( | ||||
|             f"{const.API_URL}/commands/v1/appliance/{appliance.mac_address}/favourite" | ||||
|         ) | ||||
|         async with self._hon.get(url) as response: | ||||
|             result: Dict = await response.json() | ||||
|             if not result or not result.get("payload"): | ||||
|                 return {} | ||||
|             return result["payload"]["favourites"] | ||||
|  | ||||
|     async def last_activity(self, appliance: HonAppliance) -> Dict: | ||||
|         url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity" | ||||
|         params: Dict = {"macAddress": appliance.mac_address} | ||||
| @ -140,6 +151,12 @@ class HonAPI: | ||||
|         async with self._hon.get(url, params=params) as response: | ||||
|             return (await response.json()).get("payload", {}) | ||||
|  | ||||
|     async def load_maintenance(self, appliance: HonAppliance): | ||||
|         url = f"{const.API_URL}/commands/v1/maintenance-cycle" | ||||
|         params = {"macAddress": appliance.mac_address} | ||||
|         async with self._hon.get(url, params=params) as response: | ||||
|             return (await response.json()).get("payload", {}) | ||||
|  | ||||
|     async def send_command( | ||||
|         self, | ||||
|         appliance: HonAppliance, | ||||
|  | ||||
							
								
								
									
										50
									
								
								pyhon/hon.py
									
									
									
									
									
								
							
							
						
						
									
										50
									
								
								pyhon/hon.py
									
									
									
									
									
								
							| @ -1,4 +1,5 @@ | ||||
| import asyncio | ||||
| import logging | ||||
| from types import TracebackType | ||||
| from typing import List, Optional, Dict, Any, Type | ||||
|  | ||||
| @ -8,11 +9,18 @@ from typing_extensions import Self | ||||
| from pyhon import HonAPI, exceptions | ||||
| from pyhon.appliance import HonAppliance | ||||
|  | ||||
| _LOGGER = logging.getLogger(__name__) | ||||
|  | ||||
|  | ||||
| class Hon: | ||||
|     def __init__(self, email: str, password: str, session: ClientSession | None = None): | ||||
|         self._email: str = email | ||||
|         self._password: str = password | ||||
|     def __init__( | ||||
|         self, | ||||
|         email: Optional[str] = "", | ||||
|         password: Optional[str] = "", | ||||
|         session: Optional[ClientSession] = None, | ||||
|     ): | ||||
|         self._email: Optional[str] = email | ||||
|         self._password: Optional[str] = password | ||||
|         self._session: ClientSession | None = session | ||||
|         self._appliances: List[HonAppliance] = [] | ||||
|         self._api: Optional[HonAPI] = None | ||||
| @ -34,9 +42,21 @@ class Hon: | ||||
|             raise exceptions.NoAuthenticationException | ||||
|         return self._api | ||||
|  | ||||
|     @property | ||||
|     def email(self) -> str: | ||||
|         if not self._email: | ||||
|             raise ValueError("Missing email") | ||||
|         return self._email | ||||
|  | ||||
|     @property | ||||
|     def password(self) -> str: | ||||
|         if not self._password: | ||||
|             raise ValueError("Missing password") | ||||
|         return self._password | ||||
|  | ||||
|     async def create(self) -> Self: | ||||
|         self._api = await HonAPI( | ||||
|             self._email, self._password, session=self._session | ||||
|             self.email, self.password, session=self._session | ||||
|         ).create() | ||||
|         await self.setup() | ||||
|         return self | ||||
| @ -45,17 +65,25 @@ class Hon: | ||||
|     def appliances(self) -> List[HonAppliance]: | ||||
|         return self._appliances | ||||
|  | ||||
|     @appliances.setter | ||||
|     def appliances(self, appliances) -> None: | ||||
|         self._appliances = appliances | ||||
|  | ||||
|     async def _create_appliance(self, appliance_data: Dict[str, Any], zone=0) -> None: | ||||
|         appliance = HonAppliance(self._api, appliance_data, zone=zone) | ||||
|         if appliance.mac_address == "": | ||||
|             return | ||||
|         await asyncio.gather( | ||||
|             *[ | ||||
|                 appliance.load_attributes(), | ||||
|                 appliance.load_commands(), | ||||
|                 appliance.load_statistics(), | ||||
|             ] | ||||
|         ) | ||||
|         try: | ||||
|             await asyncio.gather( | ||||
|                 *[ | ||||
|                     appliance.load_attributes(), | ||||
|                     appliance.load_commands(), | ||||
|                     appliance.load_statistics(), | ||||
|                 ] | ||||
|             ) | ||||
|         except (KeyError, ValueError, IndexError) as error: | ||||
|             _LOGGER.exception(error) | ||||
|             _LOGGER.error("Device data - %s", appliance_data) | ||||
|         self._appliances.append(appliance) | ||||
|  | ||||
|     async def setup(self) -> None: | ||||
|  | ||||
| @ -9,6 +9,8 @@ class HonParameterEnum(HonParameter): | ||||
|         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: | ||||
|             self._values.append(self._default) | ||||
|  | ||||
|     def __repr__(self) -> str: | ||||
|         return f"{self.__class__} (<{self.key}> {self.values})" | ||||
|  | ||||
| @ -32,6 +32,8 @@ class HonParameterRange(HonParameter): | ||||
|  | ||||
|     @property | ||||
|     def step(self) -> float: | ||||
|         if not self._step: | ||||
|             return 1 | ||||
|         return self._step | ||||
|  | ||||
|     @property | ||||
| @ -41,7 +43,7 @@ class HonParameterRange(HonParameter): | ||||
|     @value.setter | ||||
|     def value(self, value: float) -> None: | ||||
|         value = str_to_float(value) | ||||
|         if self._min <= value <= self._max and not value % self._step: | ||||
|         if self._min <= value <= self._max and not (value - self._min) % self._step: | ||||
|             self._value = value | ||||
|         else: | ||||
|             raise ValueError( | ||||
|  | ||||
		Reference in New Issue
	
	Block a user
	