#!/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. Most likely the modem is") print(" not forwarding to its serial port, which is what a wedged") print(" transparent-TCP session looks like: the slot is held by a") print(" connection that never closed.") print("\n Try, in order:") print(" 1. ACEmanager -> TCP Idle Timeout. If 0/disabled, a stale") print(" session holds the slot forever. 2 minutes is the value") print(" this project standardised on.") print(" 2. Reboot the modem. If that fixes it, the modem was") print(" holding state and the timeout is the permanent fix.") print(" 3. Check the unit's own screen — serial cable, power.") 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())