Compare commits

...

51 Commits

Author SHA1 Message Date
Andre Basche
327d4a1814
Update README.md 2024-01-15 10:36:18 +01:00
Andre Basche
4dc60c29ee Fix crash in loading attributs Andre0512/hon#134 2024-01-11 01:21:38 +01:00
Andre Basche
7bc9e718a0
Merge pull request #17 from MiguelAngelLV/main
Use class attribute to check active
2023-11-26 13:43:44 +01:00
Mguel Ángel
d4e5793186 Fix check active oven 2023-11-26 13:36:41 +01:00
Andre Basche
e9f2bb9f4f Send program name Andre0512/hon#124 2023-11-21 02:23:53 +01:00
Andre Basche
ea81e2891f Bump version 2023-11-20 17:51:32 +01:00
Andre Basche
c71f8f17f5 Fix error in wh 2023-11-20 17:50:28 +01:00
Andre Basche
27d974abba Fix dependencies 2023-11-20 16:38:39 +01:00
Andre Basche
ab18e45f97 Add python3.12 support 2023-11-19 23:53:21 +01:00
Andre Basche
e44b9b6773 Update mypy 2023-10-12 22:26:12 +02:00
Andre Basche
5c650d722b Bump version 2023-10-12 22:16:02 +02:00
Vadym
6ae40481e3
Issue with sync_command (#16)
* Added water heater appliance. Added ability to send only mandatory parameters

* fixed build

* formatting

* cleanup

* cleanup

* reformatting

* Added ability to send specific parameters. Useful in case the command has many not mandatory parameters and you want to send only one/few

* cleanup

* Fixed code style

* sync_command - fixed typos, skip to sync(actually reset) parameters of different types. Improved WaterHeater appliance

* cleanup

* cleanup

* clean code style

* check if base parameter is mandatory

* Reverted back sync_command, send mandatory parameters beside with specified

---------

Co-authored-by: Vadym Melnychuk <vme@primexm.com>
2023-10-12 16:43:41 +02:00
Andre Basche
ff8ae160bb Fix empty prStr 2023-10-06 01:29:38 +02:00
Andre Basche
658e80a8f4 Change dependencies to variable major version 2023-10-02 03:21:51 +02:00
Andre Basche
ffba85bf0d Bump version 2023-10-02 01:38:54 +02:00
Andre Basche
10c8d961c4 Support new style rules hon#112 2023-10-02 01:38:40 +02:00
Andre Basche
61dd470588 Set versions of dependant packages to 'compatible releases' 2023-07-27 19:16:23 +02:00
Andre Basche
1ed81c2a77 Simplify get favorites 2023-07-24 02:33:45 +02:00
Andre Basche
e4dc3cb1d0 Next try to add py.typed in package 2023-07-24 01:47:45 +02:00
Andre Basche
2523069ce9 Fix false name caused by chatgpt's wrong advice 2023-07-23 23:18:09 +02:00
Andre Basche
eeb458cb1b Add py.typed into package 2023-07-23 22:54:46 +02:00
Andre Basche
2764700bc7 Bump version 2023-07-23 21:56:16 +02:00
Andre Basche
e6c796e822 Improve type hints 2023-07-23 21:55:42 +02:00
Andre Basche
454f2d8916 Use equal mypy cnofig as home assistant 2023-07-22 12:39:50 +02:00
Andre Basche
59ca8b6caf Not loading favorite if base program renamed 2023-07-22 11:53:39 +02:00
Andre Basche
44c55c681d Update requirements 2023-07-20 23:55:40 +02:00
Andre Basche
cfee10df5f Improve logging for test api 2023-07-20 23:52:46 +02:00
Andre Basche
e0774677eb Add and apply some mypy rules 2023-07-20 23:52:07 +02:00
Andre Basche
fc60d15e60 Fix error for fridge without quickmode 2023-07-19 23:55:37 +02:00
Andre Basche
8ef8c0405d Fix empty value in settings 2023-07-19 19:52:21 +02:00
Andre Basche
5575b240e1 Bump version 2023-07-18 21:32:12 +02:00
Andre Basche
22367242a2 Add flake8 config 2023-07-18 21:31:26 +02:00
Vadym
4f7d4860db
Water Heater. Ability to send only mandatory parameters (#14)
* Added water heater appliance. Added ability to send only mandatory parameters

* fixed build

* formatting

* cleanup

* cleanup

* reformatting

* Added ability to send specific parameters. Useful in case the command has many not mandatory parameters and you want to send only one/few

* cleanup

* Fixed code style

---------

Co-authored-by: Vadym Melnychuk <vme@primexm.com>
2023-07-18 21:26:11 +02:00
Andre Basche
5a778373b6 Enable more pylint checks 2023-07-16 05:53:23 +02:00
Andre Basche
e1c8bc5835 Reduce complexity and line length for flake8 2023-07-16 04:42:29 +02:00
Andre Basche
442e7a07dd Fix pylint check 2023-07-14 00:40:48 +02:00
Andre Basche
b0e3b15ff0 Add pylint checks 2023-07-12 19:36:32 +02:00
Andre Basche
2788a3ac91 Reset commands to defaults 2023-07-12 00:05:27 +02:00
Andre Basche
bc7e8994c9 Bump version 2023-07-11 00:21:37 +02:00
Andre Basche
8ca40d7ad0 Add missing requirement typing-extensions 2023-07-10 23:59:55 +02:00
Andre Basche
9a6a07fd46 Sync enum values of commands 2023-07-10 23:58:24 +02:00
Andre Basche
f1818bbc5d Bump version 2023-07-09 23:59:27 +02:00
Andre Basche
3d5c8405ea Improve error handling 2023-07-09 23:58:55 +02:00
Andre Basche
e234ef3bbb Remove old code in ov hon#88 2023-07-09 01:36:03 +02:00
Andre Basche
e00e147ecd Bump version 2023-07-01 16:29:29 +02:00
Andre Basche
26bc35c8a6 Fix issue in locking attribute updates 2023-07-01 16:27:50 +02:00
Andre Basche
17d73cdeb8 Sync parameter to settings 2023-07-01 16:04:34 +02:00
Andre Basche
a10ab4423e Add stricter mypy rules 2023-07-01 14:59:09 +02:00
Andre Basche
0553e6c17d Improvements 2023-07-01 14:31:37 +02:00
Andre Basche
44f40c531e Fix some minor issues 2023-06-29 22:08:51 +02:00
Andre Basche
4e88bc7a9f Fix error in archive again 2023-06-29 18:49:33 +02:00
37 changed files with 545 additions and 273 deletions

3
.flake8 Normal file
View File

@ -0,0 +1,3 @@
[flake8]
max-line-length = 88
max-complexity = 7

View File

@ -13,7 +13,7 @@ jobs:
strategy:
fail-fast: false
matrix:
python-version: ["3.10", "3.11"]
python-version: ["3.10", "3.11", "3.12"]
steps:
- uses: actions/checkout@v3
@ -28,15 +28,13 @@ jobs:
python -m pip install -r requirements_dev.txt
- name: Lint with flake8
run: |
# stop the build if there are Python syntax errors or undefined names
flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
flake8 . --count --exit-zero --max-complexity=10 --max-line-length=88 --statistics
flake8 . --count --statistics
- name: Type check with mypy
run: |
mypy pyhon/
# - name: Analysing the code with pylint
# run: |
# pylint --max-line-length 88 $(git ls-files '*.py')
- name: Analysing the code with pylint
run: |
pylint $(git ls-files '*.py')
- name: Check black style
run: |
black . --check

9
.pylintrc Normal file
View File

@ -0,0 +1,9 @@
[MESSAGES CONTROL]
disable=missing-docstring
[FORMAT]
max-args=6
max-attributes=8
max-line-length=88

1
MANIFEST.in Normal file
View File

@ -0,0 +1 @@
include pyhon/py.typed

View File

@ -1,3 +1,14 @@
# Announcement: I have to take the project down in the next few days
> Dear User,
>
> We are writing to inform you that we have discovered two Home Assistant integration plug-ins developed by you ( https://github.com/Andre0512/hon and https://github.com/Andre0512/pyhOn ) that are in violation of our terms of service. Specifically, the plug-ins are using our services in an unauthorized manner which is causing significant economic harm to our Company.
> We take the protection of our intellectual property very seriously and demand that you immediately cease and desist all illegal activities related to the development and distribution of these plug-ins. We also request that you remove the plug-ins from all stores and code hosting platforms where they are currently available.
> Please be advised that we will take all necessary legal action to protect our interests if you fail to comply with this notice. We reserve the right to pursue all available remedies, including but not limited to monetary damages, injunctive relief, and attorney's fees.
> We strongly urge you to take immediate action to rectify this situation and avoid any further legal action. If you have any questions or concerns, please do not hesitate to contact us.
>
> Haier Europe Security and Governance Department
**This python package is unofficial and is not related in any way to Haier. It was developed by reversed engineered requests and can stop working at anytime!**
# pyhOn

View File

@ -1,4 +1,24 @@
[mypy]
check_untyped_defs = True
disallow_any_generics = True
disallow_untyped_defs = True
check_untyped_defs = true
disallow_any_generics = true
disallow_any_unimported = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
disable_error_code = annotation-unchecked
enable_error_code = ignore-without-code, redundant-self, truthy-iterable
follow_imports = silent
local_partial_types = true
no_implicit_optional = true
no_implicit_reexport = true
show_error_codes = true
strict_concatenate = false
strict_equality = true
warn_incomplete_stub = true
warn_redundant_casts = true
warn_return_any = true
warn_unreachable = true
warn_unused_configs = true
warn_unused_ignores = true

View File

@ -11,6 +11,7 @@ from typing import Tuple, Dict, Any
if __name__ == "__main__":
sys.path.insert(0, str(Path(__file__).parent.parent))
# pylint: disable=wrong-import-position
from pyhon import Hon, HonAPI, diagnose, printer
_LOGGER = logging.getLogger(__name__)
@ -30,13 +31,13 @@ def get_arguments() -> Dict[str, Any]:
export.add_argument("--zip", help="create zip archive", action="store_true")
export.add_argument("--anonymous", help="anonymize data", action="store_true")
export.add_argument("directory", nargs="?", default=Path().cwd())
translate = subparser.add_parser(
translation = subparser.add_parser(
"translate", help="print available translation keys"
)
translate.add_argument(
translation.add_argument(
"translate", help="language (de, en, fr...)", metavar="LANGUAGE"
)
translate.add_argument("--json", help="print as json", action="store_true")
translation.add_argument("--json", help="print as json", action="store_true")
parser.add_argument(
"-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd()
)
@ -91,15 +92,13 @@ async def main() -> None:
data = device.data.copy()
attr = "get" if args.get("all") else "pop"
print(
printer.key_print(
data["attributes"].__getattribute__(attr)("parameters")
printer.key_print(getattr(data["attributes"], attr)("parameters"))
)
)
print(printer.key_print(data.__getattribute__(attr)("appliance")))
print(printer.key_print(getattr(data, attr)("appliance")))
print(printer.key_print(data))
print(
printer.pretty_print(
printer.create_command(device.commands, concat=True)
printer.create_commands(device.commands, concat=True)
)
)
else:

View File

@ -3,13 +3,15 @@ import logging
import re
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, Dict, Any, TYPE_CHECKING, List
from typing import Optional, Dict, Any, TYPE_CHECKING, List, TypeVar, overload
from pyhon import diagnose, exceptions
from pyhon.appliances.base import ApplianceBase
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.enum import HonParameterEnum
from pyhon.parameter.range import HonParameterRange
from pyhon.typedefs import Parameter
@ -18,7 +20,10 @@ if TYPE_CHECKING:
_LOGGER = logging.getLogger(__name__)
T = TypeVar("T")
# pylint: disable=too-many-public-methods,too-many-instance-attributes
class HonAppliance:
_MINIMAL_UPDATE_INTERVAL = 5 # seconds
@ -40,30 +45,41 @@ class HonAppliance:
self._default_setting = HonParameter("", {}, "")
try:
self._extra = importlib.import_module(
self._extra: Optional[ApplianceBase] = importlib.import_module(
f"pyhon.appliances.{self.appliance_type.lower()}"
).Appliance(self)
except ModuleNotFoundError:
self._extra = None
def _get_nested_item(self, item: str) -> Any:
result: List[Any] | Dict[str, Any] = self.data
for key in item.split("."):
if all(k in "0123456789" for k in key) and isinstance(result, list):
result = result[int(key)]
elif isinstance(result, dict):
result = result[key]
return result
def __getitem__(self, item: str) -> Any:
if self._zone:
item += f"Z{self._zone}"
if "." in item:
result = self.data
for key in item.split("."):
if all(k in "0123456789" for k in key) and isinstance(result, list):
result = result[int(key)]
else:
result = result[key]
return result
return self._get_nested_item(item)
if item in self.data:
return self.data[item]
if item in self.attributes["parameters"]:
return self.attributes["parameters"][item].value
return self.info[item]
def get(self, item: str, default: Any = None) -> Any:
@overload
def get(self, item: str, default: None = None) -> Any:
...
@overload
def get(self, item: str, default: T) -> T:
...
def get(self, item: str, default: Optional[T] = None) -> Any:
try:
return self[item]
except (KeyError, IndexError):
@ -71,21 +87,22 @@ class HonAppliance:
def _check_name_zone(self, name: str, frontend: bool = True) -> str:
zone = " Z" if frontend else "_z"
if (attribute := self._info.get(name, "")) and self._zone:
attribute: str = self._info.get(name, "")
if attribute and self._zone:
return f"{attribute}{zone}{self._zone}"
return attribute
@property
def appliance_model_id(self) -> str:
return self._info.get("applianceModelId", "")
return str(self._info.get("applianceModelId", ""))
@property
def appliance_type(self) -> str:
return self._info.get("applianceTypeName", "")
return str(self._info.get("applianceTypeName", ""))
@property
def mac_address(self) -> str:
return self.info.get("macAddress", "")
return str(self.info.get("macAddress", ""))
@property
def unique_id(self) -> str:
@ -101,29 +118,31 @@ class HonAppliance:
@property
def brand(self) -> str:
return self._check_name_zone("brand")
brand = self._check_name_zone("brand")
return brand[0].upper() + brand[1:]
@property
def nick_name(self) -> str:
result = self._check_name_zone("nickName")
if not result or re.findall("^[xX\s]+$", result):
if not result or re.findall("^[xX1\\s-]+$", result):
return self.model_name
return result
@property
def code(self) -> str:
if code := self.info.get("code"):
code: str = self.info.get("code", "")
if code:
return code
serial_number = self.info.get("serialNumber", "")
serial_number: str = self.info.get("serialNumber", "")
return serial_number[:8] if len(serial_number) < 18 else serial_number[:11]
@property
def model_id(self) -> int:
return self._info.get("applianceModelId", 0)
return int(self._info.get("applianceModelId", 0))
@property
def options(self) -> Dict[str, Any]:
return self._appliance_model.get("options", {})
return dict(self._appliance_model.get("options", {}))
@property
def commands(self) -> Dict[str, HonCommand]:
@ -162,16 +181,18 @@ class HonAppliance:
self._commands = command_loader.commands
self._additional_data = command_loader.additional_data
self._appliance_model = command_loader.appliance_data
self.sync_params_to_command("settings")
async def load_attributes(self) -> None:
self._attributes = await self.api.load_attributes(self)
for name, values in self._attributes.pop("shadow").get("parameters").items():
attributes = await self.api.load_attributes(self)
for name, values in attributes.pop("shadow", {}).get("parameters", {}).items():
if name in self._attributes.get("parameters", {}):
self._attributes["parameters"][name].update(values)
else:
self._attributes.setdefault("parameters", {})[name] = HonAttribute(
values
)
self._attributes |= attributes
if self._extra:
self._attributes = self._extra.attributes(self._attributes)
@ -181,14 +202,11 @@ class HonAppliance:
async def update(self, force: bool = False) -> None:
now = datetime.now()
if (
force
or not self._last_update
or self._last_update
< now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL)
):
min_age = now - timedelta(seconds=self._MINIMAL_UPDATE_INTERVAL)
if force or not self._last_update or self._last_update < min_age:
self._last_update = now
await self.load_attributes()
self.sync_params_to_command("settings")
@property
def command_parameters(self) -> Dict[str, Dict[str, str | float]]:
@ -196,7 +214,7 @@ class HonAppliance:
@property
def settings(self) -> Dict[str, Parameter]:
result = {}
result: Dict[str, Parameter] = {}
for name, command in self._commands.items():
for key in command.setting_keys:
setting = command.settings.get(key, self._default_setting)
@ -232,32 +250,67 @@ class HonAppliance:
async def data_archive(self, path: Path) -> str:
return await diagnose.zip_archive(self, path, anonymous=True)
def sync_to_params(self, command_name: str) -> None:
def sync_command_to_params(self, command_name: str) -> None:
if not (command := self.commands.get(command_name)):
return
for key, value in self.attributes.get("parameters", {}).items():
if isinstance(value, str) and (new := command.parameters.get(key)):
for key in self.attributes.get("parameters", {}):
if new := command.parameters.get(key):
self.attributes["parameters"][key].update(
str(new.intern_value), shield=True
)
def sync_command(self, main: str, target: Optional[List[str]] = None) -> None:
def sync_params_to_command(self, command_name: str) -> None:
if not (command := self.commands.get(command_name)):
return
for key in command.setting_keys:
if (
new := self.attributes.get("parameters", {}).get(key)
) is None or new.value == "":
continue
setting = command.settings[key]
try:
if not isinstance(setting, HonParameterRange):
command.settings[key].value = str(new.value)
else:
command.settings[key].value = float(new.value)
except ValueError as error:
_LOGGER.info("Can't set %s - %s", key, error)
continue
def sync_command(
self,
main: str,
target: Optional[List[str] | str] = None,
to_sync: Optional[List[str] | bool] = None,
) -> None:
base: Optional[HonCommand] = self.commands.get(main)
if not base:
return
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):
if isinstance(base_value, HonParameterRange) and isinstance(
parameter, HonParameterRange
for name, target_param in data.parameters.items():
if not (base_param := base.parameters.get(name)):
continue
if to_sync and (
(isinstance(to_sync, list) and name not in to_sync)
or not base_param.mandatory
):
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
parameter.value = base_value.value
continue
self.sync_parameter(base_param, target_param)
def sync_parameter(self, main: Parameter, target: Parameter) -> None:
if isinstance(main, HonParameterRange) and isinstance(
target, HonParameterRange
):
target.max = main.max
target.min = main.min
target.step = main.step
elif isinstance(target, HonParameterRange):
target.max = int(main.value)
target.min = int(main.value)
target.step = 1
elif isinstance(target, HonParameterEnum):
target.values = main.values
target.value = main.value

View File

@ -1,3 +1,4 @@
# pylint: disable=duplicate-code
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase

View File

@ -1,24 +1,16 @@
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase
from pyhon.parameter.program import HonParameterProgram
class Appliance(ApplianceBase):
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["temp"].value = "0"
data["parameters"]["onOffStatus"].value = "0"
data["parameters"]["remoteCtrValid"].value = "0"
data["parameters"]["remainingTimeMM"].value = "0"
data["active"] = data["parameters"]["onOffStatus"] == "1"
if program := int(data["parameters"]["prCode"]):
if (setting := self.parent.settings["startProgram.program"]) and isinstance(
setting, HonParameterProgram
):
data["programName"] = setting.ids.get(program, "")
data["parameters"]["temp"].value = 0
data["parameters"]["onOffStatus"].value = 0
data["parameters"]["remoteCtrValid"].value = 0
data["parameters"]["remainingTimeMM"].value = 0
data["active"] = data["parameters"]["onOffStatus"].value == 1
return data

View File

@ -10,12 +10,12 @@ class Appliance(ApplianceBase):
data["modeZ1"] = "holiday"
elif data["parameters"]["intelligenceMode"] == "1":
data["modeZ1"] = "auto_set"
elif data["parameters"]["quickModeZ1"] == "1":
elif data["parameters"].get("quickModeZ1") == "1":
data["modeZ1"] = "super_cool"
else:
data["modeZ1"] = "no_mode"
if data["parameters"]["quickModeZ2"] == "1":
if data["parameters"].get("quickModeZ2") == "1":
data["modeZ2"] = "super_freeze"
elif data["parameters"]["intelligenceMode"] == "1":
data["modeZ2"] = "auto_set"

View File

@ -1,3 +1,4 @@
# pylint: disable=duplicate-code
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase

View File

@ -1,3 +1,4 @@
# pylint: disable=duplicate-code
from typing import Dict, Any
from pyhon.appliances.base import ApplianceBase

16
pyhon/appliances/wh.py Normal file
View File

@ -0,0 +1,16 @@
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase
from pyhon.parameter.base import HonParameter
class Appliance(ApplianceBase):
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data)
parameter = data.get("parameters", {}).get("onOffStatus")
is_class = isinstance(parameter, HonParameter)
data["active"] = parameter.value == 1 if is_class else parameter == 1
return data
def settings(self, settings: Dict[str, Any]) -> Dict[str, Any]:
return settings

View File

@ -1,3 +1,4 @@
# pylint: disable=duplicate-code
from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase

View File

@ -1,14 +1,15 @@
import asyncio
from contextlib import suppress
from copy import copy
from typing import Dict, Any, Optional, TYPE_CHECKING, List, Collection
from typing import Dict, Any, Optional, TYPE_CHECKING, List
from pyhon.commands import HonCommand
from pyhon.exceptions import NoAuthenticationException
from pyhon.parameter.fixed import HonParameterFixed
from pyhon.parameter.program import HonParameterProgram
if TYPE_CHECKING:
from pyhon import HonAPI, exceptions
from pyhon import HonAPI
from pyhon.appliance import HonAppliance
@ -16,12 +17,12 @@ class HonCommandLoader:
"""Loads and parses hOn command data"""
def __init__(self, api: "HonAPI", appliance: "HonAppliance") -> None:
self._api: "HonAPI" = api
self._appliance: "HonAppliance" = 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] = {}
@ -29,7 +30,7 @@ class HonCommandLoader:
def api(self) -> "HonAPI":
"""api connection object"""
if self._api is None:
raise exceptions.NoAuthenticationException("Missing hOn login")
raise NoAuthenticationException("Missing hOn login")
return self._api
@property
@ -55,7 +56,7 @@ class HonCommandLoader:
async def load_commands(self) -> None:
"""Trigger loading of command data"""
await self._load_data()
self._appliance_data = self._api_commands.pop("applianceModel")
self._appliance_data = self._api_commands.pop("applianceModel", {})
self._get_commands()
self._add_favourites()
self._recover_last_command_states()
@ -183,22 +184,44 @@ class HonCommandLoader:
def _add_favourites(self) -> None:
"""Patch program categories with favourites"""
for favourite in self._favourites:
name = favourite.get("favouriteName", {})
name, command_name, base = self._get_favourite_info(favourite)
if not base:
continue
base_command: HonCommand = copy(base)
self._update_base_command_with_data(base_command, favourite)
self._update_base_command_with_favourite(base_command)
self._update_program_categories(command_name, name, base_command)
def _get_favourite_info(
self, favourite: Dict[str, Any]
) -> tuple[str, str, HonCommand | None]:
name: str = favourite.get("favouriteName", {})
command = favourite.get("command", {})
command_name = command.get("commandName", "")
command_name: str = command.get("commandName", "")
program_name = self._clean_name(command.get("programName", ""))
base: HonCommand = copy(
self.commands[command_name].categories[program_name]
)
base_command = self.commands[command_name].categories.get(program_name)
return name, command_name, base_command
def _update_base_command_with_data(
self, base_command: HonCommand, command: Dict[str, Any]
) -> None:
for data in command.values():
if isinstance(data, str):
continue
for key, value in data.items():
if parameter := base.parameters.get(key):
if not (parameter := base_command.parameters.get(key)):
continue
with suppress(ValueError):
parameter.value = value
def _update_base_command_with_favourite(self, base_command: HonCommand) -> None:
extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom")
base.parameters.update(favourite=extra_param)
if isinstance(program := base.parameters["program"], HonParameterProgram):
base_command.parameters.update(favourite=extra_param)
def _update_program_categories(
self, command_name: str, name: str, base_command: HonCommand
) -> None:
program = base_command.parameters["program"]
if isinstance(program, HonParameterProgram):
program.set_value(name)
self.commands[command_name].categories[name] = base
self.commands[command_name].categories[name] = base_command

View File

@ -27,17 +27,16 @@ class HonCommand:
categories: Optional[Dict[str, "HonCommand"]] = None,
category_name: str = "",
):
self._api: Optional[HonAPI] = appliance.api
self._appliance: "HonAppliance" = appliance
self._name: str = name
self._api: Optional[HonAPI] = None
self._appliance: "HonAppliance" = appliance
self._categories: Optional[Dict[str, "HonCommand"]] = categories
self._category_name: str = category_name
self._description: str = attributes.pop("description", "")
self._protocol_type: str = attributes.pop("protocolType", "")
self._parameters: Dict[str, HonParameter] = {}
self._parameters: Dict[str, Parameter] = {}
self._data: Dict[str, Any] = {}
self._available_settings: Dict[str, HonParameter] = {}
self._rules: List[HonRuleSet] = []
attributes.pop("description", "")
attributes.pop("protocolType", "")
self._load_parameters(attributes)
def __repr__(self) -> str:
@ -49,6 +48,8 @@ class HonCommand:
@property
def api(self) -> "HonAPI":
if self._api is None and self._appliance:
self._api = self._appliance.api
if self._api is None:
raise exceptions.NoAuthenticationException("Missing hOn login")
return self._api
@ -76,12 +77,23 @@ class HonCommand:
result.setdefault(parameter.group, {})[name] = parameter.intern_value
return result
@property
def mandatory_parameter_groups(self) -> Dict[str, Dict[str, Union[str, float]]]:
result: Dict[str, Dict[str, Union[str, float]]] = {}
for name, parameter in self._parameters.items():
if parameter.mandatory:
result.setdefault(parameter.group, {})[name] = parameter.intern_value
return result
@property
def parameter_value(self) -> Dict[str, Union[str, float]]:
return {n: p.value for n, p in self._parameters.items()}
def _load_parameters(self, attributes: Dict[str, Dict[str, Any]]) -> None:
def _load_parameters(self, attributes: Dict[str, Dict[str, Any] | Any]) -> None:
for key, items in attributes.items():
if not isinstance(items, dict):
_LOGGER.info("Loading Attributes - Skipping %s", str(items))
continue
for name, data in items.items():
self._create_parameters(data, name, key)
for rule in self._rules:
@ -93,10 +105,12 @@ class HonCommand:
if name == "zoneMap" and self._appliance.zone:
data["default"] = self._appliance.zone
if data.get("category") == "rule":
if "fixedValue" not in data:
_LOGGER.error("Rule not supported: %s", data)
else:
if "fixedValue" in data:
self._rules.append(HonRuleSet(self, data["fixedValue"]))
elif "enumValues" in data:
self._rules.append(HonRuleSet(self, data["enumValues"]))
else:
_LOGGER.warning("Rule not supported: %s", data)
match data.get("typology"):
case "range":
self._parameters[name] = HonParameterRange(name, data, parameter)
@ -111,14 +125,33 @@ class HonCommand:
name = "program" if "PROGRAM" in self._category_name else "category"
self._parameters[name] = HonParameterProgram(name, self, "custom")
async def send(self) -> bool:
params = self.parameter_groups.get("parameters", {})
async def send(self, only_mandatory: bool = False) -> bool:
grouped_params = (
self.mandatory_parameter_groups if only_mandatory else self.parameter_groups
)
params = grouped_params.get("parameters", {})
return await self.send_parameters(params)
async def send_specific(self, param_names: List[str]) -> bool:
params: Dict[str, str | float] = {}
for key, parameter in self._parameters.items():
if key in param_names or parameter.mandatory:
params[key] = parameter.value
return await self.send_parameters(params)
async def send_parameters(self, params: Dict[str, str | float]) -> bool:
ancillary_params = self.parameter_groups.get("ancillaryParameters", {})
ancillary_params.pop("programRules", None)
self.appliance.sync_to_params(self.name)
if "prStr" in params:
params["prStr"] = self._category_name.upper()
self.appliance.sync_command_to_params(self.name)
try:
result = await self.api.send_command(
self._appliance, self._name, params, ancillary_params
self._appliance,
self._name,
params,
ancillary_params,
self._category_name,
)
if not result:
_LOGGER.error(result)
@ -169,3 +202,7 @@ class HonCommand:
else:
result[name] = parameter
return result
def reset(self) -> None:
for parameter in self._parameters.values():
parameter.reset()

View File

@ -75,8 +75,12 @@ class HonAPI:
async def load_appliances(self) -> List[Dict[str, Any]]:
async with self._hon.get(f"{const.API_URL}/commands/v1/appliance") as resp:
if result := await resp.json():
return result.get("payload", {}).get("appliances", {})
result = await resp.json()
if result:
appliances: List[Dict[str, Any]] = result.get("payload", {}).get(
"appliances", {}
)
return appliances
return []
async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]:
@ -112,7 +116,8 @@ class HonAPI:
result: Dict[str, Any] = await response.json()
if not result or not result.get("payload"):
return []
return result["payload"]["history"]
command_history: List[Dict[str, Any]] = result["payload"]["history"]
return command_history
async def load_favourites(self, appliance: HonAppliance) -> List[Dict[str, Any]]:
url: str = (
@ -122,14 +127,17 @@ class HonAPI:
result: Dict[str, Any] = await response.json()
if not result or not result.get("payload"):
return []
return result["payload"]["favourites"]
favourites: List[Dict[str, Any]] = result["payload"]["favourites"]
return favourites
async def load_last_activity(self, appliance: HonAppliance) -> Dict[str, Any]:
url: str = f"{const.API_URL}/commands/v1/retrieve-last-activity"
params: Dict[str, str] = {"macAddress": appliance.mac_address}
async with self._hon.get(url, params=params) as response:
result: Dict[str, Any] = await response.json()
if result and (activity := result.get("attributes")):
if result:
activity: Dict[str, Any] = result.get("attributes", "")
if activity:
return activity
return {}
@ -142,7 +150,10 @@ class HonAPI:
async with self._hon.get(url, params=params) as response:
result: Dict[str, Any] = await response.json()
if result:
return result.get("payload", {}).get("applianceModel", {})
appliance_data: Dict[str, Any] = result.get("payload", {}).get(
"applianceModel", {}
)
return appliance_data
return {}
async def load_attributes(self, appliance: HonAppliance) -> Dict[str, Any]:
@ -153,7 +164,8 @@ class HonAPI:
}
url: str = f"{const.API_URL}/commands/v1/context"
async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {})
attributes: Dict[str, Any] = (await response.json()).get("payload", {})
return attributes
async def load_statistics(self, appliance: HonAppliance) -> Dict[str, Any]:
params: Dict[str, str] = {
@ -162,13 +174,15 @@ class HonAPI:
}
url: str = f"{const.API_URL}/commands/v1/statistics"
async with self._hon.get(url, params=params) as response:
return (await response.json()).get("payload", {})
statistics: Dict[str, Any] = (await response.json()).get("payload", {})
return statistics
async def load_maintenance(self, appliance: HonAppliance) -> Dict[str, Any]:
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", {})
maintenance: Dict[str, Any] = (await response.json()).get("payload", {})
return maintenance
async def send_command(
self,
@ -176,6 +190,7 @@ class HonAPI:
command: str,
parameters: Dict[str, Any],
ancillary_parameters: Dict[str, Any],
program_name: str = "",
) -> bool:
now: str = datetime.utcnow().isoformat()
data: Dict[str, Any] = {
@ -194,6 +209,8 @@ class HonAPI:
"parameters": parameters,
"applianceType": appliance.appliance_type,
}
if command == "startProgram" and program_name:
data.update({"programName": program_name.upper()})
url: str = f"{const.API_URL}/commands/v1/send"
async with self._hon.post(url, json=data) as response:
json_data: Dict[str, Any] = await response.json()
@ -207,9 +224,8 @@ class HonAPI:
url: str = f"{const.API_URL}/config/v1/program-list-rules"
async with self._hon_anonymous.get(url) as response:
result: Dict[str, Any] = await response.json()
if result and (data := result.get("payload")):
data: Dict[str, Any] = result.get("payload", {})
return data
return {}
async def app_config(
self, language: str = "en", beta: bool = True
@ -223,17 +239,17 @@ class HonAPI:
}
payload: str = json.dumps(payload_data, separators=(",", ":"))
async with self._hon_anonymous.post(url, data=payload) as response:
if (result := await response.json()) and (data := result.get("payload")):
result = await response.json()
data: Dict[str, Any] = result.get("payload", {})
return data
return {}
async def translation_keys(self, language: str = "en") -> Dict[str, Any]:
config = await self.app_config(language=language)
if url := config.get("language", {}).get("jsonPath"):
async with self._hon_anonymous.get(url) as response:
if result := await response.json():
return result
if not (url := config.get("language", {}).get("jsonPath")):
return {}
async with self._hon_anonymous.get(url) as response:
result: Dict[str, Any] = await response.json()
return result
async def close(self) -> None:
if self._hon_handler is not None:
@ -250,17 +266,27 @@ class TestAPI(HonAPI):
def _load_json(self, appliance: HonAppliance, file: str) -> Dict[str, Any]:
directory = f"{appliance.appliance_type}_{appliance.appliance_model_id}".lower()
path = f"{self._path}/{directory}/{file}.json"
if not (path := self._path / directory / f"{file}.json").exists():
_LOGGER.warning("Can't open %s", str(path))
return {}
with open(path, "r", encoding="utf-8") as json_file:
return json.loads(json_file.read())
text = json_file.read()
try:
data: Dict[str, Any] = json.loads(text)
return data
except json.decoder.JSONDecodeError as error:
_LOGGER.error("%s - %s", str(path), error)
return {}
async def load_appliances(self) -> List[Dict[str, Any]]:
result = []
for appliance in self._path.glob("*/"):
with open(
appliance / "appliance_data.json", "r", encoding="utf-8"
) as json_file:
file = appliance / "appliance_data.json"
with open(file, "r", encoding="utf-8") as json_file:
try:
result.append(json.loads(json_file.read()))
except json.decoder.JSONDecodeError as error:
_LOGGER.error("%s - %s", str(file), error)
return result
async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]:
@ -296,5 +322,12 @@ class TestAPI(HonAPI):
command: str,
parameters: Dict[str, Any],
ancillary_parameters: Dict[str, Any],
program_name: str = "",
) -> bool:
_LOGGER.info(
"%s - %s - %s",
str(parameters),
str(ancillary_parameters),
str(program_name),
)
return True

