File: CLAUDE.md
Scanner (`scanner/`):
coverage, `prefer_serials_for(cfg, hz)` ranks tuners for a single-freq op.
- `bandplan.py` — freq→allocation hints.
+### Inline converters (Ham It Up / Ham It Down)
+
+An antenna row can declare an inline converter (`antennas.converter`:
+`ham_it_up` +125 MHz / 100 kHz–65 MHz passband, `ham_it_down` −1500 MHz /
+1.55–3.1 GHz — the catalog is `antenna.CONVERTERS`; picked in web Settings →
+Antennas). The invariant: **every frequency everywhere — catalog, detection,
+demod, UI — is real RF**; the LO offset is applied ONLY at the hardware-tune
+boundary. `Config.converter_offset_hz` (runtime-only, never persisted) carries
+it: `rtl.capture_iq`/`capture_iq_to_file`/`run_rtl_power` add it to `-f`
+(rtl_power also shifts returned freqs back), `wideband.peek_session`/
+`dwell_session` add it to `sess.tune`, `decoders.decode_live` to its rtl_fm.
+Neither Nooelec device inverts the spectrum, so nothing else changes. Wiring:
+the daemon gives each worker a per-tuner cfg copy (`_tuner_cfg` in `run_once`);
+the API/recorder look the offset up AFTER leasing (`_dev_cfg`/`_conv_offset` in
+`api/main.py`, `antenna.converter_offset_for`). Planning: `run_once` extends
+the pass's bands with `antenna.converter_bands` (the converter passband minus
+what the plan already covers — Ham It Up adds ~0.1–52 MHz), computed from FREE
+tuners only, and clips every tuner's share to `antenna.tunable_window` (a
+converter tuner is deaf outside its passband — the Ham It Up input low-passes
+at 65 MHz — and a bare dongle can't tune HF; `NATIVE_RANGE` = 24–2200 MHz), so
+a busy/absent converter tuner never strands HF on hardware that can't reach it.
+`coverage_score` returns 0 outside a converter's window and ≥0.9 inside, so
+single-freq ops route HF to the converter dongle automatically. `bandplan.py`
+knows the HF allocations (ham bands LSB<10 MHz/USB≥10 MHz convention in
+`guess_mode`, SWBC/WWV/CB notes, priorities, filter bands). Explicit scoped
+scans (`--start/--stop`) skip the band extension but still clip.
+
### Antenna routing (multi-tuner)
`scanner/daemon.py` scans **all present tuners in parallel** — `plan_scan`
File: api/main.py
import shutil
import subprocess
import sys
+from dataclasses import replace
from datetime import datetime, timezone
from pathlib import Path
def _antenna_map(conn) -> dict[str, str]:
return {a["serial"]: a["name"] for a in db.list_antennas(conn)}
+def _conv_offset(serial: str | None) -> float:
+ """Hardware tune offset (Hz) for the inline converter (Ham It Up/Down) on
+ `serial`'s antenna, 0 if none. Looked up AFTER leasing — the arbiter picks
+ the device, then we learn how to actually tune it."""
+ conn = _conn()
+ try:
+ return antenna.converter_offset_for(conn, serial)
+ finally:
+ conn.close()
+
+
+def _dev_cfg(serial: str | None) -> Config:
+ """Config for ops on the leased `serial`: a copy carrying its converter
+ tune offset (rtl.py applies it at the `-f` boundary), or the shared cfg."""
+ off = _conv_offset(serial)
+ return replace(cfg, converter_offset_hz=off) if off else cfg
+
+
def _antenna_for(amap: dict[str, str], serial: str | None) -> str | None:
"""Resolve a dongle serial to its antenna name; fall back to the bare serial
so the UI still shows *which* dongle when no antenna row is configured."""
def decode_live_capture(capture_id: int, decoder: str | None = None,
try:
with tuner.lease(prefer_serials=antenna.prefer_serials_for(cfg, freq),
timeout=cfg.tuner_wait_s) as dev:
- res = decoders.decode_live(cfg, dev.index, freq, which, secs)
+ res = decoders.decode_live(_dev_cfg(dev.serial), dev.index, freq,
+ which, secs)
except tuner.TunerBusy as e:
raise HTTPException(503, str(e))
def decode_live_freq(freq_mhz: float, decoder: str | None = None, secs: float |
try:
with tuner.lease(prefer_serials=antenna.prefer_serials_for(cfg, freq),
timeout=cfg.tuner_wait_s) as dev:
- return decoders.decode_live(cfg, dev.index, freq, which, secs)
+ return decoders.decode_live(_dev_cfg(dev.serial), dev.index, freq,
+ which, secs)
except tuner.TunerBusy as e:
raise HTTPException(503, str(e))
def list_antennas():
"present": serial in present,
"name": a["name"] if a else "",
"bands": bands, # [[lo_hz, hi_hz], …]
+ "converter": (a["converter"] or "") if a else "",
"configured": a is not None,
})
- return {"antennas": rows}
+ # the known inline up/down-converters, for the Settings select
+ convs = [{"key": k, "label": c["label"], "offset_mhz": c["offset_hz"] / 1e6,
+ "rf_lo_mhz": c["rf_lo"] / 1e6, "rf_hi_mhz": c["rf_hi"] / 1e6}
+ for k, c in antenna.CONVERTERS.items()]
+ return {"antennas": rows, "converters": convs}
class AntennaUpdate(BaseModel):
name: str
bands: list[tuple[float, float]] = [] # [lo_hz, hi_hz] pairs
+ converter: str = "" # antenna.CONVERTERS key, or ""
@app.put("/api/antennas/{serial}")
def put_antenna(serial: str, body: AntennaUpdate):
+ if body.converter and body.converter not in antenna.CONVERTERS:
+ raise HTTPException(400, f"unknown converter {body.converter!r}")
bands = [[float(lo), float(hi)] for lo, hi in body.bands if hi > lo]
conn = _conn()
- db.set_antenna(conn, serial, body.name.strip() or serial, json.dumps(bands))
+ db.set_antenna(conn, serial, body.name.strip() or serial, json.dumps(bands),
+ body.converter)
conn.close()
- return {"serial": serial, "name": body.name, "bands": bands}
+ return {"serial": serial, "name": body.name, "bands": bands,
+ "converter": body.converter}
@app.delete("/api/antennas/{serial}")
async def _lease_and_probe(key, freq_mhz, mode, serial, loop, exclude):
_event("stream", f"Live stream unavailable · {freq_mhz:.4f} MHz · {e}")
return None, None
- rtl_cmd = ["rtl_fm", "-d", str(dev.index), "-f", str(int(freq_mhz * 1e6)),
+ # tune through any inline converter on the leased tuner (freq stays RF)
+ tune_hz = int(freq_mhz * 1e6 + _conv_offset(dev.serial))
+ rtl_cmd = ["rtl_fm", "-d", str(dev.index), "-f", str(tune_hz),
*mflag, "-s", str(srate), "-r", str(arate), "-A", "fast", *gain, "-"]
ff_cmd = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-f", "s16le",
"-ar", str(arate), "-ac", "1", "-i", "-", "-f", "mp3", "-b:a", "128k", "-"]
File: api/recorder.py
from __future__ import annotations
import threading
+from dataclasses import replace
from datetime import datetime, timedelta, timezone
from scanner import antenna, capture, db, tuner
def _execute(job: dict) -> None:
try:
with tuner.lease(prefer_serials=antenna.prefer_serials_for(cfg, job["freq_hz"]),
timeout=cfg.tuner_wait_s) as dev:
+ # tune through any inline converter on the leased tuner's antenna
+ c = db.connect(cfg.db_path)
+ try:
+ off = antenna.converter_offset_for(c, dev.serial)
+ finally:
+ c.close()
+ rcfg = replace(cfg, converter_offset_hz=off) if off else cfg
rec = capture.record_capture(
- cfg, dev.index, job["freq_hz"], job["duration_s"],
+ rcfg, dev.index, job["freq_hz"], job["duration_s"],
notes=f"scheduled recording · {job['mode']} · {job['duration_s']:.0f}s",
serial=dev.serial,
)
