Fix connection failures

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, TypeVar, overload, Callable
+from typing import Optional, Dict, Any, TYPE_CHECKING, List, TypeVar, overload
 
 from pyhon import diagnose, exceptions
 from pyhon.appliances.base import ApplianceBase
@@ -43,7 +43,10 @@ class HonAppliance:
         self._additional_data: Dict[str, Any] = {}
         self._last_update: Optional[datetime] = None
         self._default_setting = HonParameter("", {}, "")
-        self._notify_function: Optional[Callable[[Any], None]] = None
+        self._connection = (
+            not self._attributes.get("lastConnEvent", {}).get("category", "")
+            == "DISCONNECTED"
+        )
 
         try:
             self._extra: Optional[ApplianceBase] = importlib.import_module(
@@ -91,6 +94,14 @@ class HonAppliance:
             return f"{attribute}{zone}{self._zone}"
         return attribute
 
+    @property
+    def connection(self) -> bool:
+        return self._connection
+
+    @connection.setter
+    def connection(self, connection: bool) -> None:
+        self._connection = connection
+
     @property
     def appliance_model_id(self) -> str:
         return str(self._info.get("applianceModelId", ""))
@@ -313,11 +324,3 @@ class HonAppliance:
         elif isinstance(target, HonParameterEnum):
             target.values = main.values
         target.value = main.value
-
-    def subscribe(self, notify_function: Callable[[Any], None]) -> None:
-        self._notify_function = notify_function
-
-    def notify(self) -> None:
-        self.sync_params_to_command("settings")
-        if self._notify_function:
-            self._notify_function(self.attributes)

pyhon/appliances/dw.py

@@ -7,7 +7,7 @@ from pyhon.appliances.base import ApplianceBase
 class Appliance(ApplianceBase):
     def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
         data = super().attributes(data)
-        if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
+        if not self.parent.connection:
             data["parameters"]["machMode"].value = "0"
         data["active"] = bool(data.get("activity"))
         return data

pyhon/appliances/ov.py

@@ -6,7 +6,7 @@ from pyhon.appliances.base import ApplianceBase
 class Appliance(ApplianceBase):
     def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
         data = super().attributes(data)
-        if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
+        if not self.parent.connection:
             data["parameters"]["temp"].value = 0
             data["parameters"]["onOffStatus"].value = 0
             data["parameters"]["remoteCtrValid"].value = 0

pyhon/appliances/td.py

@@ -8,7 +8,7 @@ from pyhon.parameter.fixed import HonParameterFixed
 class Appliance(ApplianceBase):
     def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
         data = super().attributes(data)
-        if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
+        if not self.parent.connection:
             data["parameters"]["machMode"].value = "0"
         data["active"] = bool(data.get("activity"))
         data["pause"] = data["parameters"]["machMode"] == "3"

pyhon/appliances/wd.py

@@ -7,7 +7,7 @@ from pyhon.appliances.base import ApplianceBase
 class Appliance(ApplianceBase):
     def attributes(self, data: Dict[str, Any]) -> Dict[str, Any]:
         data = super().attributes(data)
-        if data.get("lastConnEvent", {}).get("category", "") == "DISCONNECTED":
+        if not self.parent.connection:
             data["parameters"]["machMode"].value = "0"
         data["active"] = bool(data.get("activity"))
         data["pause"] = data["parameters"]["machMode"] == "3"

pyhon/connection/api.py

@@ -7,12 +7,10 @@ from types import TracebackType
 from typing import Dict, Optional, Any, List, no_type_check, Type
 
 from aiohttp import ClientSession
-from awscrt import mqtt5
 from typing_extensions import Self
 
 from pyhon import const, exceptions
 from pyhon.appliance import HonAppliance
-from pyhon.connection import mqtt
 from pyhon.connection.auth import HonAuth
 from pyhon.connection.handler.anonym import HonAnonymousConnectionHandler
 from pyhon.connection.handler.hon import HonConnectionHandler
@@ -40,7 +38,6 @@ class HonAPI:
         self._hon_handler: Optional[HonConnectionHandler] = None
         self._hon_anonymous_handler: Optional[HonAnonymousConnectionHandler] = None
         self._session: Optional[ClientSession] = session
-        self._mqtt_client: mqtt5.Client | None = None
 
     async def __aenter__(self) -> Self:
         return await self.create()
@@ -81,6 +78,7 @@ class HonAPI:
                 self._password,
                 session=self._session,
                 mobile_id=self._mobile_id,
+                refresh_token=self._refresh_token,
             ).create()
         return self
 
