""" 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()