Compare commits
	
		
			3 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| 7dcb34559b | |||
| 5db13a90e7 | |||
| 9ee5dbc956 | 
							
								
								
									
										0
									
								
								pyhon/__main__.py
									
									
									
									
									
										
										
										Normal file → Executable file
									
								
							
							
						
						
									
										0
									
								
								pyhon/__main__.py
									
									
									
									
									
										
										
										Normal file → Executable file
									
								
							| @ -6,6 +6,7 @@ from typing import TYPE_CHECKING | ||||
|  | ||||
| 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,7 +34,7 @@ class HonAppliance: | ||||
|         try: | ||||
|             self._extra = importlib.import_module( | ||||
|                 f"pyhon.appliances.{self.appliance_type.lower()}" | ||||
|             ).Appliance() | ||||
|             ).Appliance(self) | ||||
|         except ModuleNotFoundError: | ||||
|             self._extra = None | ||||
|  | ||||
| @ -169,7 +170,8 @@ class HonAppliance: | ||||
|     def settings(self): | ||||
|         result = {} | ||||
|         for name, command in self._commands.items(): | ||||
|             for key, setting in command.settings.items(): | ||||
|             for key in command.setting_keys: | ||||
|                 setting = command.settings.get(key, HonParameter(key, {})) | ||||
|                 result[f"{name}.{key}"] = setting | ||||
|         if self._extra: | ||||
|             return self._extra.settings(result) | ||||
| @ -187,10 +189,7 @@ class HonAppliance: | ||||
|  | ||||
|     async def load_attributes(self): | ||||
|         self._attributes = await self._api.load_attributes(self) | ||||
|         _LOGGER.warning(self._attributes) | ||||
|         for name, values in ( | ||||
|             self._attributes.pop("shadow", {}).get("parameters", {}).items() | ||||
|         ): | ||||
|         for name, values in self._attributes.pop("shadow").get("parameters").items(): | ||||
|             self._attributes.setdefault("parameters", {})[name] = values["parNewVal"] | ||||
|  | ||||
|     async def load_statistics(self): | ||||
| @ -221,4 +220,4 @@ class HonAppliance: | ||||
|             {"commands": helper.create_command(self.commands)}, | ||||
|             whitespace="\u200B \u200B ", | ||||
|         ) | ||||
|         return result.replace(self.mac_address, "12-34-56-78-90-ab") | ||||
|         return result.replace(self.mac_address, "xx-xx-xx-xx-xx-xx") | ||||
|  | ||||
| @ -1,4 +1,7 @@ | ||||
| class Appliance: | ||||
|     def __init__(self, appliance): | ||||
|         self.parent = appliance | ||||
|  | ||||
|     def data(self, data): | ||||
|         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||
|             data["attributes"]["parameters"]["machMode"] = "0" | ||||
|  | ||||
| @ -1,4 +1,7 @@ | ||||
| class Appliance: | ||||
|     def __init__(self, appliance): | ||||
|         self.parent = appliance | ||||
|  | ||||
|     def data(self, data): | ||||
|         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||
|             data["attributes"]["parameters"]["temp"] = "0" | ||||
|  | ||||
| @ -2,15 +2,21 @@ from pyhon.parameter.fixed import HonParameterFixed | ||||
|  | ||||
|  | ||||
| class Appliance: | ||||
|     def __init__(self, appliance): | ||||
|         self.parent = appliance | ||||
|  | ||||
|     def data(self, data): | ||||
|         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||
|             data["attributes"]["parameters"]["machMode"] = "0" | ||||
|         data["active"] = bool(data.get("attributes", {}).get("activity")) | ||||
|         data["pause"] = data["attributes"]["parameters"]["machMode"] == "3" | ||||
|         if program := int(data["attributes"]["parameters"]["prCode"]): | ||||
|             ids = self.parent.settings["startProgram.program"].ids | ||||
|             data["programName"] = ids.get(program, "") | ||||
|         return data | ||||
|  | ||||
|     def settings(self, settings): | ||||
|         dry_level = settings["startProgram.dryLevel"] | ||||
|         dry_level = settings.get("startProgram.dryLevel") | ||||
|         if isinstance(dry_level, HonParameterFixed) and dry_level.value == "11": | ||||
|             settings.pop("startProgram.dryLevel", None) | ||||
|         return settings | ||||
|  | ||||
| @ -1,4 +1,7 @@ | ||||
| class Appliance: | ||||
|     def __init__(self, appliance): | ||||
|         self.parent = appliance | ||||
|  | ||||
|     def data(self, data): | ||||
|         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||
|             data["attributes"]["parameters"]["machMode"] = "0" | ||||
|  | ||||
| @ -1,4 +1,7 @@ | ||||
| class Appliance: | ||||
|     def __init__(self, appliance): | ||||
|         self.parent = appliance | ||||
|  | ||||
|     def data(self, data): | ||||
|         if data["attributes"]["lastConnEvent"]["category"] == "DISCONNECTED": | ||||
|             data["attributes"]["parameters"]["machMode"] = "0" | ||||
|  | ||||
| @ -24,7 +24,7 @@ class HonCommand: | ||||
|         self._api: HonAPI = api | ||||
|         self._appliance: "HonAppliance" = appliance | ||||
|         self._name: str = name | ||||
|         self._programs: Optional[Dict[str, "HonCommand"]] = programs or {} | ||||
|         self._programs: Optional[Dict[str, "HonCommand"]] = programs | ||||
|         self._program_name: str = program_name | ||||
|         self._description: str = attributes.get("description", "") | ||||
|         self._parameters: Dict[str, HonParameter] = self._create_parameters( | ||||
|  | ||||
| @ -182,13 +182,14 @@ class HonAuth: | ||||
|             await self._error_logger(response) | ||||
|             return "" | ||||
|  | ||||
|     def _parse_token_data(self, text: str) -> None: | ||||
|     def _parse_token_data(self, text: str) -> bool: | ||||
|         if access_token := re.findall("access_token=(.*?)&", text): | ||||
|             self._access_token = access_token[0] | ||||
|         if refresh_token := re.findall("refresh_token=(.*?)&", text): | ||||
|             self._refresh_token = refresh_token[0] | ||||
|         if id_token := re.findall("id_token=(.*?)&", text): | ||||
|             self._id_token = id_token[0] | ||||
|         return True if access_token and refresh_token and id_token else False | ||||
|  | ||||
|     async def _get_token(self, url: str) -> bool: | ||||
|         async with self._request.get(url) as response: | ||||
| @ -214,7 +215,9 @@ class HonAuth: | ||||
|             if response.status != 200: | ||||
|                 await self._error_logger(response) | ||||
|                 return False | ||||
|             self._parse_token_data(await response.text()) | ||||
|             if not self._parse_token_data(await response.text()): | ||||
|                 await self._error_logger(response) | ||||
|                 return False | ||||
|         return True | ||||
|  | ||||
|     async def _api_auth(self) -> bool: | ||||
|  | ||||
| @ -1,3 +1,6 @@ | ||||
| from pyhon.parameter.base import HonParameter | ||||
|  | ||||
|  | ||||
| def key_print(data, key="", start=True): | ||||
|     result = "" | ||||
|     if isinstance(data, list): | ||||
| @ -44,12 +47,21 @@ def pretty_print(data, key="", intend=0, is_list=False, whitespace="  "): | ||||
|     return result | ||||
|  | ||||
|  | ||||
| def get_parameter(command, parameter): | ||||
|     if programs := command.programs: | ||||
|         for program in programs.values(): | ||||
|             if data := program.settings.get(parameter): | ||||
|                 return data | ||||
|     return command.settings.get(parameter) | ||||
|  | ||||
|  | ||||
| def create_command(commands, concat=False): | ||||
|     result = {} | ||||
|     for name, command in commands.items(): | ||||
|         if not concat: | ||||
|             result[name] = {} | ||||
|         for parameter, data in command.settings.items(): | ||||
|         for parameter in command.setting_keys: | ||||
|             data = get_parameter(command, parameter) | ||||
|             if data.typology == "enum": | ||||
|                 value = data.values | ||||
|             elif data.typology == "range": | ||||
|  | ||||
| @ -47,7 +47,7 @@ class Hon: | ||||
|  | ||||
|     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 is None: | ||||
|         if appliance.mac_address == "": | ||||
|             return | ||||
|         await asyncio.gather( | ||||
|             *[ | ||||
|  | ||||
| @ -1,4 +1,4 @@ | ||||
| from typing import Dict, Any | ||||
| from typing import Dict, Any, List | ||||
|  | ||||
|  | ||||
| class HonParameter: | ||||
| @ -17,6 +17,10 @@ class HonParameter: | ||||
|     def value(self) -> str | float: | ||||
|         return self._value if self._value is not None else "0" | ||||
|  | ||||
|     @property | ||||
|     def values(self) -> List[str]: | ||||
|         return list(str(self.value)) | ||||
|  | ||||
|     @property | ||||
|     def category(self) -> str: | ||||
|         return self._category | ||||
|  | ||||
| @ -1,4 +1,4 @@ | ||||
| from typing import Dict, Any, List | ||||
| from typing import Dict, Any | ||||
|  | ||||
| from pyhon.parameter.base import HonParameter | ||||
|  | ||||
| @ -19,7 +19,3 @@ class HonParameterFixed(HonParameter): | ||||
|     def value(self, value: str | float) -> None: | ||||
|         # Fixed values seems being not so fixed as thought | ||||
|         self._value = value | ||||
|  | ||||
|     @property | ||||
|     def values(self) -> List[str]: | ||||
|         return list(str(self.value)) | ||||
|  | ||||
| @ -31,3 +31,12 @@ class HonParameterProgram(HonParameterEnum): | ||||
|     def values(self) -> List[str]: | ||||
|         values = [v for v in self._programs if all(f not in v for f in self._FILTER)] | ||||
|         return sorted(values) | ||||
|  | ||||
|     @property | ||||
|     def ids(self): | ||||
|         values = { | ||||
|             int(p.parameters["prCode"].value): n | ||||
|             for i, (n, p) in enumerate(self._programs.items()) | ||||
|             if "iot_" not in n and p.parameters.get("prCode") | ||||
|         } | ||||
|         return dict(sorted(values.items())) | ||||
|  | ||||
		Reference in New Issue
	
	Block a user
	