istrain-public/scripts/iq_retune.py

#!/usr/bin/env python3
# iq_retune.py — the NO-CHURN hop. Opens dongle 1001 ONCE via librtlsdr and retunes in place
# (rtlsdr_set_center_freq) to step 452.9375 <-> 457.9375, instead of killing+reopening rtl_sdr every
# dwell. Tonight's wedge (1001 dying after a reopen, esp. through the VirtualBox USB passthrough) was
# caused by the close/reopen; retuning a live device never closes it, so it should sidestep the wedge.
#
# Reuses the PROVEN channelizer from iq_channelize.py (SubChannel: NCO mix -> FIR-LPF -> decimate ->
# in-channel-power presence -> discriminated clip), so each dwell still splits into:
#     452 -> BOT (center) + MID (±12.5k) ;  457 -> EOT (center) + MID (±12.5k)
# and feeds bot/eot/midtrain logs + the live meter exactly like iq_channelize does.
#
# RUNTIME helper — built by the maintainer, operated as a service. numpy only (DSP) + librtlsdr (ctypes).
# Bench smoke-test (no device touch):  python3 iq_retune.py --check
# Live (gated; borrows 1001 from the FM hop):  python3 iq_retune.py
#
# STATUS 2026-06-24: built, DSP reused from iq_channelize (which passed --selftest). The retune loop /
# ctypes device path is UNTESTED against hardware — the live run is the test of whether retune-in-place
# avoids the wedge. Do not trust it until it's baked on the device.
import os, sys, time, ctypes
import numpy as np

# reuse the proven DSP + meter writer (same dir; underscore name is importable, unlike scan-hop.py)
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from iq_channelize import dwell_channels, write_level, CHUNK, FS_IN  # noqa: E402

DWELL_SEC = float(os.environ.get("RETUNE_DWELL", "28"))
SERIAL    = os.environ.get("RETUNE_SERIAL", "1001")
GAIN_T    = int(os.environ.get("RETUNE_GAIN_TENTHS", "400"))   # tenths of dB (~40 dB), 0 => auto
CENTERS   = (("452", 452_937_500), ("457", 457_937_500))       # dwell tag -> center Hz
SETTLE_BUFS = 2                                                # discard N buffers right after a retune

_LIB = "librtlsdr.so.0"


def _load_lib():
    rtl = ctypes.CDLL(_LIB)
    v = ctypes.c_void_p
    rtl.rtlsdr_open.argtypes = [ctypes.POINTER(v), ctypes.c_uint32]
    rtl.rtlsdr_open.restype = ctypes.c_int
    rtl.rtlsdr_close.argtypes = [v]
    rtl.rtlsdr_set_sample_rate.argtypes = [v, ctypes.c_uint32]
    rtl.rtlsdr_set_center_freq.argtypes = [v, ctypes.c_uint32]
    rtl.rtlsdr_set_tuner_gain_mode.argtypes = [v, ctypes.c_int]
    rtl.rtlsdr_set_tuner_gain.argtypes = [v, ctypes.c_int]
    rtl.rtlsdr_reset_buffer.argtypes = [v]
    rtl.rtlsdr_read_sync.argtypes = [v, ctypes.c_void_p, ctypes.c_int, ctypes.POINTER(ctypes.c_int)]
    rtl.rtlsdr_read_sync.restype = ctypes.c_int
    # serial -> index, with graceful fallback if the symbol is absent
    try:
        rtl.rtlsdr_get_index_by_serial.argtypes = [ctypes.c_char_p]
        rtl.rtlsdr_get_index_by_serial.restype = ctypes.c_int
        rtl.rtlsdr_get_device_count.restype = ctypes.c_int
    except AttributeError:
        pass
    return rtl


