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