#!/usr/bin/env python3 # iq_hop.py — RETUNE-IN-PLACE 452/457 hop on dongle 1001, single never-closed USB handle. # # WHY THIS EXISTS. The old scan-hop tears down + respawns the capture (rtl_fm/rtl_sdr -d 1001) every # dwell. That reopen — rtlsdr_open() + USB claim/reset on the fragile Realtek 1001 — is what wedged the # dongle two nights running (06-24, 06-25): "one good read per open, stream stalls after ~70 reopens." # The retune itself was never the problem; the open/close USB transaction is. So here we open ONCE and # never close: rtl_tcp holds 1001 open forever and streams CU8 IQ on a local socket; we flip 452<->457 by # sending its SET_FREQUENCY command (a tuner I2C write on the already-claimed handle) — zero reopen. # # WHAT IT REUSES. The detection DSP is iq_channelize.py UNCHANGED (imported): per-dwell SubChannel set # (center bot/eot + the two +/-12.5k mid-train DPU sub-channels), adaptive floor+margin power presence, # FM-discriminated s16 clips for eot_scan, and the vumon-shaped level.json. Same outputs, same logs # (bot.jsonl / eot.jsonl / midtrain.jsonl) — the dashboard and decoder don't know anything changed. # # WHAT IT ADDS (the only new logic): own rtl_tcp's lifecycle, send the retune command, DRAIN a short # settle window after each retune (stale in-flight IQ from the old frequency is still buffered), route # live IQ into the active dwell's channel set, and FLUSH any burst still open at the dwell boundary so it # can't bleed across the ~30s the band is unwatched. # # Prove the routing + boundary-flush with NO radio: python3 iq_hop.py --selftest # (the DSP's channel isolation is already proven by iq_channelize.py --selftest) # # RUNTIME helper — built by the maintainer, operated as the istrain-scan-hop service. numpy only. # (a generic "one open, retune-in-place, time-division channelizer"). import os, sys, time, signal, socket, struct, select, subprocess sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import iq_channelize as iqc # the DSP: SubChannel, dwell_channels, FS_IN, CHUNK, write_level # --- knobs (mirror scan-hop's names so operating it feels the same) ---------------------------------- DWELL_452 = int(os.environ.get("HOP_DWELL_452", "7")) # BOT/head — detection only (no decoder yet); lighter DWELL_457 = int(os.environ.get("HOP_DWELL_457", "14")) # EOT/tail — the decodable ID + the departure bit; weight it SETTLE = float(os.environ.get("HOP_SETTLE", "0.4")) # post-retune drain: discard stale in-flight IQ GAIN_DB = float(os.environ.get("HOP_GAIN", "40")) # manual tuner gain (dB); rtl_tcp wants tenths FS = iqc.FS_IN # 240000 — MUST match the DSP's expected rate PORT = int(os.environ.get("HOP_TCP_PORT", "12001")) # local rtl_tcp port (mnemonic: 1001) DEV = os.environ.get("HOP_DEV", "1001") # dongle serial/index NBYTES = iqc.CHUNK * 2 # CU8 bytes per processing chunk (I,Q uint8) # (dwell tag, band center in Hz) — center == the bot/eot sub-channel; mid rides +/-12.5k off it. DWELLS = [("452", 452_937_500, DWELL_452), ("457", 457_937_500, DWELL_457)] # rtl_tcp control protocol: 1-byte command + big-endian uint32 param (struct command{u8;u32;}__packed__). SET_FREQUENCY, SET_SAMPLERATE, SET_GAIN_MODE, SET_GAIN, SET_AGC_MODE = 0x01, 0x02, 0x03, 0x04, 0x08 def send_cmd(sock, cmd, param): sock.sendall(struct.pack(">BI", cmd, param & 0xFFFFFFFF)) def recv_exact(sock, n): """Block until exactly n bytes are read; None if the stream ends (rtl_tcp died).""" buf = bytearray() while len(buf) < n: b = sock.recv(n - len(buf)) if not b: return None buf += b return bytes(buf) def drain(sock, secs): """Read and DISCARD everything arriving for `secs` — clears stale old-frequency IQ that was already in flight (USB transfer + socket buffers) when we retuned, so it can't pollute the new dwell.""" end = time.monotonic() + secs # O-6 FIX: drain length on the MONOTONIC clock — a wall-clock while True: # step (NTP slew/jump) must not collapse the drain to instant left = end - time.monotonic() if left <= 0: return True r, _, _ = select.select([sock], [], [], left) if r and not sock.recv(65536): return False # rtl_tcp closed under us def run_dwell(sock, chans, secs): """Feed live IQ into this band's SubChannel set for `secs`, then flush any open burst. Returns False if the stream died. State (adaptive floor, etc.) persists across visits to the same band by design.""" end = time.monotonic() + secs # O-6 FIX: dwell LENGTH off the monotonic clock (step-proof). while time.monotonic() < end: # the per-sample `now = time.time()` below stays wall-clock — # those are real-world data stamps for bot/eot/mid logs. raw = recv_exact(sock, NBYTES) if raw is None: return False u = iqc.np.frombuffer(raw, dtype=iqc.np.uint8).astype(iqc.np.float64) iq = ((u[0::2] - 127.5) + 1j * (u[1::2] - 127.5)) / 127.5 now = time.time() for ch in chans: ch.process(iq, now) hot = max(chans, key=lambda c: c.last_db) # meter = hottest sub-channel (vumon-shaped) iqc.write_level(now, hot.last_db, any(c.active for c in chans), hot.label, (hot.peak if hot.active else None), hot.floor) now = time.time() for ch in chans: # flush: don't let a burst span the off-period if ch.active: ch.active = False ch._emit(now) return True def connect(retries=40, delay=0.5): """rtl_tcp enumeration is racy on this VBox passthrough (airband needed 20 retries at boot) — keep trying to connect, then read+discard rtl_tcp's 12-byte 'RTL0' magic header.""" for i in range(retries): try: s = socket.create_connection(("127.0.0.1", PORT), timeout=2) hdr = recv_exact(s, 12) if not hdr or hdr[:4] != b"RTL0": s.close() raise OSError("bad/no rtl_tcp header") sys.stderr.write("[iq_hop] connected to rtl_tcp (tuner header ok)\n"); sys.stderr.flush() return s except OSError: if i == retries - 1: raise time.sleep(delay) def main(): if "--selftest" in sys.argv[1:]: sys.exit(selftest()) os.makedirs(iqc.IST, exist_ok=True) # one open, held forever: rtl_tcp claims 1001 and streams; we never reopen, only retune over the socket. rtl = subprocess.Popen( ["rtl_tcp", "-d", DEV, "-a", "127.0.0.1", "-p", str(PORT), "-s", str(FS)], preexec_fn=os.setsid) sys.stderr.write("[iq_hop] launched rtl_tcp (dev=%s fs=%d port=%d) — single open, retune-in-place\n" % (DEV, FS, PORT)); sys.stderr.flush() def shutdown(code=0): try: os.killpg(os.getpgid(rtl.pid), signal.SIGTERM) try: rtl.wait(timeout=3) except Exception: # rtl_tcp can wedge in USB teardown and shrug off SIGTERM — every stop then stalls # the full TimeoutStopSec until systemd SIGKILLs (observed 07-11 and 07-18). Finish it. os.killpg(os.getpgid(rtl.pid), signal.SIGKILL) except Exception: pass sys.exit(code) signal.signal(signal.SIGTERM, lambda *_: shutdown(0)) # systemctl stop = clean exit, no restart signal.signal(signal.SIGINT, lambda *_: shutdown(0)) try: sock = connect() send_cmd(sock, SET_SAMPLERATE, FS) send_cmd(sock, SET_GAIN_MODE, 1) # manual gain (no AGC surprises) send_cmd(sock, SET_GAIN, int(round(GAIN_DB * 10))) # rtl_tcp gain is tenths of a dB send_cmd(sock, SET_AGC_MODE, 0) chans = {tag: iqc.dwell_channels(tag) for tag, *_ in DWELLS} # persistent per-band detector sets sys.stderr.write("[iq_hop] 452/457 retune-in-place — BOT %ss / EOT %ss (weighted), %.1fs settle, gain %.0fdB\n" % (DWELL_452, DWELL_457, SETTLE, GAIN_DB)); sys.stderr.flush() while True: for tag, center, dwell in DWELLS: send_cmd(sock, SET_FREQUENCY, center) # retune on the SAME open handle — no reopen if not drain(sock, SETTLE) or not run_dwell(sock, chans[tag], dwell): sys.stderr.write("[iq_hop] rtl_tcp stream ended — exiting 1 for service restart\n") shutdown(1) except Exception as e: # Failure MUST exit nonzero or Restart=on-failure never fires and the rig stays down until a # hand starts it (2026-07-18: boot raced USB enumeration, connect() timed out, the old # `finally: shutdown()` swallowed the OSError into exit 0 — 1001 dead all morning). sys.stderr.write("[iq_hop] fatal: %s — exiting 1 so systemd retries\n" % e) shutdown(1) # ---- synthetic self-test: prove dwell-routing + boundary-flush WITHOUT a radio ---------------------- def selftest(): """Two fake dwells fed synthetic CU8 (reusing iq_channelize's burst generator). Asserts: a center burst during the 452 dwell lands in the 452/center channel and NOT the 457 one (and vice-versa), and a burst still active when the dwell ends gets flushed (emitted), not lost. DSP isolation itself is covered by iq_channelize.py --selftest; here we only exercise iq_hop's routing/flush wrapper.""" np = iqc.np def fresh_set(): cs = [iqc.SubChannel(0, "center", None, None), iqc.SubChannel(-12500, "mid_lo", None, None), iqc.SubChannel(+12500, "mid_hi", None, None)] hits = {c.label: [] for c in cs} for c in cs: def cap(now, _c=c): hits[_c.label].append((round(_c.since, 2), round(now - _c.since, 2))) c._emit = cap # capture emits in memory (no files) return cs, hits def feed(chans, sig, t0): """Push a complex baseband signal through `chans` as CU8 chunks on a synthetic clock from t0.""" u = np.empty(2 * len(sig)) u[0::2] = np.real(sig) * 127.5 + 127.5 u[1::2] = np.imag(sig) * 127.5 + 127.5 cu8 = np.clip(u, 0, 255).astype(np.uint8).tobytes() pos = 0; k = 0 while pos + NBYTES <= len(cu8): raw = cu8[pos:pos + NBYTES]; pos += NBYTES arr = np.frombuffer(raw, np.uint8).astype(np.float64) iq = ((arr[0::2] - 127.5) + 1j * (arr[1::2] - 127.5)) / 127.5 now = t0 + k * iqc.CHUNK / FS for c in chans: c.process(iq, now) k += 1 return t0 + k * iqc.CHUNK / FS # end-of-dwell clock dur = 1.2; n = int(FS * dur); t = np.arange(n) / FS rng = np.random.default_rng(7) noise = lambda: 0.05 * (rng.standard_normal(n) + 1j * rng.standard_normal(n)) set452, h452 = fresh_set() set457, h457 = fresh_set() # 452 dwell: a center burst 0.10-0.25 → must land in 452/center, never touch the 457 set. s452 = noise() + iqc._fsk_burst(t, 0.0, 0.10, 0.25) end452 = feed(set452, s452, 0.0) for c in set452: # boundary flush if c.active: c.active = False; c._emit(end452) # 457 dwell: a center burst that runs to the END (0.9→past 1.2) → must be FLUSHED at the boundary. s457 = noise() + iqc._fsk_burst(t, 0.0, 0.90, 1.5) end457 = feed(set457, s457, end452) for c in set457: if c.active: c.active = False; c._emit(end457) def near(hitlist, lo, hi): return any(lo - 0.15 <= h[0] <= hi + 0.05 for h in hitlist) ok = True def check(name, cond): nonlocal ok; print((" PASS " if cond else " FAIL ") + name); ok = ok and cond print("self-test — 452 dwell: center burst 0.10-0.25 ; 457 dwell: center burst 0.90-end (flush case)") print("452 detections:", {k: v for k, v in h452.items()}) print("457 detections:", {k: v for k, v in h457.items()}) check("452/center caught its burst", near(h452["center"], 0.10, 0.25)) check("452/mid channels stayed quiet", not h452["mid_lo"] and not h452["mid_hi"]) check("457/center caught its burst", near(h457["center"], end452 + 0.90, end452 + 0.90)) # the 457 burst runs to the end of the dwell and never goes quiet, so the ONLY thing that can emit it # is the boundary flush: emitted-and-running-to-the-edge (since + dur ~= end457) proves flush worked. flushed = bool(h457["center"]) and abs((h457["center"][0][0] + h457["center"][0][1]) - end457) < 0.15 check("boundary burst was FLUSHED, not lost", flushed) print("RESULT:", "PASS — routing + boundary-flush hold" if ok else "FAIL") return 0 if ok else 1 if __name__ == "__main__": main()