dee262e130
Every capture that is not voice, or whose voice yields five characters or fewer of transcript, now gets a PNG of the waterfall it would have painted on screen: spectrogram from the IQ where it was kept, from the demodulated audio otherwise, captioned and labelled either way. The browser shows it in the picture panel, but only when there is no transcript, Morse or decoded data to show instead. bandsaunter waterfall [PATH...] [--all] [--redraw] [--min-chars N] draws them after the fact for recordings already on disk. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016PsWPTweCT6pwxKngvVxcg
337 lines
14 KiB
Python
Executable file
337 lines
14 KiB
Python
Executable file
"""Scanner configuration: defaults, validation, and YAML profiles."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass, field, asdict, fields
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from .ranges import Lockout, ScanRange, parse_range_list
|
|
|
|
__all__ = ["ScanConfig", "DEFAULT_CONFIG_DIR", "DEFAULT_CONFIG_PATH",
|
|
"load_config", "save_config", "list_profiles", "profile_path",
|
|
"load_default", "save_default", "delete_profile",
|
|
"is_first_run", "DEFAULT_OUTPUT_DIR", "remember_lockouts"]
|
|
|
|
DEFAULT_CONFIG_DIR = Path(
|
|
os.environ.get("BANDSAUNTER_CONFIG_DIR",
|
|
Path.home() / ".config" / "bandsaunter")
|
|
)
|
|
|
|
# Settings saved here are picked up by every run, before any command-line
|
|
# option is applied. Named profiles live alongside it in the same directory.
|
|
DEFAULT_CONFIG_PATH = DEFAULT_CONFIG_DIR / "config.yaml"
|
|
|
|
# Suggested on first run. Somewhere visible rather than under a dot
|
|
# directory: these are recordings people want to browse and play.
|
|
DEFAULT_OUTPUT_DIR = "~/bandsaunter"
|
|
|
|
|
|
def is_first_run(directory: Path | None = None) -> bool:
|
|
"""True when no settings have ever been saved."""
|
|
return not (Path(directory or DEFAULT_CONFIG_DIR) / "config.yaml").exists()
|
|
|
|
|
|
@dataclass
|
|
class ScanConfig:
|
|
"""Everything that controls a scan run.
|
|
|
|
The two dwell settings the scanner is built around:
|
|
|
|
``record_seconds``
|
|
Record for at most this long on one signal before going back to
|
|
scanning, even if the signal is still up. 0 means "no limit", which
|
|
is what you want for capturing a whole conversation.
|
|
|
|
``hang_seconds``
|
|
How long the channel must stay quiet before the transmission is
|
|
treated as over. Gaps shorter than this are recorded straight
|
|
through, so a two-way exchange stays in one file across the natural
|
|
pauses between overs. "Quiet" means no real signal: silence, static
|
|
and interference all count towards it, so a burst of noise during a
|
|
pause does not hold the receiver on a finished conversation.
|
|
"""
|
|
|
|
# -- what to scan ---------------------------------------------------
|
|
ranges: list[ScanRange] = field(default_factory=list)
|
|
|
|
# -- the two headline dwell settings --------------------------------
|
|
record_seconds: float = 30.0 # max record time per hit (0 = unlimited)
|
|
hang_seconds: float = 2.0 # quiet time before the scan resumes
|
|
max_record_seconds: float = 900.0 # absolute ceiling, even when record is 0
|
|
|
|
# -- detection ------------------------------------------------------
|
|
threshold_db: float = 10.0 # dB of margin over the noise
|
|
min_record_seconds: float = 0.5 # discard blips shorter than this
|
|
dwell_seconds: float = 0.05 # capture time per tuner step while sweeping
|
|
probe_seconds: float = 0.4 # look before choosing a demodulator
|
|
resolution_hz: float = 3_000.0 # FFT bin width during the sweep
|
|
squelch_margin_db: float = 3.0 # hysteresis: drop below thr - this to end
|
|
adaptive_floor: bool = True # smooth the per-bin noise baseline
|
|
detector: str = "peak" # "peak" holds bursts, "avg" is quieter
|
|
detector_bias_db: float | None = None # None = derive it from the detector
|
|
|
|
# -- content gate: only keep captures that carry something ----------
|
|
require_signal: bool = True # discard captures with no real content
|
|
accept: list[str] = field(
|
|
default_factory=lambda: ["voice", "cw", "digital"])
|
|
min_signal_score: float = 0.45 # confidence needed to keep a capture
|
|
min_voice_score: float = 0.45 # speech-likeness needed to call it voice
|
|
verify_seconds: float = 1.5 # how often to re-check a live capture
|
|
verify_max_seconds: float = 6.0 # give up on a contentless capture by here
|
|
max_detections_per_step: int = 4
|
|
|
|
# -- trunking control channels --------------------------------------
|
|
skip_control: bool = True # spot them, abandon them, move on
|
|
control_seconds: float = 20.0 # unbroken carrier a look-alike must hold
|
|
lockout_control: bool = False # also add them to the lock-out list
|
|
|
|
# -- radio ----------------------------------------------------------
|
|
device_index: int = 0
|
|
sample_rate: int = 2_048_000
|
|
gain: float | str = "auto"
|
|
ppm: int = 0
|
|
agc: bool = False
|
|
bias_tee: bool = False
|
|
offset_tuning: bool = False
|
|
direct_sampling: int | str = "auto"
|
|
usable_fraction: float = 0.75
|
|
dc_guard_hz: float = 8_000.0 # LO offset that keeps the DC spike clear
|
|
|
|
# -- recording ------------------------------------------------------
|
|
output_dir: str = "~/bandsaunter"
|
|
save_audio: bool = True
|
|
save_iq: bool = False
|
|
audio_rate: int = 16_000
|
|
iq_format: str = "cf32" # cf32 or cs16
|
|
classify: bool = True
|
|
decode_morse: bool = True
|
|
decode_data: bool = True # read packets out of data signals
|
|
decode_images: bool = True # SSTV, weather satellites and shortwave fax
|
|
# A picture of every capture nobody can read: not voice, or voice
|
|
# that produced no words worth the name.
|
|
waterfall: bool = True
|
|
waterfall_min_chars: int = 5
|
|
|
|
# -- one file per frequency ------------------------------------------
|
|
combine_by_frequency: bool = False
|
|
combine_tolerance_hz: float = 6_250.0
|
|
combine_keep_individual: bool = False
|
|
announce_timestamps: bool = True
|
|
announce_frequency: bool = False
|
|
announce_engine: str = "auto"
|
|
|
|
# -- speech to text ---------------------------------------------------
|
|
transcribe: bool = False
|
|
transcribe_engine: str = "auto"
|
|
transcribe_model: str = "base.en"
|
|
transcribe_language: str = "en"
|
|
transcribe_min_seconds: float = 1.0
|
|
|
|
# -- callsigns and the map --------------------------------------------
|
|
callsign_lookup: bool = True # ask the licence database who they are
|
|
kml_file: str = "callsigns.kml" # map of who was heard ("" = none)
|
|
|
|
# -- behaviour ------------------------------------------------------
|
|
lockout: list[Lockout] = field(default_factory=list)
|
|
lockout_width: float = 12_500.0
|
|
save_lockouts: bool = True # write runtime lock-outs back to disk
|
|
revisit_seconds: float = 8.0 # ignore a frequency again this soon
|
|
max_cycles: int = 0 # 0 = run forever
|
|
max_runtime_seconds: float = 0.0 # 0 = no limit
|
|
quiet: bool = False
|
|
plain: bool = False # line per hit instead of the live display
|
|
log_file: str = "scan_log.jsonl"
|
|
|
|
# ------------------------------------------------------------------
|
|
def __post_init__(self):
|
|
# Lock-outs arrive as bare numbers from older settings files, as
|
|
# dicts from newer ones, and as either from callers. One type from
|
|
# here on.
|
|
self.lockout = [Lockout.coerce(v) for v in self.lockout]
|
|
|
|
def validate(self) -> list[str]:
|
|
"""Return a list of human-readable problems (empty when config is sane)."""
|
|
errs = []
|
|
if not self.ranges:
|
|
errs.append("no frequency ranges configured")
|
|
if self.record_seconds < 0:
|
|
errs.append("record_seconds cannot be negative")
|
|
if self.hang_seconds < 0:
|
|
errs.append("hang_seconds cannot be negative")
|
|
if self.min_record_seconds < 0:
|
|
errs.append("min_record_seconds cannot be negative")
|
|
if self.max_record_seconds < 0:
|
|
errs.append("max_record_seconds cannot be negative")
|
|
if self.record_seconds and self.min_record_seconds > self.record_seconds:
|
|
errs.append("min_record_seconds is longer than record_seconds, so "
|
|
"no recording would ever be kept")
|
|
if not (0.005 <= self.dwell_seconds <= 5.0):
|
|
errs.append("dwell_seconds should be between 0.005 and 5")
|
|
if self.threshold_db <= 0:
|
|
errs.append("threshold_db must be positive")
|
|
if not (0.1 <= self.usable_fraction <= 0.95):
|
|
errs.append("usable_fraction should be between 0.1 and 0.95")
|
|
if self.audio_rate < 4000:
|
|
errs.append("audio_rate is too low to carry voice")
|
|
from .quality import CATEGORIES
|
|
bad = [c for c in self.accept if c not in CATEGORIES]
|
|
if bad:
|
|
errs.append("unknown accept categories: " + ", ".join(bad)
|
|
+ " (choose from " + ", ".join(CATEGORIES) + ")")
|
|
if self.require_signal and not self.accept:
|
|
errs.append("require_signal is on but no categories are accepted, "
|
|
"so nothing could ever be recorded")
|
|
if not (0.0 <= self.min_signal_score <= 1.0):
|
|
errs.append("min_signal_score must be between 0 and 1")
|
|
if self.detector not in ("peak", "avg"):
|
|
errs.append('detector must be "peak" or "avg"')
|
|
if self.iq_format not in ("cf32", "cs16"):
|
|
errs.append("iq_format must be cf32 or cs16")
|
|
for r in self.ranges:
|
|
if r.stop <= 0 or r.start <= 0:
|
|
errs.append(f"range {r.label!r} has a non-positive frequency")
|
|
return errs
|
|
|
|
# -- serialisation ---------------------------------------------------
|
|
def to_dict(self) -> dict:
|
|
d = asdict(self)
|
|
d["ranges"] = [r.to_dict() for r in self.ranges]
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict) -> "ScanConfig":
|
|
d = dict(d or {})
|
|
raw = d.pop("ranges", [])
|
|
known = {f.name for f in fields(cls)}
|
|
unknown = set(d) - known
|
|
for k in unknown:
|
|
d.pop(k)
|
|
cfg = cls(**{k: v for k, v in d.items() if k in known})
|
|
out = []
|
|
for item in raw:
|
|
if isinstance(item, dict):
|
|
out.append(ScanRange.from_dict(item))
|
|
elif isinstance(item, str):
|
|
out.extend(parse_range_list(item))
|
|
cfg.ranges = out
|
|
if unknown:
|
|
cfg._unknown_keys = sorted(unknown) # surfaced by the CLI
|
|
return cfg
|
|
|
|
def copy(self) -> "ScanConfig":
|
|
return ScanConfig.from_dict(self.to_dict())
|
|
|
|
|
|
def profile_path(name: str, directory: Path | None = None) -> Path:
|
|
directory = Path(directory or DEFAULT_CONFIG_DIR)
|
|
if name.endswith((".yaml", ".yml")):
|
|
p = Path(name)
|
|
return p if p.is_absolute() or p.parent != Path(".") else directory / p.name
|
|
return directory / f"{name}.yaml"
|
|
|
|
|
|
def save_config(cfg: ScanConfig, name: str, directory: Path | None = None) -> Path:
|
|
path = profile_path(name, directory)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(path, "w") as fh:
|
|
yaml.safe_dump(cfg.to_dict(), fh, sort_keys=False, default_flow_style=False)
|
|
return path
|
|
|
|
|
|
def load_config(name_or_path: str, directory: Path | None = None) -> ScanConfig:
|
|
path = Path(name_or_path)
|
|
if not path.exists():
|
|
path = profile_path(name_or_path, directory)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"no such profile or config file: {name_or_path}")
|
|
with open(path) as fh:
|
|
data = yaml.safe_load(fh) or {}
|
|
cfg = ScanConfig.from_dict(data)
|
|
cfg._source_path = str(path)
|
|
return cfg
|
|
|
|
|
|
def list_profiles(directory: Path | None = None) -> list[Path]:
|
|
directory = Path(directory or DEFAULT_CONFIG_DIR)
|
|
if not directory.exists():
|
|
return []
|
|
return sorted(p for p in directory.glob("*.yaml"))
|
|
|
|
|
|
def load_default(directory: Path | None = None) -> tuple["ScanConfig", Path | None]:
|
|
"""Load the saved settings, or defaults if none have been saved.
|
|
|
|
Returns ``(config, path)`` where ``path`` is None when nothing was loaded,
|
|
so the caller can tell the user where its settings came from.
|
|
"""
|
|
path = Path(directory or DEFAULT_CONFIG_DIR) / "config.yaml"
|
|
if not path.exists():
|
|
return ScanConfig(), None
|
|
try:
|
|
with open(path) as fh:
|
|
data = yaml.safe_load(fh) or {}
|
|
except (OSError, yaml.YAMLError):
|
|
return ScanConfig(), None
|
|
cfg = ScanConfig.from_dict(data)
|
|
cfg._source_path = str(path)
|
|
return cfg, path
|
|
|
|
|
|
def save_default(cfg: "ScanConfig", directory: Path | None = None) -> Path:
|
|
"""Write the settings that every later run should start from."""
|
|
path = Path(directory or DEFAULT_CONFIG_DIR) / "config.yaml"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(".yaml.tmp")
|
|
with open(tmp, "w") as fh:
|
|
yaml.safe_dump(cfg.to_dict(), fh, sort_keys=False,
|
|
default_flow_style=False)
|
|
tmp.replace(path) # atomic, so a crash cannot leave a half file
|
|
return path
|
|
|
|
|
|
def remember_lockouts(cfg: "ScanConfig",
|
|
directory: Path | None = None) -> Path | None:
|
|
"""Write the lock-out list back to the settings file it came from.
|
|
|
|
Only that one key: a scan's config also holds whatever was passed on the
|
|
command line for this run, and saving all of it would quietly make those
|
|
one-off options permanent. So the file on disk is read, its lock-outs
|
|
replaced, and the rest left exactly as it was.
|
|
|
|
Returns where it was written, or None if it could not be -- locking out a
|
|
birdie must never be the thing that ends a scan.
|
|
|
|
A run started with --no-config has no settings file, and writing one
|
|
anyway would be the opposite of what was asked: the flag exists to leave
|
|
the saved settings alone. So without a source to write back to, nothing
|
|
is written.
|
|
"""
|
|
source = getattr(cfg, "_source_path", None)
|
|
if not source and directory is None:
|
|
return None
|
|
path = Path(source) if source else (Path(directory) / "config.yaml")
|
|
try:
|
|
data = {}
|
|
if path.exists():
|
|
with open(path) as fh:
|
|
data = yaml.safe_load(fh) or {}
|
|
data["lockout"] = [lk.to_dict() for lk in cfg.lockout]
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
with open(tmp, "w") as fh:
|
|
yaml.safe_dump(data, fh, sort_keys=False, default_flow_style=False)
|
|
tmp.replace(path)
|
|
return path
|
|
except (OSError, yaml.YAMLError):
|
|
return None
|
|
|
|
|
|
def delete_profile(name: str, directory: Path | None = None) -> bool:
|
|
path = profile_path(name, directory)
|
|
if path.exists():
|
|
path.unlink()
|
|
return True
|
|
return False
|