quick nl52 usb probe

Adds the recovery layer for the NL-43 TCP wedge failure mode (control
listener on 2255 dies while device/FTP stay alive — see
SLM-stress-test/NL43_RX55_TCP_Wedge_Investigation_2026-02-18.md).

- Failure classification: poll failures probed raw and classified as
  wedged (refused = stack alive, listener gone) vs offline (timeout).
  FTP-port probe upgrades wedge confidence to definitive.
- RecoveryManager state machine: confirm wedge -> trigger reset ->
  wait boot -> reconnect -> resume measurement via start_cycle (keeps
  overwrite protection) -> full incident trail in device logs.
- Pluggable reset backends: manual (notify, default) and webhook
  (HTTP to relay controller: Pi GPIO, Shelly, Tasmota, etc.).
- Safety: auto_recovery_enabled off by default, 2-failure confirmation
  threshold, windowed attempt limit (2 per 6h), one recovery per device.
- Endpoints: GET recovery/status, POST recovery/probe (diagnostic),
  POST recovery/trigger (manual, bypasses threshold).
- Migration: migrate_add_recovery_fields.py
- Tests: test_wedge_recovery.py — fake wedge-able NL-43 + fake relay
  webhook; 23 assertions covering classification, end-to-end recovery
  with resume, and gating. All passing.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -5,6 +5,26 @@ All notable changes to SLMM (Sound Level Meter Manager) will be documented in th
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+#### Wedge Detection & Automatic Recovery
+- **Failure classification** - Poll failures are now probed and classified: `wedged` (control port REFUSES — device stack alive, listener gone), `offline` (timeout — power/cellular loss), or `ok` (transient). The wedge signature matches the Feb 2026 investigation: refused on 2255 while FTP/21 accepts is definitive (high confidence); refused alone is medium confidence.
+- **RecoveryManager** (`app/recovery.py`) - Per-device recovery state machine: confirm wedge → trigger reset → wait for boot → reconnect → resume measurement (via existing `start_cycle` with overwrite protection) → log the full incident to device logs (category `RECOVERY`).
+- **Pluggable reset backends**:
+  - `manual` (default) - logs/flags the wedge for a human; still auto-resumes measurement when the device returns
+  - `webhook` - HTTP GET/POST to a relay controller (Pi GPIO relay, Shelly, Tasmota, etc.) that power-cycles the NL-43
+- **Safety guards** - per-device `auto_recovery_enabled` master switch (default off), confirmation threshold (`WEDGE_CONFIRM_FAILURES`, default 2 consecutive failures), windowed attempt limit (default 2 per 6 h), one recovery at a time per device.
+- **New endpoints**:
+  - `GET /api/nl43/{unit_id}/recovery/status` - connection/recovery state, wedge count, last result
+  - `POST /api/nl43/{unit_id}/recovery/probe` - diagnostic port probe + classification (no action taken)
+  - `POST /api/nl43/{unit_id}/recovery/trigger` - manually start recovery (bypasses confirmation threshold)
+- **Config fields** (via `PUT /{unit_id}/config`): `auto_recovery_enabled`, `reset_backend`, `reset_webhook_url`, `reset_webhook_method`, `recovery_boot_wait_seconds`, `recovery_reconnect_timeout`, `recovery_auto_resume`, `recovery_max_attempts`, `recovery_window_minutes`
+- **Status fields**: `connection_state`, `last_wedge_at`, `wedge_count`, `recovery_state`, `last_recovery_at`, `last_recovery_result`
+- **Migration**: `migrate_add_recovery_fields.py`
+- **Test suite**: `test_wedge_recovery.py` - fake wedge-able NL-43 + fake relay webhook; covers classification, end-to-end recovery with measurement resume, and gating (23 assertions)
+
 ## [0.3.0] - 2026-02-17
 
 ### Added

app/background_poller.py

@@ -17,6 +17,7 @@ from app.database import SessionLocal
 from app.models import NL43Config, NL43Status
 from app.services import NL43Client, persist_snapshot, sync_measurement_start_time_from_ftp
 from app.device_logger import log_device_event, cleanup_old_logs
+from app.recovery import classify_failure, recovery_manager
 
 logger = logging.getLogger(__name__)
 
@@ -231,6 +232,7 @@ class BackgroundPoller:
             status.consecutive_failures = 0
             status.last_success = datetime.utcnow()
             status.last_error = None
+            status.connection_state = "ok"
 
             db.commit()
             self._logger.info(f"✓ Successfully polled {unit_id}")
@@ -322,6 +324,61 @@ class BackgroundPoller:
 
             db.commit()
 
+            # Classify the failure: wedge (listener dead, host alive) vs offline
+            await self._classify_and_recover(cfg, status, db)
+
+    async def _classify_and_recover(self, cfg: NL43Config, status: NL43Status, db: Session):
+        """Probe the device to classify a poll failure and trigger wedge recovery.
+
+        The wedge signature (see SLM-stress-test investigation doc): control
+        port REFUSES connections (RST = host stack alive, listener gone) while
+        the device is otherwise up. A timeout means offline (power/cell loss).
+        """
+        unit_id = cfg.unit_id
+
+        # Skip while a recovery for this device is already running — its own
+        # probes manage state until it finishes.
+        if recovery_manager.is_active(unit_id):
+            return
+
+        try:
+            classification = await classify_failure(
+                cfg.host,
+                cfg.tcp_port,
+                ftp_port=(cfg.ftp_port or 21) if cfg.ftp_enabled else None,
+            )
+        except Exception as e:
+            self._logger.warning(f"Failure classification error for {unit_id}: {e}")
+            return
+
+        new_state = classification["state"]
+        prev_state = status.connection_state or "ok"
+
+        if new_state != prev_state:
+            status.connection_state = new_state
+            if new_state == "wedged":
+                status.last_wedge_at = datetime.utcnow()
+                status.wedge_count = (status.wedge_count or 0) + 1
+                log_device_event(
+                    unit_id, "ERROR", "RECOVERY",
+                    f"WEDGE detected: control port {cfg.tcp_port} refused "
+                    f"(tcp_probe={classification['tcp_probe']}, ftp_probe={classification['ftp_probe']}, "
+                    f"confidence={classification['confidence']})",
+                    db,
+                )
+                self._logger.error(f"Device {unit_id} classified as WEDGED ({classification})")
+            elif new_state == "offline":
+                log_device_event(
+                    unit_id, "WARNING", "RECOVERY",
+                    f"Device classified OFFLINE (tcp_probe={classification['tcp_probe']}) - "
+                    f"not a wedge, no reset will be triggered",
+                    db,
+                )
+            db.commit()
+
+        if new_state == "wedged":
+            recovery_manager.maybe_start_recovery(cfg, status, classification, db)
+
     def _calculate_sleep_interval(self) -> int:
         """
         Calculate the next sleep interval based on all device poll intervals.

app/main.py

@@ -76,12 +76,12 @@ app.include_router(routers.router)
 
 @app.get("/", response_class=HTMLResponse)
 def index(request: Request):
-    return templates.TemplateResponse(request, "index.html")
+    return templates.TemplateResponse("index.html", {"request": request})
 
 
 @app.get("/roster", response_class=HTMLResponse)
 def roster(request: Request):
-    return templates.TemplateResponse(request, "roster.html")
+    return templates.TemplateResponse("roster.html", {"request": request})
 
 
 @app.get("/health")

app/models.py

@@ -23,6 +23,17 @@ class NL43Config(Base):
     poll_interval_seconds = Column(Integer, nullable=True, default=60)  # Polling interval (10-3600 seconds)
     poll_enabled = Column(Boolean, default=True)  # Enable/disable background polling for this device
 
+    # Wedge recovery configuration
+    auto_recovery_enabled = Column(Boolean, default=False)  # Master switch for automatic wedge recovery
+    reset_backend = Column(String, default="manual")  # Reset actuation: "manual" (notify only) or "webhook"
+    reset_webhook_url = Column(String, nullable=True)  # URL that triggers a power cycle (relay controller)
+    reset_webhook_method = Column(String, default="POST")  # HTTP method for webhook (GET or POST)
+    recovery_boot_wait_seconds = Column(Integer, default=90)  # Wait after reset before reconnect attempts
+    recovery_reconnect_timeout = Column(Integer, default=180)  # Max seconds to wait for device after boot wait
+    recovery_auto_resume = Column(Boolean, default=True)  # Restart measurement if it was running pre-wedge
+    recovery_max_attempts = Column(Integer, default=2)  # Max recovery attempts per window
+    recovery_window_minutes = Column(Integer, default=360)  # Attempt-limit window (default 6 hours)
+
 
 class NL43Status(Base):
     """
@@ -57,6 +68,14 @@ class NL43Status(Base):
     # FTP start time sync tracking
     start_time_sync_attempted = Column(Boolean, default=False)  # True if FTP sync was attempted for current measurement
 
