istrain-public/scripts/probe-band.py

#!/usr/bin/env python3
"""probe-band — console first-light: tune a dongle to a band, snapshot the spectrum, print
what's there. The "can we hear anything?" tool, before we build decoders or dashboard panels.

  a RADIO RUNTIME, not the code author. This OPERATES a radio (tunes + samples). Run it yourself; the maintainer builds it.
  a generic band probe is engine-level (every radio project
  wants "what's on this band, and does my receiver even work"). Only the presets below are profile.

Recommended order — validate the KNOWN-GOOD first, then the target:
  1. noaa  — 162.400-162.550 MHz · NOAA Weather, always-on. If we see THIS, the receiver +
             antenna + chain all work. On 1002 (the voice stick, 18" whip — it's 162 MHz).
  2. eot   — 457.5-458.5 MHz · EOT/FRED telemetry (457.9375) on 1001 (~6" whip, matched). Sparse
             — may be quiet with no train near. Seeing the noise floor cleanly still proves the path.

Usage:
  ./probe-band.py noaa            # known-good sanity check first
  ./probe-band.py eot
  SERIAL=1002 LO=162.4M HI=162.55M GAIN=40 SECONDS=8 ./probe-band.py   # fully manual
  GAIN env: "auto" or a dB number (try 30-49 for weak signals). Take notes per run.
"""
import os, sys, glob, shutil, subprocess, tempfile, statistics

PRESETS = {
    #  name :  (serial, lo,       hi,        note)
    "noaa": ("1002", "162.400M", "162.550M", "NOAA Weather — always-on KNOWN-GOOD sanity check (voice stick)"),
    "eot":  ("1001", "457.500M", "458.500M", "EOT/FRED 457.9375 — sparse, may be quiet"),
}

preset = sys.argv[1] if len(sys.argv) > 1 and not sys.argv[1].startswith("-") else None
p = PRESETS.get(preset, (None, None, None, "manual"))

SERIAL  = os.environ.get("SERIAL", p[0] or "1001")
LO      = os.environ.get("LO", p[1] or "162.400M")
HI      = os.environ.get("HI", p[2] or "162.550M")
BIN     = os.environ.get("BIN", "3k")
SECONDS = os.environ.get("SECONDS", "8")
GAIN    = os.environ.get("GAIN", "auto")
TOPN    = int(os.environ.get("TOPN", "12"))
NOTE    = p[3]


def die(msg, code=1):
    print(f"probe-band: {msg}", file=sys.stderr); sys.exit(code)


def preflight():
    if not shutil.which("rtl_power"):
        die("rtl_power not found — install the 'rtl-sdr' package.")
    for d in glob.glob("/sys/bus/usb/devices/*"):
        try:
            if open(os.path.join(d, "idVendor")).read().strip() != "0bda": continue
            if open(os.path.join(d, "idProduct")).read().strip() != "2838": continue
            if open(os.path.join(d, "serial")).read().strip() != SERIAL: continue
        except OSError:
            continue
        for intf in glob.glob(d + ":*"):
            link = os.path.join(intf, "driver")
            if os.path.islink(link) and os.path.basename(os.readlink(link)) == "dvb_usb_rtl28xxu":
                die(f"dongle {SERIAL} is DVB-claimed — free it: "
                    f"sudo modprobe -r rtl2832_sdr dvb_usb_rtl28xxu")
        return
    die(f"dongle serial {SERIAL} not present (check the radios panel / `rtl_test`).")


def run_sweep(csv_path):
    cmd = ["rtl_power", "-d", SERIAL, "-f", f"{LO}:{HI}:{BIN}", "-i", SECONDS, "-1", csv_path]
    if GAIN != "auto":
        cmd[3:3] = ["-g", GAIN]
    print(f"# probing {LO}-{HI} @ {BIN}, {SECONDS}s, dongle :{SERIAL} (gain={GAIN})", file=sys.stderr)
    print(f"# {NOTE}", file=sys.stderr)
    try:
        subprocess.run(cmd, check=True)
    except subprocess.CalledProcessError as e:
        die(f"rtl_power exited {e.returncode} — dongle free? gain valid? band in range (24-1766 MHz)?")


def parse_bins(csv_path):
    bins = []
    with open(csv_path) as f:
        for line in f:
            parts = [p.strip() for p in line.split(",")]
            if len(parts) < 7: continue
            try:
                lo, step = float(parts[2]), float(parts[4])
                vals = [float(v) for v in parts[6:] if v]
            except ValueError:
                continue
            for i, dbm in enumerate(vals):
                bins.append(((lo + (i + 0.5) * step) / 1e6, dbm))
    return bins


def main():
    preflight()
    with tempfile.NamedTemporaryFile(suffix=".csv", delete=False) as tf:
        csv_path = tf.name
    try:
        run_sweep(csv_path)
        bins = parse_bins(csv_path)
    finally:
        try: os.unlink(csv_path)
        except OSError: pass
    if not bins:
        die("no spectrum parsed from rtl_power.")

    floor = statistics.median(d for _, d in bins)
    peak_f, peak_d = max(bins, key=lambda b: b[1])
    top = sorted(bins, key=lambda b: b[1], reverse=True)[:TOPN]
    span = peak_d - floor
    verdict = ("SIGNAL — clear carrier above the floor" if span > 10 else
               "maybe — slight rise, tune gain / dwell longer" if span > 5 else
               "just noise — nothing above the floor here")

    print(f"\nProbe {LO}-{HI}  ·  dongle :{SERIAL}  ·  {NOTE}")
    print(f"noise floor ~{floor:.1f} dBm   ·   peak {peak_d:.1f} dBm @ {peak_f:.4f} MHz "
          f"(+{span:.1f} dB)   ·   {verdict}\n")
    print(f"  {'freq (MHz)':>11}   {'dBm':>7}   {'above floor':>11}")
    print("  " + "-" * 38)
    for f_mhz, dbm in top:
        print(f"  {f_mhz:>11.4f}   {dbm:>7.1f}   {dbm - floor:>+10.1f}")
    print("\n  next: paste this back and we'll read it + pick the next gain. (write each run down)\n")


if __name__ == "__main__":
    main()