Loading upper_tester/__main__.py +31 −17 Changes for upper_tester/__main__.py: 31 added lines, 17 removed lines. Original line number Diff line number Diff line Loading @@ -4,15 +4,25 @@ @details Opens the modem, sends `ATE0` once (disable command echo so lines can be told apart from responses), and serves the upper-tester TCP protocol (see server.py) until interrupted (Ctrl-C), at which point the modem is re-initialized (AT+CFUN=1, full functionality, *no* reset flag) before the device is closed, so it isn't left in whatever state the last in-flight AT command left it in (e.g. CFUN=0/4 from a MODEM_RESTART or DEREGISTER MMI command). This deliberately does not use AT+CFUN=1,1 (module reset): on Quectel modules that reboots the cellular/USB subsystem, which drops and re-enumerates the ttyUSB device -- on the Pi that meant /dev/ttyUSB2 could vanish for the ~15 s the module takes to come back, and a relaunch before then would fail to open it. AT+CFUN=1 changes RF functionality only; the USB link is untouched. (AT+CHUP to hang up any active call, then ATZ to reset the AT command interpreter to its stored profile) before the device is closed, so it isn't left mid-call or in whatever state the last in-flight AT command left it in. Deliberately *not* AT+CFUN: even AT+CFUN=1 (no reset flag) still touches the radio functionality state, and AT+CFUN=1,1 (reset) reboots the module outright -- confirmed on real hardware (plain minicom) to drop /dev/ttyUSBx for a few seconds, independent of upper_tester. ATZ only resets the AT interpreter's own settings (echo, verbosity, ...); it never touches the radio stack or the USB enumeration. On this Quectel hardware, the ttyUSB device node can still disappear for a while around a module reset or a re-plug/re-enumeration, independent of which AT command (if any) was last sent -- Quectel's own tools (quectel-CM's AT backend, quectel-atc-proxy) don't try to prevent this either: they don't touch termios HUPCL/CLOCAL and don't send anything special on exit, they just retry opening the device with a short backoff (see quectel-atc-proxy.c's `access()`-then-`sleep(3)` loop) instead. AtModem does the same, both for the initial open here and, if the device later disappears mid-session, for a reconnect done by its own background reader thread (see modem.AtModem). """ import argparse import asyncio Loading @@ -32,7 +42,9 @@ def main() -> None: command-line default for any key it sets), `--device` (default `/dev/ttyUSB2`), `--baud` (default 115200), `--host` (default `0.0.0.0`), `--port` (default 12100), `--allow` (repeatable, an allowed client CIDR network; see server.UtServer), `--log-level` (default `INFO`). network; see server.UtServer), `--log-level` (default `INFO`), `--open-retries`/`--open-retry-delay` (default 10 attempts, 3 s apart, matching Quectel's own quectel-atc-proxy retry loop for this device). """ ap = argparse.ArgumentParser(prog="upper_tester") ap.add_argument("--config", help="TOML file; keys = option names (device, baud, host, port, allow, log_level)") Loading @@ -42,6 +54,8 @@ def main() -> None: ap.add_argument("--port", type=int, default=12100) ap.add_argument("--allow", action="append", default=[], help="allowed client network (repeatable)") ap.add_argument("--log-level", default="INFO") ap.add_argument("--open-retries", type=int, default=10, help="attempts to open the device before giving up") ap.add_argument("--open-retry-delay", type=float, default=3.0, help="seconds between open attempts") args = ap.parse_args() if args.config: with open(args.config, "rb") as f: Loading @@ -49,9 +63,9 @@ def main() -> None: setattr(args, k.replace("-", "_"), v) logging.basicConfig(level=args.log_level.upper(), format="%(asctime)s %(levelname)s %(name)s: %(message)s") modem = AtModem(args.device, args.baud) modem.open() log = logging.getLogger("ut") modem = AtModem(args.device, args.baud, args.open_retries, args.open_retry_delay) modem.open() ok, text = modem.execute("ATE0") log.info("ATE0 -> %s %r", ok, text) # Verbose numeric+text error reporting (3GPP TS 27.007 §9.1): without this, every failing AT Loading @@ -71,11 +85,11 @@ def main() -> None: except KeyboardInterrupt: log.info("Ctrl-C received: re-initializing the modem before exit") try: # AT+CFUN=1, not AT+CFUN=1,1: restore full functionality without a module reset, # which would reboot the cellular/USB subsystem and drop the ttyUSB device (see # the module docstring). ok, text = modem.execute("AT+CFUN=1") log.info("AT+CFUN=1 -> %s %r", ok, text) # AT+CHUP then ATZ, not AT+CFUN: see the module docstring for why. ok, text = modem.execute("AT+CHUP") log.info("AT+CHUP -> %s %r", ok, text) ok, text = modem.execute("ATZ") log.info("ATZ -> %s %r", ok, text) except OSError as e: log.error("modem re-initialization on exit failed: %s", e) finally: Loading upper_tester/dummy_modem.py +1 −0 Changes for upper_tester/dummy_modem.py: 1 added line, 0 removed lines. Original line number Diff line number Diff line Loading @@ -40,6 +40,7 @@ DEFAULT_SCRIPT = { "AT+CHUP": ["OK"], "ATA": ["OK"], "ATH": ["OK"], "ATZ": ["OK"], 'AT+QCFG="usbnet",0': ["OK"], 'AT+QCFG="pdp/duplicatechk",0': ["OK"], } Loading upper_tester/modem.py +116 −19 Changes for upper_tester/modem.py: 116 added lines, 19 removed lines. Original line number Diff line number Diff line Loading @@ -74,27 +74,79 @@ class AtModem: Opens the tty in raw mode, reads it on a background thread that reassembles lines (the modem may send CR, LF or CRLF), and serializes @ref execute calls with a lock so only one AT command is in flight at a time, as the AT protocol requires. Resilient to the device node (e.g. `/dev/ttyUSB2`) being transiently unavailable -- confirmed on real Quectel hardware (independent of this code, via plain `minicom`) that e.g. `AT+CFUN=1,1` makes the module reboot and the ttyUSB device disappear for a few seconds. @ref open retries with a backoff if the device isn't there yet, and the background reader thread does the same mid-session if the device vanishes and comes back while already connected (see @ref _reader / @ref _reconnect); in that case the one AT @ref execute call that was in flight when the device vanished still times out (nothing more will arrive for it), but every later command recovers automatically once the device reappears, without needing the whole process relaunched. """ def __init__(self, path: str, baud: int = 115200): def __init__(self, path: str, baud: int = 115200, open_retries: int = 10, open_retry_delay: float = 3.0): """! @brief Construct the object; does not open the device yet (see @ref open). @param path Path to the serial device (e.g. `/dev/ttyUSB2`, `/dev/quectel_at`, or a pty path such as the one FakeModem.path returns). @param baud Baud rate to configure on the tty; ignored if the device is not a real tty (e.g. a pty), see @ref open. @param path Path to the serial device (e.g. `/dev/ttyUSB2`, `/dev/quectel_at`, or a pty path such as the one FakeModem.path returns). @param baud Baud rate to configure on the tty; ignored if the device is not a real tty (e.g. a pty), see @ref open. @param open_retries Attempts to (re)open the device before giving up, both for the initial @ref open and for a later reconnect. @param open_retry_delay Seconds to wait between attempts. """ self.path = path self.baud = baud self.open_retries = open_retries self.open_retry_delay = open_retry_delay self._fd = -1 self._lines: "queue.Queue[str]" = queue.Queue() self._lock = threading.Lock() # one AT command at a time self._lock = threading.Lock() # one AT command at a time; also guards self._fd across a reconnect self._stop_r = self._stop_w = -1 self._thread = None self._closing = False def open(self) -> None: """! @brief Open the serial device, configure it and start the background reader thread. @brief Open the serial device (retrying per @ref open_retries/@ref open_retry_delay if it isn't there/accessible yet) and start the background reader thread. """ self._closing = False self._open_device_with_retry() self._stop_r, self._stop_w = os.pipe() self._thread = threading.Thread(target=self._reader, name="modem-reader", daemon=True) self._thread.start() log.info("modem opened: %s", self.path) def _open_device_with_retry(self) -> None: """! @brief Open @ref path into @ref _fd, retrying with a fixed delay on failure. @details Mirrors quectel-atc-proxy.c's own retry loop for this same device path (an `access()` check, then `sleep(3)` between attempts) rather than failing immediately: Quectel's own tooling treats a transiently missing device node as routine, not exceptional, on this hardware. """ for attempt in range(1, self.open_retries + 1): try: self._open_device() return except OSError as e: if attempt == self.open_retries: log.error("open %s failed, giving up after %d attempts: %s", self.path, self.open_retries, e) raise log.warning( "open %s failed (attempt %d/%d), retrying in %.0fs: %s", self.path, attempt, self.open_retries, self.open_retry_delay, e, ) time.sleep(self.open_retry_delay) def _open_device(self) -> None: """! @brief Low-level, no-retry open and termios configuration of @ref _fd. """ log.debug("opening modem: path=%s baud=%d", self.path, self.baud) self._fd = os.open(self.path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) Loading @@ -106,23 +158,19 @@ class AtModem: # Clear HUPCL (hang up on close) and set CLOCAL (ignore modem status lines): without # this, closing our fd -- e.g. on Ctrl-C, or just process exit -- drops DTR, which on # some USB cellular modules the firmware treats as "host went away" and responds to by # detaching/re-enumerating that USB function, i.e. /dev/ttyUSBx disappears even though # we never asked for a module reset. # detaching/re-enumerating that USB function. attrs[2] &= ~termios.HUPCL attrs[2] |= termios.CLOCAL termios.tcsetattr(self._fd, termios.TCSANOW, attrs) except (termios.error, AttributeError) as e: # e.g. pty log.debug("baud rate/HUPCL/CLOCAL not set: %s", e) self._stop_r, self._stop_w = os.pipe() self._thread = threading.Thread(target=self._reader, name="modem-reader", daemon=True) self._thread.start() log.info("modem opened: %s", self.path) def close(self) -> None: """! @brief Stop the reader thread and close the device and the internal wakeup pipe. """ log.debug("closing modem: path=%s", self.path) self._closing = True if self._thread is not None: os.write(self._stop_w, b"x") self._thread.join(2) Loading @@ -137,7 +185,9 @@ class AtModem: """! @brief Background thread body: read the device, reassemble lines and queue them. @details Blocks on @c select.select() on both the device fd and an internal pipe so @ref close can wake it up and make it exit promptly. @ref close can wake it up and make it exit promptly. On a read error (the device disappeared), tries to reconnect (see @ref _reconnect) instead of exiting outright, unless @ref close is already in progress. """ buf = b"" while True: Loading @@ -149,11 +199,27 @@ class AtModem: except BlockingIOError: continue except OSError as e: log.error("modem read error: %s", e) self._lines.put("+UT: modem read error") if self._closing: return log.error("modem read error, device likely disconnected: %s", e) self._lines.put("+UT: modem disconnected, attempting to reconnect") if not self._reconnect(): return # retries exhausted; _reconnect already logged why buf = b"" continue if not chunk: time.sleep(0.05) # select() said readable but read() returned EOF: unlike a plain "no data # yet" (BlockingIOError, handled above), this fd will never produce data # again (confirmed experimentally: the device disappearing does not # always raise an OSError here, it can just read as EOF) -- reconnect # rather than busy-waiting on a dead fd forever. if self._closing: return log.error("modem read EOF, device likely disconnected: %s", self.path) self._lines.put("+UT: modem disconnected, attempting to reconnect") if not self._reconnect(): return buf = b"" continue buf += chunk *lines, buf = buf.replace(b"\r", b"\n").split(b"\n") Loading @@ -163,6 +229,32 @@ class AtModem: log.debug("modem-reader: queued line %r", line) self._lines.put(line) def _reconnect(self) -> bool: """! @brief Close the broken fd and retry (re)opening @ref path in place. @details Runs under @ref _lock for the whole attempt, so any @ref execute call made while reconnecting simply waits for the lock (and then runs normally against the new fd) instead of racing the fd swap. Any @ref execute call already in flight when the device vanished holds the lock until its own deadline elapses (nothing more will ever arrive for it), so reconnecting only starts once that one call has timed out. @return True once reconnected; False if retries were exhausted, in which case the caller (@ref _reader) exits, as it would have before this existed. """ with self._lock: if self._fd >= 0: try: os.close(self._fd) except OSError: pass self._fd = -1 try: self._open_device_with_retry() except OSError: return False log.info("modem reconnected: %s", self.path) return True def execute(self, cmd: str, timeout: float | None = None, on_written=None) -> tuple[bool, str]: """! @brief Send one AT command and collect lines up to the final result code. Loading @@ -181,8 +273,9 @@ class AtModem: test system once the command has been submitted to the modem). @return A `(success, text)` tuple: `success` is true only when the final result code is exactly `OK`; `text` holds every received line (echo and the triggering command excluded) joined with CRLF, or ends with `"TIMEOUT"` if no final result code arrived within @p timeout. triggering command excluded) joined with CRLF, ends with `"TIMEOUT"` if no final result code arrived within @p timeout, or reports a write error if the device was unavailable (e.g. mid-reconnect) when sent. """ log.debug("execute: cmd=%r timeout=%r", cmd, timeout) data = normalize(cmd) Loading @@ -192,7 +285,11 @@ class AtModem: while not self._lines.empty(): # unsolicited results received while idle log.info("URC: %s", self._lines.get_nowait()) log.info("AT> %r (timeout %.0fs)", data, timeout) try: os.write(self._fd, data.encode("utf-8")) except OSError as e: log.error("AT write failed: %r: %s", data, e) return False, f"write error: {e}" if on_written: log.debug("execute: on_written callback fired for %r", data) on_written() Loading Loading
upper_tester/__main__.py +31 −17 Changes for upper_tester/__main__.py: 31 added lines, 17 removed lines. Original line number Diff line number Diff line Loading @@ -4,15 +4,25 @@ @details Opens the modem, sends `ATE0` once (disable command echo so lines can be told apart from responses), and serves the upper-tester TCP protocol (see server.py) until interrupted (Ctrl-C), at which point the modem is re-initialized (AT+CFUN=1, full functionality, *no* reset flag) before the device is closed, so it isn't left in whatever state the last in-flight AT command left it in (e.g. CFUN=0/4 from a MODEM_RESTART or DEREGISTER MMI command). This deliberately does not use AT+CFUN=1,1 (module reset): on Quectel modules that reboots the cellular/USB subsystem, which drops and re-enumerates the ttyUSB device -- on the Pi that meant /dev/ttyUSB2 could vanish for the ~15 s the module takes to come back, and a relaunch before then would fail to open it. AT+CFUN=1 changes RF functionality only; the USB link is untouched. (AT+CHUP to hang up any active call, then ATZ to reset the AT command interpreter to its stored profile) before the device is closed, so it isn't left mid-call or in whatever state the last in-flight AT command left it in. Deliberately *not* AT+CFUN: even AT+CFUN=1 (no reset flag) still touches the radio functionality state, and AT+CFUN=1,1 (reset) reboots the module outright -- confirmed on real hardware (plain minicom) to drop /dev/ttyUSBx for a few seconds, independent of upper_tester. ATZ only resets the AT interpreter's own settings (echo, verbosity, ...); it never touches the radio stack or the USB enumeration. On this Quectel hardware, the ttyUSB device node can still disappear for a while around a module reset or a re-plug/re-enumeration, independent of which AT command (if any) was last sent -- Quectel's own tools (quectel-CM's AT backend, quectel-atc-proxy) don't try to prevent this either: they don't touch termios HUPCL/CLOCAL and don't send anything special on exit, they just retry opening the device with a short backoff (see quectel-atc-proxy.c's `access()`-then-`sleep(3)` loop) instead. AtModem does the same, both for the initial open here and, if the device later disappears mid-session, for a reconnect done by its own background reader thread (see modem.AtModem). """ import argparse import asyncio Loading @@ -32,7 +42,9 @@ def main() -> None: command-line default for any key it sets), `--device` (default `/dev/ttyUSB2`), `--baud` (default 115200), `--host` (default `0.0.0.0`), `--port` (default 12100), `--allow` (repeatable, an allowed client CIDR network; see server.UtServer), `--log-level` (default `INFO`). network; see server.UtServer), `--log-level` (default `INFO`), `--open-retries`/`--open-retry-delay` (default 10 attempts, 3 s apart, matching Quectel's own quectel-atc-proxy retry loop for this device). """ ap = argparse.ArgumentParser(prog="upper_tester") ap.add_argument("--config", help="TOML file; keys = option names (device, baud, host, port, allow, log_level)") Loading @@ -42,6 +54,8 @@ def main() -> None: ap.add_argument("--port", type=int, default=12100) ap.add_argument("--allow", action="append", default=[], help="allowed client network (repeatable)") ap.add_argument("--log-level", default="INFO") ap.add_argument("--open-retries", type=int, default=10, help="attempts to open the device before giving up") ap.add_argument("--open-retry-delay", type=float, default=3.0, help="seconds between open attempts") args = ap.parse_args() if args.config: with open(args.config, "rb") as f: Loading @@ -49,9 +63,9 @@ def main() -> None: setattr(args, k.replace("-", "_"), v) logging.basicConfig(level=args.log_level.upper(), format="%(asctime)s %(levelname)s %(name)s: %(message)s") modem = AtModem(args.device, args.baud) modem.open() log = logging.getLogger("ut") modem = AtModem(args.device, args.baud, args.open_retries, args.open_retry_delay) modem.open() ok, text = modem.execute("ATE0") log.info("ATE0 -> %s %r", ok, text) # Verbose numeric+text error reporting (3GPP TS 27.007 §9.1): without this, every failing AT Loading @@ -71,11 +85,11 @@ def main() -> None: except KeyboardInterrupt: log.info("Ctrl-C received: re-initializing the modem before exit") try: # AT+CFUN=1, not AT+CFUN=1,1: restore full functionality without a module reset, # which would reboot the cellular/USB subsystem and drop the ttyUSB device (see # the module docstring). ok, text = modem.execute("AT+CFUN=1") log.info("AT+CFUN=1 -> %s %r", ok, text) # AT+CHUP then ATZ, not AT+CFUN: see the module docstring for why. ok, text = modem.execute("AT+CHUP") log.info("AT+CHUP -> %s %r", ok, text) ok, text = modem.execute("ATZ") log.info("ATZ -> %s %r", ok, text) except OSError as e: log.error("modem re-initialization on exit failed: %s", e) finally: Loading
upper_tester/dummy_modem.py +1 −0 Changes for upper_tester/dummy_modem.py: 1 added line, 0 removed lines. Original line number Diff line number Diff line Loading @@ -40,6 +40,7 @@ DEFAULT_SCRIPT = { "AT+CHUP": ["OK"], "ATA": ["OK"], "ATH": ["OK"], "ATZ": ["OK"], 'AT+QCFG="usbnet",0': ["OK"], 'AT+QCFG="pdp/duplicatechk",0': ["OK"], } Loading
upper_tester/modem.py +116 −19 Changes for upper_tester/modem.py: 116 added lines, 19 removed lines. Original line number Diff line number Diff line Loading @@ -74,27 +74,79 @@ class AtModem: Opens the tty in raw mode, reads it on a background thread that reassembles lines (the modem may send CR, LF or CRLF), and serializes @ref execute calls with a lock so only one AT command is in flight at a time, as the AT protocol requires. Resilient to the device node (e.g. `/dev/ttyUSB2`) being transiently unavailable -- confirmed on real Quectel hardware (independent of this code, via plain `minicom`) that e.g. `AT+CFUN=1,1` makes the module reboot and the ttyUSB device disappear for a few seconds. @ref open retries with a backoff if the device isn't there yet, and the background reader thread does the same mid-session if the device vanishes and comes back while already connected (see @ref _reader / @ref _reconnect); in that case the one AT @ref execute call that was in flight when the device vanished still times out (nothing more will arrive for it), but every later command recovers automatically once the device reappears, without needing the whole process relaunched. """ def __init__(self, path: str, baud: int = 115200): def __init__(self, path: str, baud: int = 115200, open_retries: int = 10, open_retry_delay: float = 3.0): """! @brief Construct the object; does not open the device yet (see @ref open). @param path Path to the serial device (e.g. `/dev/ttyUSB2`, `/dev/quectel_at`, or a pty path such as the one FakeModem.path returns). @param baud Baud rate to configure on the tty; ignored if the device is not a real tty (e.g. a pty), see @ref open. @param path Path to the serial device (e.g. `/dev/ttyUSB2`, `/dev/quectel_at`, or a pty path such as the one FakeModem.path returns). @param baud Baud rate to configure on the tty; ignored if the device is not a real tty (e.g. a pty), see @ref open. @param open_retries Attempts to (re)open the device before giving up, both for the initial @ref open and for a later reconnect. @param open_retry_delay Seconds to wait between attempts. """ self.path = path self.baud = baud self.open_retries = open_retries self.open_retry_delay = open_retry_delay self._fd = -1 self._lines: "queue.Queue[str]" = queue.Queue() self._lock = threading.Lock() # one AT command at a time self._lock = threading.Lock() # one AT command at a time; also guards self._fd across a reconnect self._stop_r = self._stop_w = -1 self._thread = None self._closing = False def open(self) -> None: """! @brief Open the serial device, configure it and start the background reader thread. @brief Open the serial device (retrying per @ref open_retries/@ref open_retry_delay if it isn't there/accessible yet) and start the background reader thread. """ self._closing = False self._open_device_with_retry() self._stop_r, self._stop_w = os.pipe() self._thread = threading.Thread(target=self._reader, name="modem-reader", daemon=True) self._thread.start() log.info("modem opened: %s", self.path) def _open_device_with_retry(self) -> None: """! @brief Open @ref path into @ref _fd, retrying with a fixed delay on failure. @details Mirrors quectel-atc-proxy.c's own retry loop for this same device path (an `access()` check, then `sleep(3)` between attempts) rather than failing immediately: Quectel's own tooling treats a transiently missing device node as routine, not exceptional, on this hardware. """ for attempt in range(1, self.open_retries + 1): try: self._open_device() return except OSError as e: if attempt == self.open_retries: log.error("open %s failed, giving up after %d attempts: %s", self.path, self.open_retries, e) raise log.warning( "open %s failed (attempt %d/%d), retrying in %.0fs: %s", self.path, attempt, self.open_retries, self.open_retry_delay, e, ) time.sleep(self.open_retry_delay) def _open_device(self) -> None: """! @brief Low-level, no-retry open and termios configuration of @ref _fd. """ log.debug("opening modem: path=%s baud=%d", self.path, self.baud) self._fd = os.open(self.path, os.O_RDWR | os.O_NOCTTY | os.O_NONBLOCK) Loading @@ -106,23 +158,19 @@ class AtModem: # Clear HUPCL (hang up on close) and set CLOCAL (ignore modem status lines): without # this, closing our fd -- e.g. on Ctrl-C, or just process exit -- drops DTR, which on # some USB cellular modules the firmware treats as "host went away" and responds to by # detaching/re-enumerating that USB function, i.e. /dev/ttyUSBx disappears even though # we never asked for a module reset. # detaching/re-enumerating that USB function. attrs[2] &= ~termios.HUPCL attrs[2] |= termios.CLOCAL termios.tcsetattr(self._fd, termios.TCSANOW, attrs) except (termios.error, AttributeError) as e: # e.g. pty log.debug("baud rate/HUPCL/CLOCAL not set: %s", e) self._stop_r, self._stop_w = os.pipe() self._thread = threading.Thread(target=self._reader, name="modem-reader", daemon=True) self._thread.start() log.info("modem opened: %s", self.path) def close(self) -> None: """! @brief Stop the reader thread and close the device and the internal wakeup pipe. """ log.debug("closing modem: path=%s", self.path) self._closing = True if self._thread is not None: os.write(self._stop_w, b"x") self._thread.join(2) Loading @@ -137,7 +185,9 @@ class AtModem: """! @brief Background thread body: read the device, reassemble lines and queue them. @details Blocks on @c select.select() on both the device fd and an internal pipe so @ref close can wake it up and make it exit promptly. @ref close can wake it up and make it exit promptly. On a read error (the device disappeared), tries to reconnect (see @ref _reconnect) instead of exiting outright, unless @ref close is already in progress. """ buf = b"" while True: Loading @@ -149,11 +199,27 @@ class AtModem: except BlockingIOError: continue except OSError as e: log.error("modem read error: %s", e) self._lines.put("+UT: modem read error") if self._closing: return log.error("modem read error, device likely disconnected: %s", e) self._lines.put("+UT: modem disconnected, attempting to reconnect") if not self._reconnect(): return # retries exhausted; _reconnect already logged why buf = b"" continue if not chunk: time.sleep(0.05) # select() said readable but read() returned EOF: unlike a plain "no data # yet" (BlockingIOError, handled above), this fd will never produce data # again (confirmed experimentally: the device disappearing does not # always raise an OSError here, it can just read as EOF) -- reconnect # rather than busy-waiting on a dead fd forever. if self._closing: return log.error("modem read EOF, device likely disconnected: %s", self.path) self._lines.put("+UT: modem disconnected, attempting to reconnect") if not self._reconnect(): return buf = b"" continue buf += chunk *lines, buf = buf.replace(b"\r", b"\n").split(b"\n") Loading @@ -163,6 +229,32 @@ class AtModem: log.debug("modem-reader: queued line %r", line) self._lines.put(line) def _reconnect(self) -> bool: """! @brief Close the broken fd and retry (re)opening @ref path in place. @details Runs under @ref _lock for the whole attempt, so any @ref execute call made while reconnecting simply waits for the lock (and then runs normally against the new fd) instead of racing the fd swap. Any @ref execute call already in flight when the device vanished holds the lock until its own deadline elapses (nothing more will ever arrive for it), so reconnecting only starts once that one call has timed out. @return True once reconnected; False if retries were exhausted, in which case the caller (@ref _reader) exits, as it would have before this existed. """ with self._lock: if self._fd >= 0: try: os.close(self._fd) except OSError: pass self._fd = -1 try: self._open_device_with_retry() except OSError: return False log.info("modem reconnected: %s", self.path) return True def execute(self, cmd: str, timeout: float | None = None, on_written=None) -> tuple[bool, str]: """! @brief Send one AT command and collect lines up to the final result code. Loading @@ -181,8 +273,9 @@ class AtModem: test system once the command has been submitted to the modem). @return A `(success, text)` tuple: `success` is true only when the final result code is exactly `OK`; `text` holds every received line (echo and the triggering command excluded) joined with CRLF, or ends with `"TIMEOUT"` if no final result code arrived within @p timeout. triggering command excluded) joined with CRLF, ends with `"TIMEOUT"` if no final result code arrived within @p timeout, or reports a write error if the device was unavailable (e.g. mid-reconnect) when sent. """ log.debug("execute: cmd=%r timeout=%r", cmd, timeout) data = normalize(cmd) Loading @@ -192,7 +285,11 @@ class AtModem: while not self._lines.empty(): # unsolicited results received while idle log.info("URC: %s", self._lines.get_nowait()) log.info("AT> %r (timeout %.0fs)", data, timeout) try: os.write(self._fd, data.encode("utf-8")) except OSError as e: log.error("AT write failed: %r: %s", data, e) return False, f"write error: {e}" if on_written: log.debug("execute: on_written callback fired for %r", data) on_written() Loading
