dustcouncil/bandsaunter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial commit: bandsaunter, an RTL-SDR signal scanner

Browse source 
Sweeps any set of frequency ranges, records what it finds, and works out
what kind of signal it was.

- Frequency ranges entered by hand or picked from a 135-entry US band plan,
  including whole-band and all-CW sweeps that resolve the demodulator per
  segment.
- Detection calibrated against the peak-hold detector's own noise statistics,
  so the threshold means real margin over static rather than over the floor.
- A content gate: captures are kept only if they carry voice, decodable CW,
  or an identified digital keying scheme. Speech is recognised by a pitch
  track that drifts, which static cannot imitate.
- Identification of NFM/WFM/AM/SSB, CW with Morse decoded to text, P25, DMR,
  NXDN, D-STAR, POCSAG, FLEX, ACARS, AIS, APRS, n-FSK and n-PSK.
- Gapless streaming capture, with the signal path fast enough to keep up in
  real time, so recordings play back at the right speed.
- Optional one-file-per-frequency recording with spoken timestamps, and
  speech-to-text transcription.
- Menus and command line generated from one settings table, so neither can
  offer something the other cannot; settings persist in ~/.config.

367 tests, run against synthetic signals, a built-in receiver simulator, and
real hardware.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
The Dust Council 2026-08-21 20:50:20 -07:00
commit db3e0c79b9
39 changed files with 13473 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

321
tests/test_announce.py Normal file
View file

