db3e0c79b9
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>
378 lines
16 KiB
Python
378 lines
16 KiB
Python
"""Speech to text: the engine plumbing, and how captures reach it."""
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from bandsaunter import transcribe as tr
|
|
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
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_engine(monkeypatch):
|
|
"""A recogniser that reports what it was given, so the plumbing is testable
|
|
on a machine with none installed."""
|
|
seen = []
|
|
|
|
def engine(audio, rate, model, language):
|
|
seen.append({"samples": audio.size, "rate": rate, "model": model,
|
|
"language": language})
|
|
return tr.Transcript(text="this is the transcribed text",
|
|
engine="fake", language=language or "en")
|
|
|
|
monkeypatch.setitem(tr._DISPATCH, "fake", engine)
|
|
monkeypatch.setattr(tr, "ENGINES", ("fake",) + tr.ENGINES)
|
|
monkeypatch.setattr(tr, "_is_present", lambda name: name == "fake")
|
|
return seen
|
|
|
|
|
|
def _speech(seconds=3.0, rate=16000):
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
from speech import synth_speech
|
|
return synth_speech(seconds, rate, 120, 0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Engines
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_no_engine_reports_rather_than_failing(monkeypatch):
|
|
"""Every capture failing noisily would be worse than saying so once."""
|
|
monkeypatch.setattr(tr, "_is_present", lambda name: False)
|
|
assert tr.available_engine() is None
|
|
assert tr.transcribe(np.zeros(1000, np.float32), 16000) is None
|
|
|
|
|
|
def test_engine_listing_is_complete():
|
|
listed = {name for name, _, _ in tr.describe_engines()}
|
|
assert listed == set(tr.ENGINES)
|
|
for _, _, how in tr.describe_engines():
|
|
assert how, "every engine should say how to get it"
|
|
|
|
|
|
def test_a_failing_engine_is_reported_not_raised(monkeypatch):
|
|
def boom(audio, rate, model, language):
|
|
raise RuntimeError("model file is corrupt")
|
|
monkeypatch.setitem(tr._DISPATCH, "fake", boom)
|
|
monkeypatch.setattr(tr, "_is_present", lambda name: name == "fake")
|
|
result = tr.transcribe(np.zeros(16000, np.float32), 16000, engine="fake")
|
|
assert result is not None and not result
|
|
assert "corrupt" in result.note
|
|
|
|
|
|
def test_audio_is_resampled_to_what_the_engines_expect(fake_engine):
|
|
for rate in (8000, 16000, 32000, 48000):
|
|
tr.transcribe(np.zeros(int(rate * 2), np.float32), rate, engine="fake")
|
|
assert [s["samples"] for s in fake_engine] == [32000] * 4
|
|
|
|
|
|
def test_the_model_and_language_reach_the_engine(fake_engine):
|
|
tr.transcribe(np.zeros(16000, np.float32), 16000, engine="fake",
|
|
model="small.en", language="fr")
|
|
assert fake_engine[-1]["model"] == "small.en"
|
|
assert fake_engine[-1]["language"] == "fr"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The worker
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _drain(worker, timeout=20.0):
|
|
worker.close(timeout=timeout)
|
|
|
|
|
|
def test_worker_writes_a_transcript_beside_the_recording(tmp_path, fake_engine):
|
|
worker = tr.TranscriptionWorker(engine="fake")
|
|
worker.start()
|
|
out = tmp_path / "0146.520000MHz--2026-08-21_12_00_00-nfm_transcription.txt"
|
|
worker.submit(_speech(), 16000, out, datetime(2026, 8, 21, 12, 0, 0),
|
|
146.52e6)
|
|
_drain(worker)
|
|
assert out.exists()
|
|
assert "transcribed text" in out.read_text()
|
|
assert worker.written == 1
|
|
|
|
|
|
def test_worker_appends_with_a_timestamp_when_combining(tmp_path, fake_engine):
|
|
worker = tr.TranscriptionWorker(engine="fake")
|
|
worker.start()
|
|
out = tmp_path / "0146.520000MHz_transcription.txt"
|
|
for minute in (0, 5, 9):
|
|
worker.submit(_speech(), 16000, out,
|
|
datetime(2026, 8, 21, 12, minute, 0), 146.52e6,
|
|
append=True)
|
|
_drain(worker)
|
|
lines = out.read_text().strip().split("\n")
|
|
assert len(lines) == 3
|
|
assert lines[0].startswith("[2026-08-21 12:00:00] ")
|
|
assert lines[2].startswith("[2026-08-21 12:09:00] ")
|
|
|
|
|
|
def test_nothing_recognised_writes_no_file_at_all(tmp_path, monkeypatch):
|
|
"""A directory of placeholder files is worse than no file."""
|
|
monkeypatch.setitem(tr._DISPATCH, "fake",
|
|
lambda a, r, m, l: tr.Transcript(text="", engine="fake"))
|
|
monkeypatch.setattr(tr, "_is_present", lambda name: name == "fake")
|
|
worker = tr.TranscriptionWorker(engine="fake")
|
|
worker.start()
|
|
out = tmp_path / "quiet_transcription.txt"
|
|
worker.submit(np.zeros(16000, np.float32), 16000, out, datetime.now(), 1e6)
|
|
_drain(worker)
|
|
assert not out.exists()
|
|
assert not list(tmp_path.iterdir())
|
|
assert worker.empty == 1 and worker.written == 0
|
|
|
|
|
|
def test_whitespace_only_speech_writes_no_file(tmp_path, monkeypatch):
|
|
monkeypatch.setitem(tr._DISPATCH, "fake",
|
|
lambda a, r, m, l: tr.Transcript(text=" \n ",
|
|
engine="fake"))
|
|
monkeypatch.setattr(tr, "_is_present", lambda name: name == "fake")
|
|
worker = tr.TranscriptionWorker(engine="fake")
|
|
worker.start()
|
|
out = tmp_path / "blank_transcription.txt"
|
|
worker.submit(np.zeros(16000, np.float32), 16000, out, datetime.now(), 1e6)
|
|
_drain(worker)
|
|
assert not out.exists() and worker.empty == 1
|
|
|
|
|
|
def test_an_empty_result_adds_no_line_when_combining(tmp_path, monkeypatch):
|
|
"""Combined transcripts must not fill up with empty timestamps."""
|
|
texts = iter(["something was said", "", "and something else"])
|
|
monkeypatch.setitem(tr._DISPATCH, "fake",
|
|
lambda a, r, m, l: tr.Transcript(text=next(texts),
|
|
engine="fake"))
|
|
monkeypatch.setattr(tr, "_is_present", lambda name: name == "fake")
|
|
worker = tr.TranscriptionWorker(engine="fake")
|
|
worker.start()
|
|
out = tmp_path / "0146.520000MHz_transcription.txt"
|
|
for minute in (0, 5, 9):
|
|
worker.submit(np.zeros(16000, np.float32), 16000, out,
|
|
datetime(2026, 8, 21, 12, minute, 0), 1e6, append=True)
|
|
_drain(worker)
|
|
lines = out.read_text().strip().split("\n")
|
|
assert len(lines) == 2, lines
|
|
assert "12:00:00" in lines[0] and "12:09:00" in lines[1]
|
|
|
|
|
|
def test_worker_does_not_hold_up_the_caller(tmp_path, monkeypatch):
|
|
"""Recognition takes seconds; a scan must not wait for it."""
|
|
def slow(audio, rate, model, language):
|
|
time.sleep(1.0)
|
|
return tr.Transcript(text="eventually", engine="fake")
|
|
monkeypatch.setitem(tr._DISPATCH, "fake", slow)
|
|
monkeypatch.setattr(tr, "_is_present", lambda name: name == "fake")
|
|
worker = tr.TranscriptionWorker(engine="fake")
|
|
worker.start()
|
|
started = time.perf_counter()
|
|
for i in range(3):
|
|
worker.submit(_speech(), 16000, tmp_path / f"{i}.txt", datetime.now(),
|
|
1e6)
|
|
assert time.perf_counter() - started < 0.5, "submitting blocked"
|
|
_drain(worker)
|
|
assert worker.written == 3
|
|
|
|
|
|
def test_a_full_queue_is_counted_not_blocked(tmp_path, monkeypatch):
|
|
def slow(audio, rate, model, language):
|
|
time.sleep(0.4)
|
|
return tr.Transcript(text="x", engine="fake")
|
|
monkeypatch.setitem(tr._DISPATCH, "fake", slow)
|
|
monkeypatch.setattr(tr, "_is_present", lambda name: name == "fake")
|
|
worker = tr.TranscriptionWorker(engine="fake", max_queue=2)
|
|
worker.start()
|
|
accepted = sum(worker.submit(_speech(0.5), 16000, tmp_path / f"{i}.txt",
|
|
datetime.now(), 1e6) for i in range(12))
|
|
assert accepted < 12 and worker.dropped > 0
|
|
_drain(worker)
|
|
|
|
|
|
def test_empty_audio_is_not_submitted(tmp_path, fake_engine):
|
|
worker = tr.TranscriptionWorker(engine="fake")
|
|
worker.start()
|
|
assert not worker.submit(np.zeros(0, np.float32), 16000,
|
|
tmp_path / "x.txt", datetime.now(), 1e6)
|
|
_drain(worker)
|
|
assert worker.written == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Through a scan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _scan(tmp_path, transmitters, **over):
|
|
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
|
|
output_dir=str(tmp_path), record_seconds=4.0,
|
|
hang_seconds=1.0, threshold_db=12, dwell_seconds=0.05,
|
|
max_cycles=1, revisit_seconds=0.2, transcribe=True,
|
|
transcribe_engine="fake")
|
|
for k, v in over.items():
|
|
setattr(cfg, k, v)
|
|
hits = []
|
|
scanner = Scanner(cfg, device=SimulatedDevice(transmitters=transmitters).open(),
|
|
callbacks=ScannerCallbacks(on_record_end=hits.append))
|
|
scanner.prepare()
|
|
scanner.run()
|
|
return scanner, [h for h in hits if h.kept]
|
|
|
|
|
|
def test_a_voice_capture_is_transcribed(tmp_path, fake_engine, monkeypatch):
|
|
monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake")
|
|
scanner, hits = _scan(tmp_path, [V(146_520_000, "nfm", 0.4, 12_500, "v")])
|
|
assert hits and hits[0].category == "voice"
|
|
written = list(tmp_path.glob("*_transcription.txt"))
|
|
assert written, "no transcript written"
|
|
assert written[0].stem.startswith(hits[0].filename)
|
|
assert "transcribed text" in written[0].read_text()
|
|
|
|
|
|
def test_morse_and_data_are_not_transcribed(tmp_path, fake_engine, monkeypatch):
|
|
"""Running a recogniser over CW or a data burst wastes seconds per capture."""
|
|
monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake")
|
|
scanner, hits = _scan(tmp_path, [
|
|
V(146_520_000, "fsk4", 0.4, 12_500, "data", baud=4800, deviation=1800)])
|
|
assert hits and hits[0].category == "digital"
|
|
assert not list(tmp_path.glob("*_transcription.txt"))
|
|
|
|
|
|
def test_short_captures_are_skipped(tmp_path, fake_engine, monkeypatch):
|
|
monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake")
|
|
scanner, hits = _scan(tmp_path, [V(146_520_000, "nfm", 0.4, 12_500, "v")],
|
|
transcribe_min_seconds=60.0)
|
|
assert hits
|
|
assert not list(tmp_path.glob("*_transcription.txt"))
|
|
|
|
|
|
def test_the_transcript_is_recorded_in_the_metadata(tmp_path, fake_engine,
|
|
monkeypatch):
|
|
import json
|
|
monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake")
|
|
scanner, hits = _scan(tmp_path, [V(146_520_000, "nfm", 0.4, 12_500, "v")])
|
|
meta = json.loads(Path(hits[0].meta_path).read_text())
|
|
assert meta["hit"]["transcript_path"].endswith("_transcription.txt")
|
|
assert "transcribed text" in meta["hit"]["transcript"]
|
|
|
|
|
|
def test_the_metadata_never_names_a_transcript_that_was_not_written(
|
|
tmp_path, monkeypatch):
|
|
"""Recording a path for a file that never appears would be a lie."""
|
|
import json
|
|
monkeypatch.setitem(tr._DISPATCH, "fake",
|
|
lambda a, r, m, l: tr.Transcript(text="", engine="fake"))
|
|
monkeypatch.setattr(tr, "_is_present", lambda name: name == "fake")
|
|
monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake")
|
|
scanner, hits = _scan(tmp_path, [V(146_520_000, "nfm", 0.4, 12_500, "v")])
|
|
assert hits
|
|
assert not list(tmp_path.glob("*_transcription.txt"))
|
|
meta = json.loads(Path(hits[0].meta_path).read_text())
|
|
assert not meta["hit"].get("transcript_path")
|
|
|
|
|
|
def test_combined_recordings_get_one_transcript_per_frequency(tmp_path,
|
|
fake_engine,
|
|
monkeypatch):
|
|
monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake")
|
|
scanner, hits = _scan(
|
|
tmp_path, [V(146_520_000, "nfm", 0.4, 12_500, "v")],
|
|
combine_by_frequency=True, announce_timestamps=False,
|
|
record_seconds=2.0, max_cycles=3, revisit_seconds=0.05)
|
|
assert len(hits) >= 2
|
|
written = list(tmp_path.glob("*_transcription.txt"))
|
|
assert len(written) == 1, written
|
|
lines = written[0].read_text().strip().split("\n")
|
|
assert len(lines) == len(hits)
|
|
assert all(line.startswith("[") for line in lines)
|
|
|
|
|
|
def test_transcription_off_writes_nothing(tmp_path, fake_engine, monkeypatch):
|
|
monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: "fake")
|
|
scanner, hits = _scan(tmp_path, [V(146_520_000, "nfm", 0.4, 12_500, "v")],
|
|
transcribe=False)
|
|
assert hits
|
|
assert not list(tmp_path.glob("*_transcription.txt"))
|
|
|
|
|
|
def test_missing_engine_says_so_once(tmp_path, monkeypatch):
|
|
monkeypatch.setattr("bandsaunter.scanner.available_engine", lambda: None)
|
|
notes = []
|
|
cfg = ScanConfig(ranges=parse_range_list("146.4M-146.6M"),
|
|
output_dir=str(tmp_path), record_seconds=2.0,
|
|
threshold_db=12, max_cycles=1, transcribe=True)
|
|
scanner = Scanner(cfg,
|
|
device=SimulatedDevice(transmitters=[
|
|
V(146_520_000, "nfm", 0.4, 12_500, "v")]).open(),
|
|
callbacks=ScannerCallbacks(on_status=notes.append))
|
|
scanner.prepare()
|
|
scanner.run()
|
|
assert any("no speech recogniser" in n for n in notes), notes
|
|
assert scanner.transcriber is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Installed engines, when there are any
|
|
# ---------------------------------------------------------------------------
|
|
|
|
INSTALLED = tr.available_engine()
|
|
needs_engine = pytest.mark.skipif(INSTALLED is None,
|
|
reason="no speech recogniser installed")
|
|
|
|
|
|
@needs_engine
|
|
def test_an_installed_engine_recognises_synthesised_speech():
|
|
"""Round trip: say something, then read it back off the audio."""
|
|
from bandsaunter.announce import say
|
|
audio = say("one two three four five six seven eight nine", 16000)
|
|
result = tr.transcribe(audio, 16000, engine="auto",
|
|
model="base.en", language="en")
|
|
assert result is not None
|
|
assert not result.note, result.note
|
|
# Engines are free to write numbers as digits, and whisper does.
|
|
text = result.text.lower()
|
|
spelled = ("one", "two", "three", "four", "five", "six", "seven",
|
|
"eight", "nine")
|
|
hits = sum((word in text) or (str(i) in text)
|
|
for i, word in enumerate(spelled, 1))
|
|
assert hits >= 4, f"only {hits} of nine numbers recognised: {result.text!r}"
|
|
|
|
|
|
@needs_engine
|
|
@pytest.mark.parametrize("rate", [8000, 16000, 32000])
|
|
def test_an_installed_engine_copes_with_any_rate(rate):
|
|
from bandsaunter.announce import say
|
|
audio = say("testing one two three", rate)
|
|
result = tr.transcribe(audio, rate, engine="auto", model="base.en",
|
|
language="en")
|
|
assert result is not None and not result.note, result.note
|
|
|
|
|
|
@needs_engine
|
|
def test_silence_produces_no_transcript_rather_than_invention():
|
|
result = tr.transcribe(np.zeros(16000 * 3, np.float32), 16000,
|
|
engine="auto", model="base.en", language="en")
|
|
assert result is not None
|
|
assert not result.text.strip(), f"invented {result.text!r} from silence"
|
|
|
|
|
|
@pytest.mark.skipif(not tr._is_present("vosk"), reason="vosk not installed")
|
|
def test_vosk_falls_back_when_given_a_whisper_model_name():
|
|
"""The model setting is shared with whisper, whose names are not paths."""
|
|
from bandsaunter.announce import say
|
|
result = tr.transcribe(say("one two three", 16000), 16000, engine="vosk",
|
|
model="base.en", language="en")
|
|
assert result is not None
|
|
assert not result.note, result.note
|
|
|
|
|
|
@pytest.mark.skipif(not tr._is_present("vosk"), reason="vosk not installed")
|
|
def test_vosk_accepts_the_plain_language_code():
|
|
"""Vosk names its models by region and rejects a bare "en"."""
|
|
from bandsaunter.announce import say
|
|
result = tr.transcribe(say("one two three", 16000), 16000, engine="vosk",
|
|
language="en")
|
|
assert result is not None and not result.note, result.note
|