fix(comm): persistent dongle connection (LuxPower-standard lifecycle)

The EG4/LuxPower dongle is designed for ONE long-lived TCP connection with
heartbeats (TcpFunction 0xC1). We were opening a fresh connection per read
and per write, which the dongle drops intermittently — surfacing as flaky
local reads/writes (broken pipe, 'No valid response') while the EG4 cloud
(persistent channel) stayed solid.

- New collector/comm/persistent.py: process-wide persistent socket with
  auto-reconnect, TCP keepalive, 30s heartbeats, and a thread-safe
  transact() that serializes all reads/writes and retries through drops.
- tcp_active transport now polls over the shared persistent connection.
- API read/write/multi-write helpers now use the shared connection.
- Frontend saveSchedule no longer aborts the whole batch on one failure:
  it retries each register and reports per-register results.

Verified: 6/6 AC first registers (152-157) write and stick first-try via
the API; collector snapshots stable at 105 registers.

Author: jmewing

api/__init__.py

@@ -6,6 +6,7 @@
 import json
 import logging
 import os
+import time
 import shutil
 import subprocess
 from datetime import datetime, timedelta, timezone
@@ -1450,44 +1451,30 @@ def _read_holding_block(
     count: int,
     timeout: float = 10.0,
 ) -> Tuple[bool, Dict[int, int], str]:
-    """Read a contiguous block of holding registers directly from the inverter."""
+    """Read a contiguous block of holding registers directly from the inverter.
+
+    Uses the process-wide persistent dongle connection (LuxPower-standard
+    lifecycle) instead of a fresh connection per read.
+    """
+    from collector.comm.persistent import get_shared_connection
+
     req = build_read_request(datalog_serial, inverter_serial, MODBUS_READ_HOLD, start, count)
-    sock: Optional[socket.socket] = None
-    try:
-        sock = socket.create_connection((host, port), timeout=timeout)
-        sock.settimeout(timeout)
-        sock.sendall(req)
-        deadline = time.time() + timeout
-        buffer = b""
-        result: Dict[int, int] = {}
-        while time.time() < deadline:
-            remaining = deadline - time.time()
-            if remaining <= 0:
-                break
-            sock.settimeout(remaining)
-            try:
-                chunk = sock.recv(4096)
-            except socket.timeout:
-                break
-            if not chunk:
-                break
-            buffer += chunk
-            for frame in find_frames(buffer):
-                if frame.is_error:
-                    return False, {}, f"Modbus error response: code {frame.error_code}"
-                if frame.is_read_hold and frame.register == start:
-                    for i, raw_val in enumerate(frame.values):
-                        result[start + i] = raw_val
-                    return True, result, "ok"
+    conn = get_shared_connection(host, port, datalog_serial, inverter_serial, timeout=timeout)
+
+    def match(frame) -> bool:
+        if frame.is_error:
+            return True
+        return frame.is_read_hold and frame.register == start
+
+    frame = conn.transact(req, match, timeout=timeout)
+    if frame is None:
         return False, {}, f"No valid holding-register response for {start}/{count}"
-    except Exception as exc:
-        return False, {}, f"Holding read failed: {exc}"
-    finally:
-        if sock is not None:
-            try:
-                sock.close()
-            except Exception:
-                pass
+    if frame.is_error:
+        return False, {}, f"Modbus error response: code {frame.error_code}"
+    result: Dict[int, int] = {}
+    for i, raw_val in enumerate(frame.values):
+        result[start + i] = raw_val
+    return True, result, "ok"
 
 
 def _read_all_holding_registers(dongle: dict) -> Dict[int, int]:
@@ -1668,14 +1655,23 @@ def api_holding_put(name: str, body: HoldingUpdate):
     if not dongle["datalog_serial"] or not dongle["inverter_serial"]:
         raise HTTPException(400, "datalog_serial / inverter_serial not configured")
 
-    ok, msg = _write_holding_register(
-        dongle["dongle_host"],
-        dongle["dongle_port"],
-        dongle["datalog_serial"],
-        dongle["inverter_serial"],
-        reg,
-        raw_value,
-    )
+    # The dongle intermittently drops individual write transactions (returns
+    # no valid response). Retry a few times before surfacing a 502 — a retry
+    # usually lands the write (observed needing 1-2 attempts per register).
+    ok = False
+    msg = ""
+    for attempt in range(4):
+        ok, msg = _write_holding_register(
+            dongle["dongle_host"],
+            dongle["dongle_port"],
+            dongle["datalog_serial"],
+            dongle["inverter_serial"],
+            reg,
+            raw_value,
+        )
+        if ok:
+            break
+        time.sleep(0.7)
     if not ok:
         raise HTTPException(502, f"Failed to write {name}: {msg}")
     logger.info("Wrote holding register %s (address %d) = %d", name, reg, raw_value)