View File

@ -6,7 +6,7 @@ import urllib
from contextlib import suppress
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, Optional, Any
from typing import Dict, Optional, Any, List
from urllib import parse
from urllib.parse import quote
@ -30,6 +30,14 @@ class HonLoginData:
loaded: Optional[Dict[str, Any]] = None
@dataclass
class HonAuthData:
access_token: str = ""
refresh_token: str = ""
cognito_token: str = ""
id_token: str = ""
class HonAuth:
_TOKEN_EXPIRES_AFTER_HOURS = 8
_TOKEN_EXPIRE_WARNING_HOURS = 7
@ -46,28 +54,25 @@ class HonAuth:
self._login_data = HonLoginData()
self._login_data.email = email
self._login_data.password = password
self._access_token = ""
self._refresh_token = ""
self._cognito_token = ""
self._id_token = ""
self._device = device
self._expires: datetime = datetime.utcnow()
self._auth = HonAuthData()
@property
def cognito_token(self) -> str:
return self._cognito_token
return self._auth.cognito_token
@property
def id_token(self) -> str:
return self._id_token
return self._auth.id_token
@property
def access_token(self) -> str:
return self._access_token
return self._auth.access_token
@property
def refresh_token(self) -> str:
return self._refresh_token
return self._auth.refresh_token
def _check_token_expiration(self, hours: int) -> bool:
return datetime.utcnow() >= self._expires + timedelta(hours=hours)
@ -115,7 +120,8 @@ class HonAuth:
async with self._request.get(url) as response:
text = await response.text()
self._expires = datetime.utcnow()
if not (login_url := re.findall("url = '(.+?)'", text)):
login_url: List[str] = re.findall("url = '(.+?)'", text)
if not login_url:
if "oauth/done#access_token=" in text:
self._parse_token_data(text)
raise exceptions.HonNoAuthenticationNeeded()
@ -184,18 +190,19 @@ class HonAuth:
if response.status == 200:
with suppress(json.JSONDecodeError, KeyError):
result = await response.json()
return result["events"][0]["attributes"]["values"]["url"]
url: str = result["events"][0]["attributes"]["values"]["url"]
return url
await self._error_logger(response)
return ""
def _parse_token_data(self, text: str) -> bool:
if access_token := re.findall("access_token=(.*?)&", text):
self._access_token = access_token[0]
self._auth.access_token = access_token[0]
if refresh_token := re.findall("refresh_token=(.*?)&", text):
self._refresh_token = refresh_token[0]
self._auth.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
self._auth.id_token = id_token[0]
return bool(access_token and refresh_token and id_token)
async def _get_token(self, url: str) -> bool:
async with self._request.get(url) as response:
@ -227,7 +234,7 @@ class HonAuth:
return True
async def _api_auth(self) -> bool:
post_headers = {"id-token": self._id_token}
post_headers = {"id-token": self._auth.id_token}
data = self._device.get()
async with self._request.post(
f"{const.API_URL}/auth/v1/login", headers=post_headers, json=data
@ -237,8 +244,8 @@ class HonAuth:
except json.JSONDecodeError:
await self._error_logger(response)
return False
self._cognito_token = json_data.get("cognitoUser", {}).get("Token", "")
if not self._cognito_token:
self._auth.cognito_token = json_data.get("cognitoUser", {}).get("Token", "")
if not self._auth.cognito_token:
_LOGGER.error(json_data)
raise exceptions.HonAuthenticationError()
return True
@ -260,7 +267,7 @@ class HonAuth:
async def refresh(self) -> bool:
params = {
"client_id": const.CLIENT_ID,
"refresh_token": self._refresh_token,
"refresh_token": self._auth.refresh_token,
"grant_type": "refresh_token",
}
async with self._request.post(
@ -271,14 +278,14 @@ class HonAuth:
return False
data = await response.json()
self._expires = datetime.utcnow()
self._id_token = data["id_token"]
self._access_token = data["access_token"]
self._auth.id_token = data["id_token"]
self._auth.access_token = data["access_token"]
return await self._api_auth()
def clear(self) -> None:
self._session.cookie_jar.clear_domain(const.AUTH_API.split("/")[-2])
self._request.called_urls = []
self._cognito_token = ""
self._id_token = ""
self._access_token = ""
self._refresh_token = ""
self._auth.cognito_token = ""
self._auth.id_token = ""
self._auth.access_token = ""
self._auth.refresh_token = ""

View File

@ -21,7 +21,7 @@ class HonDevice:
return self._os_version
@property
def os(self) -> str:
def os_type(self) -> str:
return self._os
@property
@ -36,7 +36,7 @@ class HonDevice:
result: Dict[str, str | int] = {
"appVersion": self.app_version,
"mobileId": self.mobile_id,
"os": self.os,
"os": self.os_type,
"osVersion": self.os_version,
"deviceModel": self.device_model,
}

View File

@ -2,7 +2,7 @@ import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from types import TracebackType
from typing import Optional, Dict, Type, Any, Protocol
from typing import Optional, Dict, Type, Any
import aiohttp
from typing_extensions import Self
@ -47,10 +47,11 @@ class ConnectionHandler:
return self
@asynccontextmanager
def _intercept(
async def _intercept(
self, method: Callback, url: str | URL, *args: Any, **kwargs: Dict[str, Any]
) -> AsyncIterator[aiohttp.ClientResponse]:
raise NotImplementedError
async with method(url, *args, **kwargs) as response:
yield response
@asynccontextmanager
async def get(
@ -59,7 +60,8 @@ class ConnectionHandler:
if self._session is None:
raise exceptions.NoSessionException()
response: aiohttp.ClientResponse
async with self._intercept(self._session.get, *args, **kwargs) as response: # type: ignore[arg-type]
args = self._session.get, *args
async with self._intercept(*args, **kwargs) as response:
yield response
@asynccontextmanager
@ -69,7 +71,8 @@ class ConnectionHandler:
if self._session is None:
raise exceptions.NoSessionException()
response: aiohttp.ClientResponse
async with self._intercept(self._session.post, *args, **kwargs) as response: # type: ignore[arg-type]
args = self._session.post, *args
async with self._intercept(*args, **kwargs) as response:
yield response
async def close(self) -> None:

View File

@ -57,7 +57,7 @@ class HonConnectionHandler(ConnectionHandler):
async def _intercept(
self, method: Callback, url: str | URL, *args: Any, **kwargs: Any
) -> AsyncIterator[aiohttp.ClientResponse]:
loop: int = kwargs.get("loop", 0)
loop: int = kwargs.pop("loop", 0)
kwargs["headers"] = await self._check_headers(kwargs.get("headers", {}))
async with method(url, *args, **kwargs) as response:
if (
@ -95,11 +95,11 @@ class HonConnectionHandler(ConnectionHandler):
try:
await response.json()
yield response
except json.JSONDecodeError:
except json.JSONDecodeError as exc:
_LOGGER.warning(
"%s - JsonDecodeError %s - %s",
response.request_info.url,
response.status,
await response.text(),
)
raise HonAuthenticationError("Decode Error")
raise HonAuthenticationError("Decode Error") from exc

View File

@ -2,9 +2,11 @@ AUTH_API = "https://account2.hon-smarthome.com"
API_URL = "https://api-iot.he.services"
API_KEY = "GRCqFhC6Gk@ikWXm1RmnSmX1cm,MxY-configuration"
APP = "hon"
# All seen id's (different accounts, different devices) are the same, so I guess this hash is static
CLIENT_ID = "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9.HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
APP_VERSION = "2.0.10"
CLIENT_ID = (
"3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9."
"HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
)
APP_VERSION = "2.4.7"
OS_VERSION = 31
OS = "android"
DEVICE_MODEL = "exynos9820"

View File

@ -70,7 +70,7 @@ async def zip_archive(
) -> str:
data = await appliance_data(appliance, path, anonymous)
archive = data[0].parent
shutil.make_archive(str(archive.parent), "zip", archive)
shutil.make_archive(str(archive), "zip", archive)
shutil.rmtree(archive)
return f"{archive.stem}.zip"
@ -89,12 +89,11 @@ def yaml_export(appliance: "HonAppliance", anonymous: bool = False) -> str:
if anonymous:
for sensible in ["serialNumber", "coords"]:
data.get("appliance", {}).pop(sensible, None)
data = {
"data": data,
"commands": printer.create_command(appliance.commands),
"rules": printer.create_rules(appliance.commands),
}
result = printer.pretty_print(data)
result = printer.pretty_print({"data": data})
if commands := printer.create_commands(appliance.commands):
result += printer.pretty_print({"commands": commands})
if rules := printer.create_rules(appliance.commands):
result += printer.pretty_print({"rules": rules})
if anonymous:
result = anonymize_data(result)
return result

View File

@ -7,9 +7,10 @@ from typing import List, Optional, Dict, Any, Type
from aiohttp import ClientSession
from typing_extensions import Self
from pyhon import HonAPI, exceptions
from pyhon.appliance import HonAppliance
from pyhon.connection.api import HonAPI
from pyhon.connection.api import TestAPI
from pyhon.exceptions import NoAuthenticationException
_LOGGER = logging.getLogger(__name__)
@ -43,7 +44,7 @@ class Hon:
@property
def api(self) -> HonAPI:
if self._api is None:
raise exceptions.NoAuthenticationException
raise NoAuthenticationException
return self._api
@property

View File

@ -7,14 +7,21 @@ if TYPE_CHECKING:
class HonParameter:
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
self._key = key
self._category: str = attributes.get("category", "")
self._typology: str = attributes.get("typology", "")
self._mandatory: int = attributes.get("mandatory", 0)
self._attributes = attributes
self._category: str = ""
self._typology: str = ""
self._mandatory: int = 0
self._value: str | float = ""
self._group: str = group
self._triggers: Dict[
str, List[Tuple[Callable[["HonRule"], None], "HonRule"]]
] = {}
self._set_attributes()
def _set_attributes(self) -> None:
self._category = self._attributes.get("category", "")
self._typology = self._attributes.get("typology", "")
self._mandatory = self._attributes.get("mandatory", 0)
@property
def key(self) -> str:
@ -61,8 +68,9 @@ class HonParameter:
self._triggers.setdefault(value, []).append((func, data))
def check_trigger(self, value: str | float) -> None:
if str(value) in self._triggers:
for trigger in self._triggers[str(value)]:
triggers = {str(k).lower(): v for k, v in self._triggers.items()}
if str(value).lower() in triggers:
for trigger in triggers[str(value)]:
func, args = trigger
func(args)
@ -85,3 +93,6 @@ class HonParameter:
param[rule.param_key] = rule.param_data.get("defaultValue", "")
return result
def reset(self) -> None:
self._set_attributes()

View File

@ -10,12 +10,19 @@ def clean_value(value: str | float) -> str:
class HonParameterEnum(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
super().__init__(key, attributes, group)
self._default = attributes.get("defaultValue")
self._value = self._default or "0"
self._values: List[str] = attributes.get("enumValues", [])
self._default: str | float = ""
self._value: str | float = ""
self._values: List[str] = []
self._set_attributes()
if self._default and clean_value(self._default.strip("[]")) not in self.values:
self._values.append(self._default)
def _set_attributes(self) -> None:
super()._set_attributes()
self._default = self._attributes.get("defaultValue", "")
self._value = self._default or "0"
self._values = self._attributes.get("enumValues", [])
def __repr__(self) -> str:
return f"{self.__class__} (<{self.key}> {self.values})"
@ -41,4 +48,4 @@ class HonParameterEnum(HonParameter):
self._value = value
self.check_trigger(value)
else:
raise ValueError(f"Allowed values {self._values}")
raise ValueError(f"Allowed values: {self._values} But was: {value}")

View File

@ -6,14 +6,19 @@ from pyhon.parameter.base import HonParameter
class HonParameterFixed(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
super().__init__(key, attributes, group)
self._value = attributes.get("fixedValue", None)
self._value: str | float = ""
self._set_attributes()
def _set_attributes(self) -> None:
super()._set_attributes()
self._value = self._attributes.get("fixedValue", "")
def __repr__(self) -> str:
return f"{self.__class__} (<{self.key}> fixed)"
@property
def value(self) -> str | float:
return self._value if self._value is not None else "0"
return self._value if self._value != "" else "0"
@value.setter
def value(self, value: str | float) -> None:

View File

@ -28,7 +28,7 @@ class HonParameterProgram(HonParameterEnum):
if value in self.values:
self._command.category = value
else:
raise ValueError(f"Allowed values {self.values}")
raise ValueError(f"Allowed values: {self.values} But was: {value}")
@property
def values(self) -> List[str]:
@ -37,17 +37,19 @@ class HonParameterProgram(HonParameterEnum):
@values.setter
def values(self, values: List[str]) -> None:
return
raise ValueError("Cant set values {values}")
@property
def ids(self) -> Dict[int, str]:
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")
and not ((fav := p.parameters.get("favourite")) and fav.value == "1")
}
values: Dict[int, str] = {}
for name, parameter in self._programs.items():
if "iot_" in name:
continue
if parameter.parameters.get("prCode"):
continue
if (fav := parameter.parameters.get("favourite")) and fav.value == "1":
continue
values[int(parameter.parameters["prCode"].value)] = name
return dict(sorted(values.items()))
def set_value(self, value: str) -> None:

View File

@ -7,11 +7,20 @@ from pyhon.parameter.base import HonParameter
class HonParameterRange(HonParameter):
def __init__(self, key: str, attributes: Dict[str, Any], group: str) -> None:
super().__init__(key, attributes, group)
self._min: float = str_to_float(attributes["minimumValue"])
self._max: float = str_to_float(attributes["maximumValue"])
self._step: float = str_to_float(attributes["incrementValue"])
self._default: float = str_to_float(attributes.get("defaultValue", self.min))
self._value: float = self._default
self._min: float = 0
self._max: float = 0
self._step: float = 0
self._default: float = 0
self._value: float = 0
self._set_attributes()
def _set_attributes(self) -> None:
super()._set_attributes()
self._min = str_to_float(self._attributes.get("minimumValue", 0))
self._max = str_to_float(self._attributes.get("maximumValue", 0))
self._step = str_to_float(self._attributes.get("incrementValue", 0))
self._default = str_to_float(self._attributes.get("defaultValue", self.min))
self._value = self._default
def __repr__(self) -> str:
return f"{self.__class__} (<{self.key}> [{self.min} - {self.max}])"
@ -49,11 +58,14 @@ class HonParameterRange(HonParameter):
@value.setter
def value(self, value: str | float) -> None:
value = str_to_float(value)
if self.min <= value <= self.max and not (value - self.min) % self.step:
if self.min <= value <= self.max and not ((value - self.min) * 100) % (
self.step * 100
):
self._value = value
self.check_trigger(value)
else:
raise ValueError(f"Allowed: min {self.min} max {self.max} step {self.step}")
allowed = f"min {self.min} max {self.max} step {self.step}"
raise ValueError(f"Allowed: {allowed} But was: {value}")
@property
def values(self) -> List[str]:

View File

@ -29,37 +29,30 @@ def pretty_print(
whitespace: str = " ",
) -> str:
result = ""
if isinstance(data, list):
if key:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
space = whitespace * intend
if isinstance(data, (dict, list)) and key:
result += f"{space}{'- ' if is_list else ''}{key}:\n"
intend += 1
if isinstance(data, list):
for i, value in enumerate(data):
result += pretty_print(
value, intend=intend, is_list=True, whitespace=whitespace
)
elif isinstance(data, dict):
if key:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}:\n"
intend += 1
for i, (key, value) in enumerate(sorted(data.items())):
if is_list and not i:
for i, (list_key, value) in enumerate(sorted(data.items())):
result += pretty_print(
value, key=key, intend=intend, is_list=True, whitespace=whitespace
)
elif is_list:
result += pretty_print(
value, key=key, intend=intend + 1, whitespace=whitespace
value,
key=list_key,
intend=intend + (is_list if i else 0),
is_list=is_list and not i,
whitespace=whitespace,
)
else:
result += pretty_print(
value, key=key, intend=intend, whitespace=whitespace
)
else:
result += f"{whitespace * intend}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n"
result += f"{space}{'- ' if is_list else ''}{key}{': ' if key else ''}{data}\n"
return result
def create_command(
def create_commands(
commands: Dict[str, "HonCommand"], concat: bool = False
) -> Dict[str, Any]:
result: Dict[str, Any] = {}

0
pyhon/py.typed Normal file
View File

View File

@ -3,6 +3,7 @@ from typing import List, Dict, TYPE_CHECKING, Any, Optional
from pyhon.parameter.enum import HonParameterEnum
from pyhon.parameter.range import HonParameterRange
from pyhon.typedefs import Parameter
if TYPE_CHECKING:
from pyhon.commands import HonCommand
@ -24,6 +25,10 @@ class HonRuleSet:
self._rules: Dict[str, List[HonRule]] = {}
self._parse_rule(rule)
@property
def rules(self) -> Dict[str, List[HonRule]]:
return self._rules
def _parse_rule(self, rule: Dict[str, Any]) -> None:
for param_key, params in rule.items():
param_key = self._command.appliance.options.get(param_key, param_key)
@ -51,6 +56,11 @@ class HonRuleSet:
extra[trigger_key] = trigger_value
for extra_key, extra_data in param_data.items():
self._parse_conditions(param_key, extra_key, extra_data, extra)
else:
param_data = {"typology": "fixed", "fixedValue": param_data}
self._create_rule(
param_key, trigger_key, trigger_value, param_data, extra
)
def _create_rule(
self,
@ -83,29 +93,47 @@ class HonRuleSet:
for rule in rules:
self._rules.setdefault(key, []).append(rule)
def _add_trigger(self, parameter: "HonParameter", data: HonRule) -> None:
def apply(rule: HonRule) -> None:
if rule.extras is not None:
def _extra_rules_matches(self, rule: HonRule) -> bool:
if rule.extras:
for key, value in rule.extras.items():
if not self._command.parameters.get(key):
return False
if str(self._command.parameters.get(key)) != str(value):
return
if param := self._command.parameters.get(rule.param_key):
if value := rule.param_data.get("fixedValue", ""):
if isinstance(param, HonParameterEnum) and set(param.values) != {
str(value)
}:
return False
return True
def _apply_fixed(self, param: Parameter, value: str | float) -> None:
if isinstance(param, HonParameterEnum) and set(param.values) != {str(value)}:
param.values = [str(value)]
param.value = str(value)
elif isinstance(param, HonParameterRange):
if float(value) < param.min:
param.min = float(value)
elif float(value) > param.max:
param.max = float(value)
param.value = float(value)
return
param.value = str(value)
elif rule.param_data.get("typology") == "enum":
if isinstance(param, HonParameterEnum):
def _apply_enum(self, param: Parameter, rule: HonRule) -> None:
if not isinstance(param, HonParameterEnum):
return
if enum_values := rule.param_data.get("enumValues"):
param.values = enum_values.split("|")
if default_value := rule.param_data.get("defaultValue"):
param.value = default_value
def _add_trigger(self, parameter: "HonParameter", data: HonRule) -> None:
def apply(rule: HonRule) -> None:
if not self._extra_rules_matches(rule):
return
if not (param := self._command.parameters.get(rule.param_key)):
return
if fixed_value := rule.param_data.get("fixedValue", ""):
self._apply_fixed(param, fixed_value)
elif rule.param_data.get("typology") == "enum":
self._apply_enum(param, rule)
parameter.add_trigger(data.trigger_value, apply, data)
def patch(self) -> None:

View File

@ -11,7 +11,7 @@ if TYPE_CHECKING:
from pyhon.parameter.range import HonParameterRange
class Callback(Protocol):
class Callback(Protocol): # pylint: disable=too-few-public-methods
def __call__(
self, url: str | URL, *args: Any, **kwargs: Any
) -> aiohttp.client._RequestContextManager:

View File

@ -1,2 +1,3 @@
aiohttp==3.8.4
yarl==1.8.2
aiohttp>=3.8
yarl>=1.8
typing-extensions>=4.8

View File

@ -1,4 +1,5 @@
black==23.3.0
flake8==6.0.0
mypy==1.2.0
pylint==2.17.2
black>=22.12
flake8>=6.0
mypy>=0.991
pylint>=2.15
setuptools>=62.3

View File

@ -2,12 +2,12 @@
from setuptools import setup, find_packages
with open("README.md", "r") as f:
with open("README.md", "r", encoding="utf-8") as f:
long_description = f.read()
setup(
name="pyhOn",
version="0.14.4",
version="0.15.15",
author="Andre Basche",
description="Control hOn devices with python",
long_description=long_description,
@ -21,7 +21,7 @@ setup(
packages=find_packages(),
include_package_data=True,
python_requires=">=3.10",
install_requires=["aiohttp"],
install_requires=["aiohttp>=3.8", "typing-extensions>=4.8", "yarl>=1.8"],
classifiers=[
"Development Status :: 4 - Beta",
"Environment :: Console",
@ -30,6 +30,7 @@ setup(
"Operating System :: OS Independent",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
],
entry_points={