Claude (homelab agent)andClaude Opus 5.5
56eb5af0ff
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
365 lines
16 KiB
Python
Executable File
365 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Generate /etc/sdr-scanner/rtl_airband.conf and channels.json from one frequency table.
|
|
|
|
Usage: gen_config.py OUT_DIR
|
|
|
|
Runs automatically on every sdr-scanner start (ExecStartPre), so frequencies promoted from the
|
|
"найденные источники" page are picked up after the next restart (the hourly discovery sweep restarts it).
|
|
|
|
Devices: DEVICES below. Today only the HackRF (scan mode). A second receiver (RTL-SDR) is a drop-in:
|
|
uncomment one of the SECOND_RECEIVER examples; every channel goes to the first device whose `take`
|
|
rule matches, the rest stays on the device with take=None.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
# `python3 -I` implies -P (no script dir on sys.path): put the package root (the parent of scanner/) there ourselves.
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from radio.config import AUTO_SNR_FILE, MAX_PROMOTED, PROMOTED_FILE # noqa: E402
|
|
from radio.util import read_json # noqa: E402
|
|
|
|
# ---------------------------------------------------------------------------------------------
|
|
# Squelch (SNR dB above the noise floor). Per-frequency overrides from the 2026-10-20 analysis.
|
|
AVIA_SNR = 12.0
|
|
NFM_SNR = 12.0 # was 10 dB: too many noise openings on NFM
|
|
LPD_SNR = 14.0 # LPD openings were mostly noise blips (433.35/.45 have constant carriers)
|
|
PMR_SNR = 13.0 # PMR 1/16 band-edge blips; spur between PMR 6/7
|
|
SNR_OVERRIDE = {
|
|
120.300: 16.0, # Pulkovo circle: 83/86 recordings were broadband noise
|
|
118.100: 14.0, # tonal birdie (steady harmonics) opened squelch 17x
|
|
129.250: 15.0, # constant +13.6 dB carrier/spur in sweep
|
|
123.650: 15.0, # constant +10.7 dB carrier/spur in sweep
|
|
135.600: 14.0,
|
|
156.600: 15.0, # 24/24 noise
|
|
156.575: 14.0,
|
|
145.500: 15.0, # 22/22 noise, long (18 s median) openings
|
|
433.500: 15.0, # 17/17 blips
|
|
# 156.675 (СУДС 73) carries weak real voice: at 12 dB it was chopped into blips -> 10 dB.
|
|
# 156.650 (СУДС 13): flapping squelch on a weak interferer held the scanner 77 of 105 min
|
|
# (02:44-04:30) at 12 dB -> 16 dB.
|
|
156.675: 10.0,
|
|
156.650: 16.0,
|
|
}
|
|
# Automatic raises written by sdr-classify (channel held the scanner > 10 min/h on noise, no voice): AUTO_SNR_FILE
|
|
|
|
|
|
def _auto_snr():
|
|
try:
|
|
with open(AUTO_SNR_FILE, encoding="utf-8") as fh:
|
|
d = json.load(fh).get("snr", {})
|
|
return {int(k): float(v) for k, v in d.items() if 8.0 <= float(v) <= 25.0}
|
|
except (OSError, ValueError, AttributeError, TypeError):
|
|
return {}
|
|
|
|
|
|
AUTO_SNR = _auto_snr()
|
|
|
|
# Scan timing (LOCAL PATCH of rtl_airband: scan_dwell_ms / scan_hold_ms, upstream = 200 / 2000).
|
|
# Measured 2026-10-20 with the Levashovo ATIS carrier: 200/150/120 ms attribute the signal to the
|
|
# right frequency, 100 and 80 ms attribute it to the NEXT entry (USB transfer ~65 ms at 2 Msps +
|
|
# 125 ms squelch batch). 150 ms keeps a safety margin.
|
|
SCAN_DWELL_MS = 150
|
|
SCAN_HOLD_MS = 2000
|
|
# LOCAL PATCH: per-frequency hold after activity, so a controller's reply on the same channel is
|
|
# not missed (ATC / VTS answer within ~2-6 s); other groups keep 2 s.
|
|
HOLD_MS = {"avia": 7000, "marine": 6000}
|
|
# ...but only after the squelch was open >= this long (noise blips keep the short 2 s hold;
|
|
# measured: a flat 7/6 s hold after every opening would have held the scanner ~47 % of the time).
|
|
LONG_HOLD_MIN_MS = 800
|
|
|
|
|
|
def fmt_mhz(x):
|
|
t = f"{x:.4f}"
|
|
return t[:-1] if t.endswith("0") else t
|
|
|
|
|
|
def mhz(x):
|
|
return int(round(x * 1e6))
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------
|
|
# Channel tables: (MHz, label, priority?) priority=True -> visited CHUNKS times per cycle.
|
|
# Sources: see the research table in the 2026-10-20 report (radioscanner.ru, vatrus, learn-atc,
|
|
# Big Port St. Petersburg port rules (Mintrans order 311), 433175.ru repeater catalogue).
|
|
AVIA = [
|
|
(118.100, "Пулково Вышка (10L/28R)", True),
|
|
(118.700, "Пулково Вышка (10R/28L)", True),
|
|
(120.300, "Пулково Круг", True),
|
|
(119.300, "Петербург Подход", True),
|
|
(125.200, "Петербург Подход 2", True),
|
|
(129.250, "Петербург Контроль R1", True),
|
|
(135.600, "Петербург Контроль R2", True),
|
|
(126.900, "Петербург Контроль R3/SW", True),
|
|
(121.500, "Авиа аварийная 121.5", True),
|
|
(123.100, "Авиа ПСО 123.1", True),
|
|
# Levashovo (ULLV/ULLA, "Бальзам") -- community sources only, no AIP:
|
|
(124.000, "124.0 Левашово? / Петербург-Контроль", True), # learn-atc/IVAO: ULLA TWR; flightplandatabase: SPb Control
|
|
# demoted 2026-10-20: zero activity in 3.5 h of night monitoring / ground stations 33 km away
|
|
(133.500, "Пулково Вышка (доп.)", False),
|
|
(121.700, "Пулково Руление", False),
|
|
(121.900, "Пулково Доставка", False),
|
|
(119.000, "Пулково Перрон (Ramp)", False),
|
|
(120.900, "Пулково Перрон (Apron)", False),
|
|
(134.100, "Петербург Контроль", False),
|
|
(135.400, "Петербург Контроль R4", False),
|
|
(123.250, "Левашово 123.25", False), # flightplandatabase TWR; radioscanner.ru 2008
|
|
(123.650, "Левашово 123.65", False), # radioscanner.ru 2006
|
|
(128.500, "Левашово Подход 128.5", False), # flightplandatabase APP; radioscanner 2005 "mil. ch 5"
|
|
# NOT scanned (continuous broadcasts would hold the squelch open forever):
|
|
# 122.475 Levashovo ATIS (seen on air, -27 dBFS), 127.4 / 127.3 Pulkovo ATIS RU/EN.
|
|
# Community-only candidates (128.0, 129.0, 131.8, 123.85, 135.0, 129.8, 131.5, 123.7, 125.5) are
|
|
# labelled by the discovery sweep instead of being scanned.
|
|
]
|
|
|
|
MARINE = [
|
|
(156.800, "Мор. 16 вызов/бедствие", True),
|
|
(156.450, "Мор. 9 Петербург-радио-17 (СУДС порта)", True), # port rules App. 8
|
|
(156.600, "Мор. 12 Петербург-радио-9", True),
|
|
(156.650, "Мор. 13 СУДС", True),
|
|
(156.500, "Мор. 10 Петербург-трафик", True),
|
|
(156.725, "Мор. 74 Петербург-трафик (РСУДС)", True), # port rules §54
|
|
(156.675, "Мор. 73 СУДС", True),
|
|
(156.575, "Мор. 71 Петербург-СКЦ", True),
|
|
(156.300, "Мор. 6 ледокол/буксир", False),
|
|
(156.400, "Мор. 8 ледокол/буксир", False),
|
|
(156.700, "Мор. 14 Петербург-радио-6", False),
|
|
(156.375, "Мор. 67 лоцман", False),
|
|
# coast-station transmit side (official port rules): pilots ch20, Volga-Balt ch23/84,
|
|
# GMDSS weather/ice broadcasts ch01/03/65/26 at 03:33/07:33/11:33/15:33/19:33/23:33
|
|
(161.600, "Мор. 20 берег (лоцман)", False),
|
|
(161.750, "Мор. 23 берег (Петербург-72)", False),
|
|
(161.825, "Мор. 84 берег (Петербург-72)", False),
|
|
(160.650, "Мор. 01 берег ГМССБ", False),
|
|
(160.750, "Мор. 03 берег ГМССБ", False),
|
|
(160.875, "Мор. 65 берег ГМССБ", False),
|
|
(161.900, "Мор. 26 берег ГМССБ", False),
|
|
]
|
|
|
|
PMR = [(446.00625 + 0.0125 * i, f"PMR {i + 1}", i == 0) for i in range(16)]
|
|
LPD = [(433.075 + 0.025 * i, f"LPD {i + 1}", i == 0) for i in range(17)] # 433.075 .. 433.475
|
|
HAM = [
|
|
(433.500, "70см вызывной 433.500", True),
|
|
(145.500, "2м вызывной 145.500", False),
|
|
(145.625, "Ретранслятор R1 RR1CB 145.625", False), # 433175.ru (in 145.025)
|
|
(145.775, "Ретранслятор R7 RD1AS 145.775", False), # 433175.ru (in 145.175)
|
|
(439.275, "Ретранслятор RR1AAA 439.275", False), # 433175.ru (in 431.675)
|
|
# 145.6-145.7625 dropped: zero activity; 145.7375 / 438.75 / 438.875-438.9 are DMR/C4FM (digital)
|
|
]
|
|
RAIL = [
|
|
(151.725, "ЖД ПРС к.1 151.725", False),
|
|
(151.825, "ЖД ПРС к.5 151.825 (основной)", False), # qrv.su 2021
|
|
(151.925, "ЖД ПРС к.9 151.925", False),
|
|
(152.000, "ЖД ПРС к.12 152.000", False),
|
|
]
|
|
|
|
GROUPS = {
|
|
"avia": "Авиа", "marine": "Морские", "pmr": "PMR446",
|
|
"lpd": "LPD433", "ham": "Любители", "rail": "ЖД", "found": "Найденные",
|
|
}
|
|
|
|
|
|
def snr(f, default):
|
|
base = SNR_OVERRIDE.get(round(f, 4), default)
|
|
return max(base, AUTO_SNR.get(mhz(f), 0.0))
|
|
|
|
|
|
# channel = (freq_hz, label, group, modulation, snr_dB, priority)
|
|
CHANNELS = [(mhz(f), l, "avia", "am", snr(f, AVIA_SNR), p) for f, l, p in AVIA] + \
|
|
[(mhz(f), l, "marine", "nfm", snr(f, NFM_SNR), p) for f, l, p in MARINE] + \
|
|
[(mhz(f), l, "pmr", "nfm", snr(f, PMR_SNR), p) for f, l, p in PMR] + \
|
|
[(mhz(f), l, "lpd", "nfm", snr(f, LPD_SNR), p) for f, l, p in LPD] + \
|
|
[(mhz(f), l, "ham", "nfm", snr(f, NFM_SNR), p) for f, l, p in HAM] + \
|
|
[(mhz(f), l, "rail", "nfm", snr(f, NFM_SNR), p) for f, l, p in RAIL]
|
|
|
|
# Frequencies promoted from the "найденные источники" page (written by web/sources.py, validated again here): PROMOTED_FILE.
|
|
FOUND_SNR = 14.0
|
|
|
|
|
|
def promoted(have):
|
|
data = read_json(PROMOTED_FILE, [])
|
|
have = set(have)
|
|
out = []
|
|
for p in data if isinstance(data, list) else []:
|
|
try:
|
|
f = int(p["freq_hz"])
|
|
mod = str(p.get("mod", "nfm"))
|
|
label = str(p.get("label", ""))[:60]
|
|
except (KeyError, TypeError, ValueError):
|
|
continue
|
|
if not (108_000_000 <= f <= 175_000_000 or 400_000_000 <= f <= 470_000_000):
|
|
continue
|
|
if mod not in ("am", "nfm") or any(abs(f - h) < 3000 for h in have):
|
|
continue
|
|
label = "".join(ch for ch in label if ch.isprintable() and ch not in '"\\') or f"{f / 1e6:.4f}"
|
|
have.add(f)
|
|
out.append((f, "Найд. " + label, "found", mod, FOUND_SNR, False))
|
|
if len(out) >= MAX_PROMOTED:
|
|
break
|
|
return out
|
|
|
|
|
|
CHANNELS += promoted(e[0] for e in CHANNELS)
|
|
|
|
# ---------------------------------------------------------------------------------------------
|
|
# Devices. `take`: None = everything not taken by another device; otherwise a dict with optional
|
|
# "groups" (list) and "window" (lo_MHz, hi_MHz). Multichannel devices need a centerfreq and take
|
|
# only channels inside centerfreq +- 0.45*sample_rate (rtl_airband's usable 90 %).
|
|
HACKRF = {
|
|
"name": "hackrf", "type": "soapysdr", "device_string": "driver=hackrf",
|
|
"gain": "LNA=32,VGA=24,AMP=0", "sample_rate": 2.0, "mode": "scan", "take": None,
|
|
}
|
|
DEVICES = [HACKRF]
|
|
|
|
# SECOND_RECEIVER examples (RTL-SDR Blog V4, 2.4 Msps max without drops on a Pi 4):
|
|
#
|
|
# (A) recommended: HackRF listens CONTINUOUSLY to the Pulkovo/Levashovo airband window
|
|
# (multichannel, 10 Msps measured: ~47 % of one Pi 4 core, no sample loss; 20 Msps loses ~25 %),
|
|
# the RTL-SDR scans everything else (control sectors 128-136, marine, LPD, PMR, ham, rail).
|
|
# HACKRF.update({"mode": "multichannel", "sample_rate": 10.0, "centerfreq": 122.600,
|
|
# "take": {"groups": ["avia"], "window": (117.6, 127.6)}})
|
|
# DEVICES.append({"name": "rtl", "type": "rtlsdr", "index": 0, "gain": 38.6, "sample_rate": 2.4,
|
|
# "mode": "scan", "take": None})
|
|
#
|
|
# (B) minimal: HackRF keeps scanning, the RTL-SDR listens continuously to the whole LPD block
|
|
# (433.05-434.79 MHz incl. 433.500) in multichannel mode.
|
|
# DEVICES.insert(0, {"name": "rtl", "type": "rtlsdr", "index": 0, "gain": 38.6, "sample_rate": 2.4,
|
|
# "mode": "multichannel", "centerfreq": 433.925,
|
|
# "take": {"groups": ["lpd", "ham"], "window": (432.95, 434.9)}})
|
|
|
|
CHUNKS = 4 # priority channels are visited CHUNKS times per scan cycle
|
|
|
|
|
|
def assign():
|
|
left = list(CHANNELS)
|
|
out = []
|
|
for d in DEVICES:
|
|
t = d["take"]
|
|
if t is None:
|
|
continue
|
|
lo, hi = t.get("window", (0, 1e9))
|
|
if d["mode"] == "multichannel":
|
|
half = d["sample_rate"] * 0.45
|
|
lo, hi = max(lo, d["centerfreq"] - half), min(hi, d["centerfreq"] + half)
|
|
mine = [e for e in left if (not t.get("groups") or e[2] in t["groups"]) and lo * 1e6 <= e[0] <= hi * 1e6]
|
|
left = [e for e in left if e not in mine]
|
|
out.append((d, mine))
|
|
for d in DEVICES:
|
|
if d["take"] is None:
|
|
out.append((d, left))
|
|
left = []
|
|
return out
|
|
|
|
|
|
def scan_order(chs):
|
|
pri = [e for e in chs if e[5]]
|
|
low = [e for e in chs if not e[5]]
|
|
if not pri:
|
|
return low
|
|
size = -(-len(low) // CHUNKS) if low else 0
|
|
order = []
|
|
for c in range(CHUNKS):
|
|
order += pri
|
|
order += low[c * size:(c + 1) * size]
|
|
return order
|
|
|
|
|
|
def q(s):
|
|
return '"' + s.replace('"', "'") + '"'
|
|
|
|
|
|
OUTPUT = """outputs:
|
|
(
|
|
{
|
|
type = "file";
|
|
directory = "/srv/sdr/recordings";
|
|
filename_template = "sdr";
|
|
dated_subdirectories = true;
|
|
split_on_transmission = true;
|
|
include_freq = true;
|
|
split_max_idle_time = 1.0;
|
|
split_max_file_time = 300.0;
|
|
}
|
|
);"""
|
|
|
|
|
|
def dev_head(d):
|
|
if d["type"] == "soapysdr":
|
|
h = f'type = "soapysdr";\n device_string = {q(d["device_string"])};\n gain = {q(d["gain"])};\n'
|
|
else:
|
|
h = f'type = {q(d["type"])};\n index = {int(d["index"])};\n gain = {float(d["gain"]):.1f};\n'
|
|
h += f' sample_rate = {int(round(d["sample_rate"] * 1e6))};\n mode = {q(d["mode"])};\n'
|
|
if d["mode"] == "multichannel":
|
|
h += f' centerfreq = {mhz(d["centerfreq"])};\n'
|
|
return h
|
|
|
|
|
|
def dev_block(d, chs):
|
|
if d["mode"] == "scan":
|
|
order = scan_order(chs)
|
|
body = f""" {{
|
|
freqs = ( {", ".join(str(e[0]) for e in order)} );
|
|
labels = ( {", ".join(q(e[1]) for e in order)} );
|
|
modulations = ( {", ".join(q(e[3]) for e in order)} );
|
|
squelch_snr_threshold = ( {", ".join(f"{e[4]:.1f}" for e in order)} );
|
|
hold_ms = ( {", ".join(str(HOLD_MS.get(e[2], SCAN_HOLD_MS)) for e in order)} );
|
|
{OUTPUT}
|
|
}}"""
|
|
n_pri = sum(1 for e in chs if e[5])
|
|
cyc = len(order) * (SCAN_DWELL_MS + 2) / 1000
|
|
note = (f"# {d['name']}: scan, {len(order)} entries ({n_pri} priority x{CHUNKS}, {len(chs) - n_pri} low x1), "
|
|
f"dwell {SCAN_DWELL_MS} ms -> cycle ~{cyc:.0f} s, priority revisit ~{cyc / CHUNKS:.1f} s")
|
|
else:
|
|
body = ",\n".join(f""" {{
|
|
freq = {e[0]}; label = {q(e[1])}; modulation = {q(e[3])}; squelch_snr_threshold = {e[4]:.1f};
|
|
{OUTPUT}
|
|
}}""" for e in chs)
|
|
note = f"# {d['name']}: multichannel {d['sample_rate']} Msps @ {d['centerfreq']} MHz, {len(chs)} channels (continuous)"
|
|
return note, f""" {{
|
|
{dev_head(d)} channels:
|
|
(
|
|
{body}
|
|
);
|
|
}}"""
|
|
|
|
|
|
def conf():
|
|
blocks = [dev_block(d, chs) for d, chs in assign() if chs]
|
|
notes = "\n".join(n for n, _ in blocks)
|
|
return f"""# GENERATED by gen_config.py -- edit the table there and regenerate.
|
|
{notes}
|
|
localtime = true;
|
|
log_scan_activity = true;
|
|
scan_dwell_ms = {SCAN_DWELL_MS}; // LOCAL PATCH (upstream: fixed 200)
|
|
scan_hold_ms = {SCAN_HOLD_MS}; // default; per-frequency hold_ms lists override
|
|
scan_long_hold_min_ms = {LONG_HOLD_MIN_MS};
|
|
stats_filepath = "/run/sdr-scanner/stats.prom"; // tmpfs, rewritten every 15 s
|
|
devices:
|
|
(
|
|
{",".join(b for _, b in blocks)}
|
|
);
|
|
"""
|
|
|
|
|
|
def channels_json():
|
|
seen = {}
|
|
for d, chs in assign():
|
|
for e in chs:
|
|
seen.setdefault(str(e[0]), {"label": e[1], "group": e[2], "mod": e[3], "snr": e[4],
|
|
"priority": e[5], "device": d["name"],
|
|
"hold_ms": HOLD_MS.get(e[2], SCAN_HOLD_MS) if d["mode"] == "scan" else 0})
|
|
return {"groups": GROUPS, "channels": seen, "scan_dwell_ms": SCAN_DWELL_MS,
|
|
"scan_hold_ms": SCAN_HOLD_MS, "long_hold_min_ms": LONG_HOLD_MIN_MS}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
out = sys.argv[1]
|
|
os.makedirs(out, exist_ok=True)
|
|
text = conf()
|
|
with open(os.path.join(out, "rtl_airband.conf"), "w") as f:
|
|
f.write(text)
|
|
with open(os.path.join(out, "channels.json"), "w") as f:
|
|
json.dump(channels_json(), f, ensure_ascii=False, indent=1)
|
|
print(text.split("\nlocaltime")[0].split("\n", 1)[1])
|