Writing a Python SDR from Scratch: IQ Pipeline, VAD Design, and the ADC Saturation Problem
ctypes bindings to libhackrf, a hand-rolled DSP pipeline (WFM/NFM/AM/SSB), PyQt5 waterfall display, dual-condition VAD, and Whisper STT β all without GNU Radio.
β οΈ Pending Before Publish
- Confirmed working on actual FM/CB traffic β specific examples?
- VFO offset bug fix confirmed in v8?
- Screenshot of waterfall + spectrum UI available?
Architecture Overview
libhackrf (C)
β ctypes bindings
HackRF stream callback (float32 IQ samples)
β
Ring buffer (NumPy circular, O(1) write via np.copyto wraparound)
β
DSP thread
ββ FFT β PSD (dBFS) β spectrum display
ββ Demodulation (WFM / NFM / AM / SSB)
β ββ Audio ring buffer β sounddevice output
ββ VAD β recording trigger β Whisper STT thread
β
PyQt5 UI (custom-painted QWidget spectrum + waterfall)
Never GNU Radio. Everything is ctypes β NumPy β PyQt5.
IQ Samples and the Complex Baseband
HackRF delivers IQ (in-phase/quadrature) samples: pairs of float32 values representing a complex number I + jQ. The signal at baseband is:
s(t) = A(t) Β· exp(jΒ·2ΟΒ·f_offsetΒ·t + Ο(t))
Where A(t) is amplitude, f_offset is frequency offset from centre, and Ο(t) is phase. The complex representation lets us manipulate frequency, phase, and amplitude independently in software.
Sample rate: 1.024 MSPS. Each FFT window is 1024 samples β frequency resolution = 1.024MHz / 1024 = 1 kHz per bin.
ctypes Bindings and the hackrf_exit() Bug
The Python SDK wraps libhackrf via ctypes. One critical quirk: never call hackrf_exit() after hackrf_open(). If the device is re-opened in the same process after calling hackrf_exit(), the next hackrf_open() call segfaults in libhackrf's internal libusb context cleanup. The workaround: open once per process, never call exit, keep the device handle alive for the process lifetime.
import ctypes
lib = ctypes.CDLL('libhackrf.so.0')
# Callback must be a C function pointer β use ctypes.CFUNCTYPE
TRANSFER_CB = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.POINTER(HackRFTransfer))
@TRANSFER_CB
def rx_callback(transfer):
samples = np.frombuffer(transfer.contents.buffer, dtype=np.float32)
# samples is interleaved IQ: [I0, Q0, I1, Q1, ...]
iq = samples[0::2] + 1j * samples[1::2]
ring_buffer.write(iq)
return 0
lib.hackrf_start_rx(device, rx_callback, None)
# NEVER call lib.hackrf_exit(device) β segfaults on reopen
FFT and Spectrum Display
window = np.hanning(FFT_SIZE) # Hann window reduces spectral leakage
fft_out = np.fft.fftshift( # reorder so DC is in centre
np.fft.fft(iq_chunk * window)
)
psd_db = 20 * np.log10(np.abs(fft_out) / FFT_SIZE + 1e-10) # dBFS
The PyQt5 spectrum widget is a custom QWidget with paintEvent overridden. Each frame redraws the dBFS line plot. The waterfall is a 2D array (time Γ frequency) maintained as a NumPy rolling buffer β new rows inserted at bottom, display scrolls upward. Colour mapping: viridis-style linear interpolation from dBFS range.
Time compression: holding Shift while scrolling allows viewing longer time windows by averaging multiple rows.
WFM Demodulation Pipeline
Wide FM broadcast (200 kHz channel width):
# 1. Frequency discriminator β FM demodulation
instantaneous_freq = np.angle(iq[1:] * np.conj(iq[:-1]))
# 2. First decimation: 1.024 MHz β 256 kHz (factor 4)
audio_decimated_1 = scipy.signal.decimate(instantaneous_freq, 4, ftype='fir')
# 3. Second decimation: 256 kHz β 48 kHz (factor ~5.3, rounded)
audio_decimated_2 = scipy.signal.decimate(audio_decimated_1, 5, ftype='fir')
# After: 48 kHz mono audio
# 4. De-emphasis: 75ΞΌs time constant (standard for FM broadcast)
# H(s) = 1/(1 + Οs) where Ο = 75ΞΌs
# Discrete: y[n] = (1-Ξ±)Β·x[n] + Ξ±Β·y[n-1], Ξ± = exp(-1/(ΟΒ·fs))
tau = 75e-6
alpha = np.exp(-1 / (tau * 48000))
# Apply as first-order IIR
# 5. Audio ring buffer β sounddevice callback
Two-stage decimation (rather than one stage at Γ21) reduces filter order requirements significantly β each stage needs only to reject above its own Nyquist.
The VFO Offset Bug and the Fix
Early versions tuned the HackRF hardware to the exact target frequency and assumed DC was at the VFO. Two problems:
- HackRF's LO (local oscillator) leaks a DC spike at the exact centre frequency β the desired signal gets masked
- Moving the hardware VFO causes retuning latency (~50ms) which makes frequency scanning sluggish
The fix: tune the hardware to centre_freq = target + VFO_OFFSET (typically +100 kHz), then mix the VFO offset back out in software:
# Digital frequency shift β move signal from +VFO_OFFSET to DC
t = np.arange(len(iq)) / SAMPLE_RATE
mixer = np.exp(-2j * np.pi * VFO_OFFSET * t)
iq_shifted = iq * mixer
This multiplication by a complex exponential shifts the entire spectrum by -VFO_OFFSET, moving the target signal to DC while the LO leak appears at +VFO_OFFSET (away from the signal of interest).
Dual-Condition VAD
The VAD must distinguish real transmissions from band noise. Two conditions must be simultaneously true:
# Condition 1: SNR ratio
noise_floor = np.percentile(power_history, 25) # rolling 25th percentile
snr = signal_power / (noise_floor + 1e-10)
condition_snr = snr > SNR_THRESHOLD # typically 3.0β4.0
# Condition 2: Absolute level gate
condition_gate = peak_dbfs > GATE_THRESHOLD # typically -45 dBFS
if condition_snr and condition_gate:
if not recording:
start_recording()
hold_frames = HOLD_FRAMES # reset hold timer on each active frame
elif hold_frames > 0:
hold_frames -= 1 # continue recording briefly after signal drops
else:
if recording:
stop_recording_and_queue_transcription()
The 25th-percentile noise floor baseline adapts to changing band conditions (time of day, local interference). A static threshold would miss weak transmissions on quiet bands or trigger continuously on noisy urban bands.
O(1) Audio Ring Buffer
Early version used collections.deque for the audio buffer. Profiling showed O(n) copy overhead on each audio callback. Replacement:
class RingBuffer:
def __init__(self, size):
self.buf = np.zeros(size, dtype=np.float32)
self.head = 0
self.size = size
def write(self, data):
n = len(data)
space = self.size - self.head
if n <= space:
np.copyto(self.buf[self.head:self.head+n], data)
else:
np.copyto(self.buf[self.head:], data[:space])
np.copyto(self.buf[:n-space], data[space:])
self.head = (self.head + n) % self.size
np.copyto with pre-allocated slices avoids any Python-level allocation. The sounddevice callback reads from the same buffer via a read pointer, achieving true circular streaming with O(1) operations on both sides.
Whisper STT Thread
import threading
from faster_whisper import WhisperModel
model = WhisperModel('tiny.en', device='cpu', compute_type='int8')
transcription_queue = queue.Queue()
def transcription_worker():
while True:
audio_clip = transcription_queue.get() # blocking
segments, _ = model.transcribe(audio_clip, language='en', vad_filter=True)
text = ' '.join(s.text for s in segments).strip()
if text:
recordings_panel.add_entry(text, timestamp=datetime.now())
threading.Thread(target=transcription_worker, daemon=True).start()
tiny.en model: ~39MB, ~0.3s transcription for a 10s clip on CPU. Whisper's built-in VAD filter (vad_filter=True) removes silence padding before transcription.
References
- HackRF One: Great Scott Gadgets. greatscottgadgets.com/hackrf
- libhackrf C library. github.com/greatscottgadgets/hackrf
- GNU Radio project. gnuradio.org
- faster-whisper: SYSTRAN. github.com/SYSTRAN/faster-whisper
- sounddevice Python library. python-sounddevice.readthedocs.io
- NumPy FFT documentation. numpy.org/doc/stable/reference/routines.fft