dustcouncil/bandsaunter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
bandsaunter/bandsaunter/announce.py
The Dust Council db3e0c79b9 Initial commit: bandsaunter, an RTL-SDR signal scanner 
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>
2026-08-21 20:50:20 -07:00

436 lines
18 KiB
Python
Raw Blame History

"""Spoken announcements, for labelling recordings with the time.
Uses an installed text-to-speech program when there is one. When there is
not -- which is the common case on a headless box -- it falls back to a
built-in formant synthesiser. The vocabulary an announcement needs is small
and fixed (numbers, month names, a few words), so synthesising it directly is
practical and keeps the feature working with nothing else installed.
"""
from __future__ import annotations
import math
import shutil
import subprocess
import tempfile
import wave
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
import numpy as np
from scipy import signal as sps
__all__ = ["say", "speak_timestamp", "timestamp_phrase", "timestamp_text",
"number_words", "available_engine", "ENGINES"]
# ---------------------------------------------------------------------------
# Phonemes. Formant targets in Hz, roughly the Peterson-Barney vowel centres
# with consonant loci from the standard synthesis-by-rule tables.
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Phone:
kind: str # vowel, nasal, fric, stop, approx, silence
f1: float = 500.0
f2: float = 1500.0
f3: float = 2500.0
duration: float = 0.09 # seconds at normal speed
voiced: bool = True
noise: float = 0.0 # fricative noise level
amp: float = 1.0
# Diphthongs and glides move towards a second target.
to: tuple[float, float, float] | None = None
P = Phone
PHONEMES: dict[str, Phone] = {
# -- vowels ----------------------------------------------------------
"iy": P("vowel", 300, 2300, 3000, 0.11),
"ih": P("vowel", 400, 2000, 2550, 0.075),
"eh": P("vowel", 550, 1850, 2500, 0.09),
"ae": P("vowel", 660, 1720, 2410, 0.11),
"aa": P("vowel", 730, 1090, 2440, 0.11),
"ao": P("vowel", 570, 840, 2410, 0.10),
"uh": P("vowel", 440, 1020, 2240, 0.075),
"uw": P("vowel", 320, 900, 2200, 0.11),
"ah": P("vowel", 640, 1190, 2390, 0.075),
"er": P("vowel", 490, 1350, 1690, 0.10),
# -- diphthongs -------------------------------------------------------
"ay": P("vowel", 730, 1090, 2440, 0.16, to=(400, 2000, 2550)),
"ey": P("vowel", 550, 1850, 2500, 0.15, to=(300, 2300, 3000)),
"ow": P("vowel", 570, 840, 2410, 0.15, to=(320, 900, 2200)),
"aw": P("vowel", 730, 1090, 2440, 0.16, to=(440, 1020, 2240)),
"oy": P("vowel", 570, 840, 2410, 0.16, to=(400, 2000, 2550)),
# -- nasals ------------------------------------------------------------
"m": P("nasal", 300, 1100, 2200, 0.07, amp=0.55),
"n": P("nasal", 300, 1600, 2600, 0.07, amp=0.55),
"ng": P("nasal", 300, 2000, 2600, 0.07, amp=0.55),
# -- approximants ------------------------------------------------------
"l": P("approx", 380, 1100, 2600, 0.07, amp=0.75),
"r": P("approx", 400, 1100, 1600, 0.07, amp=0.75),
"w": P("approx", 320, 800, 2200, 0.06, amp=0.7, to=(500, 1500, 2500)),
"y": P("approx", 300, 2300, 3000, 0.06, amp=0.7, to=(500, 1500, 2500)),
# -- fricatives --------------------------------------------------------
"s": P("fric", 500, 4500, 6500, 0.11, voiced=False, noise=1.0, amp=0.5),
"z": P("fric", 400, 4200, 6000, 0.09, voiced=True, noise=0.7, amp=0.45),
"f": P("fric", 500, 1800, 4500, 0.10, voiced=False, noise=0.7, amp=0.32),
"v": P("fric", 400, 1600, 3000, 0.07, voiced=True, noise=0.4, amp=0.35),
"th": P("fric", 500, 1700, 4000, 0.09, voiced=False, noise=0.6, amp=0.28),
"dh": P("fric", 400, 1500, 2800, 0.06, voiced=True, noise=0.35, amp=0.35),
"sh": P("fric", 500, 2400, 4000, 0.12, voiced=False, noise=1.0, amp=0.5),
"h": P("fric", 500, 1500, 2500, 0.06, voiced=False, noise=0.5, amp=0.25),
# -- stops (closure, then burst, then the transition into the vowel) ---
"p": P("stop", 400, 1100, 2200, 0.085, voiced=False, noise=0.8, amp=0.4),
"t": P("stop", 400, 1800, 2600, 0.085, voiced=False, noise=0.9, amp=0.45),
"k": P("stop", 400, 1900, 2400, 0.085, voiced=False, noise=0.85, amp=0.42),
"b": P("stop", 350, 1100, 2200, 0.07, voiced=True, noise=0.35, amp=0.4),
"d": P("stop", 350, 1800, 2600, 0.07, voiced=True, noise=0.4, amp=0.4),
"g": P("stop", 350, 1900, 2400, 0.07, voiced=True, noise=0.4, amp=0.4),
# -- affricates ---------------------------------------------------------
"ch": P("stop", 500, 2400, 4000, 0.13, voiced=False, noise=1.0, amp=0.45),
"jh": P("stop", 450, 2200, 3200, 0.11, voiced=True, noise=0.6, amp=0.45),
# -- silence -------------------------------------------------------------
"_": P("silence", duration=0.09, voiced=False, amp=0.0),
"__": P("silence", duration=0.22, voiced=False, amp=0.0),
}
# ---------------------------------------------------------------------------
# Vocabulary
# ---------------------------------------------------------------------------
WORDS: dict[str, str] = {
"zero": "z iy r ow", "one": "w ah n", "two": "t uw", "three": "th r iy",
"four": "f ao r", "five": "f ay v", "six": "s ih k s",
"seven": "s eh v ah n", "eight": "ey t", "nine": "n ay n",
"ten": "t eh n", "eleven": "ih l eh v ah n", "twelve": "t w eh l v",
"thirteen": "th er t iy n", "fourteen": "f ao r t iy n",
"fifteen": "f ih f t iy n", "sixteen": "s ih k s t iy n",
"seventeen": "s eh v ah n t iy n", "eighteen": "ey t iy n",
"nineteen": "n ay n t iy n", "twenty": "t w eh n t iy",
"thirty": "th er t iy", "forty": "f ao r t iy", "fifty": "f ih f t iy",
"sixty": "s ih k s t iy", "seventy": "s eh v ah n t iy",
"eighty": "ey t iy", "ninety": "n ay n t iy",
"hundred": "h ah n d r ah d", "thousand": "th aw z ah n d",
"oh": "ow", "point": "p oy n t",
"january": "jh ae n y uw eh r iy", "february": "f eh b y uw eh r iy",
"march": "m aa r ch", "april": "ey p r ih l", "may": "m ey",
"june": "jh uw n", "july": "jh uw l ay", "august": "ao g ah s t",
"september": "s eh p t eh m b er", "october": "aa k t ow b er",
"november": "n ow v eh m b er", "december": "d ih s eh m b er",
"a.m.": "ey eh m", "p.m.": "p iy eh m",
"megahertz": "m eh g ah h er t s", "kilohertz": "k ih l ow h er t s",
"hours": "aw er z", "at": "ae t", "on": "ao n",
}
_ONES = ("zero", "one", "two", "three", "four", "five", "six", "seven",
"eight", "nine")
_TEENS = ("ten", "eleven", "twelve", "thirteen", "fourteen", "fifteen",
"sixteen", "seventeen", "eighteen", "nineteen")
_TENS = ("", "", "twenty", "thirty", "forty", "fifty", "sixty", "seventy",
"eighty", "ninety")
MONTHS = ("january", "february", "march", "april", "may", "june", "july",
"august", "september", "october", "november", "december")
def number_words(n: int) -> list[str]:
"""Spell a whole number from 0 to 9999 the way it is normally said."""
n = int(n)
if n < 0:
return ["minus"] + number_words(-n)
if n < 10:
return [_ONES[n]]
if n < 20:
return [_TEENS[n - 10]]
if n < 100:
tens, ones = divmod(n, 10)
return [_TENS[tens]] + ([_ONES[ones]] if ones else [])
if n < 1000:
hundreds, rest = divmod(n, 100)
out = [_ONES[hundreds], "hundred"]
return out + (number_words(rest) if rest else [])
thousands, rest = divmod(n, 1000)
out = number_words(thousands) + ["thousand"]
return out + (number_words(rest) if rest else [])
def _year_words(year: int) -> list[str]:
"""Years are said in pairs: 2026 is "twenty twenty six", not "two thousand"."""
if 2000 <= year <= 2099:
rest = year - 2000
if rest == 0:
return ["two", "thousand"]
if rest < 10:
return ["two", "thousand", _ONES[rest]]
return ["twenty"] + number_words(rest)
return number_words(year)
def _two_digit_clock(value: int) -> list[str]:
"""Clock fields: 05 is "oh five", 30 is "thirty"."""
if value == 0:
return ["oh", "oh"]
if value < 10:
return ["oh", _ONES[value]]
return number_words(value)
def timestamp_phrase(when: datetime, frequency: float | None = None,
with_date: bool = True, with_seconds: bool = True) -> str:
"""The words to speak for one timestamp."""
words: list[str] = []
if with_date:
words += [MONTHS[when.month - 1]] + number_words(when.day)
words += ["_"] + _year_words(when.year) + ["_"]
words += number_words(when.hour) if when.hour >= 10 else \
_two_digit_clock(when.hour)
words += _two_digit_clock(when.minute)
if with_seconds:
words += ["_"] + _two_digit_clock(when.second)
if frequency:
words += ["__"] + _frequency_words(frequency)
return " ".join(words)
def timestamp_text(when: datetime, frequency: float | None = None,
with_date: bool = True, with_seconds: bool = True) -> str:
"""The same timestamp as ordinary text, for an installed engine.
Written the way those engines read best rather than as the word list the
built-in synthesiser needs. Punctuation is what gives them their phrasing,
and the obvious spellings are traps: espeak reads "14:38:05" as "fourteen
thirty, eight zero five", and an ISO date as "two thousand and twenty six
dash zero eight dash twenty one".
"""
parts: list[str] = []
if with_date:
parts.append(f"{MONTHS[when.month - 1].capitalize()} {when.day}")
parts.append(" ".join(_year_words(when.year)))
clock = f"{when.hour:02d} {when.minute:02d}"
if with_seconds:
clock += f" and {when.second:02d} seconds"
parts.append(f"at {clock}" if with_date else clock)
if frequency:
parts.append(f"{frequency / 1e6:.3f} megahertz")
return ", ".join(parts)
def _frequency_words(hz: float) -> list[str]:
mhz = hz / 1e6
whole = int(mhz)
frac = round((mhz - whole) * 1000) # kHz, three digits
out = number_words(whole) + ["point"]
for digit in f"{frac:03d}":
out.append(_ONES[int(digit)])
return out + ["megahertz"]
# ---------------------------------------------------------------------------
# Synthesis
# ---------------------------------------------------------------------------
def _resonator(freq: np.ndarray, bandwidth: float, fs: float,
x: np.ndarray, block: int = 64) -> np.ndarray:
"""Two-pole resonator whose centre frequency moves as the phrase does.
Processed in short blocks with the coefficients held constant across each
one. Articulation moves far more slowly than 4 ms, so this is
indistinguishable from recomputing every sample, and it runs through a
compiled filter instead of a Python loop.
"""
r = math.exp(-math.pi * bandwidth / fs)
out = np.zeros(x.size)
zi = np.zeros(2)
for start in range(0, x.size, block):
stop = min(start + block, x.size)
centre = float(np.mean(freq[start:stop]))
theta = 2.0 * math.pi * centre / fs
b1 = 2.0 * r * math.cos(theta)
b2 = -(r * r)
gain = 1.0 - b1 - b2
chunk, zi = sps.lfilter([gain], [1.0, -b1, -b2], x[start:stop], zi=zi)
out[start:stop] = chunk
return out
def _phones_of(text: str) -> list[tuple[str, Phone]]:
"""Turn a phrase into phones, looking each word up in the vocabulary."""
out: list[tuple[str, Phone]] = []
for word in text.split():
if word in ("_", "__"):
out.append((word, PHONEMES[word]))
continue
spelling = WORDS.get(word.lower())
if spelling is None:
continue
for name in spelling.split():
phone = PHONEMES.get(name)
if phone is not None:
out.append((name, phone))
out.append(("_", PHONEMES["_"]))
return out
def synthesize(text: str, fs: float = 16000.0, pitch: float = 115.0,
rate: float = 1.0) -> np.ndarray:
"""Render a phrase with the built-in formant synthesiser."""
phones = _phones_of(text)
if not phones:
return np.zeros(0, dtype=np.float32)
# Lay out the formant tracks and the source description sample by sample,
# so that transitions between phones are continuous rather than stepped.
tracks_f, tracks_a, voiced, noise = [], [], [], []
for name, ph in phones:
n = max(1, int(ph.duration / max(rate, 0.1) * fs))
start = np.array([ph.f1, ph.f2, ph.f3])
end = np.array(ph.to) if ph.to else start
ramp = np.linspace(0.0, 1.0, n)[:, None]
tracks_f.append(start[None, :] * (1 - ramp) + end[None, :] * ramp)
env = np.full(n, ph.amp)
if ph.kind == "stop":
# Closure, then a burst: silence for the first half, then a short
# noisy release. Without the silence a stop is just a fricative.
hold = int(n * 0.55)
env[:hold] = 0.0
env[hold:] = np.linspace(ph.amp, ph.amp * 0.35, n - hold)
elif ph.kind == "silence":
env[:] = 0.0
else:
edge = max(1, int(0.012 * fs))
if n > 2 * edge:
env[:edge] *= np.linspace(0, 1, edge)
env[-edge:] *= np.linspace(1, 0, edge)
tracks_a.append(env)
voiced.append(np.full(n, 1.0 if ph.voiced else 0.0))
noise.append(np.full(n, ph.noise))
f_track = np.vstack(tracks_f)
amp = np.concatenate(tracks_a)
voiced = np.concatenate(voiced)
noise = np.concatenate(noise)
total = amp.size
# Smooth the tracks: articulation is continuous, and abrupt formant jumps
# are heard as clicks.
smooth = max(3, int(0.02 * fs))
kernel = np.ones(smooth) / smooth
for i in range(3):
f_track[:, i] = np.convolve(f_track[:, i], kernel, mode="same")
voiced = np.convolve(voiced, kernel, mode="same")
# Source: a glottal pulse train with a falling pitch, plus noise.
t = np.arange(total) / fs
f0 = pitch * (1.0 - 0.18 * t / max(t[-1], 1e-6)) * \
(1.0 + 0.03 * np.sin(2 * np.pi * 3.1 * t))
phase = np.cumsum(2.0 * np.pi * f0 / fs)
pulses = np.zeros(total)
edges = np.flatnonzero(np.diff(np.floor(phase / (2 * np.pi))) > 0)
pulses[edges] = 1.0
pulses -= pulses.mean()
rng = np.random.default_rng(12345)
hiss = rng.standard_normal(total)
# A cascade, not three resonators added together. Summing them in
# parallel with fixed weights loses the first formant: each resonator is
# normalised at DC, which flatters a high formant far more than a low one,
# so F1 ends up buried. In cascade the product of the three responses is
# the vocal-tract envelope, and the relative levels come out on their own.
voiced_src = sps.lfilter([1.0], [1.0, -0.94], voiced * pulses)
voiced_src /= max(float(np.abs(voiced_src).max()), 1e-9)
src = voiced_src + noise * hiss * 0.30
out = _resonator(f_track[:, 0], 80.0, fs, src)
out = _resonator(f_track[:, 1], 110.0, fs, out)
out = _resonator(f_track[:, 2], 160.0, fs, out)
out *= amp
peak = float(np.abs(out).max())
if peak > 0:
out = out / peak * 0.7
return out.astype(np.float32)
# ---------------------------------------------------------------------------
# External engines
# ---------------------------------------------------------------------------
ENGINES = ("espeak-ng", "espeak", "pico2wave", "flite", "say")
def available_engine() -> str | None:
"""The first installed text-to-speech program, if any."""
for name in ENGINES:
if shutil.which(name):
return name
return None
def _external(text: str, fs: float, engine: str) -> np.ndarray | None:
"""Render with an installed engine, returning None if it will not play."""
words = " ".join(w for w in text.split() if w not in ("_", "__"))
with tempfile.TemporaryDirectory() as tmp:
out = Path(tmp) / "say.wav"
if engine in ("espeak-ng", "espeak"):
cmd = [engine, "-w", str(out), "-s", "150", words]
elif engine == "pico2wave":
cmd = [engine, "-w", str(out), words]
elif engine == "flite":
cmd = [engine, "-t", words, "-o", str(out)]
elif engine == "say":
cmd = [engine, "-o", str(out), "--data-format=LEI16@16000", words]
else:
return None
try:
subprocess.run(cmd, check=True, capture_output=True, timeout=20)
with wave.open(str(out)) as w:
raw = w.readframes(w.getnframes())
rate = w.getframerate()
channels = w.getnchannels()
except (OSError, subprocess.SubprocessError, wave.Error):
return None
audio = np.frombuffer(raw, dtype="<i2").astype(np.float32) / 32768.0
if channels > 1:
audio = audio.reshape(-1, channels).mean(axis=1)
if rate != fs and audio.size:
from scipy.signal import resample_poly
g = math.gcd(int(rate), int(fs))
audio = resample_poly(audio, int(fs) // g, int(rate) // g)
# Match the built-in synthesiser's level, so switching engines does not
# change how loud the announcements are against the recordings.
peak = float(np.abs(audio).max()) if audio.size else 0.0
if peak > 0:
audio = audio / peak * 0.7
return audio.astype(np.float32)
def say(text: str, fs: float = 16000.0, engine: str = "auto") -> np.ndarray:
"""Speak a phrase, preferring an installed engine over the built-in one."""
if engine != "builtin":
chosen = available_engine() if engine == "auto" else engine
if chosen:
audio = _external(text, fs, chosen)
if audio is not None and audio.size:
return audio
return synthesize(text, fs)
def speak_timestamp(when: datetime, fs: float = 16000.0,
frequency: float | None = None, engine: str = "auto",
with_date: bool = True) -> np.ndarray:
"""Audio saying the date and time, for splicing before a recording.
The wording differs by engine: an installed one is given ordinary text
with punctuation, while the built-in synthesiser is given the word list it
has pronunciations for.
"""
chosen = None if engine == "builtin" else (
available_engine() if engine == "auto" else engine)
if chosen:
audio = _external(timestamp_text(when, frequency, with_date=with_date),
fs, chosen)
if audio is not None and audio.size:
return audio
return synthesize(timestamp_phrase(when, frequency, with_date=with_date), fs)
Reference in a new issue View git blame Copy permalink