feat: add NL-52 USB bulk probe implementation

fix: jinja2 dict hash bug

CHANGELOG.md

@@ -5,52 +5,6 @@ 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).
 
-## [0.4.0] - 2026-06-22
-
-### Added
-
-#### Live Monitor (fan-out feed)
-- **Per-device fan-out monitor** - one shared, cached live feed per device. Multiple clients (dashboards, portal, charts) subscribe to the same stream instead of each fighting for the NL-43's single TCP connection: one poller reads the device, all subscribers get the same frames.
-- **WebSocket monitor** - `WS /api/nl43/{unit_id}/monitor` delivers an instant first frame from cache, then live updates.
-- **Monitor control** - `POST /api/nl43/{unit_id}/monitor/{start|stop}`, `GET /api/nl43/_monitor/status`. A persistent `monitor_enabled` flag auto-starts the keepalive on boot.
-- **Adaptive polling** - poll rate adapts to demand; unreachable devices back off; a device-offline alert fires when a monitored unit drops.
-- **De-duplication** - the background poller skips units already covered by an active monitor (no double-polling); a heartbeat keeps the feed warm.
-- **Lower latency** - the monitor caches run state, roughly halving live-feed latency; fan-out emits an instant first frame + offline status to new clients.
-
-#### Alert Engine
-- **Threshold rules** - per-device alert rules (metric + threshold + cooldown) with full CRUD: `POST/GET/PUT/DELETE /api/nl43/{unit_id}/alerts/rules[/{rule_id}]`.
-- **Events + state machine** - onset/clear tracking via `GET /api/nl43/{unit_id}/alerts/events`; acknowledge with `POST .../events/{event_id}/ack`. A `cooldown_s` is enforced between onsets.
-- **24/7 evaluation** - enabled rules pin the monitor on, so rules evaluate continuously even with no UI client connected.
-- **Resilience** - editing or deleting a rule resets its state and closes any open event; device-offline events are raised when a monitored unit goes unreachable.
-
-#### Data & History
-- **Live-chart backfill** - a downsampled DOD trail is persisted to a new `nl43_readings` table, exposed via `GET /api/nl43/{unit_id}/history` so charts can backfill recent history on load.
-- **LN1/LN2 percentiles** - L1/L10 (configurable percentiles) surfaced through SLMM in the status and live-feed payloads.
-- **measurement_start_time** included in the cached `/status` response.
-
-#### Device control
-- **Per-device disconnect** - `POST /api/nl43/{unit_id}/disconnect` drops a device's pooled connection.
-- **Deactivate / standby** - `POST /api/nl43/{unit_id}/deactivate` and global `POST /api/nl43/_system/standby` to quiesce polling/monitoring.
-
-### Changed
-- **DRD streaming reuses the pooled connection** rather than opening a separate socket, avoiding contention with the persistent pool on a single-connection device.
-- **Connection pool** - idle-TTL / max-age checks can now be disabled; pool status is logged periodically.
-
-### Fixed
-- **Measurement-start confirmation** - `/start` now recognizes the device's `Start` state. It previously waited for `Measure`, which never matched, so the start cycle ran the full retry loop and Terra-View's proxy timed out with a misleading "Unknown error" even though the device had started.
-- **Garbled reads** - corrupted measurement-state reads that produced phantom STOPPED/STARTED transitions are now ignored.
-- **DOD parsing** - corrected field parsing and stopped spurious measurement-time resets.
-- **Monitor WebSocket** - quieted a send-after-close race on client disconnect.
-
-### Database
-- **New tables** (auto-created on startup via `Base.metadata.create_all`): `alert_rules`, `alert_events`, `nl43_readings`.
-- **Migrations for existing tables** (run once per database): `migrate_add_ln_percentiles.py` (LN1/LN2 on `nl43_status`), `migrate_add_monitor_enabled.py` (`monitor_enabled` on `nl43_config`).
-
-### Notes
-- Pairs with the matching Terra-View `dev` build, which reads SLMM's `/monitor` fan-out feed for live SLM dashboards (L1/L10 lines, live-chart backfill). Ship the two together.
-
----
-
 ## [0.3.0] - 2026-02-17
 
 ### Added

README.md

@@ -1,6 +1,6 @@
 # SLMM - Sound Level Meter Manager
 
-**Version 0.4.0**
+**Version 0.3.0**
 
 Backend API service for controlling and monitoring Rion NL-43/NL-53 Sound Level Meters via TCP and FTP protocols.
 
@@ -12,9 +12,6 @@ SLMM is a standalone backend module that provides REST API routing and command t
 
 ## Features
 
-- **Live Monitor (fan-out)**: One shared cached live feed per device — many clients subscribe to the same stream instead of fighting over the meter's single TCP connection
-- **Alert Engine**: Per-device threshold rules with onset/clear events, cooldowns, acks, and 24/7 evaluation
-- **History & Percentiles**: Downsampled DOD trail + history endpoint for live-chart backfill; LN1/LN2 (L1/L10) percentiles surfaced through the feed
 - **Persistent TCP Connections**: Cached per-device connections with OS-level keepalive, tuned for cellular modem reliability
 - **Background Polling**: Continuous automatic polling of devices with configurable intervals
 - **Offline Detection**: Automatic device reachability tracking with failure counters
