diff --git a/bridges/mm_client_check.py b/bridges/mm_client_check.py index 8bd86cb..76d971f 100644 --- a/bridges/mm_client_check.py +++ b/bridges/mm_client_check.py @@ -44,7 +44,11 @@ unit, this will connect and then see nothing — that is contention, not a fault from __future__ import annotations import argparse +import errno +import os +import select import sys +import termios import time from pathlib import Path @@ -52,7 +56,72 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from micromate.client import MicromateClient, _content # noqa: E402 from micromate.protocol import ProtocolError # noqa: E402 -from minimateplus.transport import SerialTransport, TcpTransport # noqa: E402 +from minimateplus.transport import TcpTransport # noqa: E402 + + +class StdlibSerial: + """Raw serial on stdlib `termios` — no pyserial. + + `minimateplus.SerialTransport` needs pyserial, and a bench host is whatever + is to hand. On a PEP 668 distro (Mint 22, Ubuntu 24.04, Debian 12) a plain + `pip install pyserial` is refused outright, so a diagnostic that depends on + it is one you cannot run at the moment you need it. `bridges/mm_link.py` + and `scratch/fake_unit.py` already take this approach; this is the same + ~30 lines, and it means the tool runs on a stock Python 3 anywhere. + + Not a general replacement for SerialTransport — no flow control, no + parity options, Linux/macOS only. Enough for a Micromate, which is 8N1 + with no handshaking. + """ + + _BAUD = {9600: termios.B9600, 19200: termios.B19200, 38400: termios.B38400, + 57600: termios.B57600, 115200: termios.B115200} + + def __init__(self, path: str, baud: int = 115200) -> None: + if baud not in self._BAUD: + raise ValueError(f"unsupported baud {baud}; pick from {sorted(self._BAUD)}") + self.path, self.baud, self.fd = path, baud, None + + def connect(self) -> None: + if self.fd is not None: + return + self.fd = os.open(self.path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) + a = termios.tcgetattr(self.fd) + a[0] = a[1] = a[3] = 0 # raw in/out, non-canonical + a[2] = termios.CS8 | termios.CREAD | termios.CLOCAL # 8N1, ignore modem lines + a[4] = a[5] = self._BAUD[self.baud] + a[6] = list(a[6]) + a[6][termios.VMIN] = 0 + a[6][termios.VTIME] = 0 + termios.tcsetattr(self.fd, termios.TCSANOW, a) + termios.tcflush(self.fd, termios.TCIOFLUSH) + + def disconnect(self) -> None: + if self.fd is not None: + os.close(self.fd) + self.fd = None + + def is_connected(self) -> bool: + return self.fd is not None + + def read(self, n: int) -> bytes: + if self.fd is None: + return b"" + r, _, _ = select.select([self.fd], [], [], 0.05) + if not r: + return b"" + try: + return os.read(self.fd, n) + except OSError as e: + if e.errno in (errno.EAGAIN, errno.EWOULDBLOCK): + return b"" + raise + + def write(self, data: bytes) -> None: + if self.fd is None: + raise OSError("port is not open") + while data: + data = data[os.write(self.fd, data):] class _Timed: @@ -107,7 +176,9 @@ def main() -> int: ) ap.add_argument("target", help="host:port for TCP, or a serial device path") ap.add_argument("--baud", type=int, default=115200, - help="serial only; a Micromate's modem port runs at 115200") + help="serial only; the USB-A/FTDI path runs at 115200. " + "Ignored by the USB-B 'PC' port, which is CDC-ACM " + "and negotiates its own rate.") ap.add_argument("--timeout", type=float, default=10.0) ap.add_argument("--download", action="store_true", help="also download the first stored event (read-only)") @@ -120,7 +191,7 @@ def main() -> int: inner = TcpTransport(host, int(port), connect_timeout=a.timeout) path = f"TCP {host}:{port}" else: - inner = SerialTransport(a.target, baud=a.baud) + inner = StdlibSerial(a.target, baud=a.baud) path = f"serial {a.target} @ {a.baud}" transport = _Timed(inner)