Update mypy

* 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>

MANIFEST.in

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

mypy.ini

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

pyhon/appliance.py

@@ -3,7 +3,7 @@ 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
@@ -20,6 +20,8 @@ if TYPE_CHECKING:
 
 _LOGGER = logging.getLogger(__name__)
 
+T = TypeVar("T")
+
 
 # pylint: disable=too-many-public-methods,too-many-instance-attributes
 class HonAppliance:
@@ -49,24 +51,35 @@ class HonAppliance:
         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):
@@ -81,15 +94,15 @@ class HonAppliance:
 
     @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:
@@ -125,11 +138,11 @@ class HonAppliance:
 
     @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]:
@@ -250,7 +263,9 @@ class HonAppliance:
         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:
+            if (
+                new := self.attributes.get("parameters", {}).get(key)
+            ) is None or new.value == "":
                 continue
             setting = command.settings[key]
             try:
@@ -262,7 +277,12 @@ class HonAppliance:
                 _LOGGER.info("Can't set %s - %s", key, error)
                 continue
 
-    def sync_command(self, main: str, target: Optional[List[str] | str] = None) -> None:
+    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
@@ -272,7 +292,12 @@ class HonAppliance:
 
             for name, target_param in data.parameters.items():
                 if not (base_param := base.parameters.get(name)):
-                    return
+                    continue
+                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)
 
     def sync_parameter(self, main: Parameter, target: Parameter) -> None:

pyhon/appliances/ref.py

@@ -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"

pyhon/appliances/wh.py

@@ -1,11 +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)
-        data["active"] = data["parameters"]["onOffStatus"] == "1"
-
+        parameter = data["parameters"]["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

pyhon/command_loader.py

@@ -184,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

pyhon/commands.py

@@ -102,10 +102,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)
@@ -130,13 +132,15 @@ class HonCommand:
     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:
+            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)
+        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(

pyhon/connection/api.py

@@ -278,10 +278,12 @@ class TestAPI(HonAPI):
     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]:
@@ -318,4 +320,5 @@ class TestAPI(HonAPI):
         parameters: Dict[str, Any],
         ancillary_parameters: Dict[str, Any],
     ) -> bool:
+        _LOGGER.info("%s - %s", str(parameters), str(ancillary_parameters))
         return True

pyhon/const.py

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

pyhon/parameter/base.py

@@ -68,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)
 

pyhon/parameter/fixed.py

@@ -18,7 +18,7 @@ class HonParameterFixed(HonParameter):
 
     @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:

pyhon/rules.py

@@ -56,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,
@@ -102,6 +107,10 @@ class HonRuleSet:
             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)

requirements.txt

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

requirements_dev.txt

@@ -1,4 +1,4 @@
-black==23.7.0
-flake8==6.0.0
-mypy==1.4.1
-pylint==2.17.4
+black~=23.9
+flake8~=6.1
+mypy~=1.6
+pylint~=3.0

setup.py

@@ -7,7 +7,7 @@ with open("README.md", "r", encoding="utf-8") as f:
 
 setup(
     name="pyhOn",
-    version="0.15.0",
+    version="0.15.10",
     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==3.8.4", "typing-extensions==4.7.1"],
+    install_requires=["aiohttp~=3.8", "typing-extensions~=4.8", "yarl~=1.9"],
     classifiers=[
         "Development Status :: 4 - Beta",
         "Environment :: Console",