Compare commits

..

No commits in common. "main" and "v0.15.0" have entirely different histories.

19 changed files with 77 additions and 191 deletions

View File

@ -13,7 +13,7 @@ jobs:
strategy: strategy:
fail-fast: false fail-fast: false
matrix: matrix:
python-version: ["3.10", "3.11", "3.12"] python-version: ["3.10", "3.11"]
steps: steps:
- uses: actions/checkout@v3 - uses: actions/checkout@v3

View File

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

View File

@ -1,14 +1,3 @@
# 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!** **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 # pyhOn

View File

@ -1,24 +1,9 @@
[mypy] [mypy]
check_untyped_defs = true check_untyped_defs = True
disallow_any_generics = true disallow_any_generics = True
disallow_any_unimported = true disallow_untyped_defs = True
disallow_incomplete_defs = true disallow_any_unimported = True
disallow_subclassing_any = true no_implicit_optional = True
disallow_untyped_calls = true warn_return_any = True
disallow_untyped_decorators = true show_error_codes = True
disallow_untyped_defs = true warn_unused_ignores = 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

@ -3,7 +3,7 @@ import logging
import re import re
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pathlib import Path from pathlib import Path
from typing import Optional, Dict, Any, TYPE_CHECKING, List, TypeVar, overload from typing import Optional, Dict, Any, TYPE_CHECKING, List
from pyhon import diagnose, exceptions from pyhon import diagnose, exceptions
from pyhon.appliances.base import ApplianceBase from pyhon.appliances.base import ApplianceBase
@ -20,8 +20,6 @@ if TYPE_CHECKING:
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
T = TypeVar("T")
# pylint: disable=too-many-public-methods,too-many-instance-attributes # pylint: disable=too-many-public-methods,too-many-instance-attributes
class HonAppliance: class HonAppliance:
@ -51,35 +49,24 @@ class HonAppliance:
except ModuleNotFoundError: except ModuleNotFoundError:
self._extra = None 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: def __getitem__(self, item: str) -> Any:
if self._zone: if self._zone:
item += f"Z{self._zone}" item += f"Z{self._zone}"
if "." in item: if "." in item:
return self._get_nested_item(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
if item in self.data: if item in self.data:
return self.data[item] return self.data[item]
if item in self.attributes["parameters"]: if item in self.attributes["parameters"]:
return self.attributes["parameters"][item].value return self.attributes["parameters"][item].value
return self.info[item] return self.info[item]
@overload def get(self, item: str, default: Any = None) -> Any:
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: try:
return self[item] return self[item]
except (KeyError, IndexError): except (KeyError, IndexError):
@ -94,15 +81,15 @@ class HonAppliance:
@property @property
def appliance_model_id(self) -> str: def appliance_model_id(self) -> str:
return str(self._info.get("applianceModelId", "")) return self._info.get("applianceModelId", "")
@property @property
def appliance_type(self) -> str: def appliance_type(self) -> str:
return str(self._info.get("applianceTypeName", "")) return self._info.get("applianceTypeName", "")
@property @property
def mac_address(self) -> str: def mac_address(self) -> str:
return str(self.info.get("macAddress", "")) return self.info.get("macAddress", "")
@property @property
def unique_id(self) -> str: def unique_id(self) -> str:
@ -138,11 +125,11 @@ class HonAppliance:
@property @property
def model_id(self) -> int: def model_id(self) -> int:
return int(self._info.get("applianceModelId", 0)) return self._info.get("applianceModelId", 0)
@property @property
def options(self) -> Dict[str, Any]: def options(self) -> Dict[str, Any]:
return dict(self._appliance_model.get("options", {})) return self._appliance_model.get("options", {})
@property @property
def commands(self) -> Dict[str, HonCommand]: def commands(self) -> Dict[str, HonCommand]:
@ -263,9 +250,7 @@ class HonAppliance:
if not (command := self.commands.get(command_name)): if not (command := self.commands.get(command_name)):
return return
for key in command.setting_keys: for key in command.setting_keys:
if ( if (new := self.attributes.get("parameters", {}).get(key)) is None:
new := self.attributes.get("parameters", {}).get(key)
) is None or new.value == "":
continue continue
setting = command.settings[key] setting = command.settings[key]
try: try:
@ -277,12 +262,7 @@ class HonAppliance:
_LOGGER.info("Can't set %s - %s", key, error) _LOGGER.info("Can't set %s - %s", key, error)
continue continue
def sync_command( def sync_command(self, main: str, target: Optional[List[str] | str] = None) -> None:
self,
main: str,
target: Optional[List[str] | str] = None,
to_sync: Optional[List[str] | bool] = None,
) -> None:
base: Optional[HonCommand] = self.commands.get(main) base: Optional[HonCommand] = self.commands.get(main)
if not base: if not base:
return return
@ -292,12 +272,7 @@ class HonAppliance:
for name, target_param in data.parameters.items(): for name, target_param in data.parameters.items():
if not (base_param := base.parameters.get(name)): if not (base_param := base.parameters.get(name)):
continue return
if to_sync and (
(isinstance(to_sync, list) and name not in to_sync)
or not base_param.mandatory
):
continue
self.sync_parameter(base_param, target_param) self.sync_parameter(base_param, target_param)
def sync_parameter(self, main: Parameter, target: Parameter) -> None: def sync_parameter(self, main: Parameter, target: Parameter) -> None:

View File

@ -7,10 +7,10 @@ class Appliance(ApplianceBase):
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED": if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
data["parameters"]["temp"].value = 0 data["parameters"]["temp"].value = "0"
data["parameters"]["onOffStatus"].value = 0 data["parameters"]["onOffStatus"].value = "0"
data["parameters"]["remoteCtrValid"].value = 0 data["parameters"]["remoteCtrValid"].value = "0"
data["parameters"]["remainingTimeMM"].value = 0 data["parameters"]["remainingTimeMM"].value = "0"
data["active"] = data["parameters"]["onOffStatus"].value == 1 data["active"] = data["parameters"]["onOffStatus"] == "1"
return data return data

View File

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

View File

@ -1,16 +1,11 @@
from typing import Any, Dict from typing import Any, Dict
from pyhon.appliances.base import ApplianceBase from pyhon.appliances.base import ApplianceBase
from pyhon.parameter.base import HonParameter
class Appliance(ApplianceBase): class Appliance(ApplianceBase):
def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]: def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
data = super().attributes(data) data = super().attributes(data)
parameter = data.get("parameters", {}).get("onOffStatus") data["active"] = data["parameters"]["onOffStatus"] == "1"
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 data
return settings

