Bump version

Use class attribute to check active

.github/workflows/python-check.yml

@@ -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
@@ -34,7 +34,7 @@ jobs:
         mypy pyhon/
     - name: Analysing the code with pylint
       run: |
-        pylint  $(git ls-files '*.py')
+        pylint $(git ls-files '*.py')
     - name: Check black style
       run: |
         black . --check

.pylintrc

@@ -4,6 +4,6 @@ disable=missing-docstring
 
 [FORMAT]
 
-max-args=6
+max-args=7
 max-attributes=8
 max-line-length=88

MANIFEST.in

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

pyhon/__main__.py

@@ -38,9 +38,7 @@ def get_arguments() -> Dict[str, Any]:
         "translate", help="language (de, en, fr...)", metavar="LANGUAGE"
     )
     translation.add_argument("--json", help="print as json", action="store_true")
-    parser.add_argument(
+    parser.add_argument("-i", "--import", help="import pyhon data", nargs="?")
-        "-i", "--import", help="import pyhon data", nargs="?", default=Path().cwd()
-    )
     return vars(parser.parse_args())
 
 
@@ -73,9 +71,8 @@ async def main() -> None:
     if language := args.get("translate"):
         await translate(language, json_output=args.get("json", ""))
         return
-    async with Hon(
+    test_data_path = Path(path) if (path := args.get("import", "")) else None
-        *get_login_data(args), test_data_path=Path(args.get("import", ""))
+    async with Hon(*get_login_data(args), test_data_path=test_data_path) as hon:
-    ) as hon:
         for device in hon.appliances:
             if args.get("export"):
                 anonymous = args.get("anonymous", False)

pyhon/appliance.py

@@ -20,7 +20,7 @@ if TYPE_CHECKING:
 
 _LOGGER = logging.getLogger(__name__)
 
-T = TypeVar('T')
+T = TypeVar("T")
 
 
 # pylint: disable=too-many-public-methods,too-many-instance-attributes
@@ -72,12 +72,10 @@ class HonAppliance:
         return self.info[item]
 
     @overload
-    def get(self, item: str, default: None = 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: T) -> T: ...
-        ...
 
     def get(self, item: str, default: Optional[T] = None) -> Any:
         try:
@@ -94,15 +92,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:
@@ -138,11 +136,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]:
@@ -277,7 +275,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
@@ -287,7 +290,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/ov.py

@@ -7,10 +7,10 @@ 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"]["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"] == "1"
+        data["active"] = data["parameters"]["onOffStatus"].value == 1
         return data

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

pyhon/command_loader.py

@@ -109,7 +109,7 @@ class HonCommandLoader:
         categories: Optional[Dict[str, "HonCommand"]] = None,
         category_name: str = "",
     ) -> Optional[HonCommand]:
-        """Try to crate HonCommand object"""
+        """Try to create HonCommand object"""
         if not isinstance(data, dict):
             self._additional_data[command_name] = data
             return None
@@ -184,23 +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)
-            command = favourite.get("command", {})
+            if not base:
-            command_name = command.get("commandName", "")
-            program_name = self._clean_name(command.get("programName", ""))
-            if not (base := self.commands[command_name].categories.get(program_name)):
                 continue
             base_command: HonCommand = copy(base)
-            for data in command.values():
+            self._update_base_command_with_data(base_command, favourite)
-                if isinstance(data, str):
+            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: str = command.get("commandName", "")
+        program_name = self._clean_name(command.get("programName", ""))
+        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 not (parameter := base_command.parameters.get(key)):
                     continue
-                for key, value in data.items():
+                with suppress(ValueError):
-                    if parameter := base_command.parameters.get(key):
+                    parameter.value = value
-                        with suppress(ValueError):
+
-                            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_command.parameters.update(favourite=extra_param)
-            program = base_command.parameters["program"]
+
-            if isinstance(program, HonParameterProgram):
+    def _update_program_categories(
-                program.set_value(name)
+        self, command_name: str, name: str, base_command: HonCommand
-            self.commands[command_name].categories[name] = base_command
+    ) -> None:
+        program = base_command.parameters["program"]
+        if isinstance(program, HonParameterProgram):
+            program.set_value(name)
+        self.commands[command_name].categories[name] = base_command

pyhon/commands.py

@@ -89,8 +89,11 @@ class HonCommand:
     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:
@@ -102,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:
+            if "fixedValue" in data:
-                _LOGGER.error("Rule not supported: %s", data)
-            else:
                 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,17 +135,23 @@ 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(
-                self._appliance, self._name, params, ancillary_params
+                self._appliance,
+                self._name,
+                params,
+                ancillary_params,
+                self._category_name,
             )
             if not result:
                 _LOGGER.error(result)

