This is an automated email from the ASF dual-hosted git repository.
xiaoxiang781216 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nuttx-ntfc.git
The following commit(s) were added to refs/heads/main by this push:
new 960a719 device/serial: reopen the port after USB re-enumeration
960a719 is described below
commit 960a719cd0754ef1f554f76c89c9bfe45876ab1e
Author: raiden00pl <[email protected]>
AuthorDate: Fri Oct 2 16:19:35 2026 +0200
device/serial: reopen the port after USB re-enumeration
A board whose console is USB CDC/ACM drops its serial port on every reset
and re-creates it once it boots again. The serial device kept the stale
handle, so every read and write after a reboot failed, and start failed
when the port did not exist yet.
A failed read or write now closes the port and reopens it, and the health
check retries until the port is back. Start no longer requires the port
to exist; the boot wait reopens it when the device enumerates.
Signed-off-by: raiden00pl <[email protected]>
Assisted-by: Claude Code:claude-opus-5-5
---
CHANGES.md | 1 +
src/ntfc/device/serial.py | 87 ++++++++++++++++++++++++++++++++-------------
tests/device/test_serial.py | 39 ++++++++++++++++++--
3 files changed, 100 insertions(+), 27 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 56f7b7b..db752b5 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -4,6 +4,7 @@
- Add Renode device support.
- Refuse commands longer than the target CONFIG_LINE_MAX.
+- Reopen serial ports that disappear on target reset (USB re-enumeration).
## 0.0.3
diff --git a/src/ntfc/device/serial.py b/src/ntfc/device/serial.py
index 808bb98..5e7c107 100644
--- a/src/ntfc/device/serial.py
+++ b/src/ntfc/device/serial.py
@@ -43,6 +43,7 @@ class DeviceSerial(DeviceCommon):
"""Initialize sim emulator device."""
DeviceCommon.__init__(self, conf, echo=False)
self._ser = None
+ self._lost = False
def _decode_exec_args(self, args: str) -> Dict[str, Any]:
"""Decode a serial port configuration string."""
@@ -87,8 +88,41 @@ class DeviceSerial(DeviceCommon):
except Exception as e:
raise ValueError(f"Invalid format '{args}': {e}") from e
+ def _open(self) -> None:
+ """Open the serial port."""
+ path = self._conf.exec_path
+ args = self._conf.exec_args
+
+ if args: # pragma: no cover
+ self._ser = serial.Serial(
+ path, timeout=0, **self._decode_exec_args(args)
+ )
+ else:
+ self._ser = serial.Serial(path, timeout=0)
+
+ def _port_lost(self) -> None:
+ """Drop a port that failed, e.g. a USB device re-enumerating."""
+ logger.info("serial port lost, reopening")
+
+ assert self._ser
+ self._ser.close()
+ self._ser = None
+ self._lost = True
+ self._reopen()
+
+ def _reopen(self) -> None:
+ """Try to reopen a lost port."""
+ try:
+ self._open()
+ self._lost = False
+ except serial.SerialException:
+ self._ser = None
+
def _dev_is_health_priv(self) -> bool:
"""Check if the serial device is OK."""
+ if not self._ser and self._lost:
+ self._reopen()
+
if not self._ser:
return False
@@ -101,19 +135,23 @@ class DeviceSerial(DeviceCommon):
assert self._ser
- if self._conf.line_buffered:
- if data[-1] != ord("\n"):
- data += b"\n"
-
- self._ser.write(data)
- else:
- # send char by char to avoid line length full
- for c in data:
- self._ser.write(bytes([c]))
-
- # add new line if missing
- if data[-1] != ord("\n"):
- self._ser.write(b"\n") # pragma: no cover
+ try:
+ if self._conf.line_buffered:
+ if data[-1] != ord("\n"):
+ data += b"\n"
+
+ self._ser.write(data)
+ else:
+ # send char by char to avoid line length full
+ for c in data:
+ self._ser.write(bytes([c]))
+
+ # add new line if missing
+ if data[-1] != ord("\n"):
+ self._ser.write(b"\n") # pragma: no cover
+ except (serial.SerialException, OSError):
+ self._port_lost()
+ return
# read all garbage left by character echo
_ = self._read_all(timeout=0)
@@ -136,22 +174,20 @@ class DeviceSerial(DeviceCommon):
assert self._ser
- return self._ser.read(size=1024)
+ try:
+ return self._ser.read(size=1024)
+ except (serial.SerialException, OSError):
+ self._port_lost()
+ return b""
def _start_impl(self) -> None:
"""Start serial communication implementation."""
- timeout = 0
- path = self._conf.exec_path
- args = self._conf.exec_args
+ logger.info(f"serial path: {self._conf.exec_path}")
+ logger.info(f"serial args: {self._conf.exec_args}")
- logger.info(f"serial path: {path}")
- logger.info(f"serial args: {args}")
-
- if args: # pragma: no cover
- args = self._decode_exec_args(args)
- self._ser = serial.Serial(path, timeout=timeout, **args)
- else:
- self._ser = serial.Serial(path, timeout=timeout)
+ # A USB port may show up only after the device boots
+ self._lost = True
+ self._reopen()
# reboot device if possible
self.reboot()
@@ -177,6 +213,7 @@ class DeviceSerial(DeviceCommon):
if self._ser: # pragma: no cover
self._ser.close()
self._ser = None
+ self._lost = False
logger.info("serial device closed")
def _poweroff_impl(self) -> bool:
diff --git a/tests/device/test_serial.py b/tests/device/test_serial.py
index b08c56d..11f577f 100644
--- a/tests/device/test_serial.py
+++ b/tests/device/test_serial.py
@@ -123,8 +123,8 @@ def test_device_sim_init(serial_config, serial_pair):
ser = DeviceSerial(serial_config)
- # no path and args
- with pytest.raises(serial.serialutil.SerialException):
+ # no port yet (USB device not enumerated): wait for boot, then time out
+ with pytest.raises(TimeoutError):
serial_config._config["exec_path"] = ""
serial_config._config["exec_args"] = ""
ser.start()
@@ -191,3 +191,38 @@ def test_device_serial_boot_timeout(serial_pair,
monkeypatch):
# boot wait uses the configured boot_timeout
assert boot_timeouts == [9]
+
+
+def test_device_serial_reconnect(serial_config, serial_pair):
+
+ serial_config._config["exec_path"] = serial_pair[1]
+ ser = DeviceSerial(serial_config)
+ ser._open()
+ first = ser._ser
+
+ # a failed read (USB device gone) reopens the port
+ def fail(*args, **kwargs):
+ raise serial.SerialException("device disconnected")
+
+ ser._ser.read = fail
+ assert ser._read() == b""
+ assert ser._ser is not None
+ assert ser._ser is not first
+ assert ser._lost is False
+
+ # a failed write while the port is still missing leaves it lost
+ serial_config._config["exec_path"] = "/dev/ntfc-no-such-port"
+ ser._ser.write = fail
+ assert ser._write(b"a") is None
+ assert ser._ser is None
+ assert ser._lost is True
+ assert ser._dev_is_health_priv() is False
+
+ # the health check reconnects once the port is back
+ serial_config._config["exec_path"] = serial_pair[1]
+ assert ser._dev_is_health_priv() is True
+ assert ser._lost is False
+
+ ser.stop()
+ assert ser._lost is False
+ assert ser.notalive is True
