From 0fcab0d0dc95fa11c66a9a026e94653ffbd061f3 Mon Sep 17 00:00:00 2001 From: serversdown Date: Mon, 8 Jun 2026 01:02:51 +0000 Subject: [PATCH] feat: NL-42/NL-52 serial control layer (USB/RS-232, dependency-free) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a stdlib-only (termios, no pyserial) control layer for the older RION NL-42/NL-52 meters, which have no Ethernet option and speak the same ASCII command family as the NL-43 over RS-232C or USB (virtual COM port). - nl52/protocol.py: I/O-free parsers for DOD? (14 fields) and DRD? (9 fields), result codes, and level tokens. Handles '--.-' disabled-channel nulls and space-padded fixed-width fields. - nl52/client.py: termios SerialPort + NL52Client. Echo-tolerant result-code reading, buffered multi-line reads, NL-52 rate limits (200ms; 1s for DOD?). is_request = "?" in cmd (NL-52 requests can carry a param after '?', e.g. 'System Version?NL'). - nl52/cli.py: bench tool — probe/status/start/stop/monitor/poll/raw — for testing over USB on the dev server with nothing to install. - test_nl52_protocol.py: 45 parser assertions, host-runnable (no hardware). Client exchange logic separately validated via pty loopback. NL-52 quirks vs NL-43: DOD has no Lpeak and no measurement-state (query Measure? separately); DRD streaming requires the NX-42EX option. Not yet: SLMM integration (async transport / serial->TCP bridge for the RX55 PAD-mode path), DB model, validation against real hardware. Co-Authored-By: Claude Opus 4.8 (1M context) --- nl52/__init__.py | 23 ++++ nl52/cli.py | 160 +++++++++++++++++++++++ nl52/client.py | 288 ++++++++++++++++++++++++++++++++++++++++++ nl52/protocol.py | 172 +++++++++++++++++++++++++ test_nl52_protocol.py | 134 ++++++++++++++++++++ 5 files changed, 777 insertions(+) create mode 100644 nl52/__init__.py create mode 100644 nl52/cli.py create mode 100644 nl52/client.py create mode 100644 nl52/protocol.py create mode 100644 test_nl52_protocol.py diff --git a/nl52/__init__.py b/nl52/__init__.py new file mode 100644 index 0000000..21b908a --- /dev/null +++ b/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, +) diff --git a/nl52/cli.py b/nl52/cli.py new file mode 100644 index 0000000..9df1ccd --- /dev/null +++ b/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()) diff --git a/nl52/client.py b/nl52/client.py new file mode 100644 index 0000000..4c0c7df --- /dev/null +++ b/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() diff --git a/nl52/protocol.py b/nl52/protocol.py new file mode 100644 index 0000000..64cb0f9 --- /dev/null +++ b/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 diff --git a/test_nl52_protocol.py b/test_nl52_protocol.py new file mode 100644 index 0000000..60e5f44 --- /dev/null +++ b/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)