Loading upper_tester/README.md +7 −0 Changes for upper_tester/README.md: 7 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -21,6 +21,11 @@ Reply (one line): command is written. `CNF_REQUIRED`: reply after the final result code (or `TIMEOUT`, result false). - MMI commands are mapped to AT sequences in `mmi.py` (`MMI_TABLE`); unknown ones reply `result=false`. - Unsolicited result codes are only logged. - A command that reboots the modem (`AT+CFUN=<fun>,1`) is confirmed only once the modem is back: its device has reappeared, it answers `AT`, and the SIM is ready. The reply then ends with `+UT: modem restarted`. This takes the time of a modem boot. - If the modem disappears while a command is in flight, that command fails at once with `+UT: modem disconnected`; later commands wait for the device to come back. ## Data call (`ACTIVATE_BEARER` / `DEACTIVATE_BEARER`) Loading Loading @@ -61,6 +66,8 @@ Then run the service itself: ```sh $ python3.13 -m upper_tester --device /dev/ttyUSB2 --port 12100 ``` Prefer the udev symlink (`--device /dev/quectel_at`, see `deploy/99-quectel.rules`) to a `ttyUSBx` name: the numbering can change when the modem reboots. To try it without a real modem, see dummy_modem.py (`python3.13 -m upper_tester.dummy_modem`). ## Install on the Pi Loading upper_tester/__main__.py +4 −10 Changes for upper_tester/__main__.py: 4 added lines, 10 removed lines. Original line number Diff line number Diff line Loading @@ -43,8 +43,8 @@ def main() -> None: `/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`), `--open-retries`/`--open-retry-delay` (default 10 attempts, 3 s apart, matching Quectel's own quectel-atc-proxy retry loop for this device). `--open-retries`/`--open-retry-delay` (default 20 attempts, 3 s apart: long enough for the modem to come back after a restart). """ ap = argparse.ArgumentParser(prog="upper_tester") ap.add_argument("--config", help="TOML file; keys = option names (device, baud, host, port, allow, log_level)") Loading @@ -54,7 +54,7 @@ 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-retries", type=int, default=20, 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: Loading @@ -66,13 +66,7 @@ def main() -> None: 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 # command comes back as a bare "ERROR" with no cause, making the actual TTCN-side failures # (e.g. an unsupported/misapplied command) indistinguishable from a real, temporary error. ok, text = modem.execute("AT+CMEE=2") log.info("AT+CMEE=2 -> %s %r", ok, text) modem.init() # no echo, verbose +CME ERROR causes; see modem.INIT_COMMANDS async def run() -> None: """! @brief Start the TCP server and serve until cancelled. """ Loading upper_tester/dummy_modem.py +16 −0 Changes for upper_tester/dummy_modem.py: 16 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -29,6 +29,8 @@ DEFAULT_SCRIPT = { "AT+CFUN=0": ["OK"], "AT+CFUN=1": ["OK"], "AT+CFUN=1,1": ["OK"], "AT": ["OK"], "AT+CPIN?": ["+CPIN: READY", "OK"], "AT+CFUN=4": ["OK"], "AT+CGATT=1": ["OK"], "AT+CGATT=0": ["OK"], Loading Loading @@ -88,9 +90,21 @@ class FakeModem: self.echo = echo ## Every command received so far, in order, for tests to assert against. self.received: list[str] = [] self._closed = False self._t = threading.Thread(target=self._run, daemon=True) self._t.start() def close(self) -> None: """! @brief Make the modem disappear, as a real one does when it reboots or is unplugged: whoever has @ref path open then reads an error or end-of-file. """ self._closed = True os.write(self._slave, b"\r") # wake the thread up so it sees _closed self._t.join(2) os.close(self.master) os.close(self._slave) def urc(self, line: str) -> None: """! @brief Inject an unsolicited result code (URC) at any time, independent of any Loading @@ -110,6 +124,8 @@ class FakeModem: buf += os.read(self.master, 1024) except OSError: return if self._closed: return while b"\r" in buf: raw, buf = buf.split(b"\r", 1) cmd = raw.decode() Loading upper_tester/engine.py +9 −1 Changes for upper_tester/engine.py: 9 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -3,11 +3,15 @@ @brief Request execution: AT passthrough and MMI-to-AT mapping. """ import logging import re from .mmi import MMI_PROCEDURES, UnsupportedMmi, steps_for log = logging.getLogger("ut.engine") ## AT commands that reboot the modem (`AT+CFUN=<fun>,1`); see modem.AtModem.restart. RESTART_CMD = re.compile(r"AT\+CFUN=\d,1", re.IGNORECASE) ## Valid values of a request's `"cnf"` field, mirroring UpperTesterCommon.Ut_CnfReq_Type. CNF_MODES = ("NO_CNF_REQUIRED", "CNF_REQUIRED", "LOCAL_CNF_REQUIRED") Loading @@ -30,7 +34,9 @@ class Engine: def run(self, req: dict, on_written=None) -> tuple[bool, str]: """! @brief Execute one decoded request. Blocking; call from a worker thread. @details For `type: "AT"`, @p req["cmd"] is sent to the modem as-is. For @details For `type: "AT"`, @p req["cmd"] is sent to the modem as-is; if it reboots the modem (@ref RESTART_CMD), the call also waits until the modem is usable again (see modem.AtModem.restart). For `type: "MMI"`, @p req["cmd"] is run by its procedure if it has one in mmi.MMI_PROCEDURES (e.g. `ACTIVATE_BEARER`, see datacall.activate); otherwise it is resolved via mmi.steps_for into one or Loading Loading @@ -58,6 +64,8 @@ class Engine: timeout = req.get("timeout") if kind == "AT": log.debug("dispatch: AT passthrough: %r", cmd) if RESTART_CMD.fullmatch(cmd.replace("<CR>", "").strip()): return self.modem.restart(cmd, timeout, on_written) return self.modem.execute(cmd, timeout, on_written) params = {p["name"]: p["value"] for p in req.get("params") or []} log.debug("dispatch: MMI %s params=%s", cmd, params) Loading upper_tester/modem.py +104 −18 Changes for upper_tester/modem.py: 104 added lines, 18 removed lines. Original line number Diff line number Diff line Loading @@ -32,6 +32,22 @@ TIMEOUTS = ( ## Timeout, in seconds, used for any command not listed in @ref TIMEOUTS. DEFAULT_TIMEOUT = 10.0 ## Line queued by the reader thread when the device disappears; it ends the AT command in ## flight at once (see @ref AtModem.execute) instead of leaving it to time out. DISCONNECTED = "+UT: modem disconnected" ## Commands that put the modem back in the state this service expects after it (re)starts: ## no command echo, and verbose `+CME ERROR` causes (3GPP TS 27.007 clause 9.1) instead of a ## bare `ERROR`. Both settings are lost when the modem reboots. INIT_COMMANDS = ("ATE0", "AT+CMEE=2") ## Seconds @ref AtModem.restart waits for the device to disappear after the restart command. RESTART_GONE_TIMEOUT = 15.0 ## Seconds @ref AtModem.restart waits for the device to come back and answer `AT`. RESTART_BACK_TIMEOUT = 90.0 ## Seconds @ref AtModem.restart then waits for the SIM to be ready (`+CPIN: READY`). RESTART_SIM_TIMEOUT = 20.0 ## Seconds between two polls in @ref AtModem.restart. RESTART_POLL_INTERVAL = 1.0 def timeout_for(cmd: str) -> float: """! Loading Loading @@ -82,12 +98,17 @@ class AtModem: 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 one AT @ref execute call that was in flight when the device vanished fails at once (nothing more will arrive for it), and every later command recovers automatically once the device reappears, without needing the whole process relaunched. The stale descriptor is closed as soon as the device vanishes: while it stays open the kernel keeps its `ttyUSB` number reserved, so the modem's ports come back under other numbers and the configured path (e.g. `/dev/ttyUSB2`) never reappears. A udev symlink (`/dev/quectel_at`, see `deploy/99-quectel.rules`) avoids depending on the numbering. """ def __init__(self, path: str, baud: int = 115200, open_retries: int = 10, open_retry_delay: float = 3.0): def __init__(self, path: str, baud: int = 115200, open_retries: int = 20, 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`, Loading @@ -109,6 +130,8 @@ class AtModem: self._stop_r = self._stop_w = -1 self._thread = None self._closing = False self._down = threading.Event() # set when the device disappears self._up = threading.Event() # set while the device is open def open(self) -> None: """! Loading Loading @@ -165,6 +188,24 @@ class AtModem: 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._up.set() def _drop_device(self) -> None: """! @brief Handle the device disappearing: close the stale descriptor and end the AT command in flight. @details Called from the reader thread without @ref _lock, which the command in flight holds. @ref _fd is set to -1 before the descriptor is closed, so a concurrent write fails cleanly instead of reaching a reused descriptor. """ self._up.clear() fd, self._fd = self._fd, -1 try: os.close(fd) except OSError: pass self._down.set() self._lines.put(DISCONNECTED) def close(self) -> None: """! Loading Loading @@ -203,7 +244,7 @@ class AtModem: if self._closing: return log.error("modem read error, device likely disconnected: %s", e) self._lines.put("+UT: modem disconnected, attempting to reconnect") self._drop_device() if not self._reconnect(): return # retries exhausted; _reconnect already logged why buf = b"" Loading @@ -217,7 +258,7 @@ class AtModem: if self._closing: return log.error("modem read EOF, device likely disconnected: %s", self.path) self._lines.put("+UT: modem disconnected, attempting to reconnect") self._drop_device() if not self._reconnect(): return buf = b"" Loading @@ -232,23 +273,16 @@ class AtModem: def _reconnect(self) -> bool: """! @brief Close the broken fd and retry (re)opening @ref path in place. @brief Retry (re)opening @ref path in place, after @ref _drop_device. @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. normally against the new fd) instead of racing the fd swap. The @ref execute call in flight when the device vanished has already been ended by @ref _drop_device, so the lock is free almost at once. @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: Loading @@ -275,8 +309,9 @@ class AtModem: @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, 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. no final result code arrived within @p timeout, ends with @ref DISCONNECTED if the device disappeared meanwhile, 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 Loading @@ -309,6 +344,57 @@ class AtModem: if line == echo: continue lines.append(line) if line == DISCONNECTED: log.warning("AT aborted, modem disconnected: %r", data) return False, "\r\n".join(lines) if is_final(line): log.info("AT< %s", lines) return line == "OK", "\r\n".join(lines) def init(self) -> None: """! @brief Run @ref INIT_COMMANDS; call it after @ref open and after any modem restart. """ for cmd in INIT_COMMANDS: ok, text = self.execute(cmd) log.info("%s -> %s %r", cmd, ok, text) def restart(self, cmd: str, timeout: float | None = None, on_written=None) -> tuple[bool, str]: """! @brief Run an AT command that restarts the modem (e.g. `AT+CFUN=1,1`) and wait until the modem is usable again. @details The modem answers `OK` and only then reboots: its device disappears a few seconds later and comes back once it has restarted. Any command sent in between is lost with the reboot. So after the `OK` this waits for the device to disappear (@ref RESTART_GONE_TIMEOUT; carried on regardless if it does not, e.g. on a pty), to be reopened by the reader thread and to answer `AT` (@ref RESTART_BACK_TIMEOUT), runs @ref init, then waits for the SIM (@ref RESTART_SIM_TIMEOUT). @param cmd The restart command. @param timeout As for @ref execute, for @p cmd itself. @param on_written As for @ref execute. @return `(success, text)` as @ref execute does for @p cmd, followed by a `+UT:` line: `success` is false if the modem did not answer `AT` again in time. A SIM that is not ready in time is only reported, not a failure. """ self._down.clear() ok, text = self.execute(cmd, timeout, on_written) if not ok: return ok, text if not self._down.wait(RESTART_GONE_TIMEOUT): log.warning("restart: %s did not disappear within %.0fs", self.path, RESTART_GONE_TIMEOUT) deadline = time.monotonic() + RESTART_BACK_TIMEOUT while not (self._up.wait(max(0.0, deadline - time.monotonic())) and self.execute("AT", 2.0)[0]): if time.monotonic() >= deadline: log.error("restart: modem not back within %.0fs", RESTART_BACK_TIMEOUT) return False, f"{text}\r\n+UT: modem not back after restart" time.sleep(RESTART_POLL_INTERVAL) self.init() deadline = time.monotonic() + RESTART_SIM_TIMEOUT while "+CPIN: READY" not in self.execute("AT+CPIN?")[1]: if time.monotonic() >= deadline: log.warning("restart: SIM not ready within %.0fs", RESTART_SIM_TIMEOUT) return True, f"{text}\r\n+UT: modem restarted, SIM not ready" time.sleep(RESTART_POLL_INTERVAL) log.info("restart: modem ready") return True, f"{text}\r\n+UT: modem restarted" Loading
upper_tester/README.md +7 −0 Changes for upper_tester/README.md: 7 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -21,6 +21,11 @@ Reply (one line): command is written. `CNF_REQUIRED`: reply after the final result code (or `TIMEOUT`, result false). - MMI commands are mapped to AT sequences in `mmi.py` (`MMI_TABLE`); unknown ones reply `result=false`. - Unsolicited result codes are only logged. - A command that reboots the modem (`AT+CFUN=<fun>,1`) is confirmed only once the modem is back: its device has reappeared, it answers `AT`, and the SIM is ready. The reply then ends with `+UT: modem restarted`. This takes the time of a modem boot. - If the modem disappears while a command is in flight, that command fails at once with `+UT: modem disconnected`; later commands wait for the device to come back. ## Data call (`ACTIVATE_BEARER` / `DEACTIVATE_BEARER`) Loading Loading @@ -61,6 +66,8 @@ Then run the service itself: ```sh $ python3.13 -m upper_tester --device /dev/ttyUSB2 --port 12100 ``` Prefer the udev symlink (`--device /dev/quectel_at`, see `deploy/99-quectel.rules`) to a `ttyUSBx` name: the numbering can change when the modem reboots. To try it without a real modem, see dummy_modem.py (`python3.13 -m upper_tester.dummy_modem`). ## Install on the Pi Loading
upper_tester/__main__.py +4 −10 Changes for upper_tester/__main__.py: 4 added lines, 10 removed lines. Original line number Diff line number Diff line Loading @@ -43,8 +43,8 @@ def main() -> None: `/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`), `--open-retries`/`--open-retry-delay` (default 10 attempts, 3 s apart, matching Quectel's own quectel-atc-proxy retry loop for this device). `--open-retries`/`--open-retry-delay` (default 20 attempts, 3 s apart: long enough for the modem to come back after a restart). """ ap = argparse.ArgumentParser(prog="upper_tester") ap.add_argument("--config", help="TOML file; keys = option names (device, baud, host, port, allow, log_level)") Loading @@ -54,7 +54,7 @@ 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-retries", type=int, default=20, 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: Loading @@ -66,13 +66,7 @@ def main() -> None: 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 # command comes back as a bare "ERROR" with no cause, making the actual TTCN-side failures # (e.g. an unsupported/misapplied command) indistinguishable from a real, temporary error. ok, text = modem.execute("AT+CMEE=2") log.info("AT+CMEE=2 -> %s %r", ok, text) modem.init() # no echo, verbose +CME ERROR causes; see modem.INIT_COMMANDS async def run() -> None: """! @brief Start the TCP server and serve until cancelled. """ Loading
upper_tester/dummy_modem.py +16 −0 Changes for upper_tester/dummy_modem.py: 16 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -29,6 +29,8 @@ DEFAULT_SCRIPT = { "AT+CFUN=0": ["OK"], "AT+CFUN=1": ["OK"], "AT+CFUN=1,1": ["OK"], "AT": ["OK"], "AT+CPIN?": ["+CPIN: READY", "OK"], "AT+CFUN=4": ["OK"], "AT+CGATT=1": ["OK"], "AT+CGATT=0": ["OK"], Loading Loading @@ -88,9 +90,21 @@ class FakeModem: self.echo = echo ## Every command received so far, in order, for tests to assert against. self.received: list[str] = [] self._closed = False self._t = threading.Thread(target=self._run, daemon=True) self._t.start() def close(self) -> None: """! @brief Make the modem disappear, as a real one does when it reboots or is unplugged: whoever has @ref path open then reads an error or end-of-file. """ self._closed = True os.write(self._slave, b"\r") # wake the thread up so it sees _closed self._t.join(2) os.close(self.master) os.close(self._slave) def urc(self, line: str) -> None: """! @brief Inject an unsolicited result code (URC) at any time, independent of any Loading @@ -110,6 +124,8 @@ class FakeModem: buf += os.read(self.master, 1024) except OSError: return if self._closed: return while b"\r" in buf: raw, buf = buf.split(b"\r", 1) cmd = raw.decode() Loading
upper_tester/engine.py +9 −1 Changes for upper_tester/engine.py: 9 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -3,11 +3,15 @@ @brief Request execution: AT passthrough and MMI-to-AT mapping. """ import logging import re from .mmi import MMI_PROCEDURES, UnsupportedMmi, steps_for log = logging.getLogger("ut.engine") ## AT commands that reboot the modem (`AT+CFUN=<fun>,1`); see modem.AtModem.restart. RESTART_CMD = re.compile(r"AT\+CFUN=\d,1", re.IGNORECASE) ## Valid values of a request's `"cnf"` field, mirroring UpperTesterCommon.Ut_CnfReq_Type. CNF_MODES = ("NO_CNF_REQUIRED", "CNF_REQUIRED", "LOCAL_CNF_REQUIRED") Loading @@ -30,7 +34,9 @@ class Engine: def run(self, req: dict, on_written=None) -> tuple[bool, str]: """! @brief Execute one decoded request. Blocking; call from a worker thread. @details For `type: "AT"`, @p req["cmd"] is sent to the modem as-is. For @details For `type: "AT"`, @p req["cmd"] is sent to the modem as-is; if it reboots the modem (@ref RESTART_CMD), the call also waits until the modem is usable again (see modem.AtModem.restart). For `type: "MMI"`, @p req["cmd"] is run by its procedure if it has one in mmi.MMI_PROCEDURES (e.g. `ACTIVATE_BEARER`, see datacall.activate); otherwise it is resolved via mmi.steps_for into one or Loading Loading @@ -58,6 +64,8 @@ class Engine: timeout = req.get("timeout") if kind == "AT": log.debug("dispatch: AT passthrough: %r", cmd) if RESTART_CMD.fullmatch(cmd.replace("<CR>", "").strip()): return self.modem.restart(cmd, timeout, on_written) return self.modem.execute(cmd, timeout, on_written) params = {p["name"]: p["value"] for p in req.get("params") or []} log.debug("dispatch: MMI %s params=%s", cmd, params) Loading
upper_tester/modem.py +104 −18 Changes for upper_tester/modem.py: 104 added lines, 18 removed lines. Original line number Diff line number Diff line Loading @@ -32,6 +32,22 @@ TIMEOUTS = ( ## Timeout, in seconds, used for any command not listed in @ref TIMEOUTS. DEFAULT_TIMEOUT = 10.0 ## Line queued by the reader thread when the device disappears; it ends the AT command in ## flight at once (see @ref AtModem.execute) instead of leaving it to time out. DISCONNECTED = "+UT: modem disconnected" ## Commands that put the modem back in the state this service expects after it (re)starts: ## no command echo, and verbose `+CME ERROR` causes (3GPP TS 27.007 clause 9.1) instead of a ## bare `ERROR`. Both settings are lost when the modem reboots. INIT_COMMANDS = ("ATE0", "AT+CMEE=2") ## Seconds @ref AtModem.restart waits for the device to disappear after the restart command. RESTART_GONE_TIMEOUT = 15.0 ## Seconds @ref AtModem.restart waits for the device to come back and answer `AT`. RESTART_BACK_TIMEOUT = 90.0 ## Seconds @ref AtModem.restart then waits for the SIM to be ready (`+CPIN: READY`). RESTART_SIM_TIMEOUT = 20.0 ## Seconds between two polls in @ref AtModem.restart. RESTART_POLL_INTERVAL = 1.0 def timeout_for(cmd: str) -> float: """! Loading Loading @@ -82,12 +98,17 @@ class AtModem: 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 one AT @ref execute call that was in flight when the device vanished fails at once (nothing more will arrive for it), and every later command recovers automatically once the device reappears, without needing the whole process relaunched. The stale descriptor is closed as soon as the device vanishes: while it stays open the kernel keeps its `ttyUSB` number reserved, so the modem's ports come back under other numbers and the configured path (e.g. `/dev/ttyUSB2`) never reappears. A udev symlink (`/dev/quectel_at`, see `deploy/99-quectel.rules`) avoids depending on the numbering. """ def __init__(self, path: str, baud: int = 115200, open_retries: int = 10, open_retry_delay: float = 3.0): def __init__(self, path: str, baud: int = 115200, open_retries: int = 20, 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`, Loading @@ -109,6 +130,8 @@ class AtModem: self._stop_r = self._stop_w = -1 self._thread = None self._closing = False self._down = threading.Event() # set when the device disappears self._up = threading.Event() # set while the device is open def open(self) -> None: """! Loading Loading @@ -165,6 +188,24 @@ class AtModem: 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._up.set() def _drop_device(self) -> None: """! @brief Handle the device disappearing: close the stale descriptor and end the AT command in flight. @details Called from the reader thread without @ref _lock, which the command in flight holds. @ref _fd is set to -1 before the descriptor is closed, so a concurrent write fails cleanly instead of reaching a reused descriptor. """ self._up.clear() fd, self._fd = self._fd, -1 try: os.close(fd) except OSError: pass self._down.set() self._lines.put(DISCONNECTED) def close(self) -> None: """! Loading Loading @@ -203,7 +244,7 @@ class AtModem: if self._closing: return log.error("modem read error, device likely disconnected: %s", e) self._lines.put("+UT: modem disconnected, attempting to reconnect") self._drop_device() if not self._reconnect(): return # retries exhausted; _reconnect already logged why buf = b"" Loading @@ -217,7 +258,7 @@ class AtModem: if self._closing: return log.error("modem read EOF, device likely disconnected: %s", self.path) self._lines.put("+UT: modem disconnected, attempting to reconnect") self._drop_device() if not self._reconnect(): return buf = b"" Loading @@ -232,23 +273,16 @@ class AtModem: def _reconnect(self) -> bool: """! @brief Close the broken fd and retry (re)opening @ref path in place. @brief Retry (re)opening @ref path in place, after @ref _drop_device. @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. normally against the new fd) instead of racing the fd swap. The @ref execute call in flight when the device vanished has already been ended by @ref _drop_device, so the lock is free almost at once. @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: Loading @@ -275,8 +309,9 @@ class AtModem: @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, 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. no final result code arrived within @p timeout, ends with @ref DISCONNECTED if the device disappeared meanwhile, 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 Loading @@ -309,6 +344,57 @@ class AtModem: if line == echo: continue lines.append(line) if line == DISCONNECTED: log.warning("AT aborted, modem disconnected: %r", data) return False, "\r\n".join(lines) if is_final(line): log.info("AT< %s", lines) return line == "OK", "\r\n".join(lines) def init(self) -> None: """! @brief Run @ref INIT_COMMANDS; call it after @ref open and after any modem restart. """ for cmd in INIT_COMMANDS: ok, text = self.execute(cmd) log.info("%s -> %s %r", cmd, ok, text) def restart(self, cmd: str, timeout: float | None = None, on_written=None) -> tuple[bool, str]: """! @brief Run an AT command that restarts the modem (e.g. `AT+CFUN=1,1`) and wait until the modem is usable again. @details The modem answers `OK` and only then reboots: its device disappears a few seconds later and comes back once it has restarted. Any command sent in between is lost with the reboot. So after the `OK` this waits for the device to disappear (@ref RESTART_GONE_TIMEOUT; carried on regardless if it does not, e.g. on a pty), to be reopened by the reader thread and to answer `AT` (@ref RESTART_BACK_TIMEOUT), runs @ref init, then waits for the SIM (@ref RESTART_SIM_TIMEOUT). @param cmd The restart command. @param timeout As for @ref execute, for @p cmd itself. @param on_written As for @ref execute. @return `(success, text)` as @ref execute does for @p cmd, followed by a `+UT:` line: `success` is false if the modem did not answer `AT` again in time. A SIM that is not ready in time is only reported, not a failure. """ self._down.clear() ok, text = self.execute(cmd, timeout, on_written) if not ok: return ok, text if not self._down.wait(RESTART_GONE_TIMEOUT): log.warning("restart: %s did not disappear within %.0fs", self.path, RESTART_GONE_TIMEOUT) deadline = time.monotonic() + RESTART_BACK_TIMEOUT while not (self._up.wait(max(0.0, deadline - time.monotonic())) and self.execute("AT", 2.0)[0]): if time.monotonic() >= deadline: log.error("restart: modem not back within %.0fs", RESTART_BACK_TIMEOUT) return False, f"{text}\r\n+UT: modem not back after restart" time.sleep(RESTART_POLL_INTERVAL) self.init() deadline = time.monotonic() + RESTART_SIM_TIMEOUT while "+CPIN: READY" not in self.execute("AT+CPIN?")[1]: if time.monotonic() >= deadline: log.warning("restart: SIM not ready within %.0fs", RESTART_SIM_TIMEOUT) return True, f"{text}\r\n+UT: modem restarted, SIM not ready" time.sleep(RESTART_POLL_INTERVAL) log.info("restart: modem ready") return True, f"{text}\r\n+UT: modem restarted"