@@ -269,10 +267,6 @@ class HonAPI:
             result: Dict[str, Any] = await response.json()
         return result
 
-    async def subscribe_mqtt(self, appliances: list[HonAppliance]) -> None:
-        if not self._mqtt_client:
-            self._mqtt_client = await mqtt.start(self, appliances)
-
     async def close(self) -> None:
         if self._hon_handler is not None:
             await self._hon_handler.close()

pyhon/connection/handler/hon.py

@@ -62,6 +62,7 @@ class HonConnectionHandler(ConnectionHandler):
             await self.auth.refresh(self._refresh_token)
         if not (self.auth.cognito_token and self.auth.id_token):
             await self.auth.authenticate()
+        self._refresh_token = self.auth.refresh_token
         headers["cognito-token"] = self.auth.cognito_token
         headers["id-token"] = self.auth.id_token
         return self._HEADERS | headers
@@ -77,7 +78,7 @@ class HonConnectionHandler(ConnectionHandler):
                 self.auth.token_expires_soon or response.status in [401, 403]
             ) and loop == 0:
                 _LOGGER.info("Try refreshing token...")
-                await self.auth.refresh()
+                await self.auth.refresh(self._refresh_token)
                 async with self._intercept(
                     method, url, *args, loop=loop + 1, **kwargs
                 ) as result:

pyhon/connection/mqtt.py

@@ -1,5 +1,7 @@
+import asyncio
 import json
 import logging
+import secrets
 from typing import TYPE_CHECKING
 
 from awscrt import mqtt5
@@ -9,68 +11,136 @@ from pyhon import const
 from pyhon.appliance import HonAppliance
 
 if TYPE_CHECKING:
-    from pyhon import HonAPI
+    from pyhon import Hon
 
 _LOGGER = logging.getLogger(__name__)
 
-appliances: list[HonAppliance] = []
 
+class MQTTClient:
+    def __init__(self, hon: "Hon", mobile_id: str) -> None:
+        self._client: mqtt5.Client | None = None
+        self._hon = hon
+        self._mobile_id = mobile_id or const.MOBILE_ID
+        self._api = hon.api
+        self._appliances = hon.appliances
+        self._connection = False
+        self._watchdog_task: asyncio.Task[None] | None = None
 
-def on_lifecycle_stopped(lifecycle_stopped_data: mqtt5.LifecycleStoppedData) -> None:
-    _LOGGER.info("Lifecycle Stopped: %s", str(lifecycle_stopped_data))
+    @property
+    def client(self) -> mqtt5.Client:
+        if self._client is not None:
+            return self._client
+        raise AttributeError("Client is not set")
 
+    async def create(self) -> "MQTTClient":
+        await self._start()
+        self._subscribe_appliances()
+        await self.start_watchdog()
+        return self
 
-def on_lifecycle_connection_success(
-    lifecycle_connect_success_data: mqtt5.LifecycleConnectSuccessData,
-) -> None:
-    _LOGGER.info(
-        "Lifecycle Connection Success: %s", str(lifecycle_connect_success_data)
-    )
+    def _on_lifecycle_stopped(
+        self, lifecycle_stopped_data: mqtt5.LifecycleStoppedData
+    ) -> None:
+        _LOGGER.info("Lifecycle Stopped: %s", str(lifecycle_stopped_data))
 