View File

@ -184,44 +184,22 @@ class HonCommandLoader:
def _add_favourites(self) -> None: def _add_favourites(self) -> None:
"""Patch program categories with favourites""" """Patch program categories with favourites"""
for favourite in self._favourites: for favourite in self._favourites:
name, command_name, base = self._get_favourite_info(favourite) name = favourite.get("favouriteName", {})
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 = favourite.get("command", {})
command_name: str = command.get("commandName", "") command_name = command.get("commandName", "")
program_name = self._clean_name(command.get("programName", "")) program_name = self._clean_name(command.get("programName", ""))
base_command = self.commands[command_name].categories.get(program_name) base: HonCommand = copy(
return name, command_name, base_command self.commands[command_name].categories[program_name]
)
def _update_base_command_with_data(
self, base_command: HonCommand, command: Dict[str, Any]
) -> None:
for data in command.values(): for data in command.values():
if isinstance(data, str): if isinstance(data, str):
continue continue
for key, value in data.items(): for key, value in data.items():
if not (parameter := base_command.parameters.get(key)): if parameter := base.parameters.get(key):
continue
with suppress(ValueError): with suppress(ValueError):
parameter.value = value parameter.value = value
def _update_base_command_with_favourite(self, base_command: HonCommand) -> None:
extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom") extra_param = HonParameterFixed("favourite", {"fixedValue": "1"}, "custom")
base_command.parameters.update(favourite=extra_param) base.parameters.update(favourite=extra_param)
if isinstance(program := base.parameters["program"], HonParameterProgram):
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) program.set_value(name)
self.commands[command_name].categories[name] = base_command self.commands[command_name].categories[name] = base

