What We're Building
A realtime EQ receives a short block of audio, filters it, and sends it to an output before the next block needs processing. This guide builds a one-band parametric EQ that runs between an audio input and your headphones. You'll need basic Python and an input device; no DSP background is assumed.
The signal path is input → headroom gain → peaking filter → output. We'll use NumPy for arrays, SciPy for filtering, and sounddevice for audio I/O. The example processes live audio with fixed settings; the final section explains how to extend it to multiple bands and live controls.
1. Set Up the Audio Path
In a Python virtual environment, install the packages below. Run the device-list command to find an input and output. PortAudio comes with sounddevice's pip installation on macOS and Windows; Linux may need a system package such as libportaudio2.
Use headphones and start at a low listening level so the output doesn't feed back into the microphone. This script processes the selected input, so playing a song in another app won't automatically route it through the EQ. Use a line input or a configured loopback device for that.
python -m pip install numpy scipy sounddevice
python -m sounddevice2. Design One Peaking Band
A parametric band has three controls: center frequency selects where the bell sits, gain sets the boost or cut in decibels, and Q controls its width. Higher Q makes a narrower bell. At 0 dB, the band should pass the signal unchanged.
Use a peaking biquad for this. Adding independently designed low-pass, band-pass, and high-pass outputs does not guarantee a flat response at 0 dB: their phases and magnitudes interact when summed.
Save the following as eq.py. These are Robert Bristow-Johnson's peaking-EQ equations. A is 10 raised to gain_db / 40 inside this design; a standalone amplitude multiplier uses gain_db / 20. Dividing all coefficients by a0 gives SciPy the normalized filter it expects. The Q and gain limits here are practical choices for this example.
import numpy as np
from scipy.signal import sosfilt
def peaking_sos(f0, q, gain_db, sample_rate):
if not np.all(np.isfinite([f0, q, gain_db, sample_rate])):
raise ValueError("Parameters must be finite")
if sample_rate <= 0 or not 0 < f0 < sample_rate / 2:
raise ValueError("Frequency must be between 0 and Nyquist")
if not 0.1 <= q <= 20 or not -24 <= gain_db <= 24:
raise ValueError("Use Q from 0.1 to 20 and gain from -24 to 24 dB")
a = 10.0 ** (gain_db / 40.0)
w0 = 2.0 * np.pi * f0 / sample_rate
alpha = np.sin(w0) / (2.0 * q)
c = np.cos(w0)
a0 = 1.0 + alpha / a
return np.array([[
(1.0 + alpha * a) / a0,
-2.0 * c / a0,
(1.0 - alpha * a) / a0,
1.0,
-2.0 * c / a0,
(1.0 - alpha / a) / a0,
]], dtype=np.float64)3. Keep the Filter's Memory Between Blocks
A biquad is a second-order filter: its current output depends on stored history. SciPy represents each second-order section (SOS) as six coefficients: b0, b1, b2, a0, a1, a2. Stacking rows creates a series of bands without combining them into one high-order polynomial.
Append this class to eq.py. Audio has shape (frames, channels), so axis=0 filters along time. The state has shape (sections, 2, channels), keeping each channel's history separate. Initialize it once, then pass the returned state into the next call. Resetting it on every block creates discontinuities.
Zero state assumes silence before the stream starts. The process method also accepts different block lengths without changing the state shape.
class StreamingEQ:
def __init__(self, sos, channels, preamp_db=-12.0):
self.sos = np.asarray(sos, dtype=np.float64)
self.zi = np.zeros((len(self.sos), 2, channels))
self.preamp = 10.0 ** (preamp_db / 20.0)
def process(self, block):
output, self.zi = sosfilt(
self.sos, block * self.preamp, axis=0, zi=self.zi
)
return output4. Connect It to a Live Audio Stream
Save this second file as live_eq.py beside eq.py, then run python live_eq.py. It starts with a broad +6 dB boost at 1 kHz and a -12 dB preamp. Edit those values and restart to hear a different curve.
DEVICE=None uses the system defaults. To select devices explicitly, replace it with a pair of input and output IDs from the device list. Start with CHANNELS=1 for mono; use 2 when both devices support stereo. Both devices must support SAMPLE_RATE. If opening the stream fails, check these settings and microphone permission first.
The callback fills the output buffer in place. Status is recorded for reporting after shutdown; printing, file access, filter design, and UI work stay outside the callback. Press Ctrl+C to stop.
import numpy as np
import sounddevice as sd
from eq import StreamingEQ, peaking_sos
SAMPLE_RATE = 48_000
CHANNELS = 1
DEVICE = None # Or (input_device_id, output_device_id).
def main():
eq = StreamingEQ(
peaking_sos(1000, 0.707, 6.0, SAMPLE_RATE),
channels=CHANNELS,
preamp_db=-12.0,
)
last_status = None
callback_failed = False
def callback(indata, outdata, frames, time, status):
nonlocal last_status, callback_failed
if status:
last_status = status
try:
output = eq.process(indata)
# Emergency hard clipping; headroom should keep this inactive.
np.clip(output, -1.0, 1.0, out=output)
outdata[:] = output
except Exception:
outdata.fill(0)
callback_failed = True
raise sd.CallbackAbort
with sd.Stream(
samplerate=SAMPLE_RATE,
channels=CHANNELS,
device=DEVICE,
dtype="float32",
blocksize=0,
latency="low",
callback=callback,
) as stream:
print("EQ running. Press Ctrl+C to stop.")
print("Input/output latency (seconds):", stream.latency)
try:
while stream.active:
sd.sleep(100)
except KeyboardInterrupt:
pass
if last_status is not None:
print("Last audio status:", last_status)
if callback_failed:
raise RuntimeError("EQ callback failed; output was silenced")
if __name__ == "__main__":
main()5. Leave Room for Boosts and Processing Time
A +6 dB boost roughly doubles a sine wave's amplitude at the center frequency. The preamp reduces the signal before filtering to leave headroom. With several bands, boosts can overlap; -12 dB is a starting point, not a guarantee against clipping. Lower the input or preamp if the output distorts. np.clip is a last-resort hard clipper, not a transparent limiter.
Keep the stream in floating point. There's no need to convert to int16 for playback here. If you later export integer PCM, scale normalized samples to the integer range and clamp before casting; a dtype cast alone does not protect against overflow.
At 48 kHz, 256 frames span about 5.3 ms. Processing has to fit comfortably inside the available block time. That duration is not total input-to-output latency: drivers, converters, and buffering also contribute. blocksize=0 lets the host choose a suitable block size. If you hear dropouts or see underflow/overflow status, try latency='high' and reduce other CPU work.
This is a Python prototype, not a hard realtime guarantee. SciPy creates output arrays and Python execution can be delayed. For demanding low-latency use, move the audio processing into compiled code with preallocated buffers, and keep Python for controls. Measure under load on the hardware you intend to use.
6. Check the DSP Before Trusting Your Ears
Save this as check_eq.py beside eq.py and run python check_eq.py. It checks that a 0 dB band is transparent, that processing consecutive blocks matches processing the same signal in one call, and that an impulse in the left channel never leaks into the right. The preamp is set to 0 dB so it doesn't affect these comparisons.
These checks catch state and channel mistakes without opening an audio device. Then listen at low volume, check the stream status, and test for dropouts during normal computer use. Passing the math checks doesn't establish reliable hardware timing.
import numpy as np
from numpy.testing import assert_allclose
from eq import StreamingEQ, peaking_sos
rng = np.random.default_rng(0)
x = rng.normal(0, 0.05, (4096, 2))
flat = StreamingEQ(peaking_sos(1000, 1, 0, 48000), 2, 0)
assert_allclose(flat.process(x), x, atol=1e-12)
sos = peaking_sos(1000, 1, 6, 48000)
whole = StreamingEQ(sos, 2, 0).process(x)
eq = StreamingEQ(sos, 2, 0)
blocks = np.concatenate([
eq.process(x[:127]),
eq.process(x[127:1024]),
eq.process(x[1024:]),
])
assert_allclose(blocks, whole, atol=1e-12)
impulse = np.zeros((1024, 2))
impulse[0, 0] = 1.0
output = StreamingEQ(sos, 2, 0).process(impulse)
assert_allclose(output[:, 1], 0, atol=1e-12)
print("EQ checks passed")7. Add More Bands, Then Add Controls
For multiple bands, stack peaking_sos results with np.vstack and pass that array to StreamingEQ. SciPy applies the rows in series; each band gets its own state. Revisit the preamp as you add boosts.
Live sliders need a transition strategy. Replacing coefficients abruptly can click because the stored state belongs to the previous filter. Design and validate new settings outside the audio callback, then hand off a complete update at a block boundary. Never mutate an active coefficient array from the UI thread.
One approach is to run the old and new filter chains on the same input, each with its own state, and crossfade their outputs over a short ramp, such as 20–50 ms. Warm the new chain with recent input before fading it in; narrow bands may need longer to settle. Carry the fade position across blocks, budget for processing both chains, and coalesce slider events instead of creating a new chain for every movement.
Get the fixed-setting stream working first. Once its response, state handling, and timing are correct, presets and controls have a solid foundation.