+    # Wedge detection and recovery tracking
+    connection_state = Column(String, default="ok")  # ok | wedged | degraded | offline
+    last_wedge_at = Column(DateTime, nullable=True)  # When the current/most recent wedge was detected
+    wedge_count = Column(Integer, default=0)  # Lifetime count of detected wedges
+    recovery_state = Column(String, default="idle")  # idle | resetting | waiting_boot | reconnecting | resuming | awaiting_manual_reset | failed
+    last_recovery_at = Column(DateTime, nullable=True)  # When the last recovery attempt started
+    last_recovery_result = Column(Text, nullable=True)  # Outcome summary of the last recovery attempt
+
 
 class DeviceLog(Base):
     """

app/recovery.py

@@ -0,0 +1,482 @@
+"""
+Wedge detection and automatic recovery for NL-43 devices.
+
+The NL-43's TCP control listener (port 2255) can stop listening while the
+device otherwise remains healthy (LAN stack alive, FTP alive) — the "wedge"
+failure mode documented in SLM-stress-test/NL43_RX55_TCP_Wedge_Investigation_2026-02-18.md.
+Only a power cycle of the NL-43 has been proven to restore the listener.
+
+This module provides:
+
+1. Failure classification — distinguishes a wedge (connection REFUSED, i.e.
+   the device's IP stack answered with RST so the host is alive but the
+   listener is gone) from the device being offline (connect timeout — power
+   loss, cellular outage). An FTP-port probe adds confidence when available.
+
+2. RecoveryManager — a per-device state machine that, once a wedge is
+   confirmed, triggers a reset through a pluggable backend, waits for the
+   device to boot, reconnects, and resumes measurement if one was running.
+
+Reset backends:
+- "manual"  — log/flag only; a human power-cycles the device. The manager
+              keeps watching and still auto-resumes measurement when the
+              device returns.
+- "webhook" — HTTP request to a relay controller (Pi GPIO relay, Shelly,
+              Tasmota, RX55 GPIO bridge, etc.) that power-cycles the device.
+
+Recovery states (NL43Status.recovery_state):
+    idle -> resetting -> waiting_boot -> reconnecting -> resuming -> idle
+                  \\-> awaiting_manual_reset -> reconnecting -> ...
+    any step may end in "failed" (recorded in last_recovery_result).
+"""
+
+import asyncio
+import json
+import logging
+import os
+import time
+import urllib.request
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.database import SessionLocal
+from app.models import NL43Config, NL43Status
+from app.device_logger import log_device_event
+
+logger = logging.getLogger(__name__)
+
+# Number of consecutive wedge-classified poll failures required before
+# automatic recovery is triggered (manual trigger bypasses this).
+WEDGE_CONFIRM_FAILURES = int(os.getenv("WEDGE_CONFIRM_FAILURES", "2"))
+
+# Timeout for individual TCP probes during classification/reconnection
+PROBE_TIMEOUT = float(os.getenv("RECOVERY_PROBE_TIMEOUT", "5.0"))
+
+# Seconds between reconnect probes after a reset
+RECONNECT_PROBE_INTERVAL = 10
+# Seconds between probes while awaiting a manual reset
+MANUAL_WATCH_PROBE_INTERVAL = 30
+
+
+# ---------------------------------------------------------------------------
+# TCP probing and failure classification
+# ---------------------------------------------------------------------------
+
+async def probe_tcp(host: str, port: int, timeout: float = PROBE_TIMEOUT) -> str:
+    """Probe a TCP port and report what happened.
+
+    Returns one of:
+        "open"        — connect succeeded (closed again immediately)
+        "refused"     — RST received: host stack alive, no listener
+        "timeout"     — no response: host down / network unreachable / filtered
+        "unreachable" — OS-level error (no route, etc.)
+    """
+    try:
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection(host, port), timeout=timeout
+        )
+        writer.close()
+        try:
+            await writer.wait_closed()
+        except Exception:
+            pass
+        return "open"
+    except asyncio.TimeoutError:
+        return "timeout"
+    except ConnectionRefusedError:
+        return "refused"
+    except OSError:
+        return "unreachable"
+
+
+async def classify_failure(host: str, tcp_port: int, ftp_port: Optional[int] = 21) -> dict:
+    """Classify a poll failure as wedged / offline / ok.
+
+    The wedge signature (from the Feb 2026 investigation): the control port
+    actively REFUSES (RST) while the device is otherwise alive. A refused
+    connection requires a live IP stack at the far end — a powered-off device
+    behind the modem produces a timeout instead. The FTP probe upgrades
+    confidence when it accepts, but refusal of the control port alone is
+    already a strong wedge indicator.
+
+    Returns dict:
+        state       — "ok" | "wedged" | "offline"
+        confidence  — "high" | "medium" (only meaningful for "wedged")
+        tcp_probe   — raw probe result for the control port
+        ftp_probe   — raw probe result for the FTP port (None if not probed)
+    """
+    tcp_result = await probe_tcp(host, tcp_port)
+
+    result = {
+        "state": "ok",
+        "confidence": "high",
+        "tcp_probe": tcp_result,
+        "ftp_probe": None,
+    }
+
+    if tcp_result == "open":
+        # Transient failure — the listener is there now
+        return result
+
+    if tcp_result in ("timeout", "unreachable"):
+        result["state"] = "offline"
+        return result
+
+    # tcp_result == "refused": host alive, control listener gone
+    result["state"] = "wedged"
+    if ftp_port:
+        ftp_result = await probe_tcp(host, ftp_port)
+        result["ftp_probe"] = ftp_result
+        # FTP accepting while control refuses is the definitive signature
+        result["confidence"] = "high" if ftp_result == "open" else "medium"
+    else:
+        result["confidence"] = "medium"
+
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Reset backends
+# ---------------------------------------------------------------------------
+
+class ResetBackend:
+    """Base class for reset actuation backends.
+
+    Backends receive a plain dict of snapshotted config values (never a live
+    ORM row — recovery runs as a detached task after the originating DB
+    session has closed).
+    """
+
+    name = "base"
+
+    async def trigger(self, params: dict) -> dict:
+        """Attempt to power-cycle the device.
+
+        Returns dict: {"triggered": bool, "detail": str}
+        "triggered" True means a reset is expected to be in progress.
+        """
+        raise NotImplementedError
+
+
+class ManualNotifyBackend(ResetBackend):
+    """No actuation — flags the wedge for a human to power-cycle the device."""
+
+    name = "manual"
+
+    async def trigger(self, params: dict) -> dict:
+        return {
+            "triggered": False,
+            "detail": "No reset hardware configured - manual power cycle required",
+        }
+
+
+class WebhookBackend(ResetBackend):
+    """Calls an HTTP endpoint that power-cycles the device.
+
+    The endpoint is expected to perform the full cycle itself (e.g. a Shelly
+    relay with `?turn=off&timer=10`, a Tasmota PulseTime rule, or a small
+    HTTP service on a Pi driving a relay GPIO). Any 2xx response counts as
+    triggered.
+    """
+
+    name = "webhook"
+
+    async def trigger(self, params: dict) -> dict:
+        url = params.get("webhook_url")
+        if not url:
+            return {"triggered": False, "detail": "reset_backend is 'webhook' but reset_webhook_url is not set"}
+
+        method = (params.get("webhook_method") or "POST").upper()
+
+        def _call() -> dict:
+            req = urllib.request.Request(url, method=method)
+            req.add_header("User-Agent", "SLMM-wedge-recovery")
+            with urllib.request.urlopen(req, timeout=15) as resp:
+                body = resp.read(500).decode(errors="ignore")
+                return {"status": resp.status, "body": body}
+
+        try:
+            resp = await asyncio.to_thread(_call)
+            if 200 <= resp["status"] < 300:
+                return {"triggered": True, "detail": f"Webhook {method} {url} -> {resp['status']}"}
+            return {"triggered": False, "detail": f"Webhook {method} {url} -> HTTP {resp['status']}"}
+        except Exception as e:
+            return {"triggered": False, "detail": f"Webhook {method} {url} failed: {e}"}
+
+
+BACKENDS: Dict[str, ResetBackend] = {
+    "manual": ManualNotifyBackend(),
+    "webhook": WebhookBackend(),
+}
+
+
+def get_backend(name: Optional[str]) -> ResetBackend:
+    return BACKENDS.get((name or "manual").lower(), BACKENDS["manual"])
+
+
+# ---------------------------------------------------------------------------
+# Recovery manager
+# ---------------------------------------------------------------------------
+
+class RecoveryManager:
+    """Orchestrates wedge recovery, one concurrent recovery per device."""
+
+    def __init__(self):
+        self._active: Dict[str, asyncio.Task] = {}
+        # In-memory attempt timestamps per unit for windowed rate limiting.
+        # Resets on service restart; last_recovery_at in the DB provides a
+        # coarse cross-restart guard via the same window check.
+        self._attempts: Dict[str, List[float]] = {}
+
+    # -- public API ---------------------------------------------------------
+
+    def is_active(self, unit_id: str) -> bool:
+        task = self._active.get(unit_id)
+        return task is not None and not task.done()
+
+    def get_status(self, unit_id: str) -> dict:
+        now = time.time()
+        attempts = self._attempts.get(unit_id, [])
+        return {
+            "recovery_in_progress": self.is_active(unit_id),
+            "attempts_this_session": len(attempts),
+            "last_attempt_seconds_ago": round(now - attempts[-1]) if attempts else None,
+        }
+
+    def maybe_start_recovery(self, cfg: NL43Config, status: NL43Status, classification: dict, db: Session) -> bool:
+        """Start a recovery task if conditions are met. Called from the poller.
+
+        Returns True if a recovery task was started.
+        """
+        unit_id = cfg.unit_id
+
+        if not cfg.auto_recovery_enabled:
+            logger.info(f"[RECOVERY] {unit_id}: wedge detected but auto_recovery_enabled is off")
+            return False
+
+        if self.is_active(unit_id):
+            logger.debug(f"[RECOVERY] {unit_id}: recovery already in progress")
+            return False
+
+        if status.consecutive_failures < WEDGE_CONFIRM_FAILURES:
+            logger.info(
+                f"[RECOVERY] {unit_id}: wedge suspected "
+                f"({status.consecutive_failures}/{WEDGE_CONFIRM_FAILURES} confirmations)"
+            )
+            return False
+
+        if not self._attempt_allowed(cfg, status):
+            logger.warning(
+                f"[RECOVERY] {unit_id}: attempt limit reached "
+                f"({cfg.recovery_max_attempts} per {cfg.recovery_window_minutes} min) - standing down"
+            )
+            log_device_event(
+                unit_id, "ERROR", "RECOVERY",
+                f"Wedge persists but recovery attempt limit reached "
+                f"({cfg.recovery_max_attempts} per {cfg.recovery_window_minutes} min). Manual intervention required.",
+                db,
+            )
+            return False
+
+        return self.start_recovery(cfg, classification)
+
+    def start_recovery(self, cfg: NL43Config, classification: Optional[dict] = None) -> bool:
+        """Launch the recovery task (also used by the manual-trigger endpoint)."""
+        unit_id = cfg.unit_id
+        if self.is_active(unit_id):
+            return False
+
+        self._attempts.setdefault(unit_id, []).append(time.time())
+
+        # Snapshot config values so the task doesn't depend on a live DB row
+        params = {
+            "unit_id": unit_id,
+            "host": cfg.host,
+            "tcp_port": cfg.tcp_port,
+            "ftp_port": cfg.ftp_port or 21,
+            "ftp_enabled": bool(cfg.ftp_enabled),
+            "ftp_username": cfg.ftp_username,
+            "ftp_password": cfg.ftp_password,
+            "backend": (cfg.reset_backend or "manual").lower(),
+            "webhook_url": cfg.reset_webhook_url,
+            "webhook_method": cfg.reset_webhook_method,
+            "boot_wait": cfg.recovery_boot_wait_seconds or 90,
+            "reconnect_timeout": cfg.recovery_reconnect_timeout or 180,
+            "auto_resume": bool(cfg.recovery_auto_resume),
+            "window_minutes": cfg.recovery_window_minutes or 360,
+            "classification": classification or {},
+        }
+
+        task = asyncio.create_task(self._run_recovery(params))
+        self._active[unit_id] = task
+        logger.info(f"[RECOVERY] {unit_id}: recovery task started (backend={params['backend']})")
+        return True
+
+    # -- internals ----------------------------------------------------------
+
+    def _attempt_allowed(self, cfg: NL43Config, status: NL43Status) -> bool:
+        """Windowed attempt limit: max N attempts per window."""
+        window_seconds = (cfg.recovery_window_minutes or 360) * 60
+        max_attempts = cfg.recovery_max_attempts or 2
+        now = time.time()
+
+        recent = [t for t in self._attempts.get(cfg.unit_id, []) if now - t < window_seconds]
+        self._attempts[cfg.unit_id] = recent
+        if len(recent) >= max_attempts:
+            return False
+
+        # Cross-restart guard: if the in-memory history is empty (service
+        # restarted) but the DB shows a very recent attempt, respect a
+        # cooldown of one window-fraction to avoid rapid-fire cycling.
+        if not recent and status.last_recovery_at is not None:
+            elapsed = (datetime.utcnow() - status.last_recovery_at).total_seconds()
+            if elapsed < window_seconds / max_attempts / 2:
+                return False
+
+        return True
+
+    def _set_state(self, unit_id: str, recovery_state: str, result: Optional[str] = None,
+                   connection_state: Optional[str] = None, mark_recovered: bool = False):
+        """Persist recovery progress on the status row (own short-lived session)."""
+        db = SessionLocal()
+        try:
+            status = db.query(NL43Status).filter_by(unit_id=unit_id).first()
+            if not status:
+                status = NL43Status(unit_id=unit_id)
+                db.add(status)
+            status.recovery_state = recovery_state
+            if result is not None:
+                status.last_recovery_result = result[:1000]
+            if connection_state is not None:
+                status.connection_state = connection_state
+            if mark_recovered:
+                status.is_reachable = True
+                status.consecutive_failures = 0
+            db.commit()
+        except Exception as e:
+            logger.warning(f"[RECOVERY] {unit_id}: failed to persist state '{recovery_state}': {e}")
+        finally:
+            db.close()
+
+    async def _run_recovery(self, p: dict):
+        """The recovery state machine. Runs as a detached task."""
+        unit_id = p["unit_id"]
+        backend = get_backend(p["backend"])
+        started = time.time()
+
+        def _log(level: str, message: str):
+            log_device_event(unit_id, level, "RECOVERY", message)
+
+        try:
+            # Record attempt start
+            db = SessionLocal()
+            try:
+                status = db.query(NL43Status).filter_by(unit_id=unit_id).first()
+                want_resume = bool(status and status.measurement_state == "Start")
+                if status:
+                    status.last_recovery_at = datetime.utcnow()
+                    status.recovery_state = "resetting"
+                    db.commit()
+            finally:
+                db.close()
+
+            cls = p.get("classification") or {}
+            _log("ERROR",
+                 f"Wedge recovery started (backend={backend.name}, "
+                 f"tcp_probe={cls.get('tcp_probe', '?')}, ftp_probe={cls.get('ftp_probe', '?')}, "
+                 f"was_measuring={want_resume})")
+
+            # --- Step 1: trigger reset ---------------------------------------
+            trigger_result = await backend.trigger(p)
+            _log("INFO", f"Reset trigger: {trigger_result['detail']}")
+
+            if trigger_result["triggered"]:
+                # --- Step 2: wait for the device to boot ---------------------
+                self._set_state(unit_id, "waiting_boot")
+                _log("INFO", f"Reset triggered - waiting {p['boot_wait']}s for device boot")
+                await asyncio.sleep(p["boot_wait"])
+                reconnect_deadline = time.time() + p["reconnect_timeout"]
+                probe_interval = RECONNECT_PROBE_INTERVAL
+            else:
+                if backend.name == "manual":
+                    # Watch for a human to power-cycle the device
+                    self._set_state(unit_id, "awaiting_manual_reset")
+                    _log("ERROR",
+                         f"WEDGE DETECTED on {p['host']}:{p['tcp_port']} - "
+                         f"manual power cycle required. Watching for device return "
+                         f"(up to {p['window_minutes']} min).")
+                    reconnect_deadline = time.time() + p["window_minutes"] * 60
+                    probe_interval = MANUAL_WATCH_PROBE_INTERVAL
+                else:
+                    raise RuntimeError(f"Reset trigger failed: {trigger_result['detail']}")
+
+            # --- Step 3: wait for the control port to come back --------------
+            self._set_state(unit_id, "reconnecting")
+            port_open = False
+            while time.time() < reconnect_deadline:
+                probe = await probe_tcp(p["host"], p["tcp_port"])
+                if probe == "open":
+                    port_open = True
+                    break
+                await asyncio.sleep(probe_interval)
+
+            if not port_open:
+                raise RuntimeError(
+                    f"Device did not return within "
+                    f"{round(reconnect_deadline - started)}s of recovery start"
+                )
+
+            _log("INFO", f"Control port {p['tcp_port']} is accepting connections again")
+
+            # --- Step 4: flush stale pooled connection, verify protocol ------
+            from app.services import NL43Client, _connection_pool
+
+            device_key = f"{p['host']}:{p['tcp_port']}"
+            await _connection_pool.discard(device_key)
+
+            client = NL43Client(
+                p["host"], p["tcp_port"], timeout=10.0,
+                ftp_username=p["ftp_username"], ftp_password=p["ftp_password"],
+                ftp_port=p["ftp_port"],
+            )
+            state = await client.get_measurement_state()
+            _log("INFO", f"Device responding to commands - measurement state: {state}")
+
+            # --- Step 5: resume measurement if needed -------------------------
+            resumed = False
+            if want_resume and p["auto_resume"] and state != "Start":
+                self._set_state(unit_id, "resuming")
+                _log("INFO", "Measurement was running before the wedge - executing start cycle")
+                cycle = await client.start_cycle(sync_clock=True)
+                resumed = True
+                _log("INFO",
+                     f"Measurement restarted (index {cycle.get('old_index')} -> {cycle.get('new_index')})")
+            elif want_resume and state == "Start":
+                _log("INFO", "Device is already measuring after reset - no resume needed")
+                resumed = True
+
+            # --- Done ---------------------------------------------------------
+            elapsed = round(time.time() - started)
+            summary = (
+                f"Recovered in {elapsed}s via {backend.name}"
+                + (", measurement resumed" if resumed else
+                   (", measurement NOT resumed" if want_resume else ""))
+            )
+            self._set_state(unit_id, "idle", result=summary, connection_state="ok", mark_recovered=True)
+            _log("INFO", f"Wedge recovery SUCCEEDED: {summary}")
+
+        except Exception as e:
+            elapsed = round(time.time() - started)
+            summary = f"Recovery FAILED after {elapsed}s: {e}"
+            self._set_state(unit_id, "failed", result=summary)
+            _log("ERROR", summary)
+            logger.error(f"[RECOVERY] {unit_id}: {summary}")
+
+        finally:
+            self._active.pop(unit_id, None)
+
+
+# Global singleton
+recovery_manager = RecoveryManager()

app/routers.py

@@ -13,6 +13,7 @@ import asyncio
 from app.database import get_db
 from app.models import NL43Config, NL43Status
 from app.services import NL43Client, persist_snapshot
+from app.recovery import classify_failure, recovery_manager
 
 logger = logging.getLogger(__name__)
 
@@ -53,6 +54,38 @@ class ConfigPayload(BaseModel):
     poll_enabled: bool | None = None
     poll_interval_seconds: int | None = None
 
+    # Wedge recovery configuration
+    auto_recovery_enabled: bool | None = None
+    reset_backend: str | None = None
+    reset_webhook_url: str | None = None
+    reset_webhook_method: str | None = None
+    recovery_boot_wait_seconds: int | None = Field(None, ge=10, le=600)
+    recovery_reconnect_timeout: int | None = Field(None, ge=30, le=3600)
+    recovery_auto_resume: bool | None = None
+    recovery_max_attempts: int | None = Field(None, ge=1, le=10)
+    recovery_window_minutes: int | None = Field(None, ge=30, le=1440)
+
+    @field_validator("reset_backend")
+    @classmethod
+    def validate_reset_backend(cls, v):
+        if v is not None and v.lower() not in ("manual", "webhook"):
+            raise ValueError("reset_backend must be 'manual' or 'webhook'")
+        return v.lower() if v else v
+
+    @field_validator("reset_webhook_method")
+    @classmethod
+    def validate_reset_webhook_method(cls, v):
+        if v is not None and v.upper() not in ("GET", "POST"):
+            raise ValueError("reset_webhook_method must be 'GET' or 'POST'")
+        return v.upper() if v else v
+
+    @field_validator("reset_webhook_url")
+    @classmethod
+    def validate_reset_webhook_url(cls, v):
+        if v is not None and v != "" and not v.lower().startswith(("http://", "https://")):
+            raise ValueError("reset_webhook_url must start with http:// or https://")
+        return v
+
     @field_validator("host")
     @classmethod
     def validate_host(cls, v):
@@ -348,6 +381,15 @@ def get_config(unit_id: str, db: Session = Depends(get_db)):
             "ftp_username": cfg.ftp_username,
             "ftp_password": cfg.ftp_password,
             "web_enabled": cfg.web_enabled,
+            "auto_recovery_enabled": cfg.auto_recovery_enabled,
+            "reset_backend": cfg.reset_backend,
+            "reset_webhook_url": cfg.reset_webhook_url,
+            "reset_webhook_method": cfg.reset_webhook_method,
+            "recovery_boot_wait_seconds": cfg.recovery_boot_wait_seconds,
+            "recovery_reconnect_timeout": cfg.recovery_reconnect_timeout,
+            "recovery_auto_resume": cfg.recovery_auto_resume,
+            "recovery_max_attempts": cfg.recovery_max_attempts,
+            "recovery_window_minutes": cfg.recovery_window_minutes,
         },
     }
 
@@ -407,6 +449,24 @@ async def upsert_config(unit_id: str, payload: ConfigPayload, db: Session = Depe
         cfg.poll_enabled = payload.poll_enabled
     if payload.poll_interval_seconds is not None:
         cfg.poll_interval_seconds = payload.poll_interval_seconds
+    if payload.auto_recovery_enabled is not None:
+        cfg.auto_recovery_enabled = payload.auto_recovery_enabled
+    if payload.reset_backend is not None:
+        cfg.reset_backend = payload.reset_backend
+    if payload.reset_webhook_url is not None:
+        cfg.reset_webhook_url = payload.reset_webhook_url or None
+    if payload.reset_webhook_method is not None:
+        cfg.reset_webhook_method = payload.reset_webhook_method
+    if payload.recovery_boot_wait_seconds is not None:
+        cfg.recovery_boot_wait_seconds = payload.recovery_boot_wait_seconds
+    if payload.recovery_reconnect_timeout is not None:
+        cfg.recovery_reconnect_timeout = payload.recovery_reconnect_timeout
+    if payload.recovery_auto_resume is not None:
+        cfg.recovery_auto_resume = payload.recovery_auto_resume
+    if payload.recovery_max_attempts is not None:
+        cfg.recovery_max_attempts = payload.recovery_max_attempts
+    if payload.recovery_window_minutes is not None:
+        cfg.recovery_window_minutes = payload.recovery_window_minutes
 
     db.commit()
     db.refresh(cfg)
@@ -803,6 +863,89 @@ async def get_measurement_state(unit_id: str, db: Session = Depends(get_db)):
         raise HTTPException(status_code=502, detail=str(e))
 
 
+# ============================================================================
+# WEDGE RECOVERY ENDPOINTS
+# ============================================================================
+
+@router.get("/{unit_id}/recovery/status")
+def get_recovery_status(unit_id: str, db: Session = Depends(get_db)):
+    """Get wedge detection and recovery status for a device."""
+    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
+    if not cfg:
+        raise HTTPException(status_code=404, detail="NL43 config not found")
+
+    status = db.query(NL43Status).filter_by(unit_id=unit_id).first()
+    manager_status = recovery_manager.get_status(unit_id)
+
+    return {
+        "status": "ok",
+        "data": {
+            "unit_id": unit_id,
+            "connection_state": status.connection_state if status else "unknown",
+            "recovery_state": status.recovery_state if status else "idle",
+            "last_wedge_at": status.last_wedge_at.isoformat() + "Z" if status and status.last_wedge_at else None,
+            "wedge_count": status.wedge_count if status else 0,
+            "last_recovery_at": status.last_recovery_at.isoformat() + "Z" if status and status.last_recovery_at else None,
+            "last_recovery_result": status.last_recovery_result if status else None,
+            "auto_recovery_enabled": cfg.auto_recovery_enabled,
+            "reset_backend": cfg.reset_backend,
+            **manager_status,
+        },
+    }
+
+
+@router.post("/{unit_id}/recovery/probe")
+async def probe_device(unit_id: str, db: Session = Depends(get_db)):
+    """Probe the device's control and FTP ports and classify its state.
+
+    Does not trigger recovery — diagnostic only. Useful for confirming a
+    wedge before manually triggering a reset.
+    """
+    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
+    if not cfg:
+        raise HTTPException(status_code=404, detail="NL43 config not found")
+
+    classification = await classify_failure(
+        cfg.host,
+        cfg.tcp_port,
+        ftp_port=(cfg.ftp_port or 21) if cfg.ftp_enabled else None,
+    )
+    return {"status": "ok", "unit_id": unit_id, "data": classification}
+
+
+@router.post("/{unit_id}/recovery/trigger")
+async def trigger_recovery(unit_id: str, db: Session = Depends(get_db)):
+    """Manually trigger the recovery flow for a device.
+
+    Bypasses the consecutive-failure confirmation threshold but still
+    respects the one-recovery-at-a-time guard. Use after confirming a wedge
+    via /recovery/probe, or to test the reset hardware end-to-end.
+    """
+    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
+    if not cfg:
+        raise HTTPException(status_code=404, detail="NL43 config not found")
+
+    if recovery_manager.is_active(unit_id):
+        raise HTTPException(status_code=409, detail="Recovery already in progress for this device")
+
+    classification = await classify_failure(
+        cfg.host,
+        cfg.tcp_port,
+        ftp_port=(cfg.ftp_port or 21) if cfg.ftp_enabled else None,
+    )
+
+    started = recovery_manager.start_recovery(cfg, classification)
+    if not started:
+        raise HTTPException(status_code=409, detail="Could not start recovery (already in progress)")
+
+    return {
+        "status": "ok",
+        "unit_id": unit_id,
+        "message": f"Recovery started (backend={cfg.reset_backend or 'manual'})",
+        "classification": classification,
+    }
+
+
 @router.post("/{unit_id}/sleep")
 async def sleep_device(unit_id: str, db: Session = Depends(get_db)):
     """Put the device into sleep mode for battery conservation."""

migrate_add_recovery_fields.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""
+Migration script to add wedge-recovery fields to nl43_config and nl43_status tables.
+
+Adds to nl43_config:
+- auto_recovery_enabled (BOOLEAN, default 0/False)
+- reset_backend (TEXT, default 'manual')
+- reset_webhook_url (TEXT, nullable)
+- reset_webhook_method (TEXT, default 'POST')
+- recovery_boot_wait_seconds (INTEGER, default 90)
+- recovery_reconnect_timeout (INTEGER, default 180)
+- recovery_auto_resume (BOOLEAN, default 1/True)
+- recovery_max_attempts (INTEGER, default 2)
+- recovery_window_minutes (INTEGER, default 360)
+
+Adds to nl43_status:
+- connection_state (TEXT, default 'ok')
+- last_wedge_at (DATETIME, nullable)
+- wedge_count (INTEGER, default 0)
+- recovery_state (TEXT, default 'idle')
+- last_recovery_at (DATETIME, nullable)
+- last_recovery_result (TEXT, nullable)
+
+Usage:
+    python migrate_add_recovery_fields.py
+"""
+
+import sqlite3
+import sys
+from pathlib import Path
+
+CONFIG_COLUMNS = [
+    ("auto_recovery_enabled", "BOOLEAN DEFAULT 0"),
+    ("reset_backend", "TEXT DEFAULT 'manual'"),
+    ("reset_webhook_url", "TEXT"),
+    ("reset_webhook_method", "TEXT DEFAULT 'POST'"),
+    ("recovery_boot_wait_seconds", "INTEGER DEFAULT 90"),
+    ("recovery_reconnect_timeout", "INTEGER DEFAULT 180"),
+    ("recovery_auto_resume", "BOOLEAN DEFAULT 1"),
+    ("recovery_max_attempts", "INTEGER DEFAULT 2"),
+    ("recovery_window_minutes", "INTEGER DEFAULT 360"),
+]
+
+STATUS_COLUMNS = [
+    ("connection_state", "TEXT DEFAULT 'ok'"),
+    ("last_wedge_at", "DATETIME"),
+    ("wedge_count", "INTEGER DEFAULT 0"),
+    ("recovery_state", "TEXT DEFAULT 'idle'"),
+    ("last_recovery_at", "DATETIME"),
+    ("last_recovery_result", "TEXT"),
+]
+
+
+def migrate():
+    db_path = Path("data/slmm.db")
+
+    if not db_path.exists():
+        print(f"❌ Database not found at {db_path}")
+        print("   Run this script from the slmm directory")
+        return False
+
+    try:
+        conn = sqlite3.connect(db_path)
+        cursor = conn.cursor()
+
+        cursor.execute("PRAGMA table_info(nl43_config)")
+        config_columns = [row[1] for row in cursor.fetchall()]
+
+        cursor.execute("PRAGMA table_info(nl43_status)")
+        status_columns = [row[1] for row in cursor.fetchall()]
+
+        changes_made = False
+
+        for name, ddl in CONFIG_COLUMNS:
+            if name not in config_columns:
+                print(f"Adding {name} to nl43_config...")
+                cursor.execute(f"ALTER TABLE nl43_config ADD COLUMN {name} {ddl}")
+                changes_made = True
+            else:
+                print(f"✓ {name} already exists in nl43_config")
+
+        for name, ddl in STATUS_COLUMNS:
+            if name not in status_columns:
+                print(f"Adding {name} to nl43_status...")
+                cursor.execute(f"ALTER TABLE nl43_status ADD COLUMN {name} {ddl}")
+                changes_made = True
+            else:
+                print(f"✓ {name} already exists in nl43_status")
+
+        if changes_made:
+            conn.commit()
+            print("\n✓ Migration completed successfully")
+            print("  Added wedge-recovery fields to nl43_config and nl43_status")
+        else:
+            print("\n✓ All recovery fields already exist - no changes needed")
+
+        conn.close()
+        return True
+
+    except Exception as e:
+        print(f"❌ Migration failed: {e}")
+        return False
+
+
+if __name__ == "__main__":
+    success = migrate()
+    sys.exit(0 if success else 1)

