Claude (homelab agent)andClaude Opus 5.5
56eb5af0ff
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
62 lines
2.8 KiB
Python
62 lines
2.8 KiB
Python
"""Stuck-channel guard of sdr-classify: raises the squelch of channels that keep the scanner open on noise."""
|
|
import os
|
|
import time
|
|
|
|
from radio.config import AUTO_SNR_FILE, CHANNELS_FILE, CLS_INDEX, REC_DIR
|
|
from radio.recordings import FILE_RE, list_days
|
|
from radio.util import read_json, read_json_dict, write_json_compact
|
|
|
|
_last_guard = [0.0]
|
|
|
|
|
|
def stuck_guard():
|
|
"""Every 10 min: a channel that kept the scanner open on noise > 10 min in the last hour with no
|
|
voice gets +2 dB squelch (cap 20 dB) in auto_snr.json; gen_config applies it on the next
|
|
scanner start (the hourly discovery sweep restarts it)."""
|
|
now = time.time()
|
|
if now - _last_guard[0] < 600:
|
|
return
|
|
_last_guard[0] = now
|
|
cutoff = time.strftime("%Y%m%d%H%M%S", time.localtime(now - 3600))
|
|
noise, voice = {}, {}
|
|
for d in list_days()[:2]:
|
|
for n, e in read_json_dict(os.path.join(REC_DIR, d, CLS_INDEX)).items():
|
|
m = FILE_RE.match(n)
|
|
if not m or m["d"] + m["t"] < cutoff:
|
|
continue
|
|
f = int(m["f"])
|
|
if e.get("c") == "voice":
|
|
voice[f] = voice.get(f, 0) + 1
|
|
else:
|
|
noise[f] = noise.get(f, 0.0) + float((e.get("f") or {}).get("dur") or 0)
|
|
chans = read_json(CHANNELS_FILE, {}).get("channels", {})
|
|
auto = read_json_dict(AUTO_SNR_FILE).get("snr", {}) if os.path.exists(AUTO_SNR_FILE) else {}
|
|
changed = False
|
|
for f, sec in noise.items():
|
|
if sec > 600 and not voice.get(f):
|
|
cur = max(float((chans.get(str(f)) or {}).get("snr", 12.0)), float(auto.get(str(f), 0)))
|
|
new = min(20.0, cur + 2.0)
|
|
if new > float(auto.get(str(f), 0)):
|
|
auto[str(f)] = new
|
|
changed = True
|
|
print(f"stuck guard: {f / 1e6:.4f} MHz {sec:.0f} s noise/h, no voice -> squelch {new:.0f} dB "
|
|
f"(applied at next scanner start)", flush=True)
|
|
# decay: a raised channel that now has voice, or was quiet (< 2 min noise) for the last hour,
|
|
# goes back 1 dB per hour towards its configured value (gen_config takes max(base, auto))
|
|
last = read_json_dict(AUTO_SNR_FILE).get("decayed", 0) if os.path.exists(AUTO_SNR_FILE) else 0
|
|
if now - last >= 3600:
|
|
for k in list(auto):
|
|
f = int(k)
|
|
if voice.get(f) or noise.get(f, 0.0) < 120:
|
|
base = float((chans.get(k) or {}).get("snr_base", 0) or 0)
|
|
auto[k] = float(auto[k]) - 1.0
|
|
if auto[k] <= max(base, 10.0):
|
|
del auto[k]
|
|
changed = True
|
|
print(f"stuck guard: {f / 1e6:.4f} MHz calm/voice -> auto squelch "
|
|
f"{auto.get(k, 'removed')}", flush=True)
|
|
last = now
|
|
changed = True
|
|
if changed:
|
|
write_json_compact(AUTO_SNR_FILE, {"snr": auto, "updated": int(now), "decayed": int(last)})
|