Port serial layer to pyserial; add Windows COM detection and PyInstaller spec

- serial_port.py: replace termios/fcntl/select with pyserial wrapper,
  same SerialPort interface preserved for all other modules
- workflow_serial.py: detect_serial_device() uses serial.tools.list_ports
  to enumerate COM ports; removes _fix_permissions() (not needed on Windows);
  multi-device table now shows port description
- es24n_conf.py: enable VT/ANSI colour mode on Windows 10+ via ctypes;
  update docstring for Windows edition
- requirements.txt: pyserial >= 3.5, pyinstaller >= 6.0
- es24n_conf.spec: PyInstaller single-file .exe spec

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-16 15:50:17 -04:00
parent a6a0f1b246
commit 75873dd4e0
5 changed files with 103 additions and 97 deletions

View File

@@ -1,89 +1,65 @@
"""
serial_port.py — Minimal 8N1 serial port using only the Python standard library.
Replaces pyserial for TrueNAS environments where pip is unavailable.
serial_port.py — pyserial-backed 8N1 serial port for the Windows edition.
Exposes the same SerialPort interface as the TrueNAS/Linux edition so all
other modules work without modification.
"""
import fcntl
import os
import select
import termios
import time
from typing import Optional
import serial
class SerialPort:
BAUD_MAP = {
9600: termios.B9600,
19200: termios.B19200,
38400: termios.B38400,
57600: termios.B57600,
115200: termios.B115200,
}
def __init__(self, port: str, baudrate: int = 115200, timeout: float = 5.0):
self.port = port
self.baudrate = baudrate
self.timeout = timeout
self._fd: Optional[int] = None
self._saved_attrs = None
self._ser: Optional[serial.Serial] = None
# ── Open / close ──────────────────────────────────────────────────────────
def open(self):
try:
self._fd = os.open(self.port, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK)
except OSError as e:
self._ser = serial.Serial(
port = self.port,
baudrate = self.baudrate,
bytesize = serial.EIGHTBITS,
parity = serial.PARITY_NONE,
stopbits = serial.STOPBITS_ONE,
timeout = 0, # non-blocking reads; we poll manually
xonxoff = False,
rtscts = False,
dsrdtr = False,
)
except serial.SerialException as e:
raise OSError(f"Cannot open {self.port}: {e}") from e
# Switch back to blocking mode now that O_NOCTTY is set
flags = fcntl.fcntl(self._fd, fcntl.F_GETFL)
fcntl.fcntl(self._fd, fcntl.F_SETFL, flags & ~os.O_NONBLOCK)
self._saved_attrs = termios.tcgetattr(self._fd)
# Raw 8N1: iflag, oflag, cflag, lflag, ispeed, ospeed, cc
attrs = list(termios.tcgetattr(self._fd))
baud = self.BAUD_MAP.get(self.baudrate, termios.B115200)
attrs[0] = termios.IGNBRK # iflag
attrs[1] = 0 # oflag
attrs[2] = termios.CS8 | termios.CREAD | termios.CLOCAL # cflag
attrs[3] = 0 # lflag (raw)
attrs[4] = baud # ispeed
attrs[5] = baud # ospeed
attrs[6][termios.VMIN] = 0
attrs[6][termios.VTIME] = min(int(self.timeout * 10), 255)
termios.tcsetattr(self._fd, termios.TCSANOW, attrs)
termios.tcflush(self._fd, termios.TCIOFLUSH)
def close(self):
if self._fd is not None:
if self._ser is not None:
try:
if self._saved_attrs:
termios.tcsetattr(self._fd, termios.TCSADRAIN, self._saved_attrs)
os.close(self._fd)
except OSError:
self._ser.close()
except Exception:
pass
finally:
self._fd = None
self._ser = None
@property
def is_open(self) -> bool:
return self._fd is not None
return self._ser is not None and self._ser.is_open
# ── Read / write ──────────────────────────────────────────────────────────
def write(self, data: bytes):
if self._fd is None:
if self._ser is None:
raise OSError("Port is not open")
os.write(self._fd, data)
self._ser.write(data)
def read_chunk(self, size: int = 4096) -> bytes:
if self._fd is None:
if self._ser is None:
raise OSError("Port is not open")
try:
return os.read(self._fd, size)
except OSError:
return b""
waiting = self._ser.in_waiting
if waiting:
return self._ser.read(min(size, waiting))
return b""
def read_until_quiet(self, quiet_period: float = 0.5,
timeout: Optional[float] = None) -> str:
@@ -103,13 +79,12 @@ class SerialPort:
if output and (now - last_rx) >= quiet_period:
break
wait = min(deadline - now, quiet_period)
ready, _, _ = select.select([self._fd], [], [], wait)
if ready:
chunk = self.read_chunk()
if chunk:
output += chunk
last_rx = time.monotonic()
chunk = self.read_chunk()
if chunk:
output += chunk
last_rx = time.monotonic()
else:
time.sleep(0.05)
return output.decode("utf-8", errors="replace")