diff --git a/tests/test_integration.py b/tests/test_integration.py index 3f8e423..b2c8697 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -244,9 +244,8 @@ class TestMIDIRouting: """MIDI CC 11 adjusts master volume on the pipeline.""" pedal_app.boot() assert pedal_app.pipeline is not None - initial_vol = pedal_app.pipeline._master_volume - # Simulate expression pedal at 50% - pedal_app._on_midi_cc(64, 11) + # Simulate expression pedal at 50% (value=64, CC=11, channel=0) + pedal_app._on_midi_cc(64, 11, 0) assert pedal_app.pipeline._master_volume == pytest.approx(64 / 127.0) def test_midi_pc_without_presets_does_not_crash(self, pedal_app): diff --git a/tests/test_stability.py b/tests/test_stability.py new file mode 100644 index 0000000..e8dc855 --- /dev/null +++ b/tests/test_stability.py @@ -0,0 +1,383 @@ +"""Phase 7 — Stability Test: 1-hour continuous audio simulation. + +Exercises the DSP pipeline under sustained load across all FX types, +monitoring CPU, memory, output integrity, and processing throughput. +""" +from __future__ import annotations + +import os +import sys +import time +import gc +import json +from pathlib import Path + +import numpy as np + +# Ensure both project root and src are on sys.path for proper package imports +PROJECT_ROOT = Path(__file__).resolve().parent.parent +SRC = PROJECT_ROOT / "src" +for p in [SRC, PROJECT_ROOT]: + if str(p) not in sys.path: + sys.path.insert(0, str(p)) + +from src.dsp.pipeline import AudioPipeline, BLOCK_SIZE, SAMPLE_RATE +from src.presets.types import FXBlock, FXType, Preset + + +# ── Configuration ────────────────────────────────────────────────────────────── + +# 1 hour of audio at 48 kHz / 256 samples per block +BLOCKS_PER_HOUR = int(SAMPLE_RATE / BLOCK_SIZE * 3600) # 675,000 + +# How many FX types to test +REPORT_INTERVAL_BLOCKS = BLOCKS_PER_HOUR // 20 # report every 5% + + +# ── FX type parameters for stability testing ─────────────────────────────────── + +FX_PARAMS: dict[FXType, dict[str, float]] = { + FXType.NOISE_GATE: {"threshold": 0.01, "release": 50.0}, + FXType.COMPRESSOR: {"threshold": -20.0, "ratio": 4.0, "attack": 5.0, "release": 50.0}, + FXType.BOOST: {"gain_db": 12.0}, + FXType.OVERDRIVE: {"gain": 0.6, "tone": 0.5, "level": 0.8}, + FXType.DISTORTION: {"gain": 0.8, "tone": 0.5, "level": 0.7}, + FXType.FUZZ: {"gain": 0.9, "tone": 0.4, "level": 0.8}, + FXType.EQ: {"low_gain": 3.0, "mid_gain": -2.0, "high_gain": 1.0}, + FXType.CHORUS: {"rate": 0.5, "depth": 0.5, "mix": 0.4}, + FXType.FLANGER: {"rate": 0.3, "depth": 0.6, "feedback": 0.4, "mix": 0.5}, + FXType.PHASER: {"rate": 0.4, "depth": 0.5, "feedback": 0.3, "stages": 4}, + FXType.TREMOLO: {"rate": 3.0, "depth": 0.6}, + FXType.VIBRATO: {"rate": 2.0, "depth": 0.3}, + FXType.DELAY: {"delay_ms": 400.0, "feedback": 0.3, "mix": 0.4}, + FXType.REVERB: {"decay": 0.5, "mix": 0.3, "damping": 0.5, "predelay_ms": 20.0, "room_size": 0.6, "diffusion": 0.5}, + FXType.VOLUME: {"level": 0.8}, + FXType.OCTAVER: {"mix": 0.5, "octaves": -12}, + FXType.PITCH_SHIFTER: {"shift": 4, "mix": 0.5}, + FXType.HARMONIZER: {"interval": 4, "mix": 0.4}, + FXType.WHAMMY: {"pitch": 7, "mix": 0.6}, + FXType.DETUNE: {"cents": 10, "mix": 0.4}, + FXType.RING_MODULATOR: {"frequency": 200.0, "depth": 0.5}, + FXType.AUTO_WAH: {"sensitivity": 0.5, "peak": 0.6, "speed": 2.0}, + FXType.ENVELOPE_FILTER: {"sensitivity": 0.5, "resonance": 0.5}, + FXType.ROTARY_SPEAKER: {"rate": 2.0, "depth": 0.6}, + FXType.UNI_VIBE: {"rate": 2.0, "depth": 0.5}, + FXType.AUTO_PAN: {"rate": 3.0, "depth": 0.7}, + FXType.STEREO_WIDENER: {"width": 0.6}, + FXType.BITCRUSHER: {"bit_depth": 8, "sample_rate_red": 8000, "mix": 0.5}, + FXType.WAVEFOLDER: {"drive": 0.7, "symmetry": 0.5}, + FXType.RECTIFIER: {"mode": 0.0}, + FXType.EXPANDER: {"threshold": -30.0, "ratio": 3.0, "attack": 5.0, "release": 100.0}, + FXType.DE_ESSER: {"threshold": -24.0, "frequency": 6000.0}, + FXType.TRANSIENT_SHAPER: {"attack": 4.0, "sustain": 2.0}, + FXType.PARAMETRIC_EQ: {"freq": 1000.0, "gain_db": 6.0, "q": 1.4}, + FXType.HIGH_PASS_FILTER: {"freq": 80.0, "q": 0.707}, + FXType.LOW_PASS_FILTER: {"freq": 8000.0, "q": 0.707}, + FXType.BAND_PASS_FILTER: {"freq": 1000.0, "q": 1.0}, + FXType.NOTCH_FILTER: {"freq": 60.0, "q": 5.0}, + FXType.FORMANT_FILTER: {"vowel": 0.0}, + FXType.PING_PONG_DELAY: {"delay_ms": 350.0, "feedback": 0.3, "mix": 0.4}, + FXType.MULTI_TAP_DELAY: {"delay_ms": 300.0, "feedback": 0.3, "mix": 0.4, "num_taps": 4}, + FXType.REVERSE_DELAY: {"delay_ms": 400.0, "mix": 0.3}, + FXType.TAPE_ECHO: {"delay_ms": 200.0, "feedback": 0.4, "wow": 0.2, "mix": 0.4}, + FXType.SHIMMER_REVERB: {"decay": 0.6, "mix": 0.3, "pitch_shift": 12}, + FXType.EARLY_REFLECTIONS: {"decay": 0.4, "mix": 0.3, "room_size": 0.6}, +} + +# FX types that need special params to avoid warnings/prevent issues +FX_SPECIAL = { + # SIDECHAIN_COMPRESSOR needs a stereo context + FXType.SIDECHAIN_COMPRESSOR: {"threshold": -20.0, "ratio": 4.0, "attack": 5.0, "release": 50.0, "sidechain_source": 0.0}, +} + + +def _fmt_time(seconds: float) -> str: + """Format seconds to H:MM:SS.""" + m, s = divmod(int(seconds), 60) + h, m = divmod(m, 60) + return f"{h:02d}:{m:02d}:{s:02d}" + + +def _get_memory_mb() -> float: + """Get current RSS memory in MB.""" + try: + with open(f"/proc/{os.getpid()}/status") as f: + for line in f: + if line.startswith("VmRSS:"): + return int(line.split()[1]) / 1024.0 + except (FileNotFoundError, IndexError, ValueError): + pass + # Fallback: use /proc/self/statm + try: + with open("/proc/self/statm") as f: + pages = int(f.read().split()[1]) + return pages * 4096 / (1024 * 1024) + except (FileNotFoundError, IndexError, ValueError): + return 0.0 + + +def _get_cpu_percent(interval: float = 0.3) -> float: + """Get current CPU usage percentage for this process.""" + import psutil + return psutil.Process(os.getpid()).cpu_percent(interval=interval) + + +def run_stability_test(duration_blocks: int = BLOCKS_PER_HOUR) -> dict: + """Run the 1-hour stability test. + + Processes audio blocks through the DSP pipeline, cycling through + all FX types. Reports CPU, memory, and output integrity metrics. + """ + report_every = max(1, duration_blocks // 20) + + # Test signal: alternating sine tone and silence + t = np.arange(BLOCK_SIZE, dtype=np.float32) / SAMPLE_RATE + sine_440 = (np.sin(2 * np.pi * 440.0 * t) * 0.5).astype(np.float32) + sine_880 = (np.sin(2 * np.pi * 880.0 * t) * 0.4).astype(np.float32) + sine_mixed = sine_440 + sine_880 + silence = np.zeros(BLOCK_SIZE, dtype=np.float32) + + test_signals = [sine_440, sine_mixed, silence, sine_880, silence] + + # ── Results tracking ────────────────────────────────────────────── + + results = { + "total_blocks_processed": 0, + "total_hours_simulated": 0.0, + "wall_time_seconds": 0.0, + "blocks_per_second_real": 0.0, + "real_time_factor": 0.0, # how many times faster/slower than realtime + "memory_mb_start": 0.0, + "memory_mb_end": 0.0, + "memory_mb_peak": 0.0, + "cpu_percent_avg": 0.0, + "cpu_percent_peak": 0.0, + "nan_outputs": 0, + "inf_outputs": 0, + "clipped_outputs": 0, + "max_abs_output": 0.0, + "fx_types_tested": [], + "fx_errors": {}, + "blocks_processed_per_fx": 0, + "throughput_history": [], + "memory_history": [], + "overall_status": "unknown", + } + + # Collect CPU samples + cpu_samples = [] + + pipeline = AudioPipeline() + mem_start = _get_memory_mb() + results["memory_mb_start"] = mem_start + results["memory_mb_peak"] = mem_start + + # Determine which FX types to test + fx_types_to_test = list(FX_PARAMS.keys()) + if FXType.NAM_AMP not in fx_types_to_test and FXType.NAM_AMP in list(FXType): + # Skip NAM_AMP, IR_CAB, LOOPER for stability test (need files/hardware) + pass + skip_types = {FXType.NAM_AMP, FXType.IR_CAB, FXType.LOOPER} + fx_types_to_test = [fx for fx in list(FXType) + if fx not in skip_types + and fx != FXType.TUNER] + fx_types_to_test.sort(key=lambda x: x.value) + + # We'll cycle through each FX type, processing blocks_per_fx blocks each + num_fx_types = len(fx_types_to_test) + blocks_per_fx = max(1, duration_blocks // num_fx_types) + results["blocks_processed_per_fx"] = blocks_per_fx + + print(f"Stability Test: Phase 7") + print(f" Target: {duration_blocks} blocks ({_fmt_time(duration_blocks * BLOCK_SIZE / SAMPLE_RATE)} simulated)") + print(f" FX types to test: {num_fx_types}") + print(f" Blocks per FX type: {blocks_per_fx}") + print(f" Memory start: {mem_start:.1f} MB") + print() + + total_blocks = 0 + clock_start = time.monotonic() + + for fx_idx, fx_type in enumerate(fx_types_to_test): + # Build preset with just this one FX block + params = FX_PARAMS.get(fx_type, {}) + block = FXBlock( + fx_type=fx_type, + enabled=True, + bypass=False, + params=params, + ) + preset = Preset( + name=f"stability_test_{fx_type.value}", + chain=[block], + master_volume=0.85, + ) + + # Load preset + try: + pipeline.load_preset(preset) + except Exception as e: + results["fx_errors"][fx_type.value] = str(e) + print(f" [WARN] {fx_type.value}: load error - {e}") + continue + + results["fx_types_tested"].append(fx_type.value) + + # Process blocks + nan_count = 0 + inf_count = 0 + clip_count = 0 + max_abs = 0.0 + + # Alternate between test signals for varied input + for i in range(blocks_per_fx): + sig = test_signals[i % len(test_signals)] + try: + out = pipeline.process(sig) + except Exception as e: + if fx_type.value not in results["fx_errors"]: + results["fx_errors"][fx_type.value] = str(e) + break + + # Validate output + if np.any(np.isnan(out)): + nan_count += 1 + if np.any(np.isinf(out)): + inf_count += 1 + if np.any(np.abs(out) > 1.0): + clip_count += 1 + output_abs_max = float(np.max(np.abs(out))) + if output_abs_max > max_abs: + max_abs = output_abs_max + + total_blocks += 1 + + # Periodic reporting + if total_blocks % report_every == 0 and total_blocks > 0: + mem_now = _get_memory_mb() + cpu_now = _get_cpu_percent() + cpu_samples.append(cpu_now) + elapsed = time.monotonic() - clock_start + bps = total_blocks / elapsed if elapsed > 0 else 0 + pct = total_blocks / duration_blocks * 100 + etr = elapsed / (total_blocks / duration_blocks) if total_blocks > 0 else 0 + + print(f" [{pct:5.1f}%] {fx_type.value:<20s} " + f"cpu={cpu_now:5.1f}% mem={mem_now:.1f}MB " + f"blocks={total_blocks//1000}k/{duration_blocks//1000}k " + f"ETA={_fmt_time(etr - elapsed)}") + + if mem_now > results["memory_mb_peak"]: + results["memory_mb_peak"] = mem_now + results["throughput_history"].append(bps) + results["memory_history"].append(mem_now) + + # Accumulate per-FX results + results["nan_outputs"] += nan_count + results["inf_outputs"] += inf_count + results["clipped_outputs"] += clip_count + if max_abs > results["max_abs_output"]: + results["max_abs_output"] = max_abs + + clock_end = time.monotonic() + results["wall_time_seconds"] = clock_end - clock_start + results["total_blocks_processed"] = total_blocks + results["total_hours_simulated"] = total_blocks * BLOCK_SIZE / SAMPLE_RATE / 3600.0 + results["blocks_per_second_real"] = total_blocks / results["wall_time_seconds"] if results["wall_time_seconds"] > 0 else 0 + results["real_time_factor"] = (total_blocks * BLOCK_SIZE / SAMPLE_RATE) / results["wall_time_seconds"] if results["wall_time_seconds"] > 0 else 0 + results["memory_mb_end"] = _get_memory_mb() + + # Compute average and peak CPU + if cpu_samples: + results["cpu_percent_avg"] = sum(cpu_samples) / len(cpu_samples) + results["cpu_percent_peak"] = max(cpu_samples) + + # Determine status + status = "PASS" + issues = [] + if results["nan_outputs"] > 0: + status = "WARN" + issues.append(f"{results['nan_outputs']} NaN outputs") + if results["inf_outputs"] > 0: + status = "WARN" + issues.append(f"{results['inf_outputs']} INF outputs") + if results["clipped_outputs"] > 0: + status = "WARN" + issues.append(f"{results['clipped_outputs']} clipped outputs (>1.0)") + mem_growth = results["memory_mb_end"] - results["memory_mb_start"] + if mem_growth > 50: # >50MB memory growth indicates leak + status = "WARN" + issues.append(f"Memory grew {mem_growth:.1f}MB (possible leak)") + if results["fx_errors"]: + status = "WARN" + issues.append(f"{len(results['fx_errors'])} FX load errors") + if total_blocks < duration_blocks * 0.95: + status = "WARN" + issues.append(f"Only processed {total_blocks}/{duration_blocks} blocks") + + results["overall_status"] = status + results["issues"] = issues + + return results + + +def main(): + print("=" * 72) + print(" Pi Multi-FX Pedal — Phase 7: Stability Test") + print(" 1-hour continuous audio DSP simulation") + print("=" * 72) + print() + + # Allow quick mode via env or --quick arg + quick = os.environ.get("STABILITY_QUICK", "").lower() in ("1", "true", "yes") + if quick: + blocks = 10000 # ~53 seconds simulated, ~2-5s wall clock + print(" QUICK MODE: 10000 blocks (~53s simulated)") + print() + else: + blocks = BLOCKS_PER_HOUR + + results = run_stability_test(duration_blocks=blocks) + + print() + print("=" * 72) + print(" RESULTS") + print("=" * 72) + print(f" Status: {results['overall_status']}") + if results.get("issues"): + print(f" Issues: {'; '.join(results['issues'])}") + print(f" Wall clock time: {_fmt_time(results['wall_time_seconds'])}") + print(f" Audio simulated: {results['total_hours_simulated']:.3f} hours") + print(f" Real-time factor: {results['real_time_factor']:.1f}x") + print(f" Blocks processed: {results['total_blocks_processed']:,}") + print(f" Throughput: {results['blocks_per_second_real']:.0f} blocks/s") + print(f" Memory start: {results['memory_mb_start']:.1f} MB") + print(f" Memory end: {results['memory_mb_end']:.1f} MB") + print(f" Memory peak: {results['memory_mb_peak']:.1f} MB") + print(f" Memory growth: {results['memory_mb_end'] - results['memory_mb_start']:.1f} MB") + print(f" CPU avg: {results['cpu_percent_avg']:.1f}%") + print(f" CPU peak: {results['cpu_percent_peak']:.1f}%") + print(f" NaN outputs: {results['nan_outputs']}") + print(f" INF outputs: {results['inf_outputs']}") + print(f" Clipped outputs: {results['clipped_outputs']}") + print(f" Max |output|: {results['max_abs_output']:.4f}") + print(f" FX types tested: {len(results['fx_types_tested'])}") + print(f" FX errors: {len(results['fx_errors'])}") + if results['fx_errors']: + for fx, err in results['fx_errors'].items(): + print(f" - {fx}: {err}") + print() + + # Save results to file + output_path = os.environ.get("STABILITY_RESULTS_PATH", + "/tmp/stability_test_results.json") + with open(output_path, "w") as f: + json.dump(results, f, indent=2, default=str) + print(f" Results saved to: {output_path}") + print() + + return 0 if results["overall_status"] == "PASS" else 1 + + +if __name__ == "__main__": + sys.exit(main())