pyhon/connection/api.py

@@ -24,12 +24,16 @@ class HonAPI:
         email: str = "",
         password: str = "",
         anonymous: bool = False,
+        mobile_id: str = "",
+        refresh_token: str = "",
         session: Optional[ClientSession] = None,
     ) -> None:
         super().__init__()
         self._email: str = email
         self._password: str = password
         self._anonymous: bool = anonymous
+        self._mobile_id: str = mobile_id
+        self._refresh_token: str = refresh_token
         self._hon_handler: Optional[HonConnectionHandler] = None
         self._hon_anonymous_handler: Optional[HonAnonymousConnectionHandler] = None
         self._session: Optional[ClientSession] = session
@@ -69,7 +73,10 @@ class HonAPI:
         ).create()
         if not self._anonymous:
             self._hon_handler = await HonConnectionHandler(
-                self._email, self._password, self._session
+                self._email,
+                self._password,
+                session=self._session,
+                mobile_id=self._mobile_id,
             ).create()
         return self
 
@@ -190,6 +197,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] = {
@@ -208,6 +216,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()
@@ -319,6 +329,12 @@ class TestAPI(HonAPI):
         command: str,
         parameters: Dict[str, Any],
         ancillary_parameters: Dict[str, Any],
+        program_name: str = "",
     ) -> bool:
-        _LOGGER.info("%s - %s", str(parameters), str(ancillary_parameters))
+        _LOGGER.info(
+            "%s - %s - %s",
+            str(parameters),
+            str(ancillary_parameters),
+            str(program_name),
+        )
         return True

pyhon/connection/auth.py

@@ -199,7 +199,7 @@ class HonAuth:
         if access_token := re.findall("access_token=(.*?)&", text):
             self._auth.access_token = access_token[0]
         if refresh_token := re.findall("refresh_token=(.*?)&", text):
-            self._auth.refresh_token = refresh_token[0]
+            self._auth.refresh_token = parse.unquote(refresh_token[0])
         if id_token := re.findall("id_token=(.*?)&", text):
             self._auth.id_token = id_token[0]
         return bool(access_token and refresh_token and id_token)
@@ -264,7 +264,9 @@ class HonAuth:
         except exceptions.HonNoAuthenticationNeeded:
             return
 
-    async def refresh(self) -> bool:
+    async def refresh(self, refresh_token: str = "") -> bool:
+        if refresh_token:
+            self._auth.refresh_token = refresh_token
         params = {
             "client_id": const.CLIENT_ID,
             "refresh_token": self._auth.refresh_token,

pyhon/connection/device.py

@@ -1,16 +1,15 @@
-import secrets
 from typing import Dict
 
 from pyhon import const
 
 
 class HonDevice:
-    def __init__(self) -> None:
+    def __init__(self, mobile_id: str = "") -> None:
         self._app_version: str = const.APP_VERSION
         self._os_version: int = const.OS_VERSION
         self._os: str = const.OS
         self._device_model: str = const.DEVICE_MODEL
-        self._mobile_id: str = secrets.token_hex(8)
+        self._mobile_id: str = mobile_id or const.MOBILE_ID
 
     @property
     def app_version(self) -> str:

pyhon/connection/handler/hon.py

@@ -19,12 +19,18 @@ _LOGGER = logging.getLogger(__name__)
 
 class HonConnectionHandler(ConnectionHandler):
     def __init__(
-        self, email: str, password: str, session: Optional[aiohttp.ClientSession] = None
+        self,
+        email: str,
+        password: str,
+        session: Optional[aiohttp.ClientSession] = None,
+        mobile_id: str = "",
+        refresh_token: str = "",
     ) -> None:
         super().__init__(session=session)
-        self._device: HonDevice = HonDevice()
+        self._device: HonDevice = HonDevice(mobile_id)
         self._email: str = email
         self._password: str = password
+        self._refresh_token: str = refresh_token
         if not self._email:
             raise HonAuthenticationError("An email address must be specified")
         if not self._password:
@@ -43,10 +49,17 @@ class HonConnectionHandler(ConnectionHandler):
 
     async def create(self) -> Self:
         await super().create()
-        self._auth = HonAuth(self.session, self._email, self._password, self._device)
+        self._auth = HonAuth(
+            self.session,
+            self._email,
+            self._password,
+            self._device,
+        )
         return self
 
     async def _check_headers(self, headers: Dict[str, str]) -> Dict[str, str]:
+        if self._refresh_token:
+            await self.auth.refresh(self._refresh_token)
         if not (self.auth.cognito_token and self.auth.id_token):
             await self.auth.authenticate()
         headers["cognito-token"] = self.auth.cognito_token

pyhon/const.py

@@ -6,8 +6,9 @@ CLIENT_ID = (
     "3MVG9QDx8IX8nP5T2Ha8ofvlmjLZl5L_gvfbT9."
     "HJvpHGKoAS_dcMN8LYpTSYeVFCraUnV.2Ag1Ki7m4znVO6"
 )
-APP_VERSION = "2.1.2"
+APP_VERSION = "2.6.5"
-OS_VERSION = 31
+OS_VERSION = 999
 OS = "android"
-DEVICE_MODEL = "exynos9820"
+DEVICE_MODEL = "pyhOn"
-USER_AGENT = "Chrome/110.0.5481.153"
+USER_AGENT = "Chrome/999.999.999.999"
+MOBILE_ID = "pyhOn"

pyhon/hon.py

@@ -21,6 +21,8 @@ class Hon:
         email: Optional[str] = "",
         password: Optional[str] = "",
         session: Optional[ClientSession] = None,
+        mobile_id: str = "",
+        refresh_token: str = "",
         test_data_path: Optional[Path] = None,
     ):
         self._email: Optional[str] = email
