#!/usr/bin/env python3 """ mm_probe.py — answer "why can't we reach this unit?" in one command. THOR reports a failed connection as "disconnected" and nothing else. That single word covers at least four completely different faults with four different fixes, and telling them apart is the difference between a modem reboot and a site visit: * **connection refused** something answered and said no — wrong port, or the modem is refusing a further session * **connect timed out** nothing answered at all — trusted-IP whitelist, firewall, or the modem is off the network * **connected, no reply** the MODEM answered but the unit did not. The TCP path is fine; the modem is not forwarding to serial. This is the signature of a wedged transparent-TCP session, and it is the one THOR cannot distinguish from any of the others * **replied** the unit is alive; the problem is upstream software Read-only. It sends `POLL`, then optionally `SERIAL` and the state read — the same three commands THOR's own connection check uses — and never writes. Usage ----- python3 bridges/mm_probe.py 63.45.161.30:9034 python3 bridges/mm_probe.py 10.0.0.8:12345 --timeout 5 python3 bridges/mm_probe.py --slots 3 `--slots N` opens N connections at once and reports how many the far end accepts. A transparent-TCP modem typically serves **one** session; if the first succeeds and the rest are refused or hang, that confirms the single-slot behaviour and explains why a leaked session takes a unit offline until the slot frees. Works for both series: a Series III reply opens `DLE STX`, a Micromate reply opens with a bare `STX`, so the probe also tells you which one answered. """ from __future__ import annotations import argparse import socket import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from minimateplus.framing import build_bw_frame # noqa: E402 DLE, STX, ETX = 0x10, 0x02, 0x03 def destuff(raw: bytes) -> bytes: """Strip framing and DLE escapes; return the payload without its checksum.""" i = 1 if raw and raw[0] == STX else (2 if len(raw) > 1 and raw[1] == STX else 0) out = bytearray() while i < len(raw): b = raw[i] if b == DLE and i + 1 < len(raw): out.append(raw[i + 1]) i += 2 continue if b == ETX: break out.append(b) i += 1 return bytes(out[:-1]) if len(out) > 1 else b"" # Reads are two-step on Series III: a probe at offset 0, then a data read at the # block's length. THOR sends these offsets, and they also work on a Micromate. OFFSETS = {0x5B: 0x0030, 0x15: 0x000A, 0x49: 0xFFFF} def exchange(sock: socket.socket, sub: int, timeout: float) -> tuple[bytes, float]: sock.sendall(build_bw_frame(sub, OFFSETS.get(sub, 0))) t0 = time.time() buf, deadline = b"", t0 + timeout sock.settimeout(0.3) while time.time() < deadline: try: chunk = sock.recv(4096) if not chunk: break buf += chunk if buf.endswith(bytes([ETX])) and len(buf) > 8: break except TimeoutError: continue except OSError: break return buf, time.time() - t0 def step(n: int, label: str, result: str) -> None: print(f" [{n}] {label:.<28} {result}") def probe(host: str, port: int, timeout: float) -> int: print(f"\ntarget {host}:{port} (read-only: POLL, SERIAL, state)\n") # ── 1. TCP ──────────────────────────────────────────────────────────── t0 = time.time() try: sock = socket.create_connection((host, port), timeout=timeout) except ConnectionRefusedError: step(1, "TCP connect", f"REFUSED after {1000*(time.time()-t0):.0f} ms") print("\nverdict: something answered and actively refused.") print(" Not a silent firewall drop — the host is reachable.") print(" Wrong port, the service is down, or the modem is refusing") print(" an additional session because its one slot is in use.") return 2 except (TimeoutError, socket.timeout): step(1, "TCP connect", f"TIMED OUT after {time.time()-t0:.1f} s") print("\nverdict: nothing answered at all.") print(" A silent drop, which is what a trusted-IP whitelist looks") print(" like — it discards rather than refuses. Check the modem's") print(" Trusted IPs (and note a VPN changes the IP you arrive from),") print(" the firewall, and whether the modem is on the network.") return 3 except OSError as e: step(1, "TCP connect", f"FAILED: {e}") return 4 step(1, "TCP connect", f"ok ({1000*(time.time()-t0):.0f} ms)") # ── 2. POLL ─────────────────────────────────────────────────────────── try: raw, dt = exchange(sock, 0x5B, timeout) except OSError as e: step(2, "POLL", f"send failed: {e}") sock.close() return 4 if not raw: step(2, "POLL", f"NO REPLY in {timeout:.1f} s") print("\nverdict: the MODEM answered but the unit did not.") print(" TCP is fine end to end — something accepted the connection.") print(" What is missing is the serial side. Two quite different") print(" causes produce this, and they are NOT distinguishable from") print(" here:") print("\n 1. SOMEONE ELSE HOLDS THE SESSION. These modems bridge ONE") print(" TCP session to serial at a time. A second connection is") print(" accepted and then simply not forwarded. Confirmed 2026-09-26:") print(" with THOR connected this probe saw exactly this; the moment") print(" THOR disconnected the same probe returned the serial number.") print(" ** Check whether THOR (or anything else) has the unit first. **") print("\n 2. The serial path is genuinely broken — a stale session the") print(" modem never released, a cable the unit cannot enumerate, or") print(" a unit that is off.") print("\n Try, in order:") print(" 1. Disconnect any other client and re-probe. If it answers,") print(" it was contention, not a fault.") print(" 2. The cable's chipset. A Micromate drives FTDI and CDC-ACM") print(" only — a Prolific PL2303 gives it no serial port at all.") print(" lsusb: FTDI is 0403, Prolific 067b.") print(" 3. Power-cycle the UNIT with the cable attached (hold power") print(" 5 s, through the two-stage prompt). Its USB host rescans") print(" on cold boot; it may not on hot-swap.") print(" 4. AirLink OS -> TCP Idle Timeout. If 0/disabled, a stale") print(" session holds the slot indefinitely. 2 minutes is the") print(" value this project standardised on.") sock.close() return 5 series = "Series III (DLE STX)" if raw[0] == DLE else "Micromate (bare STX)" step(2, "POLL", f"reply {len(raw)} B in {1000*dt:.0f} ms") p = destuff(raw) ok = len(p) > 3 and p[2] == 0xFF - 0x5B step(3, "frame", f"{'valid' if ok else 'MALFORMED'}, {series}") if not ok: print("\nverdict: something replied, but not a seismograph.") print(" Another service is on this port, or the modem is in a mode") print(" that injects its own text (check Quiet Mode / AT echo).") print(f" first bytes: {raw[:16].hex(' ')}") sock.close() return 6 # ── 3. identity + state ─────────────────────────────────────────────── for n, (sub, label) in enumerate(((0x15, "serial"), (0x49, "state")), start=4): try: r, dt = exchange(sock, sub, timeout) d = destuff(r)[5:] if sub == 0x15: # serial is a null-terminated run; a further field follows it serial = bytes(d[11:]).split(b"\x00")[0] step(n, label, serial.decode("ascii", "replace") or "(empty)") else: step(n, label, "MONITORING" if len(d) > 11 and d[11] else "idle") except OSError: step(n, label, "no reply") sock.close() print("\nverdict: the unit is alive and answering.") print(" If THOR still shows it disconnected, the fault is in THOR, not") print(" the network or the device.") return 0 def slots(host: str, port: int, n: int, timeout: float) -> None: print(f"\nopening {n} simultaneous connections to {host}:{port}\n") held = [] for i in range(n): try: s = socket.create_connection((host, port), timeout=timeout) held.append(s) step(i + 1, f"connection {i+1}", "accepted") except ConnectionRefusedError: step(i + 1, f"connection {i+1}", "REFUSED") except (TimeoutError, socket.timeout): step(i + 1, f"connection {i+1}", "timed out") except OSError as e: step(i + 1, f"connection {i+1}", f"failed: {e}") print(f"\n{len(held)} of {n} accepted.") if len(held) == 1: print(" Single-slot behaviour confirmed — this far end serves ONE") print(" session at a time. A connection that is never closed takes") print(" the unit offline until the idle timeout frees the slot.") for s in held: s.close() def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("target", help="host:port, e.g. 63.45.161.30:9034") ap.add_argument("--timeout", type=float, default=10.0) ap.add_argument("--slots", type=int, metavar="N", help="open N simultaneous connections to test single-slot behaviour") a = ap.parse_args() host, _, port = a.target.rpartition(":") if not host: ap.error("target must be host:port") if a.slots: slots(host, int(port), a.slots, a.timeout) return 0 return probe(host, int(port), a.timeout) if __name__ == "__main__": raise SystemExit(main())