nl52/__init__.py

@@ -1,23 +0,0 @@
-"""
-RION NL-42 / NL-52 serial control layer.
-
-The NL-52 has no Ethernet option (unlike the NL-43's NX-43EX LAN card); it
-speaks the same ASCII command family over RS-232C or USB (virtual COM port).
-This package provides a dependency-free serial client + parser so the meter
-can be driven directly over USB on the bench, and later bridged onto the
-network via the RX55's serial PAD mode (or a ser2net host).
-
-Modules:
-    protocol  — pure command/response parsing (no I/O, unit-testable)
-    client    — termios-based serial transport + NL52Client command methods
-    cli       — command-line tool for bench testing over USB
-"""
-
-from nl52.protocol import (  # noqa: F401
-    DODSnapshot,
-    DRDSample,
-    ResultError,
-    parse_dod,
-    parse_drd,
-    RESULT_CODES,
-)

nl52/cli.py

@@ -1,160 +0,0 @@
-#!/usr/bin/env python3
-"""
-NL-42 / NL-52 bench CLI — talk to the meter over USB (or RS-232C), no deps.
-
-Setup on the meter first:
-    MENU -> I/O -> Communication Interface -> "USB"   (set BEFORE plugging in)
-    Connect a generic USB-A -> mini-B cable directly (no hub).
-    Find the port:  ls -l /dev/ttyUSB*   (and `dmesg | tail` after plugging in)
-
-Examples:
-    python3 -m nl52.cli probe
-    python3 -m nl52.cli --port /dev/ttyUSB0 status
-    python3 -m nl52.cli start
-    python3 -m nl52.cli stop
-    python3 -m nl52.cli monitor --seconds 10      # DRD stream (needs NX-42EX)
-    python3 -m nl52.cli poll --seconds 10         # DOD polling fallback (~1/s)
-    python3 -m nl52.cli raw "System Version?NL"
-
-Run from the slmm/ directory (so the nl52 package is importable), or just run
-this file directly — it adds slmm/ to sys.path automatically.
-"""
-
-import argparse
-import os
-import sys
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from nl52.client import NL52Client  # noqa: E402
-from nl52.protocol import DODSnapshot, DRDSample, ResultError  # noqa: E402
-
-
-def _fmt(v):
-    return "--" if v is None else (f"{v:.1f}" if isinstance(v, float) else str(v))
-
-
-def _print_dod(s: DODSnapshot):
-    print(f"  Lp   = {_fmt(s.lp)} dB     Leq  = {_fmt(s.leq)} dB    LE = {_fmt(s.le)} dB")
-    print(f"  Lmax = {_fmt(s.lmax)} dB     Lmin = {_fmt(s.lmin)} dB")
-    print(f"  Ly   = {_fmt(s.ly)}        sub Lp = {_fmt(s.sub_lp)} dB")
-    print(f"  LN1..5 = {_fmt(s.ln1)} / {_fmt(s.ln2)} / {_fmt(s.ln3)} / {_fmt(s.ln4)} / {_fmt(s.ln5)}")
-    print(f"  overload={_fmt(s.overload)}  underrange={_fmt(s.underrange)}")
-
-
-def cmd_probe(m: NL52Client, args):
-    print("[*] Probing meter (read-only)...")
-    print(f"  System Version : {m.system_version()}")
-    print(f"  Clock          : {m.get_clock()}")
-    print(f"  Measure state  : {m.measure_state()}")
-    try:
-        print(f"  Battery type   : {m.battery_type()}")
-    except Exception as e:
-        print(f"  Battery type   : (n/a: {e})")
-    try:
-        print(f"  SD free (MB?)  : {m.sd_free_size()}")
-    except Exception as e:
-        print(f"  SD free        : (n/a: {e})")
-    print("[OK] Two-way communication confirmed.")
-
-
-def cmd_status(m: NL52Client, args):
-    state = m.measure_state()
-    print(f"[*] Measure state: {state}")
-    print("[*] DOD snapshot:")
-    _print_dod(m.request_dod())
-
-
-def cmd_start(m: NL52Client, args):
-    print(f"[*] Measure,Start -> {m.measure_start()}")
-    print(f"  state now: {m.measure_state()}")
-
-
-def cmd_stop(m: NL52Client, args):
-    print(f"[*] Measure,Stop -> {m.measure_stop()}")
-    print(f"  state now: {m.measure_state()}")
-
-
-def cmd_monitor(m: NL52Client, args):
-    print(f"[*] DRD stream for {args.seconds}s (requires NX-42EX). Ctrl+C to stop.")
-
-    def on_sample(s: DRDSample):
-        print(f"  #{_fmt(s.counter):>4}  Lp={_fmt(s.lp)}  Leq={_fmt(s.leq)}  "
-              f"Lmax={_fmt(s.lmax)}  Lmin={_fmt(s.lmin)}  sub={_fmt(s.sub_lp)}"
-              f"{'  OVERLOAD' if s.overload else ''}{'  UNDER' if s.underrange else ''}")
-
-    try:
-        m.stream_drd(on_sample, duration=args.seconds)
-    except ResultError as e:
-        print(f"[!] DRD not available ({e}). The meter likely lacks the NX-42EX "
-              f"option — use `poll` instead.")
-
-
-def cmd_poll(m: NL52Client, args):
-    import time
-    print(f"[*] Polling DOD ~1/s for {args.seconds}s. Ctrl+C to stop.")
-    end = time.time() + args.seconds
-    while time.time() < end:
-        s = m.request_dod()
-        print(f"  Lp={_fmt(s.lp)}  Leq={_fmt(s.leq)}  Lmax={_fmt(s.lmax)}  "
-              f"Lmin={_fmt(s.lmin)}  sub={_fmt(s.sub_lp)}"
-              f"{'  OVERLOAD' if s.overload else ''}{'  UNDER' if s.underrange else ''}")
-
-
-def cmd_raw(m: NL52Client, args):
-    resp = m.send(args.command)
-    print(f"  > {args.command}")
-    print(f"  < {resp}")
-
-
-def main():
-    ap = argparse.ArgumentParser(description="RION NL-42/NL-52 bench CLI")
-    ap.add_argument("--port", default="/dev/ttyUSB0")
-    ap.add_argument("--baud", type=int, default=115200)
-    ap.add_argument("--timeout", type=float, default=3.0)
-    ap.add_argument("--echo-on", action="store_true",
-                    help="Don't send Echo,Off on connect")
-
-    sub = ap.add_subparsers(dest="cmd", required=True)
-    sub.add_parser("probe", help="read-only sanity check")
-    sub.add_parser("status", help="measure state + DOD snapshot")
-    sub.add_parser("start", help="Measure,Start")
-    sub.add_parser("stop", help="Measure,Stop")
-    mon = sub.add_parser("monitor", help="DRD stream (needs NX-42EX)")
-    mon.add_argument("--seconds", type=float, default=10)
-    pol = sub.add_parser("poll", help="DOD polling fallback (~1/s)")
-    pol.add_argument("--seconds", type=float, default=10)
-    raw = sub.add_parser("raw", help="send an arbitrary command")
-    raw.add_argument("command")
-
-    args = ap.parse_args()
-
-    if not os.path.exists(args.port):
-        print(f"[!] {args.port} not found. Plug in the meter (Comm Interface=USB) and check:")
-        print("    ls -l /dev/ttyUSB*   ;   dmesg | tail -20")
-        return 2
-
-    handlers = {
-        "probe": cmd_probe, "status": cmd_status, "start": cmd_start,
-        "stop": cmd_stop, "monitor": cmd_monitor, "poll": cmd_poll, "raw": cmd_raw,
-    }
-
-    try:
-        with NL52Client(args.port, baud=args.baud, timeout=args.timeout,
-                        disable_echo=not args.echo_on) as m:
-            handlers[args.cmd](m, args)
-    except PermissionError:
-        print(f"[!] Permission denied on {args.port}. Add yourself to dialout:")
-        print("    sudo usermod -aG dialout $USER   (then log out/in)")
-        return 2
-    except KeyboardInterrupt:
-        print("\n[*] Interrupted.")
-    except (TimeoutError, ResultError) as e:
-        print(f"[!] {type(e).__name__}: {e}")
-        print("    Check: Comm Interface=USB, ECO/Sleep OFF, correct port/baud.")
-        return 1
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