api/static/index.html

@@ -1056,19 +1056,38 @@ <h4 style="margin-bottom:0.5rem;">${g.title}</h4>
   try {
     // Write sequentially (not in parallel) — the dongle rejects bursts of
     // simultaneous write connections, which surfaced as intermittent 502s.
+    // Do NOT abort the whole batch on one failure: the dongle is flaky and
+    // drops individual registers, so we record failures and continue, then
+    // report exactly which registers did and did not write.
     let written = 0;
+    const failed = [];
     for (const [name, value] of Object.entries(values)) {
-      const r = await fetch(API + '/api/holding/' + encodeURIComponent(name), {
-        method:'PUT', headers:{'Content-Type':'application/json'}, body:JSON.stringify({value: Number(value)})
-      });
-      if (!r.ok) {
-        let detail = r.statusText;
-        try { const j = await r.json(); if (j.detail) detail = j.detail; } catch (e) {}
-        throw new Error(name + ': ' + detail);
+      let ok = false;
+      // Retry a few times (the dongle intermittently drops a write; a retry
+      // usually lands it — observed needing 1-2 attempts per register).
+      for (let attempt = 0; attempt < 4 && !ok; attempt++) {
+        try {
+          const r = await fetch(API + '/api/holding/' + encodeURIComponent(name), {
+            method:'PUT', headers:{'Content-Type':'application/json'}, body:JSON.stringify({value: Number(value)})
+          });
+          if (r.ok) { ok = true; break; }
+          let detail = r.statusText;
+          try { const j = await r.json(); if (j.detail) detail = j.detail; } catch (e) {}
+          if (attempt === 3) failed.push(name + ': ' + detail);
+        } catch (e) {
+          if (attempt === 3) failed.push(name + ': ' + e.message);
+        }
+        await new Promise(res => setTimeout(res, 700));
       }
-      written++;
+      if (ok) written++;
+    }
+    if (failed.length) {
+      msg.textContent = `Value set — ${written} written, ${failed.length} FAILED (${failed.join('; ')})`;
+      msg.style.color = 'var(--red-color)';
+    } else {
+      msg.textContent = `Value set — ${written} register${written === 1 ? '' : 's'} written.`;
+      msg.style.color = 'var(--green-color)';
     }
-    msg.textContent = `Value set — ${written} register${written === 1 ? '' : 's'} written.`; msg.style.color = 'var(--green-color)';
   } catch (e) {
     msg.textContent = 'Error setting value: ' + e.message; msg.style.color = 'var(--red-color)';
   }

collector/automation.py

@@ -100,45 +100,30 @@ def _write_holding_register(
     value: int,
     timeout: float = 10.0,
 ) -> Tuple[bool, str]:
-    """Send a WriteSingleRegister request and verify the echo."""
+    """Send a WriteSingleRegister request and verify the echo.
+
+    Uses the process-wide persistent dongle connection (LuxPower-standard
+    lifecycle: one long-lived socket with heartbeats) instead of opening a
+    fresh connection per write, which the dongle drops intermittently.
+    """
+    from .comm.persistent import get_shared_connection
+
     req = build_write_request(datalog_serial, inverter_serial, register, value)
-    sock: Optional[socket.socket] = None
-    try:
-        sock = socket.create_connection((host, port), timeout=timeout)
-        sock.settimeout(timeout)
-        sock.sendall(req)
-
-        deadline = time.time() + timeout
-        buffer = b""
-        while time.time() < deadline:
-            remaining = deadline - time.time()
-            if remaining <= 0:
-                break
-            sock.settimeout(remaining)
-            try:
-                chunk = sock.recv(4096)
-            except socket.timeout:
-                break
-            if not chunk:
-                break
-            buffer += chunk
-            frames = find_frames(buffer)
-            for frame in frames:
-                if frame.is_error:
-                    return False, f"Modbus error response: code {frame.error_code}"
-                if frame.device_function == MODBUS_WRITE_SINGLE and frame.register == register:
-                    if frame.values and frame.values[0] == value:
-                        return True, f"Wrote register {register} = {value}"
-                    return True, f"Write accepted (echo value {frame.values})"
+    conn = get_shared_connection(host, port, datalog_serial, inverter_serial, timeout=timeout)
+
+    def match(frame) -> bool:
+        if frame.is_error:
+            return True
+        return frame.device_function == MODBUS_WRITE_SINGLE and frame.register == register
+
+    frame = conn.transact(req, match, timeout=timeout)
+    if frame is None:
         return False, "No valid write response received"
-    except Exception as exc:
-        return False, f"Write failed: {exc}"
-    finally:
-        if sock is not None:
-            try:
-                sock.close()
-            except Exception:
-                pass
+    if frame.is_error:
+        return False, f"Modbus error response: code {frame.error_code}"
+    if frame.values and frame.values[0] == value:
+        return True, f"Wrote register {register} = {value}"
+    return True, f"Write accepted (echo value {frame.values})"
 
 
 def _write_holding_registers(
@@ -154,47 +139,28 @@ def _write_holding_registers(
 
     Used by the 7-day scheduling block (registers 500-723), which the inverter
     only accepts via multi-register writes (no single-register writes).
+    Uses the process-wide persistent dongle connection.
     """
+    from .comm.persistent import get_shared_connection
+
     req = build_write_multi_request(
         datalog_serial, inverter_serial, start_register, values
     )
-    sock: Optional[socket.socket] = None
-    try:
-        sock = socket.create_connection((host, port), timeout=timeout)
-        sock.settimeout(timeout)
-        sock.sendall(req)
-
-        deadline = time.time() + timeout
-        buffer = b""
-        while time.time() < deadline:
-            remaining = deadline - time.time()
-            if remaining <= 0:
-                break
-            sock.settimeout(remaining)
-            try:
-                chunk = sock.recv(4096)
-            except socket.timeout:
-                break
-            if not chunk:
-                break
-            buffer += chunk
-            frames = find_frames(buffer)
-            for frame in frames:
-                if frame.is_error:
-                    return False, f"Modbus error response: code {frame.error_code}"
-                if frame.device_function == MODBUS_WRITE_MULTI and frame.register == start_register:
-                    if frame.write_count == len(values):
-                        return True, f"Wrote {len(values)} registers starting at {start_register}"
-                    return True, f"Write accepted (echo count {frame.write_count})"
+    conn = get_shared_connection(host, port, datalog_serial, inverter_serial, timeout=timeout)
+
+    def match(frame) -> bool:
+        if frame.is_error:
+            return True
+        return frame.device_function == MODBUS_WRITE_MULTI and frame.register == start_register
+
+    frame = conn.transact(req, match, timeout=timeout)
+    if frame is None:
         return False, "No valid write response received"
-    except Exception as exc:
-        return False, f"Write failed: {exc}"
-    finally:
-        if sock is not None:
-            try:
-                sock.close()
-            except Exception:
-                pass
+    if frame.is_error:
+        return False, f"Modbus error response: code {frame.error_code}"
+    if frame.write_count == len(values):
+        return True, f"Wrote {len(values)} registers starting at {start_register}"
+    return True, f"Write accepted (echo count {frame.write_count})"
 
 
 def _read_holding_register(
@@ -209,41 +175,21 @@ def _read_holding_register(
     req = build_read_request(
         datalog_serial, inverter_serial, MODBUS_READ_HOLD, register, 1
     )
-    sock: Optional[socket.socket] = None
-    try:
-        sock = socket.create_connection((host, port), timeout=timeout)
-        sock.settimeout(timeout)
-        sock.sendall(req)
-
-        deadline = time.time() + timeout
-        buffer = b""
-        while time.time() < deadline:
-            remaining = deadline - time.time()
-            if remaining <= 0:
-                break
-            sock.settimeout(remaining)
-            try:
-                chunk = sock.recv(4096)
-            except socket.timeout:
-                break
-            if not chunk:
-                break
-            buffer += chunk
-            frames = find_frames(buffer)
-            for frame in frames:
-                if frame.is_error:
-                    return False, None, f"Modbus error response: code {frame.error_code}"
-                if frame.is_read_hold and frame.register == register and frame.values:
-                    return True, frame.values[0], f"Read back {register} = {frame.values[0]}"
+    from .comm.persistent import get_shared_connection
+
+    conn = get_shared_connection(host, port, datalog_serial, inverter_serial, timeout=timeout)
+
+    def match(frame) -> bool:
+        if frame.is_error:
+            return True
+        return frame.is_read_hold and frame.register == register and bool(frame.values)
+
+    frame = conn.transact(req, match, timeout=timeout)
+    if frame is None:
         return False, None, "No valid read response received"
-    except Exception as exc:
-        return False, None, f"Read failed: {exc}"
-    finally:
-        if sock is not None:
-            try:
-                sock.close()
-            except Exception:
-                pass
+    if frame.is_error:
+        return False, None, f"Modbus error response: code {frame.error_code}"
+    return True, frame.values[0], f"Read back {register} = {frame.values[0]}"
 
 
 # ── SolarAssistant setting name → lux-mon holding-register name ─────────────