View File

@ -89,11 +89,8 @@ class HonCommand:
def parameter_value(self) -> Dict[str, Union[str, float]]: def parameter_value(self) -> Dict[str, Union[str, float]]:
return {n: p.value for n, p in self._parameters.items()} return {n: p.value for n, p in self._parameters.items()}
def _load_parameters(self, attributes: Dict[str, Dict[str, Any] | Any]) -> None: def _load_parameters(self, attributes: Dict[str, Dict[str, Any]]) -> None:
for key, items in attributes.items(): 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(): for name, data in items.items():
self._create_parameters(data, name, key) self._create_parameters(data, name, key)
for rule in self._rules: for rule in self._rules:
@ -105,12 +102,10 @@ class HonCommand:
if name == "zoneMap" and self._appliance.zone: if name == "zoneMap" and self._appliance.zone:
data["default"] = self._appliance.zone data["default"] = self._appliance.zone
if data.get("category") == "rule": if data.get("category") == "rule":
if "fixedValue" in data: if "fixedValue" not in data:
self._rules.append(HonRuleSet(self, data["fixedValue"])) _LOGGER.error("Rule not supported: %s", data)
elif "enumValues" in data:
self._rules.append(HonRuleSet(self, data["enumValues"]))
else: else:
_LOGGER.warning("Rule not supported: %s", data) self._rules.append(HonRuleSet(self, data["fixedValue"]))
match data.get("typology"): match data.get("typology"):
case "range": case "range":
self._parameters[name] = HonParameterRange(name, data, parameter) self._parameters[name] = HonParameterRange(name, data, parameter)
@ -135,23 +130,17 @@ class HonCommand:
async def send_specific(self, param_names: List[str]) -> bool: async def send_specific(self, param_names: List[str]) -> bool:
params: Dict[str, str | float] = {} params: Dict[str, str | float] = {}
for key, parameter in self._parameters.items(): for key, parameter in self._parameters.items():
if key in param_names or parameter.mandatory: if key in param_names:
params[key] = parameter.value params[key] = parameter.value
return await self.send_parameters(params) return await self.send_parameters(params)
async def send_parameters(self, params: Dict[str, str | float]) -> bool: async def send_parameters(self, params: Dict[str, str | float]) -> bool:
ancillary_params = self.parameter_groups.get("ancillaryParameters", {}) ancillary_params = self.parameter_groups.get("ancillaryParameters", {})
ancillary_params.pop("programRules", None) ancillary_params.pop("programRules", None)
if "prStr" in params:
params["prStr"] = self._category_name.upper()
self.appliance.sync_command_to_params(self.name) self.appliance.sync_command_to_params(self.name)
try: try:
result = await self.api.send_command( result = await self.api.send_command(
self._appliance, self._appliance, self._name, params, ancillary_params
self._name,
params,
ancillary_params,
self._category_name,
) )
if not result: if not result:
_LOGGER.error(result) _LOGGER.error(result)

View File

@ -190,7 +190,6 @@ class HonAPI:
command: str, command: str,
parameters: Dict[str, Any], parameters: Dict[str, Any],
ancillary_parameters: Dict[str, Any], ancillary_parameters: Dict[str, Any],
program_name: str = "",
) -> bool: ) -> bool:
now: str = datetime.utcnow().isoformat() now: str = datetime.utcnow().isoformat()
data: Dict[str, Any] = { data: Dict[str, Any] = {
@ -209,8 +208,6 @@ class HonAPI:
"parameters": parameters, "parameters": parameters,
"applianceType": appliance.appliance_type, "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" url: str = f"{const.API_URL}/commands/v1/send"
async with self._hon.post(url, json=data) as response: async with self._hon.post(url, json=data) as response:
json_data: Dict[str, Any] = await response.json() json_data: Dict[str, Any] = await response.json()
@ -281,12 +278,10 @@ class TestAPI(HonAPI):
async def load_appliances(self) -> List[Dict[str, Any]]: async def load_appliances(self) -> List[Dict[str, Any]]:
result = [] result = []
for appliance in self._path.glob("*/"): for appliance in self._path.glob("*/"):
file = appliance / "appliance_data.json" with open(
with open(file, "r", encoding="utf-8") as json_file: appliance / "appliance_data.json", "r", encoding="utf-8"
try: ) as json_file:
result.append(json.loads(json_file.read())) result.append(json.loads(json_file.read()))
except json.decoder.JSONDecodeError as error:
_LOGGER.error("%s - %s", str(file), error)
return result return result
async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]: async def load_commands(self, appliance: HonAppliance) -> Dict[str, Any]:
@ -322,12 +317,5 @@ class TestAPI(HonAPI):
command: str, command: str,
parameters: Dict[str, Any], parameters: Dict[str, Any],
ancillary_parameters: Dict[str, Any], ancillary_parameters: Dict[str, Any],
program_name: str = "",
) -> bool: ) -> bool:
_LOGGER.info(
"%s - %s - %s",
str(parameters),
str(ancillary_parameters),
str(program_name),
)
return True return True

