8d94a52942
No speech recogniser is in Debian, so installing bandsaunter from a .deb left transcription to a manual pip step on every machine. A repository of one's own is not bound by archive policy, so build-repo.sh now packages faster-whisper and the base.en model alongside the application: bandsaunter the application (Architecture: all) bandsaunter-transcribe faster-whisper, vendored (amd64) bandsaunter-model-base-en the model, so nothing reaches the network The wheels land in /usr/lib/bandsaunter/vendor rather than dist-packages, and transcribe.py appends that directory to sys.path -- appends, so an apt-managed numpy or PyYAML still wins and the vendor copy only fills the gap. Duplicates of what Debian already ships are stripped from the tree. resolve_model() turns a bare "base.en" into the packaged copy when one is installed, and leaves it alone to be downloaded when none is. The app package recommends the other two, so "apt install bandsaunter" brings the lot and --no-install-recommends still gets just the scanner. Its postinst explains how to add a recogniser only when there genuinely is not one -- including the case where apt has already unpacked the recogniser package but not yet configured it. Verified with the source tree hidden and no home directory: the packaged CLI runs, and a real recording transcribes offline from the vendored engine and packaged model while numpy still resolves to the system one. apt itself resolves the repository over HTTP and plans all three. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
426 lines
18 KiB
Python
426 lines
18 KiB
Python
"""Speech to text: the engine plumbing, and how captures reach it."""
|
|
import sys
|
|
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
|
|
|
|
|
|
# --- packaged engine and models -------------------------------------------
|
|
|
|
def test_a_packaged_model_is_used_in_place_of_its_name(tmp_path, monkeypatch):
|
|
""""base.en" should mean the copy on this machine when one is installed."""
|
|
monkeypatch.setattr(tr, "MODEL_DIR", tmp_path)
|
|
(tmp_path / "base.en").mkdir()
|
|
assert tr.resolve_model("base.en") == str(tmp_path / "base.en")
|
|
|
|
|
|
def test_an_unpackaged_model_keeps_its_name_to_be_downloaded(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(tr, "MODEL_DIR", tmp_path)
|
|
assert tr.resolve_model("small.en") == "small.en"
|
|
assert tr.resolve_model("") == ""
|
|
|
|
|
|
def test_an_explicit_model_directory_wins_over_the_packaged_one(tmp_path, monkeypatch):
|
|
packaged = tmp_path / "packaged"
|
|
(packaged / "base.en").mkdir(parents=True)
|
|
monkeypatch.setattr(tr, "MODEL_DIR", packaged)
|
|
mine = tmp_path / "base.en"
|
|
mine.mkdir()
|
|
assert tr.resolve_model(str(mine)) == str(mine)
|
|
|
|
|
|
def test_the_vendored_engine_is_searched_after_the_system_one(tmp_path, monkeypatch):
|
|
"""Anything apt provides must still win; the vendor copy fills a gap."""
|
|
monkeypatch.setattr(sys, "path", list(sys.path))
|
|
monkeypatch.setattr(tr, "VENDOR_DIR", tmp_path)
|
|
tr._add_vendor_path()
|
|
assert sys.path[-1] == str(tmp_path)
|
|
|
|
|
|
def test_the_vendor_path_is_added_once(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(sys, "path", list(sys.path))
|
|
monkeypatch.setattr(tr, "VENDOR_DIR", tmp_path)
|
|
tr._add_vendor_path()
|
|
tr._add_vendor_path()
|
|
assert sys.path.count(str(tmp_path)) == 1
|
|
|
|
|
|
def test_a_missing_vendor_directory_is_not_added(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(sys, "path", list(sys.path))
|
|
monkeypatch.setattr(tr, "VENDOR_DIR", tmp_path / "absent")
|
|
tr._add_vendor_path()
|
|
assert str(tmp_path / "absent") not in sys.path
|