#!/usr/bin/env python3 """CPU benchmark for each FX block — measures per-block processing time. Usage: python scripts/benchmark_fx.py # All effects, 100 iterations python scripts/benchmark_fx.py --effect delay # Single effect python scripts/benchmark_fx.py --iters 500 # More iterations python scripts/benchmark_fx.py --csv # CSV output Results report: - Mean, min, max time (microseconds) per 256-sample block - Whether the effect meets the < 500 us (0.5 ms) target """ from __future__ import annotations import argparse import sys import time import numpy as np from src.dsp.pipeline import AudioPipeline from src.presets.types import FXBlock, FXType, Preset # ── Test tone parameters ─────────────────────────────────────────── BLOCK = ( np.sin(2 * np.pi * 440.0 * np.arange(BLOCK_SIZE) / SAMPLE_RATE) .astype(np.float32) * 0.5 ) SILENCE = np.zeros(BLOCK_SIZE, dtype=np.float32) # ── Effect parameter profiles ────────────────────────────────────── FX_PROFILES: list[tuple[str, FXType, dict]] = [ ("noise_gate", FXType.NOISE_GATE, {"threshold": 0.01, "release": 100.0}), ("compressor", FXType.COMPRESSOR, {"threshold": -20.0, "ratio": 4.0, "attack": 5.0, "release": 100.0, "gain": 1.0}), ("boost", FXType.BOOST, {"gain_db": 6.0}), ("overdrive", FXType.OVERDRIVE, {"drive": 0.5, "tone": 0.5, "gain": 1.0}), ("distortion", FXType.DISTORTION, {"drive": 0.7, "tone": 0.5, "gain": 1.0}), ("fuzz", FXType.FUZZ, {"drive": 0.8, "tone": 0.5, "gain": 1.0}), ("eq", FXType.EQ, {"bass": 6.0, "mid": 3.0, "treble": -3.0, "bass_freq": 200.0, "mid_freq": 1000.0, "treble_freq": 3500.0, "q": 0.707}), ("chorus", FXType.CHORUS, {"rate": 0.5, "depth": 0.5, "mix": 0.5, "delay": 20.0}), ("flanger", FXType.FLANGER, {"rate": 0.25, "depth": 0.7, "feedback": 0.3, "mix": 0.5, "delay": 5.0}), ("phaser", FXType.PHASER, {"rate": 0.4, "depth": 0.5, "feedback": 0.3, "mix": 0.5, "stages": 4}), ("tremolo", FXType.TREMOLO, {"rate": 4.0, "depth": 0.7, "shape": "sine"}), ("vibrato", FXType.VIBRATO, {"rate": 3.0, "depth": 0.5}), ("delay", FXType.DELAY, {"time": 400.0, "feedback": 0.3, "mix": 0.4}), ("reverb", FXType.REVERB, {"decay": 0.5, "damping": 0.4, "mix": 0.3, "predelay": 30.0}), ("volume", FXType.VOLUME, {"level": 0.8}), ] def benchmark_effect( fx_type: FXType, params: dict, iterations: int = 100, ) -> dict: """Time one effect over N iterations. Returns timing stats.""" pipeline = AudioPipeline() block = FXBlock(fx_type=fx_type, enabled=True, bypass=False, params=params) preset = Preset(name="bench", chain=[block], master_volume=1.0) pipeline.load_preset(preset) # Warm-up: process a few blocks to initialise state (delay buffers, etc.) for _ in range(5): pipeline.process(BLOCK) # Timing loop times = np.zeros(iterations, dtype=np.float64) # Alternate between tone and silence to exercise stateful effects for i in range(iterations): inp = BLOCK if i % 2 == 0 else SILENCE t0 = time.perf_counter() pipeline.process(inp) t1 = time.perf_counter() times[i] = (t1 - t0) * 1e6 # microseconds return { "mean_us": float(np.mean(times)), "min_us": float(np.min(times)), "max_us": float(np.max(times)), "std_us": float(np.std(times)), "passes": float(np.mean(times)) < 500.0, } def run_all(iterations: int, csv_mode: bool) -> None: results = [] print(f"FX Block Benchmark — {iterations} iterations per effect") print(f"Block size: {BLOCK_SIZE} samples @ {SAMPLE_RATE} Hz") print(f"Target: < 500 µs per block (< 0.5 ms)") print() print(f"{'Effect':<16} {'Mean (µs)':>10} {'Min (µs)':>10} {'Max (µs)':>10} " f"{'Std (µs)':>10} {'Pass':>6}") print("-" * 66) for name, fx_type, params in FX_PROFILES: stats = benchmark_effect(fx_type, params, iterations) results.append((name, stats)) if csv_mode: continue pass_mark = "PASS" if stats["passes"] else "FAIL" print(f"{name:<16} {stats['mean_us']:>10.1f} {stats['min_us']:>10.1f} " f"{stats['max_us']:>10.1f} {stats['std_us']:>10.1f} " f"{pass_mark:>6}") if csv_mode: print("name,mean_us,min_us,max_us,std_us,passes") for name, stats in results: print(f"{name},{stats['mean_us']:.1f},{stats['min_us']:.1f}," f"{stats['max_us']:.1f},{stats['std_us']:.1f},{stats['passes']}") # Summary passed = sum(1 for _, s in results if s["passes"]) total = len(results) print() print(f"Results: {passed}/{total} effects pass the < 500 µs target") if passed < total: failing = [n for n, s in results if not s["passes"]] print(f"Failing: {', '.join(failing)}") sys.exit(1) def main(): parser = argparse.ArgumentParser( description="Benchmark per-FX-block CPU time", ) parser.add_argument( "--effect", "-e", type=str, default=None, help="Benchmark a single effect by name (e.g. 'delay', 'reverb')", ) parser.add_argument( "--iters", "-i", type=int, default=100, help="Number of iterations per effect (default: 100)", ) parser.add_argument( "--csv", action="store_true", help="Output CSV format", ) args = parser.parse_args() if args.effect: matches = [(n, t, p) for n, t, p in FX_PROFILES if n == args.effect] if not matches: available = ", ".join(n for n, _, _ in FX_PROFILES) print(f"Unknown effect '{args.effect}'. Choose from: {available}") sys.exit(1) name, fx_type, params = matches[0] stats = benchmark_effect(fx_type, params, args.iters) print(f"{name}: mean={stats['mean_us']:.1f}µs, min={stats['min_us']:.1f}µs, " f"max={stats['max_us']:.1f}µs, std={stats['std_us']:.1f}µs, " f"{'PASS' if stats['passes'] else 'FAIL'} < 500µs target") sys.exit(0) run_all(args.iters, args.csv) if __name__ == "__main__": main()