fix(mqtt): reconnect after dropped broker connection (#787)

Author: Komzpa

custom_components/ecoflow_cloud/api/__init__.py

@@ -1,4 +1,5 @@
 import logging
+import time
 from abc import ABC, abstractmethod
 from typing import Any
 
@@ -31,6 +32,8 @@ def __init__(self):
         self.mqtt_info: EcoflowMqttInfo
         self.devices: dict[str, Any] = {}
         self.mqtt_client: EcoflowMQTTClient
+        self._mqtt_reconnect_last_attempt = 0.0
+        self._mqtt_reconnect_count = 0
 
     @abstractmethod
     async def login(self):
@@ -133,6 +136,22 @@ def start(self):
 
         self.mqtt_client = EcoflowMQTTClient(self.mqtt_info, self.devices)
 
+    def schedule_mqtt_reconnect(self, cooldown_sec: int = 60) -> int | None:
+        if self.mqtt_client.is_connected():
+            return None
+
+        now = time.monotonic()
+        if now - self._mqtt_reconnect_last_attempt < cooldown_sec:
+            return None
+
+        self._mqtt_reconnect_last_attempt = now
+        self._mqtt_reconnect_count += 1
+        return self._mqtt_reconnect_count
+
+    @property
+    def mqtt_reconnect_count(self) -> int:
+        return self._mqtt_reconnect_count
+
     def stop(self):
         _LOGGER.debug("Stopping MQTT client for %s", self.mqtt_info.client_id)
         assert self.mqtt_client is not None

custom_components/ecoflow_cloud/sensor.py

@@ -38,6 +38,7 @@
     ATTR_DATA_UPDATES,
     ATTR_MQTT_CONNECTED,
     ATTR_QUOTA_REQUESTS,
+    ATTR_STATUS_RECONNECTS,
     ATTR_STATUS_DATA_LAST_UPDATE,
     ATTR_STATUS_LAST_UPDATE,
     ATTR_STATUS_SN,
@@ -503,6 +504,7 @@ def __init__(
         self._attrs[ATTR_STATUS_SN] = self._device.device_info.sn
         self._attrs[ATTR_STATUS_DATA_LAST_UPDATE] = None
         self._attrs[ATTR_MQTT_CONNECTED] = None
+        self._attrs[ATTR_STATUS_RECONNECTS] = 0
         if poll_when_silent or scheduled_refresh_sec is not None:
             self._attrs[ATTR_QUOTA_REQUESTS] = 0
 
@@ -512,6 +514,9 @@ def _handle_coordinator_update(self) -> None:
         status = self._tracker.status
         changed = status != self._prev_status
 
+        if self._schedule_mqtt_reconnect():
+            changed = True
+
         # Active polling when device goes silent
         if self._poll_when_silent and status == OnlineStatus.ASSUME_OFFLINE:
             if (dt.utcnow() - self._last_poll).total_seconds() >= self._tracker.assume_offline_sec:
@@ -543,6 +548,21 @@ def _handle_coordinator_update(self) -> None:
                 self._actualize_attributes()
                 self.schedule_update_ha_state()
 
+    def _schedule_mqtt_reconnect(self) -> bool:
+        reconnect_count = self._client.schedule_mqtt_reconnect()
+        if reconnect_count is None:
+            return False
+
+        self._attrs[ATTR_STATUS_RECONNECTS] = reconnect_count
+        self.hass.async_create_background_task(
+            self._async_reconnect_mqtt(),
+            f"reconnect ecoflow mqtt {self._device.device_info.sn}",
+        )
+        return True
+
+    async def _async_reconnect_mqtt(self) -> None:
+        await self.hass.async_add_executor_job(self._client.mqtt_client.reconnect)
+
     def _format_age(self, timestamp: datetime | None) -> str | None:
         if timestamp is None:
             return None
@@ -558,6 +578,7 @@ def _actualize_attributes(self):
             self._attrs[ATTR_STATUS_UPDATES] = self._tracker._explicit_status_count
             self._attrs[ATTR_DATA_UPDATES] = self._tracker._data_received_count
         self._attrs[ATTR_MQTT_CONNECTED] = self._client.mqtt_client.is_connected()
+        self._attrs[ATTR_STATUS_RECONNECTS] = self._client.mqtt_reconnect_count
 
     @property
     def extra_state_attributes(self) -> Mapping[str, Any] | None: