384 lines
15 KiB
Python
384 lines
15 KiB
Python
"""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
|
|
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())
|