@ -0,0 +1,321 @@
"""Spoken announcements and the files they are appended to."""
from datetime import datetime
import numpy as np
import pytest
from bandsaunter import announce
from bandsaunter.recorder import FrequencyLog, append_wav, read_wav
from bandsaunter.quality import voice_metrics
# ---------------------------------------------------------------------------
# Words
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("n,words", [
(0, "zero"), (7, "seven"), (10, "ten"), (12, "twelve"), (19, "nineteen"),
(20, "twenty"), (21, "twenty one"), (45, "forty five"), (59, "fifty nine"),
(100, "one hundred"), (146, "one hundred forty six"),
(2026, "two thousand twenty six"),
])
def test_number_words(n, words):
assert " ".join(announce.number_words(n)) == words
def test_timestamp_phrase_reads_naturally():
phrase = announce.timestamp_phrase(datetime(2026, 8, 21, 14, 38, 5))
assert "august" in phrase
assert "twenty one" in phrase
assert "twenty twenty six" in phrase # years are said in pairs
assert "fourteen thirty eight" in phrase
assert "oh five" in phrase # 05 seconds, not "five"
def test_midnight_and_single_digits():
phrase = announce.timestamp_phrase(datetime(2026, 1, 3, 0, 5, 0))
assert "january three" in phrase
assert "oh oh" in phrase # hour 00 and second 00
def test_frequency_is_spoken_digit_by_digit():
phrase = announce.timestamp_phrase(datetime(2026, 1, 1, 0, 0, 0),
frequency=146_520_000.0)
assert "one hundred forty six point one four six" not in phrase
assert "point five two zero" in phrase
assert "megahertz" in phrase
def test_every_word_in_the_vocabulary_can_be_spoken():
"""A word with no pronunciation would be silently dropped."""
for word in announce.WORDS:
for phone in announce.WORDS[word].split():
assert phone in announce.PHONEMES, f"{word} uses unknown {phone}"
def test_all_the_words_a_timestamp_needs_are_covered():
for month in range(1, 13):
for day in (1, 9, 15, 21, 28): # 28 is safe in every month
for hour in (0, 5, 12, 23):
phrase = announce.timestamp_phrase(
datetime(2026, month, day, hour, 45, 9))
for word in phrase.split():
assert word in ("_", "__") or word in announce.WORDS, word
# ---------------------------------------------------------------------------
# Synthesis
# ---------------------------------------------------------------------------
def _lpc_formants(x, fs, order=12):
x = x - x.mean()
x = np.append(x[0], x[1:] - 0.97 * x[:-1]) * np.hamming(x.size)
r = np.correlate(x, x, "full")[x.size - 1:x.size - 1 + order + 1]
if r[0] == 0:
return []
a = np.zeros(order + 1)
a[0], e = 1.0, r[0]
for i in range(1, order + 1):
k = -(a[:i] @ r[i:0:-1]) / e
a[:i + 1] = (np.concatenate([a[:i], [0]])
+ k * np.concatenate([[0], a[i - 1::-1]]))
e *= (1 - k * k)
if e <= 0:
break
roots = [z for z in np.roots(a) if np.imag(z) > 0.01]
return [f for f in sorted(float(np.angle(z) * fs / (2 * np.pi))
for z in roots) if 120 < f < 5000]
@pytest.mark.parametrize("vowel", ["iy", "ih", "eh", "ae", "aa", "ao",
"uw", "uh", "ah", "er"])
def test_vowels_land_on_their_formant_targets(vowel):
"""Intelligibility rests on the formants being where they are meant to be.
Summing resonators in parallel instead of cascading them loses the first
formant entirely, and every vowel then sounds the same.
"""
spec = announce.PHONEMES[vowel]
announce.WORDS["_probe_"] = vowel
audio = announce.synthesize("_probe_", 16000)
seg = audio[int(audio.size * 0.35):][:800]
formants = _lpc_formants(seg, 16000)
assert len(formants) >= 2, f"{vowel}: found {formants}"
assert abs(formants[0] - spec.f1) < 130, f"{vowel} F1: {formants}"
assert abs(formants[1] - spec.f2) < 250, f"{vowel} F2: {formants}"
def test_speech_is_in_the_voice_band():
audio = announce.speak_timestamp(datetime(2026, 8, 21, 14, 38, 5))
n = 1 << 14
spec = np.abs(np.fft.rfft(audio[:n] * np.hanning(n))) ** 2
freqs = np.fft.rfftfreq(n, 1 / 16000)
band = (freqs >= 250) & (freqs < 3400)
assert spec[band].sum() / spec.sum() > 0.6
def test_the_announcement_reads_as_speech():
"""The voice detector should hear the announcement as a voice."""
audio = announce.speak_timestamp(datetime(2026, 8, 21, 14, 38, 5))
m = voice_metrics(audio, 16000)
assert m.score > 0.6, m.describe()
assert 80 < m.pitch_hz < 200
def test_synthesis_is_fast_enough_to_run_inline():
import time
t0 = time.perf_counter()
audio = announce.speak_timestamp(datetime(2026, 8, 21, 14, 38, 5))
elapsed = time.perf_counter() - t0
assert elapsed < 0.25 * audio.size / 16000, f"{elapsed:.2f}s"
def test_unknown_words_do_not_crash():
assert announce.synthesize("gibberishwordnotinvocabulary", 16000).size == 0
# ---------------------------------------------------------------------------
# Appending
# ---------------------------------------------------------------------------
def test_a_wav_stays_valid_after_every_append(tmp_path):
"""A scan can be stopped at any moment; the file so far must still play."""
path = tmp_path / "acc.wav"
for i in range(1, 5):
append_wav(path, np.full(1600, 0.1 * i, dtype=np.float32), 16000)
audio, rate = read_wav(path)
assert rate == 16000
assert audio.size == 1600 * i
import wave
with wave.open(str(path)) as w: # the standard reader must agree
assert w.getnframes() == 1600 * i
def test_appending_a_different_rate_is_refused(tmp_path):
path = tmp_path / "acc.wav"
append_wav(path, np.zeros(160, dtype=np.float32), 16000)
with pytest.raises(ValueError, match="16000"):
append_wav(path, np.zeros(160, dtype=np.float32), 32000)
def test_frequency_log_groups_nearby_receptions(tmp_path):
log = FrequencyLog(tmp_path, tolerance_hz=6250.0, announce=False)
audio = np.full(1600, 0.2, dtype=np.float32)
log.add(146_520_040.0, audio, 16000)
log.add(146_519_800.0, audio, 16000) # 240 Hz away: same channel
log.add(147_100_000.0, audio, 16000) # far away: its own file
files = sorted(p.name for p in tmp_path.glob("*.wav"))
assert len(files) == 2, files
assert log.appended == 3
def test_frequency_log_resamples_a_mismatched_capture(tmp_path):
log = FrequencyLog(tmp_path, announce=False)
log.add(146_520_000.0, np.full(16000, 0.2, dtype=np.float32), 16000)
log.add(146_520_000.0, np.full(32000, 0.2, dtype=np.float32), 32000)
audio, rate = read_wav(log.files[0])
assert rate == 16000
# one second at each rate, plus the gaps between them
assert 2.0 <= audio.size / rate <= 3.0
def test_announcement_is_placed_before_each_recording(tmp_path):
quiet = FrequencyLog(tmp_path / "a", announce=False)
spoken = FrequencyLog(tmp_path / "b", announce=True)
body = np.full(16000, 0.2, dtype=np.float32)
quiet.add(146_520_000.0, body, 16000, when=datetime(2026, 8, 21, 14, 38, 5))
spoken.add(146_520_000.0, body, 16000, when=datetime(2026, 8, 21, 14, 38, 5))
short, _ = read_wav(quiet.files[0])
long_, rate = read_wav(spoken.files[0])
assert long_.size > short.size + 2 * rate, "no announcement was inserted"
# the recording is a steady 0.2; the announcement is not, and comes first
assert voice_metrics(long_[:int(3 * rate)], rate).score > 0.5
def test_an_existing_file_from_an_earlier_run_is_continued(tmp_path):
first = FrequencyLog(tmp_path, announce=False)
first.add(146_520_000.0, np.full(1600, 0.2, dtype=np.float32), 16000)
before = read_wav(first.files[0])[0].size
second = FrequencyLog(tmp_path, announce=False)
second.add(146_520_000.0, np.full(1600, 0.2, dtype=np.float32), 16000)
after = read_wav(second.files[0])[0].size
assert after > before, "a later run started a new file instead of appending"
assert len(list(tmp_path.glob("*.wav"))) == 1
def test_the_date_is_only_spoken_when_it_changes(tmp_path):
"""Repeating the date before every over takes longer than most overs last."""
log = FrequencyLog(tmp_path, announce=True)
body = np.full(16000, 0.2, dtype=np.float32)
lengths = []
previous = 0
for when in (datetime(2026, 8, 21, 14, 38, 5),
datetime(2026, 8, 21, 14, 41, 0),
datetime(2026, 8, 22, 9, 2, 0)):
log.add(146_520_000.0, body, 16000, when=when)
size = read_wav(log.files[0])[0].size
lengths.append(size - previous)
previous = size
first, same_day, next_day = lengths
assert same_day < first * 0.7, "the date was repeated needlessly"
assert next_day > same_day * 1.5, "a new day should get the full date"
def test_time_only_announcement_is_shorter():
when = datetime(2026, 8, 21, 14, 38, 5)
full = announce.speak_timestamp(when, 16000, with_date=True)
brief = announce.speak_timestamp(when, 16000, with_date=False)
assert brief.size < full.size
assert brief.size > 8000, "the time itself must still be spoken"
# ---------------------------------------------------------------------------
# Installed engines
# ---------------------------------------------------------------------------
HAVE_ENGINE = announce.available_engine() is not None
needs_engine = pytest.mark.skipif(not HAVE_ENGINE,
reason="no text-to-speech program installed")
def test_engine_detection_does_not_throw():
engine = announce.available_engine()
assert engine is None or engine in announce.ENGINES
@pytest.mark.parametrize("when", [
datetime(2026, 8, 21, 14, 38, 5),
datetime(2026, 1, 3, 9, 5, 0),
datetime(2026, 12, 31, 23, 59, 59),
])
def test_engine_text_avoids_the_forms_engines_misread(when):
"""Punctuation is what gives an engine its phrasing, and the obvious
spellings are traps: a colon makes espeak read 14:38:05 as "fourteen
thirty, eight zero five", and an ISO date has its dashes read aloud."""
text = announce.timestamp_text(when)
assert ":" not in text
assert "-" not in text
assert text.count(",") >= 2, text # date, year and time separated
assert "twenty twenty six" in text # not "two thousand and ..."
def test_engine_text_covers_the_time_only_case():
text = announce.timestamp_text(datetime(2026, 8, 21, 9, 5, 0),
with_date=False)
assert "August" not in text
assert "09 05" in text
@needs_engine
def test_installed_engine_renders_speech():
audio = announce.speak_timestamp(datetime(2026, 8, 21, 14, 38, 5), 16000,
engine="auto")
assert audio.size > 16000, "suspiciously short"
assert audio.dtype == np.float32
m = voice_metrics(audio, 16000)
assert m.score > 0.6, m.describe()
@needs_engine
@pytest.mark.parametrize("rate", [8000, 16000, 32000, 48000])
def test_installed_engine_is_resampled_to_the_asked_for_rate(rate):
"""Engines render at their own rate; espeak-ng uses 22050 Hz."""
audio = announce.speak_timestamp(datetime(2026, 8, 21, 14, 38, 5), rate,
engine="auto")
seconds = audio.size / rate
assert 2.0 < seconds < 12.0, f"{seconds:.2f}s at {rate} Hz"
@needs_engine
def test_both_engines_produce_the_same_level():
"""Switching engines must not change how loud announcements are."""
when = datetime(2026, 8, 21, 14, 38, 5)
external = announce.speak_timestamp(when, 16000, engine="auto")
builtin = announce.speak_timestamp(when, 16000, engine="builtin")
assert abs(float(np.abs(external).max())
- float(np.abs(builtin).max())) < 0.05
@needs_engine
def test_dropping_the_date_shortens_the_announcement():
when = datetime(2026, 8, 21, 14, 38, 5)
full = announce.speak_timestamp(when, 16000, engine="auto")
brief = announce.speak_timestamp(when, 16000, engine="auto",
with_date=False)
assert brief.size < full.size * 0.75
def test_an_engine_that_is_not_installed_falls_back_to_the_builtin():
"""The feature must keep working with nothing else on the machine."""
audio = announce.speak_timestamp(datetime(2026, 8, 21, 14, 38, 5), 16000,
engine="definitely-not-installed")
assert audio.size > 0
assert voice_metrics(audio, 16000).score > 0.6
def test_a_failing_engine_falls_back_rather_than_crashing(monkeypatch):
monkeypatch.setattr(announce, "_external", lambda *a, **k: None)
audio = announce.speak_timestamp(datetime(2026, 8, 21, 14, 38, 5), 16000,
engine="auto")
assert audio.size > 0