istrain-public/scripts/iq_hop.py

#!/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()