-
-def on_publish_received(data: mqtt5.PublishReceivedData) -> None:
-    if not (data and data.publish_packet and data.publish_packet.payload):
-        return
-    payload = json.loads(data.publish_packet.payload.decode())
-    topic = data.publish_packet.topic
-    if topic and "appliancestatus" in topic:
-        appliance = next(
-            a for a in appliances if topic in a.info["topics"]["subscribe"]
+    def _on_lifecycle_connection_success(
+        self,
+        lifecycle_connect_success_data: mqtt5.LifecycleConnectSuccessData,
+    ) -> None:
+        self._connection = True
+        _LOGGER.info(
+            "Lifecycle Connection Success: %s", str(lifecycle_connect_success_data)
         )
-        for parameter in payload["parameters"]:
-            appliance.attributes["parameters"][parameter["parName"]].update(parameter)
-        appliance.notify()
-        _LOGGER.debug("%s - %s", topic, payload)
-    else:
-        _LOGGER.debug("%s - %s", topic, payload)
 
+    def _on_lifecycle_attempting_connect(
+        self,
+        lifecycle_attempting_connect_data: mqtt5.LifecycleAttemptingConnectData,
+    ) -> None:
+        _LOGGER.info(
+            "Lifecycle Attempting Connect - %s", str(lifecycle_attempting_connect_data)
+        )
 
-async def create_mqtt_client(api: "HonAPI") -> mqtt5.Client:
-    client: mqtt5.Client = mqtt5_client_builder.websockets_with_custom_authorizer(
-        endpoint=const.AWS_ENDPOINT,
-        auth_authorizer_name=const.AWS_AUTHORIZER,
-        auth_authorizer_signature=await api.load_aws_token(),
-        auth_token_key_name="token",
-        auth_token_value=api.auth.id_token,
-        client_id=const.MOBILE_ID,
-        on_lifecycle_stopped=on_lifecycle_stopped,
-        on_lifecycle_connection_success=on_lifecycle_connection_success,
-        on_publish_received=on_publish_received,
-    )
-    client.start()
-    return client
+    def _on_lifecycle_connection_failure(
+        self,
+        lifecycle_connection_failure_data: mqtt5.LifecycleConnectFailureData,
+    ) -> None:
+        self._connection = False
+        _LOGGER.info(
+            "Lifecycle Connection Failure - %s", str(lifecycle_connection_failure_data)
+        )
 
+    def _on_lifecycle_disconnection(
+        self,
+        lifecycle_disconnect_data: mqtt5.LifecycleDisconnectData,
+    ) -> None:
+        self._connection = False
+        _LOGGER.info("Lifecycle Disconnection - %s", str(lifecycle_disconnect_data))
 
-def subscribe(client: mqtt5.Client, appliance: HonAppliance) -> None:
-    for topic in appliance.info.get("topics", {}).get("subscribe", []):
-        client.subscribe(mqtt5.SubscribePacket([mqtt5.Subscription(topic)])).result(10)
-        _LOGGER.info("Subscribed to topic %s", topic)
+    def _on_publish_received(self, data: mqtt5.PublishReceivedData) -> None:
+        if not (data and data.publish_packet and data.publish_packet.payload):
+            return
+        payload = json.loads(data.publish_packet.payload.decode())
+        topic = data.publish_packet.topic
+        appliance = next(
+            a for a in self._appliances if topic in a.info["topics"]["subscribe"]
+        )
+        if topic and "appliancestatus" in topic:
+            for parameter in payload["parameters"]:
+                appliance.attributes["parameters"][parameter["parName"]].update(
+                    parameter
+                )
+            appliance.sync_params_to_command("settings")
+        elif topic and "disconnected" in topic:
+            _LOGGER.info(
+                "Disconnected %s: %s",
+                appliance.nick_name,
+                payload.get("disconnectReason"),
+            )
+            appliance.connection = False
+        elif topic and "connected" in topic:
+            appliance.connection = True
+            _LOGGER.info("Connected %s", appliance.nick_name)
+        elif topic and "discovery" in topic:
+            _LOGGER.info("Discovered %s", appliance.nick_name)
+        self._hon.notify()
+        _LOGGER.info("%s - %s", topic, payload)
 
+    async def _start(self) -> None:
+        self._client = mqtt5_client_builder.websockets_with_custom_authorizer(
+            endpoint=const.AWS_ENDPOINT,
+            auth_authorizer_name=const.AWS_AUTHORIZER,
+            auth_authorizer_signature=await self._api.load_aws_token(),
+            auth_token_key_name="token",
+            auth_token_value=self._api.auth.id_token,
+            client_id=f"{self._mobile_id}_{secrets.token_hex(8)}",
+            on_lifecycle_stopped=self._on_lifecycle_stopped,
+            on_lifecycle_connection_success=self._on_lifecycle_connection_success,
+            on_lifecycle_attempting_connect=self._on_lifecycle_attempting_connect,
+            on_lifecycle_connection_failure=self._on_lifecycle_connection_failure,
+            on_lifecycle_disconnection=self._on_lifecycle_disconnection,
+            on_publish_received=self._on_publish_received,
+        )
+        self.client.start()
 
-async def start(api: "HonAPI", app: list[HonAppliance]) -> mqtt5.Client:
-    client = await create_mqtt_client(api)
-    global appliances  # pylint: disable=global-statement
-    appliances = app
-    for appliance in appliances:
-        subscribe(client, appliance)
-    return client
+    def _subscribe_appliances(self) -> None:
+        for appliance in self._appliances:
+            self._subscribe(appliance)
+
+    def _subscribe(self, appliance: HonAppliance) -> None:
+        for topic in appliance.info.get("topics", {}).get("subscribe", []):
+            self.client.subscribe(
+                mqtt5.SubscribePacket([mqtt5.Subscription(topic)])
+            ).result(10)
+            _LOGGER.info("Subscribed to topic %s", topic)
+
+    async def start_watchdog(self) -> None:
+        if not self._watchdog_task or self._watchdog_task.done():
+            self._watchdog_task = asyncio.create_task(self._watchdog())
+
+    async def _watchdog(self) -> None:
+        while True:
+            await asyncio.sleep(5)
+            if not self._connection:
+                _LOGGER.info("Restart mqtt connection")
+                await self._start()
+                self._subscribe_appliances()

pyhon/hon.py

@@ -1,8 +1,7 @@
-import asyncio
 import logging
 from pathlib import Path
 from types import TracebackType
-from typing import List, Optional, Dict, Any, Type
+from typing import List, Optional, Dict, Any, Type, Callable
 
 from aiohttp import ClientSession
 from typing_extensions import Self
@@ -10,11 +9,13 @@ from typing_extensions import Self
 from pyhon.appliance import HonAppliance
 from pyhon.connection.api import HonAPI
 from pyhon.connection.api import TestAPI
+from pyhon.connection.mqtt import MQTTClient
 from pyhon.exceptions import NoAuthenticationException
 
 _LOGGER = logging.getLogger(__name__)
 
 
+# pylint: disable=too-many-instance-attributes
 class Hon:
     def __init__(
         self,
@@ -33,6 +34,8 @@ class Hon:
         self._test_data_path: Path = test_data_path or Path().cwd()
         self._mobile_id: str = mobile_id
         self._refresh_token: str = refresh_token
+        self._mqtt_client: MQTTClient | None = None
+        self._notify_function: Optional[Callable[[Any], None]] = None
 
     async def __aenter__(self) -> Self:
         return await self.create()
@@ -89,13 +92,9 @@ class Hon:
         if appliance.mac_address == "":
             return
         try:
-            await asyncio.gather(
-                *[
-                    appliance.load_attributes(),
-                    appliance.load_commands(),
-                    appliance.load_statistics(),
-                ]
-            )
+            await appliance.load_commands()
+            await appliance.load_attributes()
+            await appliance.load_statistics()
         except (KeyError, ValueError, IndexError) as error:
             _LOGGER.exception(error)
             _LOGGER.error("Device data - %s", appliance_data)
@@ -120,7 +119,15 @@ class Hon:
             api = TestAPI(test_data)
             for appliance in await api.load_appliances():
                 await self._create_appliance(appliance, api)
-        await self.api.subscribe_mqtt(self.appliances)
+        if not self._mqtt_client:
+            self._mqtt_client = await MQTTClient(self, self._mobile_id).create()
+
+    def subscribe_updates(self, notify_function: Callable[[Any], None]) -> None:
+        self._notify_function = notify_function
+
+    def notify(self) -> None:
+        if self._notify_function:
+            self._notify_function(None)
 
     async def close(self) -> None:
         await self.api.close()

pyhon/parameter/program.py

@@ -45,7 +45,7 @@ class HonParameterProgram(HonParameterEnum):
         for name, parameter in self._programs.items():
             if "iot_" in name:
                 continue
-            if parameter.parameters.get("prCode"):
+            if not parameter.parameters.get("prCode"):
                 continue
             if (fav := parameter.parameters.get("favourite")) and fav.value == "1":
                 continue

setup.py

@@ -7,7 +7,7 @@ with open("README.md", "r", encoding="utf-8") as f:
 
 setup(
     name="pyhOn",
-    version="0.17.0",
+    version="0.17.4",
     author="Andre Basche",
     description="Control hOn devices with python",
     long_description=long_description,