diff --git a/kota_tech/serialio/manager.py b/kota_tech/serialio/manager.py index 93095fd..18692a9 100644 --- a/kota_tech/serialio/manager.py +++ b/kota_tech/serialio/manager.py @@ -10,6 +10,7 @@ from __future__ import annotations import enum import queue +import sys import threading import time from collections.abc import Callable @@ -92,13 +93,21 @@ class SerialManager: ser = serial.serial_for_url( info.device, baudrate=self._baud, timeout=READ_TIMEOUT_S, do_not_open=True ) - # Deassert both lines before open. On USB-Serial-JTAG, RTS=1/DTR=0 - # holds the chip in reset and DTR=1 during reset selects download - # mode; pyserial would otherwise assert both on open (Windows). - ser.dtr = False + # On USB-Serial-JTAG, RTS=1/DTR=0 holds the chip in reset and DTR=1 + # during reset selects download mode. + # Windows: pyserial applies both lines atomically at open, so + # deassert both beforehand (it would otherwise assert both). + # macOS/Linux: the OS asserts both lines on open and pyserial then + # writes DTR before RTS, passing through RTS=1/DTR=0 and resetting + # the box on every connect. Drop RTS first, then DTR. + posix = sys.platform != "win32" + if not posix: + ser.dtr = False ser.rts = False try: ser.open() + if posix: + ser.dtr = False except (serial.SerialException, OSError): return None self._target.device = info.device # COM number / node may change diff --git a/tests/test_manager.py b/tests/test_manager.py index ad11fdf..8b832ba 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -86,6 +86,46 @@ def test_hard_reset_pushes_rts_via_dtr_write(monkeypatch): assert all(c[1] is False for c in ser.calls if c[0] == "dtr") +class _PosixOpenSerial: + """pyserial on macOS/Linux: the OS asserts both lines on open, then + pyserial writes the requested DTR, then RTS. Records every line state.""" + + def __init__(self, *args, **kwargs): + self.port = args[0] if args else "fake" + self.dtr = self.rts = True # pyserial defaults + self.states = [] + + def __setattr__(self, name, value): + object.__setattr__(self, name, value) + if name in ("dtr", "rts") and getattr(self, "is_open", False): + self.states.append((self.dtr, self.rts)) + + def open(self): + want_dtr, want_rts = self.dtr, self.rts + object.__setattr__(self, "dtr", True) + object.__setattr__(self, "rts", True) + self.is_open = True + self.states.append((True, True)) + self.dtr = want_dtr + self.rts = want_rts + + +def test_open_on_posix_never_passes_through_reset(monkeypatch): + monkeypatch.setattr(mgr_mod.sys, "platform", "darwin") + monkeypatch.setattr(mgr_mod, "find_port", lambda sn, dev: LOOP) + opened = [] + monkeypatch.setattr( + mgr_mod.serial, "serial_for_url", + lambda *a, **k: opened.append(_PosixOpenSerial(*a, **k)) or opened[-1], + ) + m = SerialManager(lambda ln: None, lambda s, d: None) + m._target = mgr_mod._Target(LOOP.device, None) + ser = m._open() + # (DTR, RTS) = (False, True) resets the chip on connect. + assert (False, True) not in ser.states + assert ser.states[-1] == (False, False) + + def test_waits_when_port_missing(monkeypatch): monkeypatch.setattr(mgr_mod, "find_port", lambda sn, dev: None) states, lines = [], []