feat: surface LN1/LN2 (L1/L10) percentiles through SLMM

Completes the SLMM side of the L1/L10 live-display contract. The NL-43's
DOD response carries percentile slots LN1-LN5 (channel 1, parts[5]/[6]);
parse the first two and expose them as ln1/ln2 end to end:

- NL43Snapshot dataclass: ln1/ln2 fields
- NL43Status model: ln1/ln2 columns (+ migrate_add_ln_percentiles.py)
- DOD parser: snap.ln1=parts[5], snap.ln2=parts[6]
- persist_snapshot writes them
- all /status data dicts, StatusPayload, and the DRD stream payload emit
  ln1/ln2 (null on the DRD stream itself, which doesn't carry percentiles)

Labels: device LN1 defaults to L5, not L1 — Terra-View defaults the label
to L1/L10, so the device's Ln1/Ln2 slots must be set to 1%/10% for the
labels to be accurate (dynamic label emission is a follow-up).

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

SLM-stress-test/nl52_usb_probe.py

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

@@ -41,6 +41,8 @@ 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)

app/routers.py

@@ -450,6 +450,8 @@ def get_status(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,
@@ -472,6 +474,8 @@ 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
@@ -504,6 +508,8 @@ def upsert_status(unit_id: str, payload: StatusPayload, db: Session = Depends(ge
             "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,
@@ -1205,6 +1211,8 @@ 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:
@@ -1876,6 +1884,8 @@ 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,6 +46,8 @@ 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
@@ -69,10 +71,16 @@ def persist_snapshot(s: NL43Snapshot, db: Session):
 
         logger.info(f"State transition check for {s.unit_id}: '{previous_state}' -> '{new_state}'")
 
-        # 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"
+        # 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
+        # every single time (the "elapsed time keeps resetting / shows wrong value on
+        # another computer" bug — and each extra viewer made it worse).
+        MEASURING_STATES = {"Start", "Measure"}
+        is_measuring = new_state in MEASURING_STATES
+        was_measuring = previous_state in MEASURING_STATES
 
         if not was_measuring and is_measuring:
             # Measurement just started - record the start time
@@ -102,6 +110,8 @@ 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
@@ -691,22 +701,29 @@ class NL43Client:
 
         snap = NL43Snapshot(unit_id="", raw_payload=resp, measurement_state=measurement_state)
 
-        # 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, ...
+        # 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).
         try:
-            # Capture d0 (counter) for timer synchronization
             if len(parts) >= 1:
-                snap.counter = parts[0]  # d0: Measurement interval counter (1-600)
+                snap.lp = parts[0]      # Lp: instantaneous sound pressure level
             if len(parts) >= 2:
-                snap.lp = parts[1]     # d1: Instantaneous sound pressure level
-            if len(parts) >= 3:
-                snap.leq = parts[2]    # d2: Equivalent continuous sound level
+                snap.leq = parts[1]     # Leq: equivalent continuous level
+            # parts[2] = LE (sound exposure level) — not currently surfaced
             if len(parts) >= 4:
-                snap.lmax = parts[3]   # d3: Maximum level
+                snap.lmax = parts[3]    # Lmax
             if len(parts) >= 5:
-                snap.lmin = parts[4]   # d4: Minimum level
+                snap.lmin = parts[4]    # Lmin
+            if len(parts) >= 11:
+                snap.lpeak = parts[10]  # Lpeak  (parts[5] is LN1, NOT Lpeak)
             if len(parts) >= 6:
-                snap.lpeak = parts[5]  # d5: Peak level
+                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)
         except (IndexError, ValueError) as e:
             logger.warning(f"Error parsing DOD data points: {e}")
 
@@ -896,15 +913,20 @@ 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 = await _connection_pool._open_connection(
-                    self.host, self.port, self.timeout
+                reader, writer, from_cache = await _connection_pool.acquire(
+                    self.device_key, self.host, self.port, self.timeout
                 )
             except ConnectionError:
                 logger.error(f"DRD stream connection failed to {self.device_key}")
@@ -981,16 +1003,20 @@ class NL43Client:
                         break
 
             finally:
-                # Send SUB character to stop streaming
+                # 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.
                 try:
                     writer.write(b"\x1A")
                     await writer.drain()
                 except Exception:
                     pass
 
-                writer.close()
-                with contextlib.suppress(Exception):
-                    await writer.wait_closed()
+                await _connection_pool.release(
+                    self.device_key, reader, writer, self.host, self.port
+                )
 
                 logger.info(f"DRD stream ended for {self.device_key}")
 

migrate_add_ln_percentiles.py

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

nl52/__init__.py

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

nl52/cli.py

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

nl52/client.py

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

nl52/protocol.py

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

nl52/usb_bulk_probe.py

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

test_nl52_protocol.py

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