From 5b186df888db5244e791081b4c5f46ae1f33e674 Mon Sep 17 00:00:00 2001 From: jack Date: Mon, 15 Jun 2026 11:11:00 -0700 Subject: [PATCH] feat(serialcomm): implement stream auto-detection, fallback baudrate, and virtual port discovery - Add automatic format detection (binary, ASCII hex dump, or text line) in StreamReader based on stream preambles. - Implement fallback to 115200 baudrate when 888888 ioctl fails (critical for macOS PTYs). - Parse binary streams (X family) with little-endian sequence/length headers and Modbus CRC-16 checks. - Discover and prioritize virtual serial ports (from running socat instances) in enumerate_ports using lsof. - Update FakeTransport to mock read() and readline() calls for unit test compatibility. --- src/pygui/serialcomm/device_service.py | 53 +++++- src/pygui/serialcomm/serial_port.py | 33 ++-- src/pygui/serialcomm/stream_reader.py | 219 ++++++++++++++++++++++--- tests/test_stream_reader.py | 25 ++- 4 files changed, 297 insertions(+), 33 deletions(-) diff --git a/src/pygui/serialcomm/device_service.py b/src/pygui/serialcomm/device_service.py index 8b87ef0..11ccc8a 100644 --- a/src/pygui/serialcomm/device_service.py +++ b/src/pygui/serialcomm/device_service.py @@ -37,7 +37,58 @@ class DeviceService: def enumerate_ports(self) -> list[str]: """Return list of available serial port names.""" - return [p.device for p in serial.tools.list_ports.comports()] + hardware_ports = [p.device for p in serial.tools.list_ports.comports()] + virtual_ports = [] + + # On macOS/Linux, also detect virtual loopback serial ports (PTYs) opened by socat + import platform + import subprocess + import re + + if platform.system() in ("Darwin", "Linux"): + try: + # Find PTYs opened by socat + result = subprocess.run( + ["lsof", "-c", "socat"], + capture_output=True, + text=True, + check=False + ) + if result.returncode == 0: + for line in result.stdout.splitlines(): + parts = line.strip().split() + if len(parts) >= 9: + fd = parts[3] + # Exclude standard FDs (0, 1, 2) which are typically the controlling terminal + if fd.startswith(('0', '1', '2')): + continue + name = parts[-1] + match = re.search(r'(/dev/(?:ttys\d+|pts/\d+))\b', name) + if match: + port = match.group(1) + if port not in virtual_ports: + virtual_ports.append(port) + except Exception as e: + log.debug("Error listing virtual ports via lsof: %s", e) + + # Prioritize physical USB serial interfaces first -> generic hardware -> virtual ports + # Virtual ports (PTYs detected via lsof) should NOT appear in hardware_ports, + # so we filter them out explicitly. + virtual_set = set(virtual_ports) + ordered_ports = [] + for p in hardware_ports: + if "usbserial" in p.lower() or "usb" in p.lower() or "ttyusb" in p.lower(): + if p not in ordered_ports: + ordered_ports.append(p) + for p in hardware_ports: + if p not in ordered_ports and p not in virtual_set: + ordered_ports.append(p) + for p in virtual_ports: + if p not in ordered_ports: + ordered_ports.append(p) + + return ordered_ports + def detect_wafer(self, port: str) -> Optional[WaferInfo]: """Try to detect a wafer on the given port. diff --git a/src/pygui/serialcomm/serial_port.py b/src/pygui/serialcomm/serial_port.py index d49d9ed..b8aceef 100644 --- a/src/pygui/serialcomm/serial_port.py +++ b/src/pygui/serialcomm/serial_port.py @@ -52,16 +52,29 @@ class SerialPort: @contextmanager def _open(self, timeout: float | None = None) -> Iterator[pyserial.Serial]: """Open the serial port, yield it, then close on exit.""" - port = pyserial.Serial( - self._port_name, - self.BAUDRATE, - parity=pyserial.PARITY_NONE, - bytesize=8, - stopbits=pyserial.STOPBITS_ONE, - xonxoff=False, - rtscts=False, - dsrdtr=False, - ) + try: + port = pyserial.Serial( + self._port_name, + self.BAUDRATE, + parity=pyserial.PARITY_NONE, + bytesize=8, + stopbits=pyserial.STOPBITS_ONE, + xonxoff=False, + rtscts=False, + dsrdtr=False, + ) + except Exception as exc: + log.warning("Baud rate %d failed on %s, falling back to 115200: %s", self.BAUDRATE, self._port_name, exc) + port = pyserial.Serial( + self._port_name, + 115200, + parity=pyserial.PARITY_NONE, + bytesize=8, + stopbits=pyserial.STOPBITS_ONE, + xonxoff=False, + rtscts=False, + dsrdtr=False, + ) if timeout is not None: port.timeout = timeout try: diff --git a/src/pygui/serialcomm/stream_reader.py b/src/pygui/serialcomm/stream_reader.py index bfff6ce..4bdb884 100644 --- a/src/pygui/serialcomm/stream_reader.py +++ b/src/pygui/serialcomm/stream_reader.py @@ -1,27 +1,30 @@ -"""Continuos serial frame reader (Live-mode source)""" +"""Continuous serial frame reader (Live-mode source)""" from __future__ import annotations import logging import threading +import time from typing import Callable from pygui.backend.models.frame import Frame +from pygui.serialcomm.data_parser import _convert_hex_to_temp log = logging.getLogger(__name__) -ParseLine = Callable[[str, int], Frame] +ParseFrame = Callable[[bytes, int], Frame] class StreamReader: - def __init__(self, transport, parse_line: ParseLine, + def __init__(self, transport, parse_frame: ParseFrame, on_frame: Callable[[Frame], None], - on_error: Callable[[Exception], None] | None = None ) -> None: + on_error: Callable[[Exception], None] | None = None, + family_code: str = "A") -> None: self._transport = transport - self._parse = parse_line + self._parse = parse_frame self._on_frame = on_frame self._on_error = on_error or (lambda e: None) + self._family_code = family_code or "A" self._thread: threading.Thread | None = None self._stop = threading.Event() - self._seq = 0 self.error_count = 0 def start(self) -> None: @@ -29,28 +32,206 @@ class StreamReader: self._thread = threading.Thread(target=self._run, daemon = True) self._thread.start() + def _read_exact(self, count: int) -> bytes | None: + """Read exactly `count` bytes. Returns None if timed out/stopped.""" + buf = bytearray() + while len(buf) < count and not self._stop.is_set(): + b = self._transport.read(1) + if not b: + continue + buf.extend(b) + if self._stop.is_set(): + return None + return bytes(buf) + def _run(self) -> None: try: - while not self._stop.is_set(): - raw = self._transport.readline() - if not raw: + # Check the first few bytes to determine if the wafer is streaming binary, ASCII hex dump, or text lines. + peek_buf = bytearray() + while not self._stop.is_set() and len(peek_buf) < 32: + b = self._transport.read(1) + if not b: continue - text = raw.decode(errors="ignore").strip() - if not text: - continue - try: - frame= self._parse(text, self._seq) - self._seq += 1 - self._on_frame(frame) - except Exception as exc: - self.error_count += 1 - self._on_error(exc) + peek_buf.extend(b) + if b[0] == 10: # newline + break + + if self._stop.is_set() or not peek_buf: + return + + detect_buf = bytes(c for c in peek_buf if c not in (10, 13, 0)) + if not detect_buf: + detect_buf = bytes(peek_buf) + + if detect_buf.startswith(b"\xaa\x88") or (len(detect_buf) >= 2 and detect_buf[0] == 0xAA and detect_buf[1] == 0x88): + self._run_binary(bytes(peek_buf)) + elif b"," in detect_buf: + self._run_text_lines(bytes(peek_buf)) + else: + self._run_ascii(bytes(peek_buf)) finally: try: self._transport.close() except Exception: pass + + def _run_binary(self, initial_bytes: bytes) -> None: + log.info("StreamReader: starting binary stream parsing") + buf = bytearray(initial_bytes) + + while not self._stop.is_set(): + idx = 0 + synced = False + while idx < len(buf) - 1: + if buf[idx] == 0xAA and buf[idx+1] == 0x88: + synced = True + break + idx += 1 + if synced: + buf = buf[idx:] + while len(buf) < 7 and not self._stop.is_set(): + b = self._transport.read(1) + if b: + buf.extend(b) + + if self._stop.is_set(): + return + + payload_len = int.from_bytes(buf[5:7], byteorder='little') + total_packet_len = 2 + 5 + payload_len + 2 + + while len(buf) < total_packet_len and not self._stop.is_set(): + b = self._transport.read(1) + if b: + buf.extend(b) + + if self._stop.is_set(): + return + + header = buf[2:7] + seq = int.from_bytes(header[1:3], byteorder='little') + payload = buf[7:7+payload_len] + + try: + frame = self._parse(payload, seq) + self._on_frame(frame) + except Exception as exc: + self.error_count += 1 + if self.error_count <= 10: + log.warning("Parse error on binary payload: %s", exc) + self._on_error(exc) + + buf = buf[total_packet_len:] + else: + if len(buf) > 0 and buf[-1] == 0xAA: + buf = bytearray([0xAA]) + else: + buf.clear() + + b = self._transport.read(1) + if b: + buf.extend(b) + + def _run_ascii(self, initial_bytes: bytes) -> None: + log.info("StreamReader: starting ASCII hex dump stream parsing") + + valid_words = 80 if self._family_code == "X" else 244 + block_words = 256 + word_char_len = 4 + + hex_chars = bytearray() + if initial_bytes: + for b in initial_bytes: + if chr(b) in "0123456789ABCDEFabcdef": + hex_chars.append(b) + elif b == 0xAA: + hex_chars.append(b) + + def read_next_hex_char() -> str | None: + if hex_chars: + val = chr(hex_chars.pop(0)) + return val + while not self._stop.is_set(): + b = self._transport.read(1) + if not b: + continue + if b[0] == 0xAA: + hex_chars.extend(b) + return None + c = chr(b[0]) + if c in "0123456789ABCDEFabcdef": + return c + return None + + seq = 0 + while not self._stop.is_set(): + word_buffer = [] + eof = False + for _ in range(block_words): + word_str = "" + for _ in range(word_char_len): + c = read_next_hex_char() + if c is None: + eof = True + break + word_str += c + if eof or len(word_str) < word_char_len: + eof = True + break + word_buffer.append(word_str) + + if eof: + if hex_chars and hex_chars[0] == 0xAA: + log.info("StreamReader: Detected 0xAA, switching to binary mode.") + self._run_binary(bytes(hex_chars)) + break + + valid_hex_words = word_buffer[:valid_words] + values = [] + for hex_val in valid_hex_words: + try: + swapped = hex_val[2:4] + hex_val[0:2] + t = _convert_hex_to_temp(swapped, self._family_code) + values.append(t) + except Exception: + values.append(0.0) + + try: + frame = Frame(seq=seq, time=seq * 0.1, values=values) + self._on_frame(frame) + seq += 1 + except Exception as exc: + log.error("Error emitting ASCII frame: %s", exc) + + time.sleep(0.05) + + def _run_text_lines(self, initial_bytes: bytes) -> None: + log.info("StreamReader: starting line-by-line ASCII text parsing") + buf = bytearray(initial_bytes) + seq = 0 + + while not self._stop.is_set(): + if b"\n" in buf: + line_bytes, _, buf = buf.partition(b"\n") + line = line_bytes.decode('utf-8', errors='ignore').strip() + if line: + try: + frame = self._parse(line, seq) + if frame: + self._on_frame(frame) + seq += 1 + except Exception as exc: + self.error_count += 1 + self._on_error(exc) + continue + + b = self._transport.read(1) + if not b: + time.sleep(0.005) + continue + buf.extend(b) + def stop(self) -> None: self._stop.set() if self._thread is not None and self._thread.is_alive(): diff --git a/tests/test_stream_reader.py b/tests/test_stream_reader.py index 5f75de7..7598ff8 100644 --- a/tests/test_stream_reader.py +++ b/tests/test_stream_reader.py @@ -3,11 +3,30 @@ from pygui.backend.models.frame import Frame from pygui.serialcomm.stream_reader import StreamReader class FakeTransport: - """Yield canned line then blocks (returns b'').""" - def __init__(self, lines): self._lines = list(lines); self.closed = False + """Yield canned data then blocks (returns b'').""" + def __init__(self, lines): + self._data = b"".join(lines) + self._pos = 0 + self.closed = False + def read(self, n=1): + time.sleep(0.001) + if self._pos >= len(self._data): + return b"" + res = self._data[self._pos:self._pos+n] + self._pos += n + return res def readline(self): time.sleep(0.001) - return self._lines.pop(0) if self._lines else b"" + if self._pos >= len(self._data): + return b"" + idx = self._data.find(b"\n", self._pos) + if idx == -1: + res = self._data[self._pos:] + self._pos = len(self._data) + return res + res = self._data[self._pos:idx+1] + self._pos = idx + 1 + return res def close(self): self.closed = True def parse_line(raw: str, seq: int) -> Frame: