Add gap reconciliation for MQTT disconnects

Author: Feberdin

README.md

@@ -18,6 +18,7 @@ Inoffizielle Home Assistant Integration für die **EcoFlow PowerOcean Plus** Pho
 - **Energie-Dashboard** — kWh-Zähler direkt integriert, kein YAML nötig
 - **Verbindungsstatus** — MQTT-Verbindung als Sensor für Automationen
 - **Options Flow** — Anzahl Batterie-Packs jederzeit änderbar ohne Neueinrichtung
+- **Gap-Reconciliation** — bei kurzer Internet-Unterbrechung wird die Energielücke beim Reconnect transparent geschätzt
 
 ---
 
@@ -182,6 +183,9 @@ Die kWh-Sensoren sind direkt einsatzbereit. Navigiere zu *Einstellungen → Dash
 **Hinweise:**
 - Zähler starten mit der ersten MQTT-Nachricht — historische Werte werden nicht rückwirkend berechnet
 - Werte bleiben über HA-Neustarts erhalten
+- Bei MQTT-/Internet-Lücken wird beim Reconnect eine Schätzung angewendet
+  (Trapezregel aus letzter Leistung vor Disconnect und erster Leistung nach Reconnect)
+- Sehr lange Unterbrechungen werden aus Sicherheitsgründen nicht automatisch nachgerechnet
 - Kleine Messschwankungen (±5 W) können gleichzeitig minimale Bezugs- und Einspeisungswerte erzeugen — physikalisch normal, Einfluss auf Monatssummen vernachlässigbar
 
 ---
@@ -194,6 +198,7 @@ Die kWh-Sensoren sind direkt einsatzbereit. Navigiere zu *Einstellungen → Dash
 2. HA-Netzwerkverbindung prüfen
 3. Logs prüfen: *Einstellungen → System → Logs → „ecoflow"*
 4. Verbindungsstatus-Sensor prüfen: zeigt er `disconnected`?
+5. Im Verbindungsstatus-Sensor die Attribute `last_gap_*` prüfen (Start/Ende/Dauer der letzten Lücke)
 
 ### Login schlägt fehl
 

custom_components/ecoflow_powerocean/const.py

@@ -66,6 +66,13 @@
 MQTT_FIRST_DATA_TIMEOUT = 20
 """Maximale Wartezeit in Sekunden auf erste Nutzdaten nach dem Verbindungsaufbau."""
 
+GAP_RECONCILIATION_MIN_SECONDS = 60
+"""Mindestdauer einer MQTT-Unterbrechung, ab der eine Lücken-Korrektur berechnet wird."""
+
+GAP_RECONCILIATION_MAX_SECONDS = 6 * 60 * 60
+"""Maximale Dauer (in Sekunden), die für eine automatische Lücken-Korrektur berücksichtigt wird.
+Längere Unterbrechungen werden aus Sicherheitsgründen nicht automatisch nachgerechnet."""
+
 TOPIC_DEVICE_PROPERTY = "/app/device/property/{sn}"
 """MQTT-Topic, auf dem das Gerät regelmäßig seinen Zustand veröffentlicht."""
 

custom_components/ecoflow_powerocean/coordinator.py

@@ -33,7 +33,7 @@
 import ssl
 import threading
 import uuid
-from datetime import timedelta
+from datetime import datetime, timedelta
 from typing import Any
 
 import aiohttp