@@ -29,6 +31,8 @@ class Hon:
         self._appliances: List[HonAppliance] = []
         self._api: Optional[HonAPI] = None
         self._test_data_path: Path = test_data_path or Path().cwd()
+        self._mobile_id: str = mobile_id
+        self._refresh_token: str = refresh_token
 
     async def __aenter__(self) -> Self:
         return await self.create()
@@ -61,7 +65,11 @@ class Hon:
 
     async def create(self) -> Self:
         self._api = await HonAPI(
-            self.email, self.password, session=self._session
+            self.email,
+            self.password,
+            session=self._session,
+            mobile_id=self._mobile_id,
+            refresh_token=self._refresh_token,
         ).create()
         await self.setup()
         return self
@@ -103,8 +111,12 @@ class Hon:
                     )
             await self._create_appliance(appliance, self.api)
         if (
-            test_data := self._test_data_path / "hon-test-data" / "test_data"
+            self._test_data_path
-        ).exists() or (test_data := test_data / "test_data").exists():
+            and (
+                test_data := self._test_data_path / "hon-test-data" / "test_data"
+            ).exists()
+            or (test_data := test_data / "..").exists()
+        ):
             api = TestAPI(test_data)
             for appliance in await api.load_appliances():
                 await self._create_appliance(appliance, api)

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:
+        triggers = {str(k).lower(): v for k, v in self._triggers.items()}
-            for trigger in self._triggers[str(value)]:
+        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/parameter/range.py

@@ -69,4 +69,9 @@ class HonParameterRange(HonParameter):
 
     @property
     def values(self) -> List[str]:
-        return [str(i) for i in range(int(self.min), int(self.max) + 1, int(self.step))]
+        result = []
+        i = self.min
+        while i <= self.max:
+            result.append(str(i))
+            i += self.step
+        return result

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)

pyhon/typedefs.py

@@ -14,8 +14,7 @@ if TYPE_CHECKING:
 class Callback(Protocol):  # pylint: disable=too-few-public-methods
     def __call__(
         self, url: str | URL, *args: Any, **kwargs: Any
-    ) -> aiohttp.client._RequestContextManager:
+    ) -> aiohttp.client._RequestContextManager: ...
-        ...
 
 
 Parameter = Union[

requirements.txt

@@ -1,3 +1,3 @@
-aiohttp==3.8.5
+aiohttp>=3.8.6
-yarl==1.9.2
+yarl>=1.8
-typing-extensions==4.7.1
+typing-extensions>=4.8

requirements_dev.txt

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

setup.py

@@ -7,7 +7,7 @@ with open("README.md", "r", encoding="utf-8") as f:
 
 setup(
     name="pyhOn",
-    version="0.15.3",
+    version="0.16.1",
     author="Andre Basche",
     description="Control hOn devices with python",
     long_description=long_description,
@@ -20,9 +20,8 @@ setup(
     platforms="any",
     packages=find_packages(),
     include_package_data=True,
-    package_data={"pyhon": ["py.typed"]},
     python_requires=">=3.10",
-    install_requires=["aiohttp==3.8.5", "typing-extensions==4.7.1"],
+    install_requires=["aiohttp>=3.8.6", "typing-extensions>=4.8", "yarl>=1.8"],
     classifiers=[
         "Development Status :: 4 - Beta",
         "Environment :: Console",
@@ -31,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={