dustcouncil/bandsaunter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
bandsaunter/tests/test_scanner.py
The Dust Council 44c98b11e3 Remember runtime lock-outs, and let a lock-out be a span 
Pressing `l` during a scan locked a frequency out for that run only, so the
same birdie had to be locked out again on every later one. It now writes
back to the settings file the run started from -- only that one key, since
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 permanent. The file is
read, its lock-outs replaced, the rest left as it was. `save_lockouts`
turns it off for anyone who would rather their config were never touched.

Lock-outs were also single frequencies only. They are now a list of
frequencies and spans -- "162.55M, 450M-455M, 88M to 108M" -- which is what
a pager band or a noisy stretch of spectrum actually is. A point is still
widened by the lock-out width; a span is taken exactly as written, because
whoever typed it already said how wide it is.

The scanner matched lock-outs by rounding a frequency into a bucket of the
lock-out width, which cannot express a span and was never exact at the
edges. It now holds intervals and tests them directly.

Settings files that predate this hold a bare number per lock-out, and still
mean the same thing: Lockout.coerce takes numbers, strings, pairs and dicts,
so old profiles load untouched.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-21 23:44:24 -07:00

569 lines
24 KiB
Python
Raw Blame History

"""End-to-end scanner tests driven by the built-in simulator."""
import json
import re
import wave
from pathlib import Path
import numpy as np
import pytest
from bandsaunter.config import ScanConfig
from bandsaunter.ranges import parse_range_list
from bandsaunter.scanner import Scanner, ScannerCallbacks
from bandsaunter.simulator import SimulatedDevice, VirtualTransmitter as V
def run_scan(tmp_path, transmitters, ranges, **over):
dev = SimulatedDevice(transmitters=transmitters).open()
cfg = ScanConfig(ranges=parse_range_list(ranges), output_dir=str(tmp_path),
record_seconds=3.0, hang_seconds=1.0,
min_record_seconds=0.3, threshold_db=12,
dwell_seconds=0.05, max_cycles=2, revisit_seconds=0.2)
cfg.accept = list(over.pop("accept", cfg.accept))
for k, v in over.items():
setattr(cfg, k, v)
hits = []
s = Scanner(cfg, device=dev,
callbacks=ScannerCallbacks(on_record_end=hits.append))
s.prepare()
s.run()
return s, [h for h in hits if h.kept]
def test_finds_and_identifies_narrowband_fm(tmp_path):
tx = [V(146_520_000, "nfm", 0.4, 12_500, "voice", ctcss=100.0)]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M")
assert hits, "nothing recorded"
h = hits[0]
assert h.frequency == pytest.approx(146_520_000, abs=8_000)
assert h.category == "voice"
assert h.family == "nfm", h.classification
assert "voice" in h.classification.lower()
assert h.ctcss_hz == pytest.approx(100.0, abs=0.5)
def test_record_seconds_is_honoured_exactly(tmp_path):
tx = [V(146_520_000, "nfm", 0.4, 12_500, "voice")]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=2.0)
assert hits
for h in hits:
assert h.duration == pytest.approx(2.0, abs=0.15)
assert "record limit" in h.stop_reason
def test_hang_seconds_ends_a_finished_transmission(tmp_path):
"""A transmitter that stops must release the scanner after the hang time.
What is under test is the ending, not the catching, so the burst is a
generous 40% of the cycle over six sweeps: landing on it is meant to be
easy, and only the release afterwards should be able to fail this.
"""
tx = [V(146_520_000, "nfm", 0.4, 12_500, "burst",
period_seconds=5.0, on_seconds=2.0)]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M",
record_seconds=20.0, hang_seconds=0.8, max_cycles=6)
assert hits, "nothing recorded"
ended_on_silence = [h for h in hits if "quiet" in h.stop_reason]
assert ended_on_silence, [h.stop_reason for h in hits]
for h in ended_on_silence:
assert h.duration < 20.0
def test_min_record_seconds_discards_blips(tmp_path):
"""A transmission shorter than the minimum must leave nothing behind."""
tx = [V(146_520_000, "nfm", 0.5, 12_500, "blip",
period_seconds=3.0, on_seconds=0.35)]
dev = SimulatedDevice(transmitters=tx).open()
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
output_dir=str(tmp_path), record_seconds=20.0,
min_record_seconds=6.0, hang_seconds=0.5,
threshold_db=12, dwell_seconds=0.05, max_cycles=4,
revisit_seconds=0.1)
all_hits = []
s = Scanner(cfg, device=dev,
callbacks=ScannerCallbacks(on_record_end=all_hits.append))
s.prepare()
s.run()
assert all_hits, "the blip was never detected"
assert not any(h.kept for h in all_hits)
assert s.stats.discarded >= 1
# Nothing should have been left behind on disk.
assert not list(tmp_path.glob("*/*/audio.wav"))
def test_invalid_config_is_rejected_before_touching_the_radio(tmp_path):
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
output_dir=str(tmp_path),
record_seconds=5.0, min_record_seconds=9.0)
s = Scanner(cfg, device=SimulatedDevice().open())
with pytest.raises(ValueError, match="min_record_seconds"):
s.prepare()
def test_decodes_morse_from_a_cw_beacon(tmp_path):
tx = [V(144_100_000, "cw", 0.35, 500, "beacon", wpm=18,
message="VVV DE W1AW")]
# Several sweeps, because a 50 ms dwell can easily land in a key-up gap.
s, hits = run_scan(tmp_path, tx, "144.05M-144.15M",
record_seconds=10.0, hang_seconds=4.0, max_cycles=8,
revisit_seconds=0.1)
assert hits
h = hits[0]
assert h.mode == "cw", "should have chosen the CW demodulator"
assert "W1AW" in h.morse_text or "VVV" in h.morse_text, h.morse_text
assert h.morse_wpm == pytest.approx(18, rel=0.15)
def test_lockout_is_respected(tmp_path):
tx = [V(146_520_000, "nfm", 0.4, 12_500, "voice")]
dev = SimulatedDevice(transmitters=tx).open()
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
output_dir=str(tmp_path), record_seconds=2.5,
threshold_db=12, dwell_seconds=0.05, max_cycles=2,
lockout=[146_520_000.0])
s = Scanner(cfg, device=dev)
s.prepare()
s.run()
assert s.stats.recordings == 0
def test_writes_audio_iq_and_metadata(tmp_path):
tx = [V(146_520_000, "nfm", 0.4, 12_500, "voice")]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M",
record_seconds=2.5, save_iq=True, max_cycles=1)
assert hits
h = hits[0]
with wave.open(h.audio_path) as w:
assert w.getnframes() / w.getframerate() == pytest.approx(2.5, abs=0.2)
assert w.getnchannels() == 1
iq = np.fromfile(h.iq_path, dtype=np.complex64)
assert iq.size > 0
meta = json.loads(open(h.meta_path).read())
assert meta["frequency_hz"] == pytest.approx(h.frequency)
assert meta["hit"]["classification"]
sigmf = json.loads(open(h.iq_path.replace(".cf32", ".sigmf-meta")).read())
assert sigmf["global"]["core:datatype"] == "cf32_le"
# the run log
assert (tmp_path / "scan_log.jsonl").exists()
lines = (tmp_path / "scan_log.csv").read_text().strip().split("\n")
assert len(lines) >= 2
def test_multiple_signals_in_one_step_all_get_visited(tmp_path):
tx = [V(146_500_000, "nfm", 0.40, 12_500, "a"),
V(146_800_000, "nfm", 0.30, 12_500, "b"),
V(147_100_000, "carrier", 0.25, 1_000, "c")]
s, hits = run_scan(tmp_path, tx, "146.4M-147.2M", record_seconds=2.5,
max_cycles=2, accept=["voice", "cw", "digital", "carrier"])
found = {round(h.frequency / 1e5) for h in hits}
assert len(found) >= 3, f"only found {found}"
def test_unreachable_frequencies_are_skipped_not_fatal(tmp_path):
tx = [V(146_520_000, "nfm", 0.4, 12_500, "voice")]
# 2.5 GHz is beyond the tuner; the scan must still run the reachable part.
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M, 2500M-2600M",
record_seconds=2.5, max_cycles=1)
assert hits
# ---------------------------------------------------------------------------
# The content gate: only voice, CW and digital should reach the disk.
# ---------------------------------------------------------------------------
def test_static_is_never_recorded(tmp_path):
"""An empty band must produce no files, whatever the squelch does."""
dev = SimulatedDevice(transmitters=[], noise_amplitude=0.05).open()
cfg = ScanConfig(ranges=parse_range_list("146M-148M"),
output_dir=str(tmp_path), record_seconds=4.0,
hang_seconds=1.0, threshold_db=2.0, # deliberately wide open
dwell_seconds=0.05, max_cycles=2, revisit_seconds=0.1)
hits = []
s = Scanner(cfg, device=dev,
callbacks=ScannerCallbacks(on_record_end=hits.append))
s.prepare()
s.run()
assert not any(h.kept for h in hits), "static was recorded"
assert not list(tmp_path.glob("*/*/audio.wav"))
def test_bare_carrier_is_rejected_by_default(tmp_path):
tx = [V(146_520_000, "carrier", 0.4, 1_000, "dead carrier")]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=3.0)
assert not hits, "an unmodulated carrier is not a signal worth keeping"
assert not list(tmp_path.glob("*/*/audio.wav"))
def test_bare_carrier_kept_when_asked_for(tmp_path):
tx = [V(146_520_000, "carrier", 0.4, 1_000, "dead carrier")]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=3.0,
accept=["voice", "cw", "digital", "carrier"])
assert hits and hits[0].category == "carrier"
@pytest.mark.parametrize("mode,bw,category", [
("nfm", 12_500, "voice"),
("am", 8_000, "voice"),
("fsk4", 12_500, "digital"),
("fsk2", 12_500, "digital"),
])
def test_real_signals_are_categorised_and_kept(tmp_path, mode, bw, category):
tx = [V(146_520_000, mode, 0.4, bw, "tx", baud=4800 if mode == "fsk4" else 1200,
deviation=1800 if mode.startswith("fsk") else 2500)]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=4.0)
assert hits, f"{mode} was rejected"
assert hits[0].category == category, \
f"{mode} -> {hits[0].category}: {hits[0].content_reason}"
def test_keep_everything_disables_the_gate(tmp_path):
"""--keep-everything restores plain power-threshold behaviour."""
dev = SimulatedDevice(transmitters=[], noise_amplitude=0.05).open()
cfg = ScanConfig(ranges=parse_range_list("146M-148M"),
output_dir=str(tmp_path), record_seconds=1.0,
hang_seconds=0.5, threshold_db=1.0, min_record_seconds=0.2,
dwell_seconds=0.05, max_cycles=2, revisit_seconds=0.1,
require_signal=False, detector_bias_db=0.0)
hits = []
s = Scanner(cfg, device=dev,
callbacks=ScannerCallbacks(on_record_end=hits.append))
s.prepare()
s.run()
assert s.stats.detections > 0, "threshold was not wide open enough"
def test_filenames_are_flat_dated_and_named_for_the_modulation(tmp_path):
tx = [V(146_520_000, "nfm", 0.4, 12_500, "voice")]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=3.0)
assert hits
wavs = list(tmp_path.glob("*.wav"))
assert wavs, "no wav written at the top level"
assert not list(tmp_path.glob("*/*.wav")), "files should not be in subfolders"
name = wavs[0].name
assert re.match(r"^\d{4}\.\d{6}MHz--\d{4}-\d{2}-\d{2}_"
r"\d{2}_\d{2}_\d{2}-[a-z0-9-]+\.wav$", name), name
assert name.endswith("-nfm.wav"), name
assert name.startswith("0146.5"), name
# every artefact of one capture shares a stem
stem = wavs[0].stem
assert (tmp_path / f"{stem}.json").exists()
def test_recorded_length_matches_the_requested_record_time(tmp_path):
"""The WAV must hold as many seconds as the scanner says it captured.
A mismatch here is what makes a recording play at the wrong speed.
"""
tx = [V(146_520_000, "nfm", 0.4, 12_500, "voice")]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=3.0)
assert hits
h = hits[0]
with wave.open(h.audio_path) as w:
seconds = w.getnframes() / w.getframerate()
assert seconds == pytest.approx(3.0, abs=0.15)
assert seconds == pytest.approx(h.duration, abs=0.02)
# ---------------------------------------------------------------------------
# Two-way conversations: natural pauses must not end the capture.
# ---------------------------------------------------------------------------
def _two_way(freq=146_520_000):
"""Two stations alternating on one frequency with a 1.5 s gap between overs.
A is up for [0, 2.5), B for [4, 6.5), repeating every 8 s.
"""
return [
V(freq, "nfm", 0.40, 12_500, "station A",
period_seconds=8.0, on_seconds=2.5, phase_offset=0.0),
V(freq, "nfm", 0.35, 12_500, "station B", pitch_hz=190.0,
period_seconds=8.0, on_seconds=2.5, phase_offset=4.0),
]
def test_a_pause_between_overs_does_not_end_the_capture(tmp_path):
"""With hang longer than the gap, both overs land in one recording."""
s, hits = run_scan(tmp_path, _two_way(), "146.4M-146.6M",
record_seconds=0.0, hang_seconds=3.0,
max_record_seconds=14.0, max_cycles=1)
assert hits, "nothing recorded"
h = hits[0]
# Must span the 1.5 s gap and reach the second station's over.
assert h.duration > 6.0, \
f"capture ended after {h.duration:.1f}s, so it cut at the pause"
assert len(list(tmp_path.glob("*.wav"))) == 1, \
"the exchange was split into separate files"
def test_a_short_hang_stops_at_the_first_pause(tmp_path):
"""With hang shorter than the gap, the capture ends when the over does."""
s, hits = run_scan(tmp_path, _two_way(), "146.4M-146.6M",
record_seconds=0.0, hang_seconds=0.8,
max_record_seconds=14.0, max_cycles=1)
assert hits
h = hits[0]
assert h.duration < 5.0, f"ran {h.duration:.1f}s past a {0.8}s hang"
assert "quiet" in h.stop_reason or "no signal" in h.stop_reason
def test_an_established_conversation_is_never_abandoned_as_noise(tmp_path):
"""Once speech is heard, a quiet spell must not trigger the content abort."""
s, hits = run_scan(tmp_path, _two_way(), "146.4M-146.6M",
record_seconds=0.0, hang_seconds=3.0,
verify_max_seconds=2.0, # would fire during the gap
max_record_seconds=14.0, max_cycles=1)
assert hits, "an established conversation was thrown away"
assert hits[0].category == "voice"
def test_max_record_seconds_bounds_an_unlimited_capture(tmp_path):
"""--record 0 must still not run forever."""
tx = [V(146_520_000, "nfm", 0.4, 12_500, "continuous")]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=0.0,
hang_seconds=5.0, max_record_seconds=5.0, max_cycles=1)
assert hits
assert hits[0].duration == pytest.approx(5.0, abs=0.3)
assert "safety limit" in hits[0].stop_reason
def test_interference_during_a_gap_does_not_hold_the_capture_open(tmp_path):
"""Static breaking squelch mid-pause must still count as quiet.
Otherwise a burst of noise after the conversation ends keeps the receiver
parked on a dead channel indefinitely.
"""
tx = [V(146_520_000, "nfm", 0.40, 12_500, "voice",
period_seconds=40.0, on_seconds=3.0, phase_offset=0.0),
V(146_520_000, "carrier", 0.35, 1_000, "interference",
period_seconds=40.0, on_seconds=8.0, phase_offset=36.0)]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=0.0,
hang_seconds=2.0, max_record_seconds=20.0, max_cycles=1)
assert hits
h = hits[0]
# The carrier holds squelch open until 12 s; the capture must not wait it
# out. Some lag is unavoidable -- deciding a signal carries nothing takes
# a couple of seconds of evidence -- but it must be well short of that.
assert h.duration < 10.0, f"ran {h.duration:.1f}s on an empty carrier"
assert h.duration > 3.0, "cut the real speech short"
assert "silence, static or interference" in h.stop_reason
def test_being_cut_off_mid_transmission_is_reported(tmp_path):
"""The record limit silently truncating a live signal must be visible.
Otherwise the file just ends and there is nothing to say the hang time was
never the thing that stopped it.
"""
tx = [V(146_520_000, "nfm", 0.4, 12_500, "continuous")]
notes = []
dev = SimulatedDevice(transmitters=tx).open()
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
output_dir=str(tmp_path), record_seconds=4.0,
hang_seconds=3.0, threshold_db=12, dwell_seconds=0.05,
max_cycles=1, revisit_seconds=0.2)
hits = []
s = Scanner(cfg, device=dev,
callbacks=ScannerCallbacks(on_record_end=hits.append,
on_status=notes.append))
s.prepare()
s.run()
assert hits and "record limit" in hits[0].stop_reason
assert s.stats.truncated == 1
assert any("--record 0" in n for n in notes), notes
def test_no_warning_when_the_signal_really_ended(tmp_path):
tx = [V(146_520_000, "nfm", 0.4, 12_500, "burst",
period_seconds=5.0, on_seconds=2.0)]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=20.0,
hang_seconds=1.0, max_cycles=6, revisit_seconds=0.1)
# Without a recording there is nothing to warn about either way, so the
# count alone would pass whether or not the burst was ever heard.
assert hits, "nothing recorded"
assert s.stats.truncated == 0
def test_record_zero_follows_a_transmission_past_the_old_limit(tmp_path):
tx = [V(146_520_000, "nfm", 0.4, 12_500, "continuous")]
s, hits = run_scan(tmp_path, tx, "146.4M-146.6M", record_seconds=0.0,
hang_seconds=3.0, max_record_seconds=9.0, max_cycles=1)
assert hits
assert hits[0].duration > 8.0, "stopped early despite no record limit"
assert s.stats.truncated == 0
# ---------------------------------------------------------------------------
# One file per frequency
# ---------------------------------------------------------------------------
def _two_channels():
return [V(146_520_000, "nfm", 0.40, 12_500, "A",
period_seconds=6.0, on_seconds=2.0, phase_offset=0.0),
V(147_100_000, "nfm", 0.35, 12_500, "B", pitch_hz=190.0,
period_seconds=6.0, on_seconds=2.0, phase_offset=3.0)]
def test_combining_gives_one_file_per_frequency(tmp_path):
s, hits = run_scan(tmp_path, _two_channels(), "146.4M-147.2M",
record_seconds=3.0, hang_seconds=1.0, max_cycles=4,
revisit_seconds=0.1, combine_by_frequency=True)
assert len(hits) >= 3, "not enough transmissions to test combining"
wavs = sorted(p.name for p in tmp_path.glob("*.wav"))
assert len(wavs) == 2, wavs
assert all(name.endswith("MHz.wav") for name in wavs), wavs
# every capture went into one of them
assert all(h.combined_path for h in hits)
def test_combining_removes_the_separate_files_by_default(tmp_path):
s, hits = run_scan(tmp_path, _two_channels(), "146.4M-147.2M",
record_seconds=3.0, max_cycles=3, revisit_seconds=0.1,
combine_by_frequency=True)
assert hits
dated = [p for p in tmp_path.glob("*.wav") if "--" in p.name]
assert not dated, f"per-transmission files left behind: {dated}"
# the metadata for each capture is still there
assert len(list(tmp_path.glob("*.json"))) >= len(hits)
def test_combining_can_keep_the_separate_files_too(tmp_path):
s, hits = run_scan(tmp_path, _two_channels(), "146.4M-147.2M",
record_seconds=3.0, max_cycles=3, revisit_seconds=0.1,
combine_by_frequency=True, combine_keep_individual=True)
assert hits
dated = [p for p in tmp_path.glob("*.wav") if "--" in p.name]
assert len(dated) == len(hits)
def test_a_combined_file_grows_with_each_reception(tmp_path):
from bandsaunter.recorder import read_wav
# Continuously active, so each sweep produces another capture and the
# combined file has to grow.
tx = [V(146_520_000, "nfm", 0.4, 12_500, "A")]
sizes = []
def note(hit):
if hit.kept and hit.combined_path:
sizes.append(read_wav(Path(hit.combined_path))[0].size)
dev = SimulatedDevice(transmitters=tx).open()
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
output_dir=str(tmp_path), record_seconds=2.0,
hang_seconds=1.0, threshold_db=12, dwell_seconds=0.05,
max_cycles=3, revisit_seconds=0.05,
combine_by_frequency=True, announce_timestamps=False)
s = Scanner(cfg, device=dev, callbacks=ScannerCallbacks(on_record_end=note))
s.prepare()
s.run()
assert len(sizes) >= 2, sizes
assert sizes == sorted(sizes) and sizes[-1] > sizes[0], sizes
def test_combining_off_keeps_the_old_layout(tmp_path):
s, hits = run_scan(tmp_path, _two_channels(), "146.4M-147.2M",
record_seconds=3.0, max_cycles=2, revisit_seconds=0.1)
assert hits
assert not [p for p in tmp_path.glob("*.wav") if p.name.endswith("MHz.wav")]
assert all(not h.combined_path for h in hits)
def test_a_complete_band_demodulates_each_segment_correctly(tmp_path):
"""One sweep of 2 m must handle its CW, SSB and FM segments each properly.
That is the whole point of a complete-band entry: the band is not one
mode, so scanning it as one would demodulate two thirds of it wrongly.
"""
tx = [V(144_050_000, "cw", 0.35, 500, "beacon", wpm=18,
message="VVV DE W1AW"),
V(144_200_000, "usb", 0.35, 3_000, "ssb"),
V(146_520_000, "nfm", 0.40, 12_500, "fm", ctcss=100.0)]
s, hits = run_scan(tmp_path, tx, "2m-complete", record_seconds=6.0,
hang_seconds=2.0, max_cycles=2, revisit_seconds=0.2)
def segment(freq):
if freq < 144.1e6:
return "cw"
return "ssb" if freq < 144.3e6 else "fm"
seen = {segment(h.frequency): (h.mode, h.category) for h in hits}
assert seen.get("cw") == ("cw", "cw"), seen
assert seen.get("ssb") == ("usb", "voice"), seen
assert seen.get("fm") == ("nfm", "voice"), seen
def test_frequencies_are_padded_so_names_sort_by_frequency(tmp_path):
"""Unpadded, a directory listing puts 1090 MHz before 146 MHz."""
from bandsaunter.recorder import build_stem, safe_freq_name
import datetime
assert safe_freq_name(146_520_000) == "0146.520000MHz"
assert safe_freq_name(1_090_000_000) == "1090.000000MHz"
assert safe_freq_name(1_800_000) == "0001.800000MHz"
when = datetime.datetime(2026, 8, 21, 19, 35, 48).timestamp()
names = [build_stem(when, hz, "nfm") for hz in
(1_090_000_000, 146_520_000, 98_295_546, 1_800_000, 14_058_000)]
# sorted as text must equal sorted by frequency
by_text = sorted(names)
by_freq = [build_stem(when, hz, "nfm") for hz in
sorted((1_090_000_000, 146_520_000, 98_295_546, 1_800_000,
14_058_000))]
assert by_text == by_freq
def test_combined_files_are_padded_too(tmp_path):
from bandsaunter.recorder import FrequencyLog
log = FrequencyLog(tmp_path, announce=False)
log.add(1_090_000_000.0, np.full(1600, 0.2, dtype=np.float32), 16000)
log.add(146_520_000.0, np.full(1600, 0.2, dtype=np.float32), 16000)
names = sorted(p.name for p in tmp_path.glob("*.wav"))
assert names == ["0146.520000MHz.wav", "1090.000000MHz.wav"]
def test_a_span_of_spectrum_can_be_locked_out(tmp_path):
"""Not just single frequencies: a whole noisy stretch, as one entry."""
tx = [V(146_520_000, "nfm", 0.4, 12_500, "voice"),
V(147_000_000, "nfm", 0.4, 12_500, "voice")]
s, hits = run_scan(tmp_path, tx, "146.4M-147.1M",
lockout=["146.4M-146.8M"], max_cycles=3)
assert hits, "nothing recorded at all"
assert all(h.frequency > 146_800_000 for h in hits), \
[h.frequency for h in hits]
def test_a_lock_out_made_during_a_scan_is_remembered(tmp_path):
"""It goes back to the settings file, so the next run starts with it."""
import yaml
from bandsaunter.config import load_default, save_default
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
output_dir=str(tmp_path), record_seconds=4.0)
save_default(cfg, tmp_path)
loaded, path = load_default(tmp_path)
s = Scanner(loaded, device=SimulatedDevice().open())
s.prepare()
s.lockout(146_520_000.0)
saved = yaml.safe_load(path.read_text())
assert saved["lockout"] == [{"start": 146_520_000.0, "stop": 146_520_000.0}]
# Only that one key: the rest of the file is as it was.
assert saved["record_seconds"] == 4.0
again, _ = load_default(tmp_path)
assert [lk.start for lk in again.lockout] == [146_520_000.0]
def test_lock_outs_can_be_kept_to_the_current_run(tmp_path):
from bandsaunter.config import load_default, save_default
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
output_dir=str(tmp_path), save_lockouts=False)
save_default(cfg, tmp_path)
loaded, path = load_default(tmp_path)
s = Scanner(loaded, device=SimulatedDevice().open())
s.prepare()
s.lockout(146_520_000.0)
assert s._is_locked(146_520_000.0), "still locked out for this run"
again, _ = load_default(tmp_path)
assert not again.lockout, "but not written back"
Reference in a new issue View git blame Copy permalink