View File

@ -6,7 +6,7 @@ CLIENT_ID = (
"3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9." "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9."
"HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6" "HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
) )
APP_VERSION = "2.4.7" APP_VERSION = "2.1.2"
OS_VERSION = 31 OS_VERSION = 31
OS = "android" OS = "android"
DEVICE_MODEL = "exynos9820" DEVICE_MODEL = "exynos9820"

View File

@ -68,9 +68,8 @@ class HonParameter:
self._triggers.setdefault(value, []).append((func, data)) self._triggers.setdefault(value, []).append((func, data))
def check_trigger(self, value: str | float) -> None: def check_trigger(self, value: str | float) -> None:
triggers = {str(k).lower(): v for k, v in self._triggers.items()} if str(value) in self._triggers:
if str(value).lower() in triggers: for trigger in self._triggers[str(value)]:
for trigger in triggers[str(value)]:
func, args = trigger func, args = trigger
func(args) func(args)

View File

@ -18,7 +18,7 @@ class HonParameterFixed(HonParameter):
@property @property
def value(self) -> str | float: def value(self) -> str | float:
return self._value if self._value != "" else "0" return self._value if self._value is not None else "0"
@value.setter @value.setter
def value(self, value: str | float) -> None: def value(self, value: str | float) -> None:

View File

View File

@ -56,11 +56,6 @@ class HonRuleSet:
extra[trigger_key] = trigger_value extra[trigger_key] = trigger_value
for extra_key, extra_data in param_data.items(): for extra_key, extra_data in param_data.items():
self._parse_conditions(param_key, extra_key, extra_data, extra) 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( def _create_rule(
self, self,
@ -107,10 +102,6 @@ class HonRuleSet:
param.values = [str(value)] param.values = [str(value)]
param.value = str(value) param.value = str(value)
elif isinstance(param, HonParameterRange): 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) param.value = float(value)
return return
param.value = str(value) param.value = str(value)

View File

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

View File

@ -1,5 +1,4 @@
black>=22.12 black==23.7.0
flake8>=6.0 flake8==6.0.0
mypy>=0.991 mypy==1.4.1
pylint>=2.15 pylint==2.17.4
setuptools>=62.3

View File

@ -7,7 +7,7 @@ with open("README.md", "r", encoding="utf-8") as f:
setup( setup(
name="pyhOn", name="pyhOn",
version="0.15.15", version="0.15.0",
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,
@ -21,7 +21,7 @@ setup(
packages=find_packages(), packages=find_packages(),
include_package_data=True, include_package_data=True,
python_requires=">=3.10", python_requires=">=3.10",
install_requires=["aiohttp>=3.8", "typing-extensions>=4.8", "yarl>=1.8"], install_requires=["aiohttp==3.8.4", "typing-extensions==4.7.1"],
classifiers=[ classifiers=[
"Development Status :: 4 - Beta", "Development Status :: 4 - Beta",
"Environment :: Console", "Environment :: Console",
@ -30,7 +30,6 @@ setup(
"Operating System :: OS Independent", "Operating System :: OS Independent",
"Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules", "Topic :: Software Development :: Libraries :: Python Modules",
], ],
entry_points={ entry_points={