def _open_device(rtl):
    idx = -1
    try:
        idx = rtl.rtlsdr_get_index_by_serial(SERIAL.encode())
    except Exception:
        idx = -1
    if idx < 0:
        idx = 1  # 1001 enumerated as index 1 alongside 1002; fallback only
        sys.stderr.write("[iq_retune] serial %s not matched, falling back to index %d\n" % (SERIAL, idx))
    dev = ctypes.c_void_p()
    if rtl.rtlsdr_open(ctypes.byref(dev), idx) != 0:
        raise RuntimeError("rtlsdr_open failed for index %d" % idx)
    rtl.rtlsdr_set_sample_rate(dev, FS_IN)
    if GAIN_T > 0:
        rtl.rtlsdr_set_tuner_gain_mode(dev, 1)
        rtl.rtlsdr_set_tuner_gain(dev, GAIN_T)
    else:
        rtl.rtlsdr_set_tuner_gain_mode(dev, 0)
    return dev


def _to_complex(buf, n):
    u = np.frombuffer(bytes(buf[:n]), dtype=np.uint8).astype(np.float64)
    return ((u[0::2] - 127.5) + 1j * (u[1::2] - 127.5)) / 127.5


def run():
    rtl = _load_lib()
    dev = _open_device(rtl)
    sets = {tag: dwell_channels(tag) for tag, _ in CENTERS}   # build channelizers once; reuse across retunes
    nbytes = CHUNK * 2
    buf = (ctypes.c_ubyte * nbytes)()
    nread = ctypes.c_int()
    sys.stderr.write("[iq_retune] open OK; retune-hop %s, dwell %.0fs, fs %d — device stays open\n"
                     % ("/".join(t for t, _ in CENTERS), DWELL_SEC, FS_IN)); sys.stderr.flush()
    try:
        while True:
            for tag, center in CENTERS:
                rtl.rtlsdr_set_center_freq(dev, center)     # IN-PLACE retune — no close/reopen
                rtl.rtlsdr_reset_buffer(dev)
                chans = sets[tag]
                discard = SETTLE_BUFS
                t_end = time.time() + DWELL_SEC
                sys.stderr.write("[iq_retune] tune %s (%d Hz) for %.0fs\n" % (tag, center, DWELL_SEC))
                sys.stderr.flush()
                while time.time() < t_end:
                    if rtl.rtlsdr_read_sync(dev, buf, nbytes, ctypes.byref(nread)) < 0:
                        break
                    if nread.value < nbytes:
                        continue
                    if discard > 0:                          # drop the post-retune settling buffers
                        discard -= 1
                        continue
                    iq = _to_complex(buf, nread.value)
                    now = time.time()
                    for ch in chans:
                        ch.process(iq, now)
                    hot = max(chans, key=lambda c: c.last_db)
                    write_level(now, hot.last_db, any(c.active for c in chans),
                                hot.label, (hot.peak if hot.active else None), hot.floor)
    finally:
        rtl.rtlsdr_close(dev)


def check():
    """Bench smoke-test: lib loads, symbols resolve, DSP imports — no device opened."""
    ok = True
    try:
        rtl = _load_lib()
        for sym in ("rtlsdr_open", "rtlsdr_set_center_freq", "rtlsdr_read_sync", "rtlsdr_reset_buffer"):
            present = hasattr(rtl, sym)
            print(("  PASS " if present else "  FAIL ") + "librtlsdr." + sym)
            ok = ok and present
    except OSError as e:
        print("  FAIL load librtlsdr:", e); ok = False
    try:
        s = dwell_channels("452")
        print("  PASS dwell_channels('452') -> %d sub-channels (%s)" % (len(s), ",".join(c.label for c in s)))
    except Exception as e:
        print("  FAIL channelizer import:", e); ok = False
    print("RESULT:", "PASS — bench wiring OK (device path still needs a live bake)" if ok else "FAIL")
    return 0 if ok else 1


if __name__ == "__main__":
    if "--check" in sys.argv[1:]:
        sys.exit(check())
    run()