Phase 7: Stability test + fix MIDI test signature
- tests/test_stability.py — 1-hour continuous DSP pipeline stability test covering all 46 FX types with CPU/memory/NaN monitoring - tests/test_integration.py — fixed test_midi_cc_maps_expression to use correct _on_midi_cc(value, cc_number, channel) signature matching the Phase 6 MIDI fix
This commit is contained in:
@@ -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):
|
||||
|
||||
@@ -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())
|
||||
Reference in New Issue
Block a user