@@ -51,6 +51,7 @@
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+from homeassistant.util import dt as dt_util
 
 from .const import (
     API_CERT_URL,
@@ -64,6 +65,8 @@
     DATA_EMS_HEARTBEAT,
     DATA_ENERGY_STREAM,
     DOMAIN,
+    GAP_RECONCILIATION_MAX_SECONDS,
+    GAP_RECONCILIATION_MIN_SECONDS,
     MQTT_HOST,
     MQTT_KEEPALIVE,
     MQTT_PORT,
@@ -119,6 +122,17 @@ def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
         self._mqtt_client: mqtt.Client | None = None
         self._mqtt_connected: bool = False
         self._mqtt_lock = threading.Lock()
+        self._last_message_at: datetime | None = None
+
+        # Gap-Reconciliation:
+        # Erfasst Verbindungsunterbrechungen und stellt Metadaten für die
+        # Energie-Akkumulatoren bereit, damit diese eine kontrollierte
+        # Schätzung für Offline-Phasen anwenden können.
+        self._disconnect_started_at: datetime | None = None
+        self._last_gap_started_at: datetime | None = None
+        self._last_gap_ended_at: datetime | None = None
+        self._last_gap_seconds: float = 0.0
+        self._gap_event_id: int = 0
 
         # Initialer Datensatz
         self.data: dict[str, Any] = {
@@ -292,7 +306,35 @@ def _on_mqtt_connect(self, client, userdata, flags, reason_code, properties=None
         # reason_code: 0 (v1) oder ReasonCode.SUCCESS (v2)
         rc = reason_code if isinstance(reason_code, int) else reason_code.value
         if rc == 0:
+            now = dt_util.utcnow()
             self._mqtt_connected = True
+            if self._disconnect_started_at is not None:
+                gap_seconds = max(
+                    0.0,
+                    (now - self._disconnect_started_at).total_seconds(),
+                )
+                self._disconnect_started_at = None
+
+                if (
+                    GAP_RECONCILIATION_MIN_SECONDS
+                    <= gap_seconds
+                    <= GAP_RECONCILIATION_MAX_SECONDS
+                ):
+                    self._last_gap_seconds = gap_seconds
+                    self._last_gap_started_at = now - timedelta(seconds=gap_seconds)
+                    self._last_gap_ended_at = now
+                    self._gap_event_id += 1
+                    _LOGGER.info(
+                        "MQTT-Lücke erkannt (%.0f s) — Energie-Akkumulatoren wenden Gap-Schätzung an",
+                        gap_seconds,
+                    )
+                elif gap_seconds > GAP_RECONCILIATION_MAX_SECONDS:
+                    _LOGGER.warning(
+                        "MQTT-Lücke zu lang für automatische Schätzung (%.0f s > %.0f s), "
+                        "Gap-Korrektur wird übersprungen",
+                        gap_seconds,
+                        GAP_RECONCILIATION_MAX_SECONDS,
+                    )
             topic = TOPIC_DEVICE_PROPERTY.format(sn=self.serial_number)
             client.subscribe(topic, qos=1)
             _LOGGER.info("MQTT verbunden, abonniere Topic: %s", topic)
@@ -306,6 +348,8 @@ def _on_mqtt_disconnect(self, client, userdata, disconnect_flags=None, reason_co
 
         paho-mqtt versucht automatisch erneut zu verbinden (loop_start() + reconnect_delay_set).
         """
+        if self._mqtt_connected and self._disconnect_started_at is None:
+            self._disconnect_started_at = dt_util.utcnow()
         self._mqtt_connected = False
         _LOGGER.warning("MQTT-Verbindung getrennt (code=%s), Reconnect läuft…", reason_code)
 
@@ -329,6 +373,7 @@ def _on_mqtt_message(self, client, userdata, msg):
 
         # Vorhandene Daten aktualisieren (nicht überschreiben)
         with self._mqtt_lock:
+            self._last_message_at = dt_util.utcnow()
             new_batteries = dict(self.data.get(DATA_BATTERIES, {}))
             for pack in battery_packs:
                 new_batteries[pack.pack_index] = pack
@@ -416,3 +461,24 @@ async def async_shutdown(self) -> None:
             await self.hass.async_add_executor_job(self._mqtt_client.disconnect)
             self._mqtt_client = None
             self._mqtt_connected = False
+            self._disconnect_started_at = None
+
+    @property
+    def gap_event_id(self) -> int:
+        """Monotoner Zähler: erhöht sich bei jeder relevanten Disconnect-Lücke."""
+        return self._gap_event_id
+
+    @property
+    def last_gap_seconds(self) -> float:
+        """Dauer der zuletzt erkannten MQTT-Lücke in Sekunden."""
+        return self._last_gap_seconds
+
+    @property
+    def last_gap_started_at(self) -> datetime | None:
+        """Startzeitpunkt der zuletzt erkannten MQTT-Lücke (UTC)."""
+        return self._last_gap_started_at
+
+    @property
+    def last_gap_ended_at(self) -> datetime | None:
+        """Endzeitpunkt der zuletzt erkannten MQTT-Lücke (UTC)."""
+        return self._last_gap_ended_at

custom_components/ecoflow_powerocean/diagnostics.py

@@ -77,6 +77,10 @@ async def async_get_config_entry_diagnostics(
             getattr(coordinator, "_mqtt_user", None)
             and getattr(coordinator, "_mqtt_password", None)
         ),
+        "gap_event_id": getattr(coordinator, "gap_event_id", None),
+        "last_gap_seconds": getattr(coordinator, "last_gap_seconds", None),
+        "last_gap_started_at": _to_jsonable(getattr(coordinator, "last_gap_started_at", None)),
+        "last_gap_ended_at": _to_jsonable(getattr(coordinator, "last_gap_ended_at", None)),
         "data": _to_jsonable(getattr(coordinator, "data", None)),
     }
 

custom_components/ecoflow_powerocean/manifest.json

@@ -1,7 +1,7 @@
 {
   "domain": "ecoflow_powerocean",
   "name": "EcoFlow PowerOcean",
-  "version": "0.3.5",
+  "version": "0.3.6",
   "codeowners": [],
   "config_flow": true,
   "documentation": "https://github.com/Feberdin/ecoflow-powerocean-ha",

custom_components/ecoflow_powerocean/sensor.py

@@ -950,6 +950,7 @@ def __init__(self, coordinator, description, device_info, serial):
         self._accumulated_kwh: float = 0.0
         self._last_update: datetime | None = None
         self._last_power_w: float = 0.0
+        self._seen_gap_event_id: int = 0
 
     async def async_added_to_hass(self) -> None:
         await super().async_added_to_hass()
@@ -968,7 +969,32 @@ def _handle_coordinator_update(self) -> None:
         power_w = self._get_power_w()
         if self._last_update is not None:
             dt_hours = (now - self._last_update).total_seconds() / 3600.0
-            self._accumulated_kwh += (self._last_power_w / 1000.0) * dt_hours
+            gap_hours = 0.0
+            if self.coordinator.gap_event_id != self._seen_gap_event_id:
+                # Für die erste Aktualisierung nach Reconnect:
+                # - Nicht mit "letzter Leistung über gesamte Offline-Zeit" integrieren.
+                # - Stattdessen Trapez-Schätzung zwischen letzter und erster Leistung.
+                gap_hours = max(self.coordinator.last_gap_seconds, 0.0) / 3600.0
+                self._seen_gap_event_id = self.coordinator.gap_event_id
+
+            active_hours = max(dt_hours - gap_hours, 0.0)
+            self._accumulated_kwh += (self._last_power_w / 1000.0) * active_hours
+
+            if gap_hours > 0.0:
+                estimated_gap_power_w = max(
+                    (self._last_power_w + power_w) / 2.0,
+                    0.0,
+                )
+                gap_kwh = (estimated_gap_power_w / 1000.0) * gap_hours
+                self._accumulated_kwh += gap_kwh
+                _LOGGER.info(
+                    "Gap-Reconciliation %s: +%.4f kWh (Lücke %.1f min, P_alt=%.1f W, P_neu=%.1f W)",
+                    self.entity_description.key,
+                    gap_kwh,
+                    gap_hours * 60.0,
+                    self._last_power_w,
+                    power_w,
+                )
         self._last_update = now
         self._last_power_w = power_w
         super()._handle_coordinator_update()
@@ -1026,6 +1052,20 @@ def icon(self) -> str:
     def available(self) -> bool:
         return True  # Immer verfügbar — zeigt connected/disconnected
 
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Zeigt zuletzt erkannte Verbindungs-Lücke für transparentes Debugging."""
+        attrs: dict[str, Any] = {}
+        if self.coordinator.last_gap_started_at is not None:
+            attrs["last_gap_started_at"] = self.coordinator.last_gap_started_at.isoformat()
+        if self.coordinator.last_gap_ended_at is not None:
+            attrs["last_gap_ended_at"] = self.coordinator.last_gap_ended_at.isoformat()
+        if self.coordinator.last_gap_seconds > 0:
+            attrs["last_gap_seconds"] = round(self.coordinator.last_gap_seconds, 1)
+            attrs["last_gap_minutes"] = round(self.coordinator.last_gap_seconds / 60.0, 2)
+            attrs["gap_event_id"] = self.coordinator.gap_event_id
+        return attrs
+
 
 # ── Hilfsfunktionen ───────────────────────────────────────────────────────────