@@ -47,30 +44,6 @@ SLMM is a standalone backend module that provides REST API routing and command t
                                   └──────────────┘
 ```
 
-### Live Monitor — Fan-Out Feed (v0.4.0)
-
-The NL-43 allows only one TCP control connection at a time, so multiple clients
-polling the same device directly would contend for it. The monitor solves this
-with a single shared, cached feed per device:
-
-- **One reader, many subscribers**: a single poller reads the device; every
-  WebSocket subscriber (`WS /api/nl43/{unit_id}/monitor`) receives the same
-  frames — an instant first frame from cache, then live updates.
-- **Persistent + auto-start**: a `monitor_enabled` flag keeps the feed running
-  and auto-starts it on boot. Enabled alert rules pin the monitor on for 24/7
-  evaluation even with no UI connected.
-- **Adaptive & deduplicated**: poll rate adapts to demand, unreachable devices
-  back off, and the background poller skips units already covered by a monitor.
-
-### Alert Engine (v0.4.0)
-
-Per-device threshold alerting evaluated against the live feed:
-
-- **Rules**: metric + threshold + `cooldown_s`, full CRUD per device
-- **Events**: onset/clear state machine, acknowledgement, and a device-offline
-  alert when a monitored unit drops
-- **Robust**: editing/deleting a rule resets its state and closes open events
-
 ### Persistent TCP Connection Pool (v0.3.0)
 
 SLMM maintains persistent TCP connections to devices with OS-level keepalive, designed for reliable operation over cellular modems:
@@ -172,32 +145,8 @@ Logs are written to:
 |--------|----------|-------------|
 | GET | `/api/nl43/{unit_id}/status` | Get cached measurement snapshot (updated by background poller) |
 | GET | `/api/nl43/{unit_id}/live` | Request fresh DOD data from device (bypasses cache) |
-| GET | `/api/nl43/{unit_id}/history` | Downsampled DOD trail for live-chart backfill |
 | WS | `/api/nl43/{unit_id}/stream` | WebSocket stream for real-time DRD data |
 
-### Live Monitor (fan-out feed)
-
-| Method | Endpoint | Description |
-|--------|----------|-------------|
-| WS | `/api/nl43/{unit_id}/monitor` | Subscribe to the shared cached live feed (instant first frame) |
-| POST | `/api/nl43/{unit_id}/monitor/start` | Start the device's monitor feed |
-| POST | `/api/nl43/{unit_id}/monitor/stop` | Stop the device's monitor feed |
-| GET | `/api/nl43/_monitor/status` | Global monitor status across devices |
-| POST | `/api/nl43/{unit_id}/disconnect` | Drop the device's pooled TCP connection |
-| POST | `/api/nl43/{unit_id}/deactivate` | Quiesce polling/monitoring for one device |
-| POST | `/api/nl43/_system/standby` | Global standby — quiesce all polling/monitoring |
-
-### Alerts
-
-| Method | Endpoint | Description |
-|--------|----------|-------------|
-| GET | `/api/nl43/{unit_id}/alerts/rules` | List alert rules for a device |
-| POST | `/api/nl43/{unit_id}/alerts/rules` | Create an alert rule (metric, threshold, cooldown) |
-| PUT | `/api/nl43/{unit_id}/alerts/rules/{rule_id}` | Update a rule (resets its state, closes open events) |
-| DELETE | `/api/nl43/{unit_id}/alerts/rules/{rule_id}` | Delete a rule |
-| GET | `/api/nl43/{unit_id}/alerts/events` | List alert events (onset/clear) |
-| POST | `/api/nl43/{unit_id}/alerts/events/{event_id}/ack` | Acknowledge an event |
-
 ### Background Polling
 
 | Method | Endpoint | Description |
@@ -324,35 +273,11 @@ Caches latest measurement snapshot:
 - `sd_remaining_mb`: Free SD card space (MB)
 - `sd_free_ratio`: SD card free space ratio
 - `raw_payload`: Raw device response data
-- `is_reachable`: Device reachability status (Boolean)
-- `consecutive_failures`: Count of consecutive poll failures
-- `last_poll_attempt`: Last time background poller attempted to poll
-- `last_success`: Last successful poll timestamp
-- `last_error`: Last error message (truncated to 500 chars)
-- `ln1` / `ln2`: LN1/LN2 (L1/L10) percentile levels ⭐ v0.4.0
-
-### NL43Readings Table ⭐ v0.4.0
-Downsampled DOD trail backing the live-chart history endpoint (one row/minute,
-pruned to a retention window — viewing only, not the report source):
-- `id` (PK), `unit_id`, `timestamp`
-- `lp` / `leq` / `lmax` / `ln1` / `ln2`: cached level samples
-
-### AlertRule Table ⭐ v0.4.0
-Per-device threshold alert rules:
-- `id` (PK), `unit_id`, `name`, `enabled`
-- `metric`, `comparison` (above/below), `threshold_db`, `clear_margin_db` (hysteresis)
-- `duration_s` (sustained), `cooldown_s` (min seconds between onsets)
-- `channels` / `recipients`, optional `schedule_start`/`schedule_end`/`schedule_days`
-
-### AlertEvent Table ⭐ v0.4.0
-Alert onset/clear events for history, inbox, and acknowledgement:
-- `id` (PK), `unit_id`, `rule_id`, `rule_name`, `metric`, `threshold_db`
-- `onset_at` / `onset_value`, `peak_value`, `clear_at`, `status` (active/cleared)
-- `acknowledged_at` / `acknowledged_by`, `notes`
-
-> New tables (`alert_rules`, `alert_events`, `nl43_readings`) auto-create on
-> startup. Existing-table columns ship with migrations:
-> `migrate_add_ln_percentiles.py`, `migrate_add_monitor_enabled.py`.
+- `is_reachable`: Device reachability status (Boolean) ⭐ NEW
+- `consecutive_failures`: Count of consecutive poll failures ⭐ NEW
+- `last_poll_attempt`: Last time background poller attempted to poll ⭐ NEW
+- `last_success`: Last successful poll timestamp ⭐ NEW
+- `last_error`: Last error message (truncated to 500 chars) ⭐ NEW
 
 ## Protocol Details
 

SLM-stress-test/nl52_usb_probe.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""
+RION NL-42 / NL-52 USB serial probe — zero dependencies (stdlib termios only).
+
+The NL-52's USB port enumerates as a virtual COM port ("RION USB to RS232C
+Converter"). On Linux that is almost always handled by an in-kernel USB-serial
+driver (ftdi_sio / cp210x / ch341) which creates /dev/ttyUSB*. This script
+opens that port and sends a few harmless REQUEST commands (no settings are
+changed) to confirm two-way communication.
+
+Protocol (NL-42/NL-52 Serial Interface Manual 55779):
+  Request:  "<Command>?"          + CRLF
+  Setting:  "$<Command>,<param>"  + CRLF      (NOT used here — read-only probe)
+  Reply:    result code "R+0000"  + CRLF, then data line(s) for requests.
+
+Before running:
+  1. On the meter: MENU -> I/O -> Communication Interface -> "USB"
+     (set this BEFORE plugging in the cable).
+  2. Connect a generic USB-A -> mini-B cable directly (no hub).
+  3. Find the port:  ls -l /dev/ttyUSB*   (and `dmesg | tail` after plugging in)
+
+Usage:
+  python3 nl52_usb_probe.py                      # defaults to /dev/ttyUSB0
+  python3 nl52_usb_probe.py --port /dev/ttyUSB0 --baud 115200
+  python3 nl52_usb_probe.py --cmd "System Version?" --cmd "Clock?"
+"""
+
+import argparse
+import os
+import select
+import sys
+import termios
+import time
+
+# Safe, read-only probe commands (all are pure requests).
+DEFAULT_COMMANDS = [
+    "System Version?",   # firmware version — proves the link end-to-end
+    "Clock?",            # current date/time
+    "SD Card Free Size?",
+    "DOD?",              # snapshot of currently displayed values
+]
+
+BAUD_CONSTANTS = {
+    9600: termios.B9600,
+    19200: termios.B19200,
+    38400: termios.B38400,
+    57600: termios.B57600,
+    115200: termios.B115200,
+}
+
+
+def open_serial(port: str, baud: int) -> int:
+    """Open a serial port in raw 8N1, no flow control. Returns an fd."""
+    if baud not in BAUD_CONSTANTS:
+        raise ValueError(f"Unsupported baud {baud}; choose from {sorted(BAUD_CONSTANTS)}")
+
+    fd = os.open(port, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
+
+    attrs = termios.tcgetattr(fd)
+    iflag, oflag, cflag, lflag, ispeed, ospeed, cc = attrs
+
+    # Raw mode
+    iflag = 0
+    oflag = 0
+    lflag = 0
+    # 8 data bits, enable receiver, ignore modem control lines
+    cflag = termios.CS8 | termios.CREAD | termios.CLOCAL
+    # (no PARENB = no parity, no CSTOPB = 1 stop bit, no CRTSCTS = no flow control)
+
+    bconst = BAUD_CONSTANTS[baud]
+    ispeed = bconst
+    ospeed = bconst
+
+    termios.tcsetattr(fd, termios.TCSANOW, [iflag, oflag, cflag, lflag, ispeed, ospeed, cc])
+    termios.tcflush(fd, termios.TCIOFLUSH)
+    return fd
+
+
+def send(fd: int, line: str):
+    os.write(fd, (line + "\r\n").encode("ascii"))
+
+
+def read_reply(fd: int, timeout: float = 3.0) -> bytes:
+    """Read whatever arrives within `timeout` seconds (idle-gap terminated)."""
+    buf = bytearray()
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        r, _, _ = select.select([fd], [], [], 0.3)
+        if r:
+            try:
+                chunk = os.read(fd, 4096)
+            except BlockingIOError:
+                continue
+            if chunk:
+                buf.extend(chunk)
+                # Once we've seen a CRLF and there's a brief idle, stop early
+                deadline = min(deadline, time.time() + 0.4)
+    return bytes(buf)
+
+
+def main():
+    ap = argparse.ArgumentParser(description="RION NL-42/NL-52 USB serial probe")
+    ap.add_argument("--port", default="/dev/ttyUSB0")
+    ap.add_argument("--baud", type=int, default=115200,
+                    help="USB CDC usually ignores baud, but RS-232C needs a match")
+    ap.add_argument("--cmd", action="append", dest="cmds",
+                    help="Override probe command(s); repeatable")
+    ap.add_argument("--timeout", type=float, default=3.0)
+    args = ap.parse_args()
+
+    commands = args.cmds or DEFAULT_COMMANDS
+
+    if not os.path.exists(args.port):
+        print(f"[!] {args.port} does not exist.")
+        print("    Plug in the meter (Comm Interface = USB) and check: ls -l /dev/ttyUSB*")
+        print("    Also check the kernel saw it:  dmesg | tail -20")
+        return 2
+
+    try:
+        fd = open_serial(args.port, args.baud)
+    except PermissionError:
+        print(f"[!] Permission denied on {args.port}.")
+        print("    Add yourself to the 'dialout' group, or run with sudo:")
+        print(f"      sudo usermod -aG dialout $USER   (then log out/in)")
+        return 2
+    except Exception as e:
+        print(f"[!] Could not open {args.port}: {e}")
+        return 2
+
+    print(f"[*] Opened {args.port} @ {args.baud} 8N1 (raw, no flow control)")
+    print(f"[*] Sending {len(commands)} read-only request command(s)\n")
+
+    ok = 0
+    try:
+        for cmd in commands:
+            send(fd, cmd)
+            reply = read_reply(fd, args.timeout)
+            decoded = reply.decode("ascii", errors="replace").replace("\r", "\\r").replace("\n", "\\n\n")
+            if reply:
+                ok += 1
+                print(f"  > {cmd}")
+                for line in decoded.splitlines():
+                    print(f"      {line}")
+            else:
+                print(f"  > {cmd}")
+                print(f"      (no response within {args.timeout}s)")
+            print()
+            time.sleep(1.0)  # NL-series likes >=1s between commands
+    finally:
+        os.close(fd)
+
+    if ok == 0:
+        print("[!] No responses. Things to check:")
+        print("    - Meter's Communication Interface is set to USB (not RS-232C)")
+        print("    - ECO / Sleep mode is OFF (both disable the comm interface)")
+        print("    - Right port (try other /dev/ttyUSB* or /dev/ttyACM*)")
+        print("    - For RS-232C path, baud must match the meter's setting")
+        return 1
+
+    print(f"[OK] {ok}/{len(commands)} commands answered — two-way comms confirmed.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

app/alerts.py

@@ -1,322 +0,0 @@
-"""
-Threshold alert engine.
-
-Each unit can have any number of AlertRules. A rule is evaluated against the
-unit's live monitor snapshots via a small per-(unit, rule) state machine:
-
-    IDLE  --(metric exceeds threshold for duration_s)-->  ACTIVE   (fire ONSET)
-    ACTIVE --(metric recovers past hysteresis for duration_s)--> IDLE (fire CLEAR)
-
-duration_s debounces both edges; clear_margin_db adds hysteresis so a level
-hovering at the threshold doesn't flap. Onset and clear are distinct events.
-
-The state-machine logic (`_evaluate_step`) is intentionally pure — no DB, no
-real clock — so it can be unit-tested with a synthetic level series and a fake
-clock. The AlertEvaluator wraps it with rule loading, scheduling, persistence,
-and dispatch. Dispatch is a server log for now (POC); the seam to POST events to
-a Terra-View webhook (email/SMS) is _dispatch().
-"""
-
-import asyncio
-import logging
-import os
-from dataclasses import dataclass
-from datetime import datetime, timedelta
-from typing import Dict, List, Optional, Tuple
-
-logger = logging.getLogger(__name__)
-
-# Local timezone offset for schedule windows (same env var services.py uses).
-_TZ_OFFSET_HOURS = float(os.getenv("TIMEZONE_OFFSET", "-5"))
-
-# How long to cache a unit's rules before re-querying the DB (rules change rarely).
-_RULE_CACHE_TTL_S = 15.0
-
-
-@dataclass
-class RuleState:
-    """In-memory runtime state for one (unit, rule)."""
-    phase: str = "idle"                 # "idle" | "active"
-    edge_since: Optional[float] = None  # when the current edge condition began (clock time)
-    peak: float = 0.0
-    event_id: Optional[int] = None      # the open AlertEvent row (for the clear update)
-    last_onset: Optional[float] = None  # time of the last onset (for cooldown)
-
-
-def _exceeds(value: float, rule) -> bool:
-    if rule.comparison == "below":
-        return value < rule.threshold_db
-    return value > rule.threshold_db
-
-
-def _recovered(value: float, rule) -> bool:
-    margin = rule.clear_margin_db or 0.0
-    if rule.comparison == "below":
-        return value > rule.threshold_db + margin
-    return value < rule.threshold_db - margin
-
-
-def _evaluate_step(state: RuleState, value: float, now: float, rule) -> Optional[str]:
-    """Advance the state machine by one reading.
-
-    Pure: mutates `state`, returns 'onset' | 'clear' | None. `now` is injected so
-    tests can drive a fake clock.
-    """
-    duration = rule.duration_s or 0
-
-    if state.phase == "idle":
-        if _exceeds(value, rule):
-            if state.edge_since is None:
-                state.edge_since = now
-            if now - state.edge_since >= duration:
-                # Cooldown: suppress a new onset within cooldown_s of the last one
-                # (stops a repeatedly-breaching signal from flooding the history).
-                # Hold edge_since so it fires the moment cooldown lapses if still
-                # breaching — don't reset it here.
-                cooldown = getattr(rule, "cooldown_s", 0) or 0
-                if state.last_onset is not None and (now - state.last_onset) < cooldown:
-                    return None
-                state.phase = "active"
-                state.edge_since = None
-                state.peak = value
-                state.last_onset = now
-                return "onset"
-        else:
-            state.edge_since = None
-        return None
-
-    # active
-    if rule.comparison == "below":
-        state.peak = min(state.peak, value)
-    else:
-        state.peak = max(state.peak, value)
-
-    if _recovered(value, rule):
-        if state.edge_since is None:
-            state.edge_since = now
-        if now - state.edge_since >= duration:
-            state.phase = "idle"
-            state.edge_since = None
-            return "clear"
-    else:
-        state.edge_since = None
-    return None
-
-
-def _in_window(now_minutes: int, start: str, end: str) -> bool:
-    """Is now_minutes (minutes since local midnight) within [start, end)?
-    Handles wraparound windows like 22:00–07:00."""
-    def _m(s: str) -> int:
-        h, m = s.split(":")
-        return int(h) * 60 + int(m)
-    s, e = _m(start), _m(end)
-    if s == e:
-        return True
-    if s < e:
-        return s <= now_minutes < e
-    return now_minutes >= s or now_minutes < e  # wraparound
-
-
-class AlertEvaluator:
-    def __init__(self):
-        self._states: Dict[Tuple[str, int], RuleState] = {}
-        self._rule_cache: Dict[str, Tuple[float, list]] = {}  # unit_id -> (fetched_at, rules)
-        self._offline_events: Dict[str, int] = {}  # unit_id -> open connectivity AlertEvent id
-        logger.info("[ALERT] rule-based evaluator ready")
-
-    async def evaluate(self, unit_id: str, snap) -> None:
-        """Evaluate every enabled rule for this unit against one snapshot."""
-        rules = self._get_rules(unit_id)
-        if not rules:
-            return
-        now = asyncio.get_running_loop().time()
-        for rule in rules:
-            if not self._in_schedule(rule):
-                continue
-            raw = getattr(snap, rule.metric, None)
-            try:
-                value = float(raw)
-            except (TypeError, ValueError):
-                continue  # missing / non-numeric ("-.-")
-            state = self._states.setdefault((unit_id, rule.id), RuleState())
-            action = _evaluate_step(state, value, now, rule)
-            if action == "onset":
-                await self._on_onset(unit_id, rule, value, state)
-            elif action == "clear":
-                await self._on_clear(unit_id, rule, value, state)
-
-    # -- rule loading (cached) ----------------------------------------------
-
-    def _get_rules(self, unit_id: str) -> list:
-        loop_now = asyncio.get_running_loop().time()
-        cached = self._rule_cache.get(unit_id)
-        if cached and loop_now - cached[0] < _RULE_CACHE_TTL_S:
-            return cached[1]
-        rules = self._load_rules(unit_id)
-        self._rule_cache[unit_id] = (loop_now, rules)
-        return rules
-
-    def _load_rules(self, unit_id: str) -> list:
-        from app.database import SessionLocal
-        from app.models import AlertRule
-        db = SessionLocal()
-        try:
-            return db.query(AlertRule).filter_by(unit_id=unit_id, enabled=True).all()
-        except Exception as e:
-            logger.warning(f"[ALERT] failed to load rules for {unit_id}: {e}")
-            return []
-        finally:
-            db.close()
-
-    def invalidate(self, unit_id: Optional[str] = None) -> None:
-        """Drop cached rules so a change is picked up immediately."""
-        if unit_id is None:
-            self._rule_cache.clear()
-        else:
-            self._rule_cache.pop(unit_id, None)
-
-    def forget_rule(self, unit_id: str, rule_id: int) -> None:
-        """Drop a rule's per-(unit, rule) state machine after the rule is edited or
-        deleted, so a stale 'active' phase / open event_id from the old config
-        doesn't bleed into the new one (mis-firing a clear or suppressing an onset)."""
-        self._states.pop((unit_id, rule_id), None)
-
-    # -- scheduling ----------------------------------------------------------
-
-    def _in_schedule(self, rule) -> bool:
-        if not rule.schedule_start or not rule.schedule_end:
-            day_ok = self._day_ok(rule)
-            return day_ok
-        local = datetime.utcnow() + timedelta(hours=_TZ_OFFSET_HOURS)
-        if not self._day_ok(rule, local):
-            return False
-        return _in_window(local.hour * 60 + local.minute, rule.schedule_start, rule.schedule_end)
-
-    @staticmethod
-    def _day_ok(rule, local: Optional[datetime] = None) -> bool:
-        if not rule.schedule_days:
-            return True
-        if local is None:
-            local = datetime.utcnow() + timedelta(hours=_TZ_OFFSET_HOURS)
-        allowed = {int(d) for d in str(rule.schedule_days).split(",") if d.strip() != ""}
-        return local.weekday() in allowed  # Mon=0
-
-    # -- event persistence + dispatch ---------------------------------------
-
-    async def _on_onset(self, unit_id: str, rule, value: float, state: RuleState) -> None:
-        from app.database import SessionLocal
-        from app.models import AlertEvent
-        db = SessionLocal()
-        try:
-            evt = AlertEvent(
-                rule_id=rule.id, unit_id=unit_id, rule_name=rule.name,
-                metric=rule.metric, threshold_db=rule.threshold_db,
-                onset_value=value, peak_value=value, status="active",
-            )
-            db.add(evt)
-            db.commit()
-            db.refresh(evt)
-            state.event_id = evt.id
-        except Exception as e:
-            logger.warning(f"[ALERT] failed to record onset for {unit_id}: {e}")
-        finally:
-            db.close()
-        await self._dispatch(
-            "ONSET", unit_id, rule,
-            f"{rule.metric.upper()}={value:.1f} dB "
-            f"{'<' if rule.comparison == 'below' else '>'} {rule.threshold_db:.1f} dB"
-            f"{f' for {rule.duration_s}s' if rule.duration_s else ''}",
-        )
-
-    async def _on_clear(self, unit_id: str, rule, value: float, state: RuleState) -> None:
-        peak = state.peak
-        from app.database import SessionLocal
-        from app.models import AlertEvent
-        db = SessionLocal()
-        try:
-            if state.event_id is not None:
-                evt = db.query(AlertEvent).filter_by(id=state.event_id).first()
-                if evt:
-                    evt.clear_at = datetime.utcnow()
-                    evt.peak_value = peak
-                    evt.status = "cleared"
-                    db.commit()
-        except Exception as e:
-            logger.warning(f"[ALERT] failed to record clear for {unit_id}: {e}")
-        finally:
-            db.close()
-        state.event_id = None
-        await self._dispatch(
-            "CLEAR", unit_id, rule,
-            f"recovered to {value:.1f} dB (peak {peak:.1f} dB)",
-        )
-
-    # -- connectivity (device offline/online) -------------------------------
-    #
-    # Raised by the live monitor when it loses / regains contact with a device.
-    # Persisted as an AlertEvent (sentinel rule_id=0, metric="connectivity") so it
-    # lands in the same events/inbox/ack pipeline as threshold alerts. The in-memory
-    # map dedupes; the DB query also dedupes across a process restart.
-
-    async def device_offline(self, unit_id: str) -> None:
-        if unit_id in self._offline_events:
-            return  # already flagged offline
-        from app.database import SessionLocal
-        from app.models import AlertEvent
-        db = SessionLocal()
-        try:
-            existing = db.query(AlertEvent).filter_by(
-                unit_id=unit_id, metric="connectivity", status="active").first()
-            if existing:  # already open in the DB (e.g. carried across a restart)
-                self._offline_events[unit_id] = existing.id
-                return
-            evt = AlertEvent(
-                rule_id=0, unit_id=unit_id, rule_name="Device unreachable",
-                metric="connectivity", threshold_db=0.0, status="active",
-            )
-            db.add(evt)
-            db.commit()
-            db.refresh(evt)
-            self._offline_events[unit_id] = evt.id
-        except Exception as e:
-            logger.warning(f"[ALERT] failed to record offline for {unit_id}: {e}")
-        finally:
-            db.close()
-        await self._dispatch_raw("OFFLINE", unit_id, "Device unreachable",
-                                 "live monitor lost contact with the device")
-
-    async def device_online(self, unit_id: str) -> None:
-        self._offline_events.pop(unit_id, None)
-        from app.database import SessionLocal
-        from app.models import AlertEvent
-        db = SessionLocal()
-        cleared = 0
-        try:
-            opened = db.query(AlertEvent).filter_by(
-                unit_id=unit_id, metric="connectivity", status="active").all()
-            for evt in opened:
-                evt.clear_at = datetime.utcnow()
-                evt.status = "cleared"
-                cleared += 1
-            if cleared:
-                db.commit()
-        except Exception as e:
-            logger.warning(f"[ALERT] failed to record online for {unit_id}: {e}")
-        finally:
-            db.close()
-        if cleared:  # only announce recovery if it was actually flagged offline
-            await self._dispatch_raw("ONLINE", unit_id, "Device recovered",
-                                     "live monitor regained contact with the device")
-
-    # -- event persistence + dispatch ---------------------------------------
-
-    async def _dispatch(self, kind: str, unit_id: str, rule, detail: str) -> None:
-        await self._dispatch_raw(kind, unit_id, rule.name, detail)
-
-    async def _dispatch_raw(self, kind: str, unit_id: str, name: str, detail: str) -> None:
-        """POC dispatch: server log. Swap in a Terra-View webhook (email/SMS) here."""
-        logger.warning(f"[ALERT:{kind}] {unit_id} '{name}': {detail}")
-
-
-# Module-level singleton (the monitor calls alert_evaluator.evaluate per snapshot)
-alert_evaluator = AlertEvaluator()

app/background_poller.py

@@ -8,7 +8,6 @@ for fast API access without querying devices on every request.
 
 import asyncio
 import logging
-import os
 from datetime import datetime, timedelta
 from typing import Optional
 
@@ -21,11 +20,6 @@ from app.device_logger import log_device_event, cleanup_old_logs
 
 logger = logging.getLogger(__name__)
 
-# Global polling default. Set SLMM_POLLING_ENABLED=false to start an instance in
-# standby (running but not polling and not holding device connections) — e.g. a
-# dev box that must not latch onto a device that a prod instance owns.
-POLLING_ENABLED_DEFAULT = os.getenv("SLMM_POLLING_ENABLED", "true").lower() == "true"
-
 
 class BackgroundPoller:
     """
@@ -45,7 +39,6 @@ class BackgroundPoller:
         self._logger = logger
         self._last_cleanup = None  # Track last log cleanup time
         self._last_pool_log = None  # Track last connection pool heartbeat log
-        self._active = POLLING_ENABLED_DEFAULT  # Global polling on/off (standby toggle)
 
     async def start(self):
         """Start the background polling task."""
@@ -78,48 +71,15 @@ class BackgroundPoller:
 
         self._logger.info("Background poller stopped")
 
-    def is_active(self) -> bool:
-        """Whether background polling is currently active (vs standby)."""
-        return self._active
-
-    async def set_active(self, active: bool):
-        """Globally enable/disable polling at runtime.
-
-        When deactivated, the loop stays alive but polls nothing and releases all
-        device connections, so this SLMM instance stops occupying the devices'
-        single connection slots (e.g. so a prod instance can take over). Runtime
-        state only — on restart the instance returns to SLMM_POLLING_ENABLED.
-        """
-        self._active = active
-        if active:
-            self._logger.info("[SYSTEM] Background polling ACTIVATED")
-        else:
-            self._logger.info("[SYSTEM] Background polling DEACTIVATED (standby) — releasing connections")
-            await self._release_all_connections()
-
-    async def _release_all_connections(self):
-        """Gracefully close every pooled device connection (no-op if none)."""
-        from app.services import _connection_pool
-        for device_key in list(_connection_pool.get_stats().get("connections", {})):
-            await _connection_pool.discard(device_key)
-
     async def _poll_loop(self):
         """Main polling loop that runs continuously."""
         self._logger.info("Background polling loop started")
 
         while self._running:
-            if self._active:
             try:
                 await self._poll_all_devices()
             except Exception as e:
                 self._logger.error(f"Error in poll loop: {e}", exc_info=True)
-            else:
-                # Standby: poll nothing, and keep holding no device connection slots
-                # so another SLMM instance (e.g. prod) can talk to the devices.
-                try:
-                    await self._release_all_connections()
-                except Exception as e:
-                    self._logger.warning(f"Standby connection release failed: {e}")
 
             # Run log cleanup once per hour
             try:
@@ -178,19 +138,10 @@ class BackgroundPoller:
             now = datetime.utcnow()
             polled_count = 0
 
-            from app.monitor import monitor_manager
-
             for cfg in configs:
                 if not self._running:
                     break
 
-                # Skip units with an active live monitor: it polls them at ~1Hz and
-                # keeps the status cache fresh, so a redundant background poll would just
-                # add load/lock-contention on the device's single connection.
-                if monitor_manager.is_active(cfg.unit_id):
-                    self._logger.debug(f"Skipping {cfg.unit_id} — live monitor active")
-                    continue
-
                 # Get current status
                 status = db.query(NL43Status).filter_by(unit_id=cfg.unit_id).first()
 

app/main.py

@@ -38,25 +38,6 @@ async def lifespan(app: FastAPI):
     await poller.start()
     logger.info("Background poller started")
 
-    # Auto-start keepalive live monitors for units configured for 24/7 monitoring
-    # (monitor_enabled). This is what keeps alerting running unattended across
-    # restarts — without it a feed only runs while someone has the live view open.
-    try:
-        from app.monitor import monitor_manager
-        from app.database import SessionLocal
-        from app.models import NL43Config
-        db = SessionLocal()
-        try:
-            units = db.query(NL43Config).filter_by(monitor_enabled=True, tcp_enabled=True).all()
-            for cfg in units:
-                m = await monitor_manager.get(cfg.unit_id)
-                await m.set_keepalive(True)
-                logger.info(f"Auto-started keepalive monitor for {cfg.unit_id}")
-        finally:
-            db.close()
-    except Exception as e:
-        logger.error(f"Failed to auto-start monitors: {e}")
-
     yield  # Application runs
 
     # Shutdown
@@ -71,7 +52,7 @@ async def lifespan(app: FastAPI):
 app = FastAPI(
     title="SLMM NL43 Addon",
     description="Standalone module for NL43 configuration and status APIs with background polling",
-    version="0.4.0",
+    version="0.3.0",
     lifespan=lifespan,
 )
 

app/models.py

@@ -1,4 +1,4 @@
-from sqlalchemy import Column, String, DateTime, Boolean, Integer, Float, Text, func
+from sqlalchemy import Column, String, DateTime, Boolean, Integer, Text, func
 from app.database import Base
 
 
@@ -23,10 +23,6 @@ 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
 
-    # Live monitor (fan-out DOD feed). Keepalive runs it 24/7 even with no viewer,
-    # which is what makes alerting continuous. On by default; toggleable from the UI.
-    monitor_enabled = Column(Boolean, default=True)
-
 
 class NL43Status(Base):
     """
@@ -45,8 +41,6 @@ class NL43Status(Base):
     lmax = Column(String, nullable=True)  # Maximum level
     lmin = Column(String, nullable=True)  # Minimum level
     lpeak = Column(String, nullable=True)  # Peak level
-    ln1 = Column(String, nullable=True)  # Percentile slot LN1 (configurable; device default L5, contract L1)
-    ln2 = Column(String, nullable=True)  # Percentile slot LN2 (configurable; device default L10)
     battery_level = Column(String, nullable=True)
     power_source = Column(String, nullable=True)
     sd_remaining_mb = Column(String, nullable=True)
@@ -78,74 +72,3 @@ class DeviceLog(Base):
     level = Column(String, default="INFO")  # DEBUG, INFO, WARNING, ERROR
     category = Column(String, default="GENERAL")  # TCP, FTP, POLL, COMMAND, STATE, SYNC
     message = Column(Text, nullable=False)
-
-
-class AlertRule(Base):
-    """A threshold-alert rule evaluated against a unit's live monitor feed.
-
-    Source-agnostic: today it runs over the DOD monitor; the same rule transfers
-    unchanged if a unit's feed is later sourced from FTP intervals.
-    """
-
-    __tablename__ = "alert_rules"
-
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    unit_id = Column(String, index=True, nullable=False)
-    name = Column(String, nullable=False, default="Alert")
-    metric = Column(String, nullable=False, default="lp")  # lp/leq/lmax/lmin/lpeak/ln1/ln2
-    comparison = Column(String, nullable=False, default="above")  # above | below
-    threshold_db = Column(Float, nullable=False)
-    duration_s = Column(Integer, nullable=False, default=0)       # sustained seconds (0 = instant)
-    clear_margin_db = Column(Float, nullable=False, default=2.0)  # hysteresis band
-    cooldown_s = Column(Integer, nullable=False, default=300)     # min seconds between onsets
-    # Optional time-of-day scoping (local time). schedule_start/end as "HH:MM";
-    # null = always active. schedule_days = CSV of 0-6 (Mon=0); null = every day.
-    schedule_start = Column(String, nullable=True)
-    schedule_end = Column(String, nullable=True)
-    schedule_days = Column(String, nullable=True)
-    channels = Column(String, nullable=False, default="log")  # CSV: log,email,sms
-    recipients = Column(Text, nullable=True)                  # CSV of emails/phones
-    enabled = Column(Boolean, default=True)
-    created_at = Column(DateTime, default=func.now())
-
-
-class AlertEvent(Base):
-    """A fired alert (onset → clear), for history / inbox / acknowledgement."""
-
-    __tablename__ = "alert_events"
-
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    rule_id = Column(Integer, index=True, nullable=False)
-    unit_id = Column(String, index=True, nullable=False)
-    rule_name = Column(String, nullable=True)
-    metric = Column(String, nullable=False)
-    threshold_db = Column(Float, nullable=False)
-    onset_at = Column(DateTime, default=func.now(), index=True)
-    onset_value = Column(Float, nullable=True)
-    peak_value = Column(Float, nullable=True)
-    clear_at = Column(DateTime, nullable=True)
-    status = Column(String, default="active")  # active | cleared
-    acknowledged_at = Column(DateTime, nullable=True)
-    acknowledged_by = Column(String, nullable=True)
-    notes = Column(Text, nullable=True)
-
-
-class NL43Reading(Base):
-    """Downsampled time-series of live-monitor readings, for the live-chart
-    backfill (so a viewer sees recent trend on open, not a blank chart).
-
-    Viewing only — NOT the report source. Reports use the device's authoritative
-    FTP .rnd intervals. This is a short, capped trail (one row/minute, pruned to
-    a retention window) fed by the monitor's keepalive poll loop.
-    """
-
-    __tablename__ = "nl43_readings"
-
-    id = Column(Integer, primary_key=True, autoincrement=True)
-    unit_id = Column(String, index=True, nullable=False)
-    timestamp = Column(DateTime, default=func.now(), index=True)
-    lp = Column(String, nullable=True)
-    leq = Column(String, nullable=True)
-    lmax = Column(String, nullable=True)
-    ln1 = Column(String, nullable=True)
-    ln2 = Column(String, nullable=True)

app/monitor.py

@@ -1,322 +0,0 @@
-"""
-Per-device live monitor (fan-out hub).
-
-ONE DOD poll loop per device, broadcast to many subscribers:
-- browser WebSocket clients (live view) — they no longer each open their own
-  device stream, so the NL43's single-connection limit stops causing the
-  "second viewer sees nothing" contention.
-- the alert evaluator (threshold alerts), which can keep a device's feed running
-  even with no browser attached.
-- persistence (each snapshot is written to NL43Status, like the poller does).
-
-The device's one TCP connection is respected: every poll goes through the same
-per-device lock + connection pool in services.py, so the monitor, the background
-poller, and on-demand commands all serialize safely.
-"""
-
-import asyncio
-import logging
-import os
-from datetime import datetime
-from typing import Dict, Optional, Set
-
-from app.database import SessionLocal
-from app.models import NL43Config, NL43Status
-from app.services import NL43Client, persist_snapshot
-from app.alerts import alert_evaluator
-
-logger = logging.getLogger(__name__)
-
-# Extra idle between DOD polls WHEN A BROWSER IS WATCHING. The 1s device rate-limit
-# already paces consecutive DOD? commands, so this just needs to be small — the
-# rate-limit is the real floor (~1.25s/poll effective).
-MONITOR_POLL_INTERVAL = float(os.getenv("MONITOR_POLL_INTERVAL", "0.25"))
-
-# Idle cadence when NO browser is subscribed and the feed is only kept alive for
-# alerting. Same data, ~8x fewer polls -> ~8x less cellular traffic on a metered
-# SIM (~1 GB/device/month at full rate -> ~125 MB). NOTE: this also sets the alert
-# sampling resolution when nobody is watching, so keep it <= the smallest alert
-# duration_s you rely on (default 10s comfortably catches a "sustained 30/60s" rule).
-MONITOR_IDLE_POLL_INTERVAL = float(os.getenv("MONITOR_IDLE_POLL_INTERVAL", "10"))
-
-# Exponential backoff once the device is unreachable, so a powered-off / asleep /
-# out-of-signal device stops churning reconnects every cycle (log spam + a trickle
-# of wasted cellular data on failed SYNs). delay = min(BASE * 2**(fails-1), MAX),
-# reset to full-rate on the first good poll. While a browser is actively watching we
-# cap the backoff lower (WATCHED_MAX) so a recovery surfaces quickly for the viewer.
-MONITOR_BACKOFF_BASE_S = float(os.getenv("MONITOR_BACKOFF_BASE_S", "1"))
-MONITOR_BACKOFF_MAX_S = float(os.getenv("MONITOR_BACKOFF_MAX_S", "60"))
-MONITOR_BACKOFF_WATCHED_MAX_S = float(os.getenv("MONITOR_BACKOFF_WATCHED_MAX_S", "5"))
-
-# How often to refresh the run state (Measure?). It changes rarely, so we cache it
-# and skip that second rate-limited command on most polls — roughly halving the
-# per-update latency (~2.5s -> ~1.3s).
-MONITOR_STATE_REFRESH_S = float(os.getenv("MONITOR_STATE_REFRESH_S", "30"))
-
-# Downsampled trail for the live-chart backfill: store one reading per
-# TRAIL_SAMPLE_S and keep TRAIL_RETENTION_HOURS of it (pruned). Viewing only —
-# reports use the device's FTP .rnd data, not this.
-TRAIL_SAMPLE_S = float(os.getenv("MONITOR_TRAIL_SAMPLE_S", "60"))
-TRAIL_RETENTION_HOURS = float(os.getenv("MONITOR_TRAIL_RETENTION_HOURS", "24"))
-
-# If nothing has been broadcast in this many seconds (e.g. device offline and
-# silent), send a keepalive frame so reverse proxies don't drop the idle WS.
-MONITOR_HEARTBEAT_S = float(os.getenv("MONITOR_HEARTBEAT_S", "25"))
-
-
-def _snapshot_payload(snap, unit_id: str, measurement_start_time) -> dict:
-    """Build the broadcast payload — same shape as the DRD stream, but DOD-sourced
-    so it carries ln1/ln2 (which DRD cannot)."""
-    return {
-        "unit_id": unit_id,
-        "timestamp": datetime.utcnow().isoformat(),
-        "measurement_state": snap.measurement_state,
-        "measurement_start_time": measurement_start_time,
-        "counter": snap.counter,
-        "lp": snap.lp,
-        "leq": snap.leq,
-        "lmax": snap.lmax,
-        "lmin": snap.lmin,
-        "lpeak": snap.lpeak,
-        "ln1": snap.ln1,
-        "ln2": snap.ln2,
-        "raw_payload": snap.raw_payload,
-    }
-
-
-class DeviceMonitor:
-    """Owns a single DOD poll loop for one device and fans each snapshot out to
-    all subscribers. Runs while it has at least one browser subscriber OR the
-    server-side keep-alive (alerting) flag is set."""
-
-    def __init__(self, unit_id: str):
-        self.unit_id = unit_id
-        self._subscribers: Set[asyncio.Queue] = set()
-        self._keepalive = False
-        self._task: Optional[asyncio.Task] = None
-        self._lock = asyncio.Lock()
-        self._last_payload: Optional[dict] = None  # replayed to new subscribers
-        self._consec_fail = 0
-        self._reachable = True  # last broadcast reachability (for transition frames)
-        self._cached_state: Optional[str] = None  # run state, refreshed periodically
-        self._last_state_refresh = 0.0
-        self._last_trail_store = 0.0  # downsample throttle for the backfill trail
-
-    @property
-    def running(self) -> bool:
-        return self._task is not None and not self._task.done()
-
-    def subscriber_count(self) -> int:
-        return len(self._subscribers)
-
-    def _has_demand(self) -> bool:
-        return bool(self._subscribers) or self._keepalive
-
-    def _ensure_task(self) -> None:
-        if self._task is None or self._task.done():
-            self._task = asyncio.create_task(self._run())
-
-    async def subscribe(self) -> asyncio.Queue:
-        q: asyncio.Queue = asyncio.Queue(maxsize=5)
-        async with self._lock:
-            self._subscribers.add(q)
-            # Replay the last frame so a client connecting mid-stream sees data
-            # (or the current 'unreachable' state) immediately, not after a poll.
-            if self._last_payload is not None:
-                try:
-                    q.put_nowait(self._last_payload)
-                except asyncio.QueueFull:
-                    pass
-            self._ensure_task()
-        return q
-
-    async def unsubscribe(self, q: asyncio.Queue) -> None:
-        async with self._lock:
-            self._subscribers.discard(q)
-
-    async def set_keepalive(self, on: bool) -> None:
-        async with self._lock:
-            self._keepalive = on
-            if on:
-                self._ensure_task()
-
-    async def _run(self) -> None:
-        logger.info(f"[MONITOR] {self.unit_id}: feed started")
-        loop = asyncio.get_running_loop()
-        last_send = loop.time()
-        try:
-            while self._has_demand():
-                snap, mst = await self._poll_once()
-                if snap is not None:
-                    if not self._reachable:
-                        # Recovered from an outage — clear the connectivity alert.
-                        try:
-                            await alert_evaluator.device_online(self.unit_id)
-                        except Exception as e:
-                            logger.warning(f"[MONITOR] {self.unit_id}: online alert failed: {e}")
-                    self._consec_fail = 0
-                    self._reachable = True
-                    payload = _snapshot_payload(snap, self.unit_id, mst)
-                    payload["feed_status"] = "ok"
-                    self._broadcast(payload)
-                    last_send = loop.time()
-                    try:
-                        await alert_evaluator.evaluate(self.unit_id, snap)
-                    except Exception as e:
-                        logger.warning(f"[MONITOR] {self.unit_id}: alert eval failed: {e}")
-                else:
-                    # Tell clients the device went offline — once, on transition, after a
-                    # few failures so a momentary blip doesn't flap the UI. Same edge
-                    # raises the device-offline alert.
-                    self._consec_fail += 1
-                    if self._reachable and self._consec_fail >= 3:
-                        self._reachable = False
-                        self._broadcast({
-                            "unit_id": self.unit_id,
-                            "timestamp": datetime.utcnow().isoformat(),
-                            "feed_status": "unreachable",
-                        })
-                        last_send = loop.time()
-                        try:
-                            await alert_evaluator.device_offline(self.unit_id)
-                        except Exception as e:
-                            logger.warning(f"[MONITOR] {self.unit_id}: offline alert failed: {e}")
-
-                # Heartbeat: during quiet/offline stretches, send a keepalive so an
-                # idle WS isn't dropped by a reverse proxy. Not cached (new subscribers
-                # should still get the last real frame, not a heartbeat).
-                if loop.time() - last_send >= MONITOR_HEARTBEAT_S:
-                    self._broadcast({
-                        "unit_id": self.unit_id,
-                        "timestamp": datetime.utcnow().isoformat(),
-                        "feed_status": "ok" if self._reachable else "unreachable",
-                        "heartbeat": True,
-                    }, cache=False)
-                    last_send = loop.time()
-
-                await asyncio.sleep(self._next_delay())
-        finally:
-            logger.info(f"[MONITOR] {self.unit_id}: feed stopped")
-
-    def _next_delay(self) -> float:
-        """Inter-poll delay: exponential backoff while unreachable, full-rate while a
-        browser is watching, relaxed cadence when the feed is keepalive-only."""
-        if self._consec_fail > 0:
-            shift = min(self._consec_fail - 1, 6)  # cap growth at 2**6 = 64x base
-            delay = min(MONITOR_BACKOFF_BASE_S * (2 ** shift), MONITOR_BACKOFF_MAX_S)
-            if self._subscribers:
-                delay = min(delay, MONITOR_BACKOFF_WATCHED_MAX_S)
-            return delay
-        if self._subscribers:
-            return MONITOR_POLL_INTERVAL       # a browser is watching — smooth chart
-        return MONITOR_IDLE_POLL_INTERVAL      # keepalive-only (alerting) — save data
-
-    async def _poll_once(self):
-        """One DOD poll: read, persist, return (snapshot, measurement_start_iso)."""
-        db = SessionLocal()
-        try:
-            cfg = db.query(NL43Config).filter_by(unit_id=self.unit_id).first()
-            if not cfg or not cfg.tcp_enabled:
-                return None, None
-            client = NL43Client(
-                cfg.host, cfg.tcp_port,
-                ftp_username=cfg.ftp_username, ftp_password=cfg.ftp_password,
-                ftp_port=cfg.ftp_port or 21,
-            )
-            # Refresh the run state only every MONITOR_STATE_REFRESH_S; reuse the
-            # cached state otherwise so most polls send just DOD? (one rate-limited
-            # command) instead of DOD? + Measure?.
-            now = asyncio.get_running_loop().time()
-            refresh_state = (self._cached_state is None
-                             or now - self._last_state_refresh >= MONITOR_STATE_REFRESH_S)
-            snap = await client.request_dod(
-                measurement_state=None if refresh_state else self._cached_state
-            )
-            if refresh_state:
-                self._cached_state = snap.measurement_state
-                self._last_state_refresh = now
-            snap.unit_id = self.unit_id
-            persist_snapshot(snap, db)
-            db.commit()
-            # Append to the downsampled backfill trail (~one row per TRAIL_SAMPLE_S).
-            if now - self._last_trail_store >= TRAIL_SAMPLE_S:
-                self._last_trail_store = now
-                self._store_trail(snap, db)
-            status = db.query(NL43Status).filter_by(unit_id=self.unit_id).first()
-            mst = (status.measurement_start_time.isoformat()
-                   if status and status.measurement_start_time else None)
-            return snap, mst
-        except Exception as e:
-            logger.warning(f"[MONITOR] {self.unit_id}: poll failed: {e}")
-            return None, None
-        finally:
-            db.close()
-
-    def _store_trail(self, snap, db) -> None:
-        """Append one downsampled reading to the backfill trail and prune old rows."""
-        from datetime import datetime, timedelta
-        from app.models import NL43Reading
-        try:
-            db.add(NL43Reading(
-                unit_id=self.unit_id, timestamp=datetime.utcnow(),
-                lp=snap.lp, leq=snap.leq, lmax=snap.lmax, ln1=snap.ln1, ln2=snap.ln2,
-            ))
-            cutoff = datetime.utcnow() - timedelta(hours=TRAIL_RETENTION_HOURS)
-            db.query(NL43Reading).filter(
-                NL43Reading.unit_id == self.unit_id,
-                NL43Reading.timestamp < cutoff,
-            ).delete()
-            db.commit()
-        except Exception as e:
-            logger.warning(f"[MONITOR] {self.unit_id}: trail store failed: {e}")
-
-    def _broadcast(self, payload: dict, cache: bool = True) -> None:
-        if cache:
-            self._last_payload = payload  # replayed to new subscribers
-        for q in list(self._subscribers):
-            try:
-                q.put_nowait(payload)
-            except asyncio.QueueFull:
-                # Slow consumer — drop this frame rather than stall the whole feed.
-                pass
-
-
-class MonitorManager:
-    """Registry of per-device monitors (one per unit_id)."""
-
-    def __init__(self):
-        self._monitors: Dict[str, DeviceMonitor] = {}
-        self._lock = asyncio.Lock()
-
-    async def get(self, unit_id: str) -> DeviceMonitor:
-        async with self._lock:
-            m = self._monitors.get(unit_id)
-            if m is None:
-                m = DeviceMonitor(unit_id)
-                self._monitors[unit_id] = m
-            return m
-
-    def is_active(self, unit_id: str) -> bool:
-        """True if this unit has a running monitor feed (so the background poller
-        can skip it — the monitor already polls it more often)."""
-        m = self._monitors.get(unit_id)
-        return m is not None and m.running
-
-    def status(self) -> dict:
-        return {
-            uid: {
-                "running": m.running,
-                "subscribers": m.subscriber_count(),
-                "keepalive": m._keepalive,
-                "reachable": m._reachable,
-                # what cadence the loop is currently using, for observability
-                "mode": ("backoff" if m._consec_fail > 0
-                         else "watched" if m._subscribers
-                         else "idle"),
-            }
-            for uid, m in self._monitors.items()
-        }
-
-
-# Module-level singleton
-monitor_manager = MonitorManager()

app/routers.py

@@ -11,7 +11,7 @@ import os
 import asyncio
 
 from app.database import get_db
-from app.models import NL43Config, NL43Status, AlertRule, AlertEvent, NL43Reading
+from app.models import NL43Config, NL43Status
 from app.services import NL43Client, persist_snapshot
 
 logger = logging.getLogger(__name__)
@@ -121,392 +121,6 @@ async def flush_connection_pool():
     return {"status": "ok", "message": "All cached connections closed"}
 
 
-@router.post("/{unit_id}/disconnect")
-async def disconnect_device(unit_id: str, db: Session = Depends(get_db)):
-    """Cleanly close SLMM's persistent TCP connection to a single device.
-
-    Gracefully closes (TCP FIN + wait_closed) the pooled connection for this
-    device and removes it from the pool, freeing the NL43's single connection
-    slot. Idempotent — a no-op if no connection is currently cached.
-
-    Note: this releases the *idle* pooled connection. It does not interrupt an
-    in-progress DRD stream or an in-flight command (those have the socket
-    checked out of the pool) — close the stream WebSocket to end a live stream.
-    """
-    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
-    if not cfg:
-        raise HTTPException(status_code=404, detail="NL43 config not found")
-
-    from app.services import _connection_pool
-
-    device_key = f"{cfg.host}:{cfg.tcp_port}"
-    had_conn = device_key in _connection_pool.get_stats().get("connections", {})
-
-    await _connection_pool.discard(device_key)
-
-    return {
-        "status": "ok",
-        "unit_id": unit_id,
-        "device_key": device_key,
-        "disconnected": had_conn,
-        "message": "Connection closed" if had_conn else "No cached connection to close",
-    }
-
-
-@router.post("/{unit_id}/deactivate")
-async def deactivate_device(unit_id: str, db: Session = Depends(get_db)):
-    """Make a single unit dormant: stop background polling for it AND drop its
-    connection, freeing the device's connection slot. poll_enabled=False is
-    persisted, so the unit stays dormant across restarts until /activate.
-    """
-    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
-    if not cfg:
-        raise HTTPException(status_code=404, detail="NL43 config not found")
-
-    cfg.poll_enabled = False
-    db.commit()
-
-    from app.services import _connection_pool, _get_device_lock
-
-    device_key = f"{cfg.host}:{cfg.tcp_port}"
-
-    # Wait briefly for any in-flight poll/command to finish (so its connection is
-    # back in the pool), then drop it. If a long-lived stream holds the lock we
-    # don't block forever — discard the pooled connection regardless.
-    lock = await _get_device_lock(device_key)
-    acquired = False
-    try:
-        await asyncio.wait_for(lock.acquire(), timeout=10.0)
-        acquired = True
-    except asyncio.TimeoutError:
-        acquired = False
-    try:
-        await _connection_pool.discard(device_key)
-    finally:
-        if acquired:
-            lock.release()
-
-    return {
-        "status": "ok",
-        "unit_id": unit_id,
-        "poll_enabled": False,
-        "message": "Polling disabled and connection closed for this unit",
-    }
-
-
-@router.post("/{unit_id}/activate")
-async def activate_device(unit_id: str, db: Session = Depends(get_db)):
-    """Resume background polling for a unit previously deactivated."""
-    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
-    if not cfg:
-        raise HTTPException(status_code=404, detail="NL43 config not found")
-
-    cfg.poll_enabled = True
-    db.commit()
-
-    return {
-        "status": "ok",
-        "unit_id": unit_id,
-        "poll_enabled": True,
-        "message": "Polling enabled for this unit",
-    }
-
-
-@router.get("/_system/status")
-async def system_status():
-    """Report whether this SLMM instance is actively polling or in standby."""
-    from app.background_poller import poller
-    from app.services import _connection_pool
-    return {
-        "status": "ok",
-        "mode": "active" if poller.is_active() else "standby",
-        "polling_active": poller.is_active(),
-        "active_connections": _connection_pool.get_stats().get("active_connections", 0),
-    }
-
-
-@router.post("/_system/standby")
-async def system_standby():
-    """Put this SLMM instance into standby: stop polling ALL devices and release
-    every connection, so it stops occupying device slots (e.g. so a prod instance
-    can take over). Runtime-only — on restart the instance returns to its
-    SLMM_POLLING_ENABLED default.
-    """
-    from app.background_poller import poller
-    await poller.set_active(False)
-    return {"status": "ok", "mode": "standby",
-            "message": "Polling stopped and all device connections released"}
-
-
-@router.post("/_system/resume")
-async def system_resume():
-    """Resume polling after standby (global)."""
-    from app.background_poller import poller
-    await poller.set_active(True)
-    return {"status": "ok", "mode": "active", "message": "Polling resumed"}
-
-
-# ============================================================================
-# LIVE MONITOR (fan-out) — one DOD feed per device, broadcast to many clients
-# ============================================================================
-
-@router.websocket("/{unit_id}/monitor")
-async def monitor_stream(websocket: WebSocket, unit_id: str):
-    """Subscribe a browser to the device's shared 1 Hz DOD feed.
-
-    Any number of clients can attach without each opening its own device
-    connection (one poll loop per device, fanned out). Same JSON shape as the
-    DRD stream, but DOD-sourced so it includes ln1/ln2 (L1/L10).
-    """
-    await websocket.accept()
-    from app.monitor import monitor_manager
-
-    monitor = await monitor_manager.get(unit_id)
-    queue = await monitor.subscribe()
-    logger.info(f"Monitor subscriber attached for {unit_id} ({monitor.subscriber_count()} total)")
-
-    async def _watch_disconnect():
-        # Completes when the client disconnects, so an idle feed (no data) still
-        # detects the drop and we don't leak a subscription that keeps the device
-        # feed (and its connection) alive.
-        try:
-            while True:
-                msg = await websocket.receive()
-                if msg.get("type") == "websocket.disconnect":
-                    return
-        except Exception:
-            return
-
-    gone = asyncio.ensure_future(_watch_disconnect())
-    try:
-        while not gone.done():
-            try:
-                payload = await asyncio.wait_for(queue.get(), timeout=1.0)
-            except asyncio.TimeoutError:
-                continue  # re-check gone.done()
-            if gone.done():
-                break  # client disconnected while we waited — don't send into a closing socket
-            await websocket.send_json(payload)
-    except WebSocketDisconnect:
-        logger.info(f"Monitor subscriber disconnected for {unit_id}")
-    except Exception as e:
-        # A frame that races the close (client vanished mid-send) surfaces as
-        # "Unexpected ASGI message 'websocket.send' after ... websocket.close".
-        # That's expected on disconnect (the portal closes the socket on every tab
-        # switch), not an error — log it quietly.
-        msg = str(e)
-        if "after sending" in msg or "websocket.close" in msg or "response already completed" in msg:
-            logger.debug(f"Monitor stream for {unit_id} closed mid-send (client gone)")
-        else:
-            logger.warning(f"Monitor stream error for {unit_id}: {e}")
-    finally:
-        gone.cancel()
-        await monitor.unsubscribe(queue)
-
-
-@router.post("/{unit_id}/monitor/start")
-async def monitor_start(unit_id: str, db: Session = Depends(get_db)):
-    """Enable 24/7 keepalive monitoring: persist monitor_enabled and start the feed
-    now, so alerting evaluates continuously even with no viewer. Survives restarts
-    (auto-started on boot from the persisted flag)."""
-    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
-    if cfg:
-        cfg.monitor_enabled = True
-        db.commit()
-    from app.monitor import monitor_manager
-    monitor = await monitor_manager.get(unit_id)
-    await monitor.set_keepalive(True)
-    return {"status": "ok", "unit_id": unit_id, "monitor_enabled": True, "running": monitor.running}
-
-
-@router.post("/{unit_id}/monitor/stop")
-async def monitor_stop(unit_id: str, db: Session = Depends(get_db)):
-    """Disable keepalive monitoring: persist monitor_enabled=False and drop the
-    keepalive (the feed stops once no browser subscribers remain)."""
-    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
-    if cfg:
-        cfg.monitor_enabled = False
-        db.commit()
-    from app.monitor import monitor_manager
-    monitor = await monitor_manager.get(unit_id)
-    await monitor.set_keepalive(False)
-    return {"status": "ok", "unit_id": unit_id, "monitor_enabled": False}
-
-
-@router.get("/_monitor/status")
-async def monitor_status():
-    """Status of every device monitor (running, subscriber count, keep-alive)."""
-    from app.monitor import monitor_manager
-    return {"status": "ok", "monitors": monitor_manager.status()}
-
-
-@router.get("/{unit_id}/history")
-def get_monitor_history(unit_id: str, hours: float = 2.0, db: Session = Depends(get_db)):
-    """Recent downsampled monitor readings (the DOD trail) for the live-chart
-    backfill. Viewing only — NOT the FTP report data."""
-    from datetime import timedelta
-    hours = max(0.1, min(hours, 48.0))
-    cutoff = datetime.utcnow() - timedelta(hours=hours)
-    rows = (db.query(NL43Reading)
-            .filter(NL43Reading.unit_id == unit_id, NL43Reading.timestamp >= cutoff)
-            .order_by(NL43Reading.timestamp.asc()).all())
-    return {
-        "status": "ok",
-        "unit_id": unit_id,
-        "hours": hours,
-        "count": len(rows),
-        "readings": [
-            {
-                "timestamp": r.timestamp.isoformat() if r.timestamp else None,
-                "lp": r.lp, "leq": r.leq, "lmax": r.lmax, "ln1": r.ln1, "ln2": r.ln2,
-            }
-            for r in rows
-        ],
-    }
-
-
-# ============================================================================
-# ALERTS — threshold rules + fired events
-# ============================================================================
-
-class AlertRulePayload(BaseModel):
-    name: str = "Alert"
-    metric: str = "lp"            # lp/leq/lmax/lmin/lpeak/ln1/ln2
-    comparison: str = "above"     # above | below
-    threshold_db: float
-    duration_s: int = 0           # sustained seconds before firing (0 = instant)
-    clear_margin_db: float = 2.0  # hysteresis band
-    cooldown_s: int = 300
-    schedule_start: str | None = None  # "HH:MM" local; null = always
-    schedule_end: str | None = None
-    schedule_days: str | None = None   # CSV of 0-6 (Mon=0); null = every day
-    channels: str = "log"
-    recipients: str | None = None
-    enabled: bool = True
-
-
-def _rule_dict(r: AlertRule) -> dict:
-    return {
-        "id": r.id, "unit_id": r.unit_id, "name": r.name, "metric": r.metric,
-        "comparison": r.comparison, "threshold_db": r.threshold_db,
-        "duration_s": r.duration_s, "clear_margin_db": r.clear_margin_db,
-        "cooldown_s": r.cooldown_s, "schedule_start": r.schedule_start,
-        "schedule_end": r.schedule_end, "schedule_days": r.schedule_days,
-        "channels": r.channels, "recipients": r.recipients, "enabled": r.enabled,
-    }
-
-
-def _event_dict(e: AlertEvent) -> dict:
-    return {
-        "id": e.id, "rule_id": e.rule_id, "unit_id": e.unit_id,
-        "rule_name": e.rule_name, "metric": e.metric, "threshold_db": e.threshold_db,
-        "onset_at": e.onset_at.isoformat() if e.onset_at else None,
-        "onset_value": e.onset_value, "peak_value": e.peak_value,
-        "clear_at": e.clear_at.isoformat() if e.clear_at else None,
-        "status": e.status,
-        "acknowledged_at": e.acknowledged_at.isoformat() if e.acknowledged_at else None,
-        "acknowledged_by": e.acknowledged_by,
-    }
-
-
-async def _sync_keepalive_to_rules(unit_id: str, db: Session):
-    """Keep a unit's monitor running while it has enabled alert rules, so the
-    evaluator runs 24/7 even with no browser watching. Turns keepalive ON (and
-    persists monitor_enabled so it survives a restart via the boot auto-start)
-    when enabled rules exist; never turns it OFF — a device may be kept alive for
-    other reasons, so operators control that on /admin/slmm."""
-    has_enabled = (db.query(AlertRule)
-                   .filter_by(unit_id=unit_id, enabled=True).first() is not None)
-    if not has_enabled:
-        return
-    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
-    if cfg and not cfg.monitor_enabled:
-        cfg.monitor_enabled = True
-        db.commit()
-    from app.monitor import monitor_manager
-    m = await monitor_manager.get(unit_id)
-    await m.set_keepalive(True)
-
-
-def _reset_rule_runtime(unit_id: str, rule_id: int, db: Session):
-    """After a rule edit/delete: drop its evaluator state machine and close any open
-    event, so a stale 'active' phase doesn't mis-evaluate against the new config and
-    the client portal doesn't stay 'in alarm' on a rule that changed or is gone."""
-    from app.alerts import alert_evaluator
-    alert_evaluator.forget_rule(unit_id, rule_id)
-    now = datetime.utcnow()
-    for evt in db.query(AlertEvent).filter_by(unit_id=unit_id, rule_id=rule_id, status="active").all():
-        evt.clear_at = now
-        evt.status = "cleared"
-    db.commit()
-
-
-@router.post("/{unit_id}/alerts/rules")
-async def create_alert_rule(unit_id: str, payload: AlertRulePayload, db: Session = Depends(get_db)):
-    rule = AlertRule(unit_id=unit_id, **payload.model_dump())
-    db.add(rule)
-    db.commit()
-    db.refresh(rule)
-    from app.alerts import alert_evaluator
-    alert_evaluator.invalidate(unit_id)
-    await _sync_keepalive_to_rules(unit_id, db)
-    return {"status": "ok", "rule": _rule_dict(rule)}
-
-
-@router.get("/{unit_id}/alerts/rules")
-def list_alert_rules(unit_id: str, db: Session = Depends(get_db)):
-    rules = db.query(AlertRule).filter_by(unit_id=unit_id).all()
-    return {"status": "ok", "rules": [_rule_dict(r) for r in rules]}
-
-
-@router.put("/{unit_id}/alerts/rules/{rule_id}")
-async def update_alert_rule(unit_id: str, rule_id: int, payload: AlertRulePayload, db: Session = Depends(get_db)):
-    rule = db.query(AlertRule).filter_by(id=rule_id, unit_id=unit_id).first()
-    if not rule:
-        raise HTTPException(status_code=404, detail="Alert rule not found")
-    for field, value in payload.model_dump().items():
-        setattr(rule, field, value)
-    db.commit()
-    db.refresh(rule)
-    from app.alerts import alert_evaluator
-    alert_evaluator.invalidate(unit_id)
-    _reset_rule_runtime(unit_id, rule_id, db)
-    await _sync_keepalive_to_rules(unit_id, db)
-    return {"status": "ok", "rule": _rule_dict(rule)}
-
-
-@router.delete("/{unit_id}/alerts/rules/{rule_id}")
-async def delete_alert_rule(unit_id: str, rule_id: int, db: Session = Depends(get_db)):
-    rule = db.query(AlertRule).filter_by(id=rule_id, unit_id=unit_id).first()
-    if not rule:
-        raise HTTPException(status_code=404, detail="Alert rule not found")
-    db.delete(rule)
-    db.commit()
-    from app.alerts import alert_evaluator
-    alert_evaluator.invalidate(unit_id)
-    _reset_rule_runtime(unit_id, rule_id, db)   # close its open event so the portal doesn't stay red
-    await _sync_keepalive_to_rules(unit_id, db)  # no-op if no enabled rules remain
-    return {"status": "ok", "deleted": rule_id}
-
-
-@router.get("/{unit_id}/alerts/events")
-def list_alert_events(unit_id: str, limit: int = 50, db: Session = Depends(get_db)):
-    events = (db.query(AlertEvent).filter_by(unit_id=unit_id)
-              .order_by(AlertEvent.onset_at.desc()).limit(limit).all())
-    return {"status": "ok", "events": [_event_dict(e) for e in events]}
-
-
-@router.post("/{unit_id}/alerts/events/{event_id}/ack")
-def ack_alert_event(unit_id: str, event_id: int, by: str | None = None, db: Session = Depends(get_db)):
-    evt = db.query(AlertEvent).filter_by(id=event_id, unit_id=unit_id).first()
-    if not evt:
-        raise HTTPException(status_code=404, detail="Alert event not found")
-    evt.acknowledged_at = datetime.utcnow()
-    evt.acknowledged_by = by
-    db.commit()
-    return {"status": "ok", "acknowledged": event_id}
-
-
 # ============================================================================
 # GLOBAL POLLING STATUS ENDPOINT (must be before /{unit_id} routes)
 # ============================================================================
@@ -583,7 +197,6 @@ def get_roster(db: Session = Depends(get_db)):
             "web_enabled": cfg.web_enabled,
             "poll_enabled": cfg.poll_enabled,
             "poll_interval_seconds": cfg.poll_interval_seconds,
-            "monitor_enabled": cfg.monitor_enabled,
             "status": None
         }
 
@@ -832,14 +445,11 @@ def get_status(unit_id: str, db: Session = Depends(get_db)):
             "unit_id": unit_id,
             "last_seen": status.last_seen.isoformat() if status.last_seen else None,
             "measurement_state": status.measurement_state,
-            "measurement_start_time": status.measurement_start_time.isoformat() if status.measurement_start_time else None,
             "lp": status.lp,
             "leq": status.leq,
             "lmax": status.lmax,
             "lmin": status.lmin,
             "lpeak": status.lpeak,
-            "ln1": status.ln1,
-            "ln2": status.ln2,
             "battery_level": status.battery_level,
             "power_source": status.power_source,
             "sd_remaining_mb": status.sd_remaining_mb,
@@ -862,8 +472,6 @@ class StatusPayload(BaseModel):
     lmax: str | None = None
     lmin: str | None = None
     lpeak: str | None = None
-    ln1: str | None = None
-    ln2: str | None = None
     battery_level: str | None = None
     power_source: str | None = None
     sd_remaining_mb: str | None = None
@@ -891,14 +499,11 @@ def upsert_status(unit_id: str, payload: StatusPayload, db: Session = Depends(ge
             "unit_id": unit_id,
             "last_seen": status.last_seen.isoformat(),
             "measurement_state": status.measurement_state,
-            "measurement_start_time": status.measurement_start_time.isoformat() if status.measurement_start_time else None,
             "lp": status.lp,
             "leq": status.leq,
             "lmax": status.lmax,
             "lmin": status.lmin,
             "lpeak": status.lpeak,
-            "ln1": status.ln1,
-            "ln2": status.ln2,
             "battery_level": status.battery_level,
             "power_source": status.power_source,
             "sd_remaining_mb": status.sd_remaining_mb,
@@ -939,7 +544,7 @@ async def start_measurement(unit_id: str, db: Session = Depends(get_db)):
             db.expire_all()
             status = db.query(NL43Status).filter_by(unit_id=unit_id).first()
             logger.info(f"State check: measurement_state={status.measurement_state if status else 'None'}, start_time={status.measurement_start_time if status else 'None'}")
-            if status and status.measurement_state in ("Start", "Measure") and status.measurement_start_time:
+            if status and status.measurement_state == "Measure" and status.measurement_start_time:
                 logger.info(f"✓ Measurement state confirmed for {unit_id} with start time {status.measurement_start_time}")
                 break
 
@@ -1600,8 +1205,6 @@ async def stream_live(websocket: WebSocket, unit_id: str):
                     "lmax": snap.lmax,  # Maximum level
                     "lmin": snap.lmin,  # Minimum level
                     "lpeak": snap.lpeak,  # Peak level
-                    "ln1": snap.ln1,    # LN1 percentile (L1/L10 contract); null on DRD stream
-                    "ln2": snap.ln2,    # LN2 percentile; null on DRD stream
                     "raw_payload": snap.raw_payload,
                 })
             except Exception as e:
@@ -2273,8 +1876,6 @@ async def run_diagnostics(unit_id: str, db: Session = Depends(get_db)):
             "lmax": status.lmax,
             "lmin": status.lmin,
             "lpeak": status.lpeak,
-            "ln1": status.ln1,
-            "ln2": status.ln2,
             "battery_level": status.battery_level,
             "power_source": status.power_source,
             "sd_remaining_mb": status.sd_remaining_mb,

app/services.py

@@ -46,8 +46,6 @@ class NL43Snapshot:
     lmax: Optional[str] = None  # Maximum level
     lmin: Optional[str] = None  # Minimum level
     lpeak: Optional[str] = None  # Peak level
-    ln1: Optional[str] = None  # Percentile slot LN1 (configurable; device default L5, contract L1)
-    ln2: Optional[str] = None  # Percentile slot LN2 (configurable; device default L10)
     battery_level: Optional[str] = None
     power_source: Optional[str] = None
     sd_remaining_mb: Optional[str] = None
@@ -71,27 +69,10 @@ def persist_snapshot(s: NL43Snapshot, db: Session):
 
         logger.info(f"State transition check for {s.unit_id}: '{previous_state}' -> '{new_state}'")
 
-        # The device reports "Start" while measuring; the DOD path uses that string,
-        # but the DRD stream path tags snapshots "Measure" (and the DOD fallback also
-        # uses "Measure"). Treat ALL of these as "measuring" — otherwise opening and
-        # closing the live stream flips state "Start"->"Measure"->"Start", which the
-        # old equality check misread as stop-then-start and reset measurement_start_time.
-        #
-        # Also: only act on RECOGNIZED states. A buffer desync on the shared connection
-        # (e.g. right after a DRD/DOD test) can make a Measure? read return a stray,
-        # garbled value; treating that as "not measuring" produced constant phantom
-        # "STOPPED -> STARTED" log pairs and reset the timer. Ignore unknown reads.
-        MEASURING_STATES = {"Start", "Measure"}
-        STOPPED_STATES = {"Stop"}
-        was_measuring = previous_state in MEASURING_STATES
-
-        if new_state in MEASURING_STATES:
-            is_measuring = True
-        elif new_state in STOPPED_STATES:
-            is_measuring = False
-        else:
-            logger.warning(f"Ignoring unrecognized measurement state for {s.unit_id}: {new_state!r}")
-            is_measuring = was_measuring  # garbled/unknown read — no transition
+        # Device returns "Start" when measuring, "Stop" when stopped
+        # Normalize to previous behavior for backward compatibility
+        is_measuring = new_state == "Start"
+        was_measuring = previous_state == "Start"
 
         if not was_measuring and is_measuring:
             # Measurement just started - record the start time
@@ -114,9 +95,6 @@ def persist_snapshot(s: NL43Snapshot, db: Session):
             except Exception:
                 pass
 
-        # Only persist a recognized state so one garbled read can't poison the next
-        # transition check (which would manufacture the phantom STOPPED/STARTED pair).
-        if new_state in MEASURING_STATES or new_state in STOPPED_STATES:
         row.measurement_state = new_state
         row.counter = s.counter
         row.lp = s.lp
@@ -124,8 +102,6 @@ def persist_snapshot(s: NL43Snapshot, db: Session):
         row.lmax = s.lmax
         row.lmin = s.lmin
         row.lpeak = s.lpeak
-        row.ln1 = s.ln1
-        row.ln2 = s.ln2
         row.battery_level = s.battery_level
         row.power_source = s.power_source
         row.sd_remaining_mb = s.sd_remaining_mb
@@ -680,12 +656,10 @@ class NL43Client:
         else:
             raise ValueError(f"Unknown result code: {result_code}")
 
-    async def request_dod(self, measurement_state: Optional[str] = None) -> NL43Snapshot:
+    async def request_dod(self) -> NL43Snapshot:
         """Request DOD (Data Output Display) snapshot from device.
 
-        Returns parsed measurement data from the device display. Pass
-        measurement_state to reuse a cached run state and skip the extra Measure?
-        round-trip (the state changes rarely); leave it None to query it.
+        Returns parsed measurement data from the device display.
         """
         # _send_command now handles result code validation and returns the data line
         resp = await self._send_command("DOD?\r\n")
@@ -708,10 +682,7 @@ class NL43Client:
 
         logger.info(f"Parsed {len(parts)} data points from DOD response")
 
-        # DOD doesn't include the run state. Query it only when not supplied by the
-        # caller — the monitor passes a cached state most cycles and refreshes it
-        # occasionally, avoiding a second rate-limited command on every poll.
-        if measurement_state is None:
+        # Query actual measurement state (DOD doesn't include this information)
         try:
             measurement_state = await self.get_measurement_state()
         except Exception as e:
@@ -720,29 +691,22 @@ class NL43Client:
 
         snap = NL43Snapshot(unit_id="", raw_payload=resp, measurement_state=measurement_state)
 
-        # Parse DOD positional fields. DOD's layout is DIFFERENT from DRD: it has NO
-        # leading counter and it includes LE plus LN1–LN5. The device returns 4 channels
-        # of 16 fields each — [Lp, Leq, LE, Lmax, Lmin, LN1, LN2, LN3, LN4, LN5, Lpeak,
-        # LIeq, Leq_mov, Ltm5, over, under] — and channel 1 (parts[0:16]) is the main
-        # display. The previous code reused the DRD map (treating parts[0] as a counter),
-        # which shifted everything: Lp was reported as the counter, Leq as Lp, LE as Leq,
-        # and LN1 as Lpeak (you could spot it because "Lpeak" came out < Lmax).
+        # Parse known positions (based on NL43 communication guide - DRD format)
+        # DRD format: d0=counter, d1=Lp, d2=Leq, d3=Lmax, d4=Lmin, d5=Lpeak, d6=LIeq, ...
         try:
+            # Capture d0 (counter) for timer synchronization
             if len(parts) >= 1:
-                snap.lp = parts[0]      # Lp: instantaneous sound pressure level
+                snap.counter = parts[0]  # d0: Measurement interval counter (1-600)
             if len(parts) >= 2:
-                snap.leq = parts[1]     # Leq: equivalent continuous level
-            # parts[2] = LE (sound exposure level) — not currently surfaced
+                snap.lp = parts[1]     # d1: Instantaneous sound pressure level
+            if len(parts) >= 3:
+                snap.leq = parts[2]    # d2: Equivalent continuous sound level
             if len(parts) >= 4:
-                snap.lmax = parts[3]    # Lmax
+                snap.lmax = parts[3]   # d3: Maximum level
             if len(parts) >= 5:
-                snap.lmin = parts[4]    # Lmin
-            if len(parts) >= 11:
-                snap.lpeak = parts[10]  # Lpeak  (parts[5] is LN1, NOT Lpeak)
+                snap.lmin = parts[4]   # d4: Minimum level
             if len(parts) >= 6:
-                snap.ln1 = parts[5]     # LN1 percentile slot (device default L5; contract L1)
-            if len(parts) >= 7:
-                snap.ln2 = parts[6]     # LN2 percentile slot (device default L10)
+                snap.lpeak = parts[5]  # d5: Peak level
         except (IndexError, ValueError) as e:
             logger.warning(f"Error parsing DOD data points: {e}")
 
@@ -932,20 +896,15 @@ class NL43Client:
         # Acquire per-device lock - held for entire streaming session
         device_lock = await _get_device_lock(self.device_key)
         async with device_lock:
+            # Evict any cached connection — streaming needs its own dedicated socket
+            await _connection_pool.discard(self.device_key)
             await self._enforce_rate_limit()
 
             logger.info(f"Starting DRD stream for {self.device_key}")
 
-            # Reuse the pooled connection instead of discard()+reopen. The NL43
-            # allows only ONE TCP connection at a time, and on a cellular link the
-            # device does not free its single slot fast enough for an immediate
-            # reconnect — so a fresh connect times out (the DRD stream failure).
-            # The per-device lock is held for the whole session, so it already
-            # blocks the poller; reusing the warm socket keeps us at exactly one
-            # connection and lets the stream start on the slot commands already use.
             try:
-                reader, writer, from_cache = await _connection_pool.acquire(
-                    self.device_key, self.host, self.port, self.timeout
+                reader, writer = await _connection_pool._open_connection(
+                    self.host, self.port, self.timeout
                 )
             except ConnectionError:
                 logger.error(f"DRD stream connection failed to {self.device_key}")
@@ -1022,20 +981,16 @@ class NL43Client:
                         break
 
             finally:
-                # Stop streaming on the device (SUB = 0x1A), then return the warm
-                # connection to the pool so subsequent commands reuse this single
-                # socket instead of opening a second one. release() returns healthy
-                # sockets to the pool and closes dead ones; the next acquire()
-                # drains any residual stop output before reuse.
+                # Send SUB character to stop streaming
                 try:
                     writer.write(b"\x1A")
                     await writer.drain()
                 except Exception:
                     pass
 
-                await _connection_pool.release(
-                    self.device_key, reader, writer, self.host, self.port
-                )
+                writer.close()
+                with contextlib.suppress(Exception):
+                    await writer.wait_closed()
 
                 logger.info(f"DRD stream ended for {self.device_key}")
 

migrate_add_ln_percentiles.py

@@ -1,58 +0,0 @@
-#!/usr/bin/env python3
-"""
-Migration script to add ln1 and ln2 percentile columns to the nl43_status table.
-
-The NL-43 DOD response carries percentile slots LN1-LN5; the live SLM display
-(Terra-View) shows two of them (default L1/L10). This adds storage for the two
-surfaced slots. Run once per database to update existing schema.
-"""
-
-import sqlite3
-import sys
-from pathlib import Path
-
-DB_PATH = Path(__file__).parent / "data" / "slmm.db"
-
-
-def migrate():
-    """Add ln1 and ln2 columns to the nl43_status table."""
-
-    if not DB_PATH.exists():
-        print(f"Database not found at {DB_PATH}")
-        print("No migration needed - database will be created with new schema")
-        return
-
-    conn = sqlite3.connect(DB_PATH)
-    cursor = conn.cursor()
-
-    try:
-        cursor.execute("PRAGMA table_info(nl43_status)")
-        columns = [row[1] for row in cursor.fetchall()]
-
-        if "ln1" in columns and "ln2" in columns:
-            print("✓ ln1/ln2 columns already exist, no migration needed")
-            return
-
-        if "ln1" not in columns:
-            print("Adding ln1 column...")
-            cursor.execute("ALTER TABLE nl43_status ADD COLUMN ln1 TEXT")
-            print("✓ Added ln1 column")
-
-        if "ln2" not in columns:
-            print("Adding ln2 column...")
-            cursor.execute("ALTER TABLE nl43_status ADD COLUMN ln2 TEXT")
-            print("✓ Added ln2 column")
-
-        conn.commit()
-        print("\n✓ Migration completed successfully!")
-
-    except Exception as e:
-        conn.rollback()
-        print(f"✗ Migration failed: {e}", file=sys.stderr)
-        sys.exit(1)
-    finally:
-        conn.close()
-
-
-if __name__ == "__main__":
-    migrate()

migrate_add_monitor_enabled.py

@@ -1,48 +0,0 @@
-#!/usr/bin/env python3
-"""
-Migration: add monitor_enabled column to nl43_config.
-
-Controls whether the live fan-out DOD monitor is kept alive 24/7 for a unit
-(which is what makes alerting continuous). Defaults to enabled. Run once per DB.
-"""
-
-import sqlite3
-import sys
-from pathlib import Path
-
-DB_PATH = Path(__file__).parent / "data" / "slmm.db"
-
-
-def migrate():
-    if not DB_PATH.exists():
-        print(f"Database not found at {DB_PATH}")
-        print("No migration needed - database will be created with new schema")
-        return
-
-    conn = sqlite3.connect(DB_PATH)
-    cursor = conn.cursor()
-    try:
-        cursor.execute("PRAGMA table_info(nl43_config)")
-        columns = [row[1] for row in cursor.fetchall()]
-
-        if "monitor_enabled" in columns:
-            print("✓ monitor_enabled column already exists, no migration needed")
-            return
-
-        print("Adding monitor_enabled column (default enabled)...")
-        # SQLite stores booleans as 0/1; default 1 = enabled.
-        cursor.execute("ALTER TABLE nl43_config ADD COLUMN monitor_enabled BOOLEAN DEFAULT 1")
-        conn.commit()
-        print("✓ Added monitor_enabled column")
-        print("\n✓ Migration completed successfully!")
-
-    except Exception as e:
-        conn.rollback()
-        print(f"✗ Migration failed: {e}", file=sys.stderr)
-        sys.exit(1)
-    finally:
-        conn.close()
-
-
-if __name__ == "__main__":
-    migrate()

nl52/__init__.py

@@ -0,0 +1,23 @@
+"""
+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

@@ -0,0 +1,160 @@
+#!/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

@@ -0,0 +1,288 @@
+"""
+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

@@ -0,0 +1,172 @@
+"""
+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

@@ -0,0 +1,128 @@
+#!/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_alert_evaluator.py

@@ -1,68 +0,0 @@
-"""
-Synthetic unit test for the alert state machine — no DB, no device.
-
-Drives `_evaluate_step` with a fake clock + a level series and checks that
-onset/clear fire with the right debounce + hysteresis. Run:
-
-    docker compose exec -T slmm python3 test_alert_evaluator.py
-    # or, if app.alerts imports cleanly standalone:  python3 test_alert_evaluator.py
-"""
-
-from types import SimpleNamespace
-from app.alerts import RuleState, _evaluate_step
-
-
-def rule(**kw):
-    base = dict(threshold_db=85.0, duration_s=3, clear_margin_db=2.0, comparison="above")
-    base.update(kw)
-    return SimpleNamespace(**base)
-
-
-def run(series, r):
-    st = RuleState()
-    events = [(now, a) for value, now in series
-              if (a := _evaluate_step(st, value, now, r))]
-    return events, st
-
-
-def main():
-    failures = 0
-
-    def check(label, cond, detail=""):
-        nonlocal failures
-        print(("PASS" if cond else "FAIL"), label, detail)
-        if not cond:
-            failures += 1
-
-    # 1) sustained exceedance -> onset after duration; recovery -> clear after duration
-    r = rule(threshold_db=85, duration_s=3, clear_margin_db=2)
-    ev, _ = run([(80, 0), (86, 1), (87, 2), (88, 3), (88, 4),
-                 (88, 5), (82, 6), (82, 7), (82, 8), (82, 9)], r)
-    onsets = [t for t, a in ev if a == "onset"]
-    clears = [t for t, a in ev if a == "clear"]
-    check("1 sustained onset@4 / clear@9", onsets == [4] and clears == [9], str(ev))
-
-    # 2) brief spike under duration -> no onset (debounce)
-    ev, _ = run([(80, 0), (90, 1), (90, 2), (80, 3), (80, 4)], rule(duration_s=3))
-    check("2 brief spike debounced", ev == [], str(ev))
-
-    # 3) hysteresis: a dip into the margin (below threshold, above threshold-margin)
-    #    does NOT clear
-    r = rule(threshold_db=85, duration_s=0, clear_margin_db=3)
-    ev, st = run([(86, 0), (84, 1), (84, 2), (84, 3)], r)
-    check("3 hysteresis holds ACTIVE", ev == [(0, "onset")] and st.phase == "active",
-          f"{ev} phase={st.phase}")
-
-    # 4) 'below' comparison (device too quiet) -> onset when value < threshold
-    ev, _ = run([(30, 0), (15, 1)], rule(threshold_db=20, duration_s=0,
-                                         clear_margin_db=2, comparison="below"))
-    check("4 below-comparison onset@1", ev == [(1, "onset")], str(ev))
-
-    print()
-    print("ALL PASS" if failures == 0 else f"{failures} FAILURE(S)")
-    return failures
-
-
-if __name__ == "__main__":
-    import sys
-    sys.exit(1 if main() else 0)

test_nl52_protocol.py

@@ -0,0 +1,134 @@
+#!/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)