homelab-radio/sdr/classify/classify.py

#!/usr/bin/env python3
"""Voice/noise post-filter for SDR recordings (runs as user sdr, loop).

For every finished /srv/sdr/recordings/YYYY-MM-DD/*.mp3 not yet classified:
  decode (ffmpeg, 8 kHz mono) -> features -> voice | noise.
  noise -> moved to YYYY-MM-DD/noise/ (hidden in the UI, deleted by sdr-retention after 2 days).
Results (class, features, 120-point peak envelope for the UI waveform) go to
YYYY-MM-DD/.cls.json (atomic rewrite).

Rule (tuned on 272 recordings of 2026-10-20, "keep on doubt"):
  voice  <=> open-squelch audio >= 1.0 s
             AND voiced fraction (autocorr pitch strength > 0.45 in 80-400 Hz) >= 0.10
             AND pitch varies (std of log F0 >= 0.10)  -- rejects steady birdies/hum/tones
"""
import os
import subprocess
import sys
import time

# `python3 -I` implies -P (no script dir on sys.path): put the package root (the parent of classify/) there ourselves.
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from classify.adsb import adsb_snapshot, is_avia  # noqa: E402
from classify.audio import classify, features, pcm, peaks  # noqa: E402
from classify.guard import stuck_guard  # noqa: E402
from radio.config import CLS_INDEX, REC_DIR  # noqa: E402
from radio.recordings import FILE_RE, list_days  # noqa: E402
from radio.util import read_json_dict, write_json_compact  # noqa: E402

MIN_AGE = 2          # s since last write (rtl_airband renames .tmp -> .mp3 on close anyway)
LOOP = 3             # s (UI gets new recordings pushed within ~1-5 s)
DAYS_BACK = 3        # only look at recent day dirs in steady state
VERSION = 1


_last_gc = {}


def process_day(day, budget):
    ddir = os.path.join(REC_DIR, day)
    idx_path = os.path.join(ddir, CLS_INDEX)
    idx = read_json_dict(idx_path)
    changed = False
    try:
        names = sorted(n for n in os.listdir(ddir) if FILE_RE.match(n))
    except OSError:
        return 0
    now = time.time()
    done = 0
    for n in names:
        if n in idx or done >= budget:
            continue
        p = os.path.join(ddir, n)
        try:
            if now - os.stat(p).st_mtime < MIN_AGE:
                continue
            x = pcm(p)
        except (OSError, subprocess.SubprocessError):
            continue
        f = features(x)
        c = classify(f)
        idx[n] = {"c": c, "f": f, "pk": peaks(x), "v": VERSION}
        if c == "voice" and is_avia(n) and now - os.stat(p).st_mtime < 600:
            try:
                idx[n]["ac"] = adsb_snapshot()
            except Exception as e:  # noqa: BLE001 - context only, never block classification
                idx[n]["ac"] = {"err": type(e).__name__}
        if c == "noise":
            nd = os.path.join(ddir, "noise")
            os.makedirs(nd, exist_ok=True)
            try:
                os.replace(p, os.path.join(nd, n))
            except OSError:
                pass
        changed = True
        done += 1
        if done % 20 == 0:
            write_json_compact(idx_path, idx)
    # forget entries whose files are gone everywhere (retention) -- every 10 min
    if now - _last_gc.get(day, 0) > 600:
        _last_gc[day] = now
    for n in (list(idx) if _last_gc.get(day) == now else ()):
        if not os.path.exists(os.path.join(ddir, n)) and not os.path.exists(os.path.join(ddir, "noise", n)):
            del idx[n]
            changed = True
    if changed:
        write_json_compact(idx_path, idx)
    return done


def days(all_days):
    ds = list_days()
    return ds if all_days else ds[:DAYS_BACK]


def main():
    once = "--once" in sys.argv
    first = True
    while True:
        total = 0
        for d in days(first or once):
            total += process_day(d, 400)
        if total:
            print(f"classified {total}", flush=True)
        first = False
        try:
            adsb_snapshot()   # keeps adsb_latest.json fresh (cached: <= 1 request/min)
        except Exception:  # noqa: BLE001
            pass
        try:
            stuck_guard()
        except Exception as e:  # noqa: BLE001 - never kill the classifier
            print(f"stuck guard error: {e}", flush=True)
        if once:
            return
        time.sleep(LOOP)


if __name__ == "__main__":
    main()