nl52/client.py

@@ -1,288 +0,0 @@
-"""
-NL-42 / NL-52 serial client — dependency-free (stdlib termios).
-
-Opens the meter's virtual COM port (USB) or RS-232C port and exchanges
-ASCII commands. No pyserial required, so it runs on the dev server with
-nothing to install.
-
-Usage:
-    from nl52.client import NL52Client
-    with NL52Client("/dev/ttyUSB0") as m:
-        print(m.system_version())
-        print(m.measure_state())
-        snap = m.request_dod()
-        print(snap.lp, snap.leq)
-
-Integration note: this client is synchronous for a robust bench tool. The
-parsing in nl52.protocol is I/O-free and reused as-is when this is later
-wrapped for SLMM (async transport, or a serial->TCP bridge behind the
-existing NL43-style TCP client).
-"""
-
-import os
-import select
-import termios
-import time
-from typing import Callable, List, Optional
-
-from nl52.protocol import (
-    CRLF,
-    SUB,
-    DODSnapshot,
-    DRDSample,
-    ResultError,
-    is_result_code,
-    parse_dod,
-    parse_drd,
-)
-
-_BAUD = {
-    9600: termios.B9600,
-    19200: termios.B19200,
-    38400: termios.B38400,
-    57600: termios.B57600,
-    115200: termios.B115200,
-}
-
-# Inter-command spacing (Serial Interface Manual "Rated Values"):
-#   - wait >=200 ms after a reply before the next command
-#   - wait >=1 s between DOD? requests
-MIN_GAP_DEFAULT = 0.2
-MIN_GAP_DOD = 1.0
-
-
-class SerialPort:
-    """Minimal raw 8N1 serial port over a tty fd (no flow control)."""
-
-    def __init__(self, device: str, baud: int = 115200):
-        if baud not in _BAUD:
-            raise ValueError(f"Unsupported baud {baud}; choose {sorted(_BAUD)}")
-        self.device = device
-        self.baud = baud
-        self.fd: Optional[int] = None
-        self._buf = bytearray()  # holds bytes read past the last returned line
-
-    def open(self):
-        fd = os.open(self.device, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
-        iflag, oflag, cflag, lflag, ispeed, ospeed, cc = termios.tcgetattr(fd)
-        iflag = 0
-        oflag = 0
-        lflag = 0
-        cflag = termios.CS8 | termios.CREAD | termios.CLOCAL
-        ispeed = ospeed = _BAUD[self.baud]
-        termios.tcsetattr(fd, termios.TCSANOW,
-                          [iflag, oflag, cflag, lflag, ispeed, ospeed, cc])
-        termios.tcflush(fd, termios.TCIOFLUSH)
-        self.fd = fd
-
-    def close(self):
-        if self.fd is not None:
-            try:
-                os.close(self.fd)
-            finally:
-                self.fd = None
-
-    def flush_input(self):
-        self._buf.clear()
-        if self.fd is not None:
-            termios.tcflush(self.fd, termios.TCIFLUSH)
-
-    def write(self, data: bytes):
-        assert self.fd is not None
-        os.write(self.fd, data)
-
-    def read_line(self, timeout: float) -> Optional[str]:
-        """Read one CRLF/LF-terminated line. Returns the line without the
-        terminator, or None on timeout with no complete line.
-
-        Bytes received past the newline are retained in self._buf so the next
-        call returns the next line — handles multiple lines arriving in a
-        single read (e.g. result code + data line together)."""
-        assert self.fd is not None
-        deadline = time.time() + timeout
-        while True:
-            if b"\n" in self._buf:
-                line, _, rest = self._buf.partition(b"\n")
-                self._buf = bytearray(rest)
-                return line.decode("ascii", errors="replace").strip()
-            remaining = deadline - time.time()
-            if remaining <= 0:
-                break
-            r, _, _ = select.select([self.fd], [], [], remaining)
-            if not r:
-                break
-            try:
-                chunk = os.read(self.fd, 256)
-            except BlockingIOError:
-                continue
-            if chunk:
-                self._buf.extend(chunk)
-        # Timeout: surface any buffered (un-terminated) bytes as a last resort
-        if self._buf:
-            line = bytes(self._buf)
-            self._buf = bytearray()
-            return line.decode("ascii", errors="replace").strip()
-        return None
-
-
-class NL52Client:
-    def __init__(self, device: str = "/dev/ttyUSB0", baud: int = 115200,
-                 timeout: float = 3.0, disable_echo: bool = True):
-        self.port = SerialPort(device, baud)
-        self.timeout = timeout
-        self.disable_echo = disable_echo
-        self._last_cmd_time = 0.0
-
-    # -- lifecycle ----------------------------------------------------------
-
-    def __enter__(self):
-        self.connect()
-        return self
-
-    def __exit__(self, *exc):
-        self.close()
-
-    def connect(self):
-        self.port.open()
-        if self.disable_echo:
-            # Best-effort: turn echo-back off so replies aren't prefixed with
-            # the command. Ignore failures (some firmware defaults to off).
-            try:
-                self.send("$Echo,Off")
-            except Exception:
-                pass
-
-    def close(self):
-        self.port.close()
-
-    # -- core exchange ------------------------------------------------------
-
-    def _rate_limit(self, command: str):
-        gap = MIN_GAP_DOD if command.strip().upper().startswith("DOD") else MIN_GAP_DEFAULT
-        elapsed = time.time() - self._last_cmd_time
-        if elapsed < gap:
-            time.sleep(gap - elapsed)
-
-    def send(self, command: str) -> str:
-        """Send one command and return its response.
-
-        For request commands (containing '?') returns the data line.
-        For setting commands returns the result code (e.g. 'R+0000').
-        Raises ResultError on R+0001..0004, TimeoutError if no reply.
-
-        Note: NL-52 request commands may carry a parameter *after* the '?'
-        (e.g. 'System Version?NL'), so detection is "contains ?", not
-        "ends with ?". Setting commands start with '$' and never contain '?'.
-        """
-        is_request = "?" in command
-        self._rate_limit(command)
-        self.port.flush_input()
-        self.port.write((command + CRLF).encode("ascii"))
-
-        result_code = self._read_result_code(command)
-
-        try:
-            if result_code != "R+0000":
-                raise ResultError(result_code)
-            if is_request:
-                data = self.port.read_line(self.timeout)
-                if data is None:
-                    raise TimeoutError(f"No data line after {command!r}")
-                return data
-            return result_code
-        finally:
-            self._last_cmd_time = time.time()
-
-    def _read_result_code(self, command: str) -> str:
-        """Read lines until a result code, skipping echo / '$' prompt lines."""
-        sent = command.strip().lstrip("$").strip().lower()
-        deadline = time.time() + self.timeout
-        while time.time() < deadline:
-            line = self.port.read_line(max(0.1, deadline - time.time()))
-            if line is None:
-                continue
-            stripped = line.strip()
-            if not stripped:
-                continue
-            if is_result_code(stripped):
-                return stripped
-            # Skip an echoed command or a bare '$' prompt
-            norm = stripped.lstrip("$").strip().lower()
-            if norm == sent or stripped == "$":
-                continue
-            # Unexpected line — keep looking until timeout
-        raise TimeoutError(f"No result code after {command!r}")
-
-    # -- convenience commands ----------------------------------------------
-
-    def system_version(self, option: str = "NL") -> str:
-        return self.send(f"System Version?{option}")
-
-    def get_clock(self) -> str:
-        return self.send("Clock?")
-
-    def set_clock_now(self):
-        t = time.localtime()
-        # Clock,YYYY/MM/DD HH:MM:SS
-        stamp = time.strftime("%Y/%m/%d %H:%M:%S", t)
-        return self.send(f"$Clock,{stamp}")
-
-    def measure_state(self) -> str:
-        """Returns 'Start' or 'Stop'."""
-        return self.send("Measure?")
-
-    def measure_start(self):
-        return self.send("$Measure,Start")
-
-    def measure_stop(self):
-        return self.send("$Measure,Stop")
-
-    def battery_type(self) -> str:
-        return self.send("Battery Type?")
-
-    def sd_free_size(self) -> str:
-        return self.send("SD Card Free Size?")
-
-    def request_dod(self) -> DODSnapshot:
-        return parse_dod(self.send("DOD?"))
-
-    # -- DRD streaming (requires NX-42EX) -----------------------------------
-
-    def stream_drd(self, on_sample: Callable[[DRDSample], None],
-                   duration: Optional[float] = None,
-                   max_samples: Optional[int] = None):
-        """Start DRD continuous output and call on_sample for each line.
-
-        Stops after `duration` seconds or `max_samples`, then sends SUB to
-        halt the stream. Requires the NX-42EX option on the meter.
-        """
-        self._rate_limit("DRD?")
-        self.port.flush_input()
-        self.port.write(("DRD?" + CRLF).encode("ascii"))
-
-        # First line should be the result code
-        rc = self._read_result_code("DRD?")
-        if rc != "R+0000":
-            self._last_cmd_time = time.time()
-            raise ResultError(rc)
-
-        count = 0
-        deadline = time.time() + duration if duration else None
-        try:
-            while True:
-                if deadline and time.time() >= deadline:
-                    break
-                if max_samples and count >= max_samples:
-                    break
-                line = self.port.read_line(timeout=2.0)
-                if line is None:
-                    continue
-                if is_result_code(line):
-                    continue
-                on_sample(parse_drd(line))
-                count += 1
-        finally:
-            self.port.write(SUB)
-            time.sleep(0.3)
-            self.port.flush_input()
-            self._last_cmd_time = time.time()

nl52/protocol.py

@@ -1,172 +0,0 @@
-"""
-NL-42 / NL-52 serial protocol — pure parsing, no I/O.
-
-Reference: RION NL-42/NL-52 Serial Interface Manual (No. 55779).
-
-Command grammar (shared with the NL-43 family):
-    Setting:  "$" + name + "," + param + CRLF      e.g. "$Measure,Start"
-    Request:  name + "?" + CRLF                     e.g. "DOD?"
-    Reply:    result code "R+0000" + CRLF, then for requests a data line.
-    Stop DRD stream: SUB (0x1A).
-
-This module is intentionally I/O-free so it can be unit-tested without a
-device and reused unchanged behind any transport (serial, or a serial->TCP
-bridge).
-"""
-
-from dataclasses import dataclass, field
-from typing import Dict, List, Optional
-
-CR = "\r"
-LF = "\n"
-CRLF = "\r\n"
-SUB = b"\x1a"  # stop DRD streaming
-
-# Result codes (Serial Interface Manual, "Result code")
-RESULT_CODES: Dict[str, str] = {
-    "R+0000": "Normal end",
-    "R+0001": "Command error (command not recognized)",
-    "R+0002": "Parameter error (bad count/type of parameters)",
-    "R+0003": "Designation error (setting sent to request-only cmd, or vice versa)",
-    "R+0004": "Status error (command not valid in current state)",
-}
-
-# DOD? response field order (d1..d14). Main channel unless noted.
-DOD_FIELDS: List[str] = [
-    "lp", "leq", "le", "lmax", "lmin",
-    "ly",                       # additional processing value (e.g. LCpeak)
-    "ln1", "ln2", "ln3", "ln4", "ln5",
-    "sub_lp",                   # sub channel Lp
-    "overload", "underrange",   # 0/1 flags
-]
-
-# DRD? response field order (d0..d8). Requires NX-42EX.
-DRD_FIELDS: List[str] = [
-    "counter",                  # d0: 1..600
-    "lp", "leq", "lmax", "lmin",
-    "ly",
-    "sub_lp",
-    "overload", "underrange",
-]
-
-# Token the meter returns for a disabled/unavailable display channel.
-_NULL_TOKENS = {"--.-", "-.-", "---.-", ""}
-
-
-class ResultError(Exception):
-    """Raised when the meter returns a non-OK result code (R+0001..0004)."""
-
-    def __init__(self, code: str):
-        self.code = code
-        self.meaning = RESULT_CODES.get(code, "Unknown result code")
-        super().__init__(f"{code}: {self.meaning}")
-
-
-def is_result_code(line: str) -> bool:
-    line = line.strip()
-    return line.startswith("R+") and len(line) == 6 and line[2:].isdigit()
-
-
-def parse_level(token: str) -> Optional[float]:
-    """Parse a space-padded level token into a float, or None if disabled.
-
-    The meter pads to 5 chars and returns '--.-' (leading space) for channels
-    whose display is OFF.
-    """
-    t = token.strip()
-    if t in _NULL_TOKENS or set(t) <= set("-. "):
-        return None
-    try:
-        return float(t)
-    except ValueError:
-        return None
-
-
-def _parse_flag(token: str) -> Optional[bool]:
-    t = token.strip()
-    if t == "1":
-        return True
-    if t == "0":
-        return False
-    return None
-
-
-@dataclass
-class DODSnapshot:
-    """Parsed DOD? snapshot (displayed values). Levels are dB or None if the
-    channel's display is OFF. Measurement state is NOT part of DOD — query
-    Measure? separately."""
-
-    lp: Optional[float] = None
-    leq: Optional[float] = None
-    le: Optional[float] = None
-    lmax: Optional[float] = None
-    lmin: Optional[float] = None
-    ly: Optional[float] = None
-    ln1: Optional[float] = None
-    ln2: Optional[float] = None
-    ln3: Optional[float] = None
-    ln4: Optional[float] = None
-    ln5: Optional[float] = None
-    sub_lp: Optional[float] = None
-    overload: Optional[bool] = None
-    underrange: Optional[bool] = None
-    raw: str = ""
-
-
-@dataclass
-class DRDSample:
-    """Parsed DRD? sample (continuous output, ~every 100 ms)."""
-
-    counter: Optional[int] = None
-    lp: Optional[float] = None
-    leq: Optional[float] = None
-    lmax: Optional[float] = None
-    lmin: Optional[float] = None
-    ly: Optional[float] = None
-    sub_lp: Optional[float] = None
-    overload: Optional[bool] = None
-    underrange: Optional[bool] = None
-    raw: str = ""
-
-
-def _split(resp: str) -> List[str]:
-    return [p for p in resp.strip().split(",")]
-
-
-def parse_dod(resp: str) -> DODSnapshot:
-    """Parse a DOD? data line into a DODSnapshot.
-
-    Tolerant of trailing/short field counts — only maps what is present.
-    """
-    parts = _split(resp)
-    snap = DODSnapshot(raw=resp.strip())
-
-    for idx, name in enumerate(DOD_FIELDS):
-        if idx >= len(parts):
-            break
-        if name in ("overload", "underrange"):
-            setattr(snap, name, _parse_flag(parts[idx]))
-        else:
-            setattr(snap, name, parse_level(parts[idx]))
-
-    return snap
-
-
-def parse_drd(resp: str) -> DRDSample:
-    """Parse a single DRD? data line into a DRDSample."""
-    parts = _split(resp)
-    sample = DRDSample(raw=resp.strip())
-
-    for idx, name in enumerate(DRD_FIELDS):
-        if idx >= len(parts):
-            break
-        if name == "counter":
-            t = parts[idx].strip()
-            sample.counter = int(t) if t.isdigit() else None
-        elif name in ("overload", "underrange"):
-            setattr(sample, name, _parse_flag(parts[idx]))
-        else:
-            setattr(sample, name, parse_level(parts[idx]))
-
-    return sample

nl52/usb_bulk_probe.py

@@ -1,128 +0,0 @@
-#!/usr/bin/env python3
-"""
-RION NL-42/NL-52 USB bulk probe — talks to the meter via libusb directly
-(ctypes, no pyusb/pip needed).
-
-The NL-52's USB serial interface (interface 0) is a plain vendor class with
-two bulk endpoints and NO control protocol — not FTDI/CDC, so no kernel
-serial driver applies. We just claim interface 0 and move ASCII over the
-bulk pipe: write command -> EP 0x03 OUT, read reply <- EP 0x84 IN.
-
-Needs write access to the USB device node (root, or a udev rule granting the
-plugdev group). Run:  sudo python3 nl52/usb_bulk_probe.py
-"""
-
-import ctypes as C
-import time
-
-VID, PID = 0x0EA3, 0x000F
-EP_OUT, EP_IN = 0x03, 0x84
-IFACE = 0
-
-LIBUSB_ERROR_TIMEOUT = -7
-
-lib = C.CDLL("libusb-1.0.so.0")
-lib.libusb_init.argtypes = [C.POINTER(C.c_void_p)]
-lib.libusb_open_device_with_vid_pid.argtypes = [C.c_void_p, C.c_uint16, C.c_uint16]
-lib.libusb_open_device_with_vid_pid.restype = C.c_void_p
-lib.libusb_set_auto_detach_kernel_driver.argtypes = [C.c_void_p, C.c_int]
-lib.libusb_claim_interface.argtypes = [C.c_void_p, C.c_int]
-lib.libusb_release_interface.argtypes = [C.c_void_p, C.c_int]
-lib.libusb_set_interface_alt_setting.argtypes = [C.c_void_p, C.c_int, C.c_int]
-lib.libusb_clear_halt.argtypes = [C.c_void_p, C.c_ubyte]
-lib.libusb_close.argtypes = [C.c_void_p]
-lib.libusb_exit.argtypes = [C.c_void_p]
-lib.libusb_bulk_transfer.argtypes = [
-    C.c_void_p, C.c_ubyte, C.POINTER(C.c_ubyte), C.c_int, C.POINTER(C.c_int), C.c_uint
-]
-lib.libusb_strerror.argtypes = [C.c_int]
-lib.libusb_strerror.restype = C.c_char_p
-
-
-def err(code):
-    return lib.libusb_strerror(code).decode(errors="replace")
-
-
-def main():
-    ctx = C.c_void_p()
-    if lib.libusb_init(C.byref(ctx)) != 0:
-        print("[!] libusb_init failed")
-        return 1
-
-    h = lib.libusb_open_device_with_vid_pid(ctx, VID, PID)
-    if not h:
-        print(f"[!] open {VID:04x}:{PID:04x} failed — device present? running as root?")
-        lib.libusb_exit(ctx)
-        return 1
-
-    lib.libusb_set_auto_detach_kernel_driver(h, 1)
-    rc = lib.libusb_claim_interface(h, IFACE)
-    if rc != 0:
-        print(f"[!] claim_interface({IFACE}) failed: {err(rc)}")
-        lib.libusb_close(h); lib.libusb_exit(ctx)
-        return 1
-
-    print(f"[*] Claimed interface {IFACE} on {VID:04x}:{PID:04x} — bulk OUT 0x{EP_OUT:02x}, IN 0x{EP_IN:02x}")
-
-    # Select alt setting 0 explicitly and clear any endpoint halts left over
-    # from prior (wrong-driver) probing.
-    rc = lib.libusb_set_interface_alt_setting(h, IFACE, 0)
-    print(f"[*] set_interface_alt_setting(0): {err(rc) if rc else 'ok'}")
-    for ep in (EP_OUT, EP_IN):
-        rc = lib.libusb_clear_halt(h, ep)
-        print(f"[*] clear_halt(0x{ep:02x}): {err(rc) if rc else 'ok'}")
-    print()
-
-    def bulk_out(data: bytes, timeout=2000):
-        buf = (C.c_ubyte * len(data)).from_buffer_copy(data)
-        actual = C.c_int(0)
-        rc = lib.libusb_bulk_transfer(h, EP_OUT, buf, len(data), C.byref(actual), timeout)
-        return rc, actual.value
-
-    def bulk_in(n=512, timeout=2000):
-        buf = (C.c_ubyte * n)()
-        actual = C.c_int(0)
-        rc = lib.libusb_bulk_transfer(h, EP_IN, buf, n, C.byref(actual), timeout)
-        return rc, bytes(buf[:actual.value])
-
-    def exchange(cmd: str):
-        rc, n = bulk_out((cmd + "\r\n").encode("ascii"))
-        if rc != 0:
-            print(f"  > {cmd!r:24} OUT failed: {err(rc)}")
-            return
-        # Read until idle (collect result code + data lines)
-        chunks = bytearray()
-        end = time.time() + 2.0
-        while time.time() < end:
-            rrc, data = bulk_in(512, 600)
-            if rrc == 0 and data:
-                chunks.extend(data)
-                end = min(end, time.time() + 0.4)  # brief idle window then stop
-            elif rrc == LIBUSB_ERROR_TIMEOUT:
-                if chunks:
-                    break
-            else:
-                if rrc != 0:
-                    break
-        print(f"  > {cmd!r:24} -> {bytes(chunks)!r}")
-
-    # Read-first sanity check: is the IN endpoint alive / does the device send
-    # anything unsolicited?
-    rrc, data = bulk_in(512, 1000)
-    print(f"[*] read-first IN: rc={err(rrc) if rrc else 'ok'} data={data!r}\n")
-
-    try:
-        for cmd in ["System Version?NL", "Measure?", "DOD?", "Battery Type?"]:
-            exchange(cmd)
-            time.sleep(0.3)
-    finally:
-        lib.libusb_release_interface(h, IFACE)
-        lib.libusb_close(h)
-        lib.libusb_exit(ctx)
-
-    print("\n[done]")
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

test_nl52_protocol.py

@@ -1,134 +0,0 @@
-#!/usr/bin/env python3
-"""
-Unit tests for nl52.protocol — DOD/DRD parsing. No hardware required.
-
-Run:  python3 test_nl52_protocol.py
-"""
-
-import sys
-
-from nl52.protocol import (
-    parse_dod,
-    parse_drd,
-    parse_level,
-    is_result_code,
-    ResultError,
-    RESULT_CODES,
-)
-
-PASS = 0
-FAIL = 0
-
-
-def check(name, cond, detail=""):
-    global PASS, FAIL
-    if cond:
-        PASS += 1
-        print(f"  ✓ {name}")
-    else:
-        FAIL += 1
-        print(f"  ✗ {name} {('— ' + detail) if detail else ''}")
-
-
-def test_level_tokens():
-    print("\n[1] Level token parsing")
-    check("plain value", parse_level("55.5") == 55.5)
-    check("space-padded value", parse_level("  60.1") == 60.1)
-    check("disabled channel '--.-' -> None", parse_level(" --.-") is None)
-    check("dashes-only -> None", parse_level("---.-") is None)
-    check("empty -> None", parse_level("   ") is None)
-    check("negative value", parse_level("-3.2") == -3.2)
-
-
-def test_result_codes():
-    print("\n[2] Result code recognition")
-    check("R+0000 is a result code", is_result_code("R+0000"))
-    check("R+0004 is a result code", is_result_code(" R+0004 "))
-    check("data line is not a result code", not is_result_code("55.5,54.2"))
-    check("all 5 codes documented", set(RESULT_CODES) == {
-        "R+0000", "R+0001", "R+0002", "R+0003", "R+0004"})
-    err = ResultError("R+0004")
-    check("ResultError carries meaning", "Status error" in err.meaning, err.meaning)
-
-
-def test_dod_full():
-    print("\n[3] DOD? full 14-field parse")
-    # d1..d14: Lp,Leq,LE,Lmax,Lmin,Ly,LN1..5,subLp,overload,underrange
-    resp = "55.5,54.2,60.1,50.3,45.2,12.3,48.1,50.0,52.3,44.1,43.0,40.2,0,0"
-    s = parse_dod(resp)
-    check("lp", s.lp == 55.5, str(s.lp))
-    check("leq", s.leq == 54.2)
-    check("le", s.le == 60.1)
-    check("lmax", s.lmax == 50.3)
-    check("lmin", s.lmin == 45.2)
-    check("ly", s.ly == 12.3)
-    check("ln1", s.ln1 == 48.1)
-    check("ln5", s.ln5 == 43.0)
-    check("sub_lp", s.sub_lp == 40.2)
-    check("overload False", s.overload is False)
-    check("underrange False", s.underrange is False)
-    check("raw preserved", s.raw == resp)
-
-
-def test_dod_disabled_channels():
-    print("\n[4] DOD? with disabled display channels ('--.-')")
-    # When display is OFF, d2..d12 come back as ' --.-'
-    resp = "55.5, --.-, --.-, --.-, --.-, --.-, --.-, --.-, --.-, --.-, --.-, --.-,1,0"
-    s = parse_dod(resp)
-    check("lp still present", s.lp == 55.5)
-    check("leq disabled -> None", s.leq is None)
-    check("sub_lp disabled -> None", s.sub_lp is None)
-    check("overload True", s.overload is True)
-    check("underrange False", s.underrange is False)
-
-
-def test_dod_space_padded():
-    print("\n[5] DOD? space-padded fixed-width fields")
-    resp = "   55.5,   54.2,   60.1,   50.3,   45.2,   12.3,   48.1,   50.0,   52.3,   44.1,   43.0,   40.2, 0, 1"
-    s = parse_dod(resp)
-    check("padded lp", s.lp == 55.5)
-    check("padded sub_lp", s.sub_lp == 40.2)
-    check("padded underrange True", s.underrange is True)
-
-
-def test_drd():
-    print("\n[6] DRD? sample parse (d0..d8)")
-    # d0=counter,d1=Lp,d2=Leq,d3=Lmax,d4=Lmin,d5=Ly,d6=subLp,d7=ovl,d8=under
-    resp = "12,55.5,54.2,60.1,50.3, --.-,40.2,0,1"
-    s = parse_drd(resp)
-    check("counter int", s.counter == 12, str(s.counter))
-    check("lp", s.lp == 55.5)
-    check("leq", s.leq == 54.2)
-    check("lmax", s.lmax == 60.1)
-    check("lmin", s.lmin == 50.3)
-    check("ly disabled -> None", s.ly is None)
-    check("sub_lp", s.sub_lp == 40.2)
-    check("overload False", s.overload is False)
-    check("underrange True", s.underrange is True)
-
-
-def test_short_field_counts():
-    print("\n[7] Tolerance of short/odd field counts")
-    s = parse_dod("55.5,54.2")  # only first two present
-    check("partial DOD: lp", s.lp == 55.5)
-    check("partial DOD: leq", s.leq == 54.2)
-    check("partial DOD: missing -> None", s.lmax is None)
-    d = parse_drd("7")  # only counter
-    check("partial DRD: counter", d.counter == 7)
-    check("partial DRD: missing lp -> None", d.lp is None)
-
-
-def main():
-    test_level_tokens()
-    test_result_codes()
-    test_dod_full()
-    test_dod_disabled_channels()
-    test_dod_space_padded()
-    test_drd()
-    test_short_field_counts()
-    print(f"\n{'='*50}\nResults: {PASS} passed, {FAIL} failed")
-    return FAIL == 0
-
-
-if __name__ == "__main__":
-    sys.exit(0 if main() else 1)

test_wedge_recovery.py

@@ -0,0 +1,377 @@
+#!/usr/bin/env python3
+"""
+End-to-end test for wedge detection and recovery (app/recovery.py).
+
+Spins up a fake NL-43 (control + FTP listeners speaking just enough of the
+ASCII protocol) that can be "wedged" — control listener killed while FTP
+stays up, exactly the failure signature from the Feb 2026 investigation.
+A local HTTP server stands in for the relay webhook; hitting it "power
+cycles" the fake device.
+
+Runs against a throwaway SQLite DB in a temp directory — does not touch
+data/slmm.db.
+
+Usage:
+    python3 test_wedge_recovery.py
+
+Exit code 0 = all tests passed.
+"""
+
+import asyncio
+import os
+import sys
+import tempfile
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+
+# --- Run from a temp dir so app.database creates a throwaway DB ------------
+REPO_ROOT = Path(__file__).resolve().parent
+sys.path.insert(0, str(REPO_ROOT))
+WORKDIR = tempfile.mkdtemp(prefix="slmm-wedge-test-")
+os.chdir(WORKDIR)
+
+# Confirm fast in tests
+os.environ.setdefault("WEDGE_CONFIRM_FAILURES", "2")
+
+from app.database import Base, engine, SessionLocal  # noqa: E402
+from app.models import NL43Config, NL43Status  # noqa: E402
+from app.recovery import classify_failure, probe_tcp, recovery_manager  # noqa: E402
+
+Base.metadata.create_all(bind=engine)
+
+PASS = 0
+FAIL = 0
+
+
+def check(name: str, condition: bool, detail: str = ""):
+    global PASS, FAIL
+    if condition:
+        PASS += 1
+        print(f"  ✓ {name}")
+    else:
+        FAIL += 1
+        print(f"  ✗ {name} {('— ' + detail) if detail else ''}")
+
+
+# ---------------------------------------------------------------------------
+# Fake NL-43 device
+# ---------------------------------------------------------------------------
+
+class FakeNL43:
+    """Minimal NL-43: control port speaks the ASCII protocol, FTP port
+    just accepts. Can be wedged (control listener killed, FTP alive)."""
+
+    def __init__(self):
+        self.control_server = None
+        self.ftp_server = None
+        self.control_port = None
+        self.ftp_port = None
+        self.measurement_state = "Stop"
+        self.index = 1
+        self.commands = []  # every command line received
+
+    async def start(self):
+        self.ftp_server = await asyncio.start_server(self._handle_ftp, "127.0.0.1", 0)
+        self.ftp_port = self.ftp_server.sockets[0].getsockname()[1]
+        await self.start_control()
+
+    async def start_control(self):
+        self.control_server = await asyncio.start_server(
+            self._handle_control, "127.0.0.1", 0 if self.control_port is None else self.control_port
+        )
+        self.control_port = self.control_server.sockets[0].getsockname()[1]
+
+    async def wedge(self):
+        """Kill the control listener (existing + new connections die). FTP stays up."""
+        if self.control_server:
+            self.control_server.close()
+            await self.control_server.wait_closed()
+            self.control_server = None
+
+    async def power_cycle(self, boot_delay: float = 2.0):
+        """Simulate a power cycle: everything down, then back up after boot_delay.
+        Power loss stops any running measurement."""
+        await self.wedge()
+        self.measurement_state = "Stop"
+        await asyncio.sleep(boot_delay)
+        await self.start_control()
+
+    async def stop(self):
+        for srv in (self.control_server, self.ftp_server):
+            if srv:
+                srv.close()
+                await srv.wait_closed()
+        self.control_server = None
+        self.ftp_server = None
+
+    async def _handle_ftp(self, reader, writer):
+        try:
+            writer.write(b"220 Connection Ready\r\n")
+            await writer.drain()
+            await reader.read(1024)
+        except Exception:
+            pass
+        finally:
+            writer.close()
+
+    async def _handle_control(self, reader, writer):
+        try:
+            while True:
+                line = await reader.readuntil(b"\n")
+                cmd = line.decode(errors="ignore").strip()
+                if not cmd:
+                    continue
+                self.commands.append(cmd)
+                writer.write(self._respond(cmd))
+                await writer.drain()
+        except (asyncio.IncompleteReadError, ConnectionResetError):
+            pass
+        finally:
+            try:
+                writer.close()
+            except Exception:
+                pass
+
+    def _respond(self, cmd: str) -> bytes:
+        ok = b"R+0000\r\n"
+        if cmd == "Measure?":
+            return ok + f"{self.measurement_state}\r\n".encode()
+        if cmd == "Measure,Start":
+            self.measurement_state = "Start"
+            return ok
+        if cmd == "Measure,Stop":
+            self.measurement_state = "Stop"
+            return ok
+        if cmd.startswith("Clock,"):
+            return ok
+        if cmd == "Store Name?":
+            return ok + f"{self.index:04d}\r\n".encode()
+        if cmd.startswith("Store Name,"):
+            self.index = int(cmd.split(",")[1])
+            return ok
+        if cmd == "Overwrite?":
+            return ok + b"None\r\n"
+        if cmd == "DOD?":
+            return ok + b"1,55.5,54.2,60.1,50.3,72.8\r\n"
+        if cmd.startswith("Sleep Mode"):
+            return ok + (b"Off\r\n" if cmd.endswith("?") else b"")
+        return b"R+0001\r\n"  # command error
+
+
+# ---------------------------------------------------------------------------
+# Fake relay webhook
+# ---------------------------------------------------------------------------
+
+class FakeRelay:
+    """HTTP server standing in for a relay controller. A request to /cycle
+    power-cycles the fake device."""
+
+    def __init__(self, device: FakeNL43, loop: asyncio.AbstractEventLoop):
+        self.device = device
+        self.loop = loop
+        self.hits = 0
+        relay = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def _handle(self):
+                relay.hits += 1
+                # Schedule the device power cycle on the asyncio loop
+                asyncio.run_coroutine_threadsafe(relay.device.power_cycle(boot_delay=2.0), relay.loop)
+                self.send_response(200)
+                self.end_headers()
+                self.wfile.write(b"cycling")
+
+            do_GET = _handle
+            do_POST = _handle
+
+            def log_message(self, *args):
+                pass
+
+        self.server = HTTPServer(("127.0.0.1", 0), Handler)
+        self.port = self.server.server_address[1]
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+
+    @property
+    def url(self):
+        return f"http://127.0.0.1:{self.port}/cycle"
+
+    def stop(self):
+        self.server.shutdown()
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+async def test_classifier(device: FakeNL43):
+    print("\n[1] Failure classification")
+
+    # Healthy device
+    c = await classify_failure("127.0.0.1", device.control_port, device.ftp_port)
+    check("healthy device classifies as ok", c["state"] == "ok", str(c))
+
+    # Wedged: control refused, FTP alive — the definitive signature
+    await device.wedge()
+    c = await classify_failure("127.0.0.1", device.control_port, device.ftp_port)
+    check("wedged device classifies as wedged", c["state"] == "wedged", str(c))
+    check("wedge confidence is high (FTP alive)", c["confidence"] == "high", str(c))
+
+    # Wedged without FTP probe available
+    c = await classify_failure("127.0.0.1", device.control_port, None)
+    check("wedge without FTP probe is medium confidence",
+          c["state"] == "wedged" and c["confidence"] == "medium", str(c))
+
+    # Offline: unroutable address times out (TEST-NET-1)
+    c = await classify_failure("192.0.2.1", 2255, None)
+    check("unreachable host classifies as offline", c["state"] == "offline", str(c))
+
+    await device.start_control()
+    probe = await probe_tcp("127.0.0.1", device.control_port)
+    check("control port back open after un-wedge", probe == "open", probe)
+
+
+async def test_webhook_recovery(device: FakeNL43, relay: FakeRelay):
+    print("\n[2] End-to-end webhook recovery (wedge → relay cycle → reconnect → resume)")
+
+    db = SessionLocal()
+    cfg = NL43Config(
+        unit_id="TEST-NL43",
+        host="127.0.0.1",
+        tcp_port=device.control_port,
+        ftp_port=device.ftp_port,
+        tcp_enabled=True,
+        ftp_enabled=True,
+        auto_recovery_enabled=True,
+        reset_backend="webhook",
+        reset_webhook_url=relay.url,
+        reset_webhook_method="POST",
+        recovery_boot_wait_seconds=10,   # validator min; device boots in 2s
+        recovery_reconnect_timeout=60,
+        recovery_auto_resume=True,
+        recovery_max_attempts=3,
+        recovery_window_minutes=360,
+    )
+    db.add(cfg)
+    # Device was measuring before the wedge
+    status = NL43Status(unit_id="TEST-NL43", measurement_state="Start", consecutive_failures=2)
+    db.add(status)
+    db.commit()
+
+    # Wedge it
+    device.measurement_state = "Start"
+    await device.wedge()
+
+    classification = await classify_failure("127.0.0.1", device.control_port, device.ftp_port)
+    check("pre-recovery classification is wedged", classification["state"] == "wedged")
+
+    started = recovery_manager.start_recovery(cfg, classification)
+    check("recovery task started", started)
+
+    # Wait for the recovery task to finish (boot wait 10s + commands ≈ 20s)
+    deadline = time.time() + 90
+    while recovery_manager.is_active("TEST-NL43") and time.time() < deadline:
+        await asyncio.sleep(1)
+    check("recovery task completed", not recovery_manager.is_active("TEST-NL43"))
+
+    check("relay webhook was hit", relay.hits == 1, f"hits={relay.hits}")
+
+    db.expire_all()
+    status = db.query(NL43Status).filter_by(unit_id="TEST-NL43").first()
+    check("recovery_state is idle", status.recovery_state == "idle", status.recovery_state)
+    check("connection_state is ok", status.connection_state == "ok", status.connection_state)
+    check("consecutive_failures reset", status.consecutive_failures == 0)
+    check("last_recovery_result records success",
+          status.last_recovery_result and "Recovered" in status.last_recovery_result,
+          str(status.last_recovery_result))
+
+    check("measurement resumed on device", device.measurement_state == "Start", device.measurement_state)
+    check("device received Measure,Start", "Measure,Start" in device.commands)
+    check("start cycle used overwrite protection", "Overwrite?" in device.commands)
+
+    db.close()
+
+
+async def test_gating(device: FakeNL43):
+    print("\n[3] Recovery gating (confirmation threshold + attempt limit)")
+
+    db = SessionLocal()
+    cfg = NL43Config(
+        unit_id="TEST-GATE",
+        host="127.0.0.1",
+        tcp_port=device.control_port,
+        ftp_port=device.ftp_port,
+        tcp_enabled=True,
+        auto_recovery_enabled=True,
+        reset_backend="webhook",
+        reset_webhook_url="http://127.0.0.1:1/unreachable",  # fails fast
+        recovery_max_attempts=1,
+        recovery_window_minutes=360,
+    )
+    db.add(cfg)
+    status = NL43Status(unit_id="TEST-GATE", consecutive_failures=1)
+    db.add(status)
+    db.commit()
+
+    classification = {"state": "wedged", "confidence": "high", "tcp_probe": "refused", "ftp_probe": "open"}
+
+    # Below confirmation threshold (1 < 2) — no recovery
+    started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
+    check("below threshold does not trigger", not started)
+
+    # At threshold — triggers (and fails fast on the dead webhook)
+    status.consecutive_failures = 2
+    db.commit()
+    started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
+    check("at threshold triggers recovery", started)
+
+    deadline = time.time() + 30
+    while recovery_manager.is_active("TEST-GATE") and time.time() < deadline:
+        await asyncio.sleep(0.5)
+
+    db.expire_all()
+    status = db.query(NL43Status).filter_by(unit_id="TEST-GATE").first()
+    check("failed webhook marks recovery failed", status.recovery_state == "failed", status.recovery_state)
+    check("failure recorded in result",
+          status.last_recovery_result and "FAILED" in status.last_recovery_result,
+          str(status.last_recovery_result))
+
+    # Attempt limit (max 1 per window) — second attempt blocked
+    started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
+    check("attempt limit blocks repeat recovery", not started)
+
+    # Disabled flag blocks everything
+    cfg.auto_recovery_enabled = False
+    db.commit()
+    started = recovery_manager.maybe_start_recovery(cfg, status, classification, db)
+    check("auto_recovery_enabled=False blocks recovery", not started)
+
+    db.close()
+
+
+async def main():
+    print(f"Wedge recovery test — temp workdir: {WORKDIR}")
+
+    device = FakeNL43()
+    await device.start()
+    relay = FakeRelay(device, asyncio.get_running_loop())
+    print(f"Fake NL-43: control=127.0.0.1:{device.control_port}, ftp=127.0.0.1:{device.ftp_port}")
+    print(f"Fake relay webhook: {relay.url}")
+
+    try:
+        await test_classifier(device)
+        await test_webhook_recovery(device, relay)
+        await test_gating(device)
+    finally:
+        relay.stop()
+        await device.stop()
+
+    print(f"\n{'='*50}\nResults: {PASS} passed, {FAIL} failed")
+    return FAIL == 0
+
+
+if __name__ == "__main__":
+    ok = asyncio.run(main())
+    sys.exit(0 if ok else 1)