bandsaunter/tests/signals.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

76 lines
2.7 KiB
Python

"""Synthetic test signals shared by the test modules."""
import numpy as np
from scipy.signal import butter, hilbert, lfilter
FS = 32000.0
def _noise(x, snr_db, rng):
p = float(np.mean(np.abs(x) ** 2))
n = np.sqrt(p / (2 * 10 ** (snr_db / 10.0)))
return (x + n * (rng.standard_normal(x.size)
+ 1j * rng.standard_normal(x.size))).astype(np.complex64)
def voice(n, fs=FS, seed=0):
"""Band-limited noise with a syllabic envelope -- a good speech stand-in."""
rng = np.random.default_rng(seed)
b, a = butter(4, [300 / (fs / 2), 2700 / (fs / 2)], btype="band")
v = lfilter(b, a, rng.standard_normal(n))
v /= max(np.abs(v).max(), 1e-9)
t = np.arange(n) / fs
return v * (0.4 + 0.6 * np.abs(np.sin(2 * np.pi * 1.7 * t)))
def make(kind, n=64000, fs=FS, snr_db=30.0, seed=3):
rng = np.random.default_rng(seed)
t = np.arange(n) / fs
v = voice(n, fs, seed)
if kind == "nfm":
msg = v + 0.15 * np.sin(2 * np.pi * 100.0 * t)
x = np.exp(1j * np.cumsum(2 * np.pi * 2500 * msg / fs))
elif kind == "wfm":
x = np.exp(1j * np.cumsum(2 * np.pi * 3000 * v / fs))
elif kind == "am":
x = ((1 + 0.6 * v) * np.exp(2j * np.pi * 30 * t))
elif kind == "usb":
x = 0.5 * hilbert(v)
elif kind == "lsb":
x = 0.5 * np.conj(hilbert(v))
elif kind == "carrier":
x = np.exp(2j * np.pi * 137 * t)
elif kind == "cw":
dot = 0.08
pat = [1, 0, 1, 1, 1, 0, 0, 0, 1, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0]
key = np.zeros(n)
i = 0
while i < n:
for b in pat:
m = int(dot * fs)
if i + m > n:
break
key[i:i + m] = b
i += m
env = np.convolve(key, np.hanning(int(0.005 * fs)), "same")
env /= max(env.max(), 1e-9)
x = env * np.exp(2j * np.pi * 300 * t)
elif kind.startswith("fsk"):
levels = int(kind[3])
baud, dev = (1200.0, 2400.0) if levels == 2 else (4800.0, 1800.0)
sp = int(fs / baud)
sym = rng.integers(0, levels, n // sp + 1)
lv = (sym - (levels - 1) / 2) / max(1, (levels - 1) / 2)
f = np.resize(np.repeat(lv, sp) * dev, n)
x = np.exp(1j * np.cumsum(2 * np.pi * f / fs))
elif kind.startswith("psk"):
m = int(kind[3])
sp = int(fs / 4800.0)
sym = rng.integers(0, m, n // sp + 1)
x = np.exp(1j * np.resize(np.repeat(2 * np.pi * sym / m, sp), n))
elif kind == "noise":
return (0.01 * (rng.standard_normal(n)
+ 1j * rng.standard_normal(n))).astype(np.complex64)
else:
raise ValueError(kind)
return _noise(np.asarray(x, dtype=np.complex128), snr_db, rng)