refactor: normalize data directory naming, and consolidate serial connection handling

This commit is contained in:
jack
2026-06-15 12:39:16 -07:00
parent 5b1c924d35
commit 7d32fb4b65
5 changed files with 40 additions and 34 deletions
+28 -25
View File
@@ -45,6 +45,33 @@ class SerialPort:
def __init__(self, port_name: str) -> None:
self._port_name = port_name
# ------------------------------------------------------------------
# Port opening (shared between scoped and long-lived use)
# ------------------------------------------------------------------
@staticmethod
def open_port(port_name: str, timeout: float | None = None) -> pyserial.Serial:
"""Open a serial port with baud-rate fallback.
Tries SerialPort.BAUDRATE first, falls back to 115200 on failure.
The caller owns the lifecycle of the returned Serial object."""
kwargs = dict(
parity=pyserial.PARITY_NONE,
bytesize=8,
stopbits=pyserial.STOPBITS_ONE,
xonxoff=False,
rtscts=False,
dsrdtr=False,
)
if timeout is not None:
kwargs["timeout"] = timeout
try:
return pyserial.Serial(port_name, SerialPort.BAUDRATE, **kwargs)
except Exception as exc:
log.warning("Baud rate %d failed on %s, falling back to 115200: %s", SerialPort.BAUDRATE, port_name, exc)
return pyserial.Serial(port_name, 115200, **kwargs)
# ------------------------------------------------------------------
# Context manager for scoped port access
# ------------------------------------------------------------------
@@ -52,31 +79,7 @@ class SerialPort:
@contextmanager
def _open(self, timeout: float | None = None) -> Iterator[pyserial.Serial]:
"""Open the serial port, yield it, then close on exit."""
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
port = self.open_port(self._port_name, timeout=timeout)
try:
yield port
finally: