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.
This commit is contained in:
jack
2026-06-15 11:11:00 -07:00
parent f89a735d51
commit 5b186df888
4 changed files with 297 additions and 33 deletions
+52 -1
View File
@@ -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.