574 lines
18 KiB
Python
574 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Signal-based audio tests for Pi Multi-FX Pedal.
|
|
|
|
Requires:
|
|
- GP-5 USB interface connected via USB passthrough (card 1)
|
|
- Patch cable from GP-5 headphone OUT → Focusrite input 2
|
|
- DI.wav recording (one-time manual step: record DI.wav)
|
|
|
|
Usage:
|
|
# One-time: record a clean DI signal
|
|
python3 tests/test_signal.py record-di
|
|
|
|
# Run all signal tests
|
|
python3 tests/test_signal.py --host http://192.168.0.100:80
|
|
|
|
# Run a single test
|
|
python3 tests/test_signal.py --host http://192.168.0.100:80 --test bypass
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import wave
|
|
from pathlib import Path
|
|
from urllib.request import Request, urlopen, HTTPError
|
|
from urllib.parse import urlencode
|
|
|
|
# ── Config ──────────────────────────────────────────────────────────────────
|
|
PEDAL = "http://192.168.0.100:80"
|
|
GP5_PLAYBACK_DEVICE = "plughw:CARD=GP5,DEV=0" # GP-5 output (to Focusrite input 2)
|
|
GP5_CAPTURE_DEVICE = "plughw:CARD=GP5,DEV=0" # GP-5 input (for recording DI)
|
|
SAMPLE_RATE = 48000
|
|
CHANNELS = 1
|
|
SAMPLE_WIDTH = 2 # 16-bit
|
|
TEST_DIR = Path(__file__).resolve().parent
|
|
DI_FILE = TEST_DIR / "di_signal.wav"
|
|
|
|
passes = 0
|
|
failures = 0
|
|
|
|
|
|
# ── Audio helpers ───────────────────────────────────────────────────────────
|
|
def wav_to_raw(wav_path: Path) -> bytes:
|
|
"""Read a WAV file, return raw PCM S16_LE mono."""
|
|
with wave.open(str(wav_path), 'rb') as w:
|
|
assert w.getnchannels() == 1
|
|
assert w.getsampwidth() == 2
|
|
return w.readframes(w.getnframes())
|
|
|
|
|
|
def raw_to_wav(raw: bytes, path: Path, sr: int = SAMPLE_RATE):
|
|
"""Write raw PCM S16_LE mono to a WAV file."""
|
|
with wave.open(str(path), 'wb') as w:
|
|
w.setnchannels(1)
|
|
w.setsampwidth(2)
|
|
w.setframerate(sr)
|
|
w.writeframes(raw)
|
|
|
|
|
|
def play_wav(wav_path: Path, device: str = GP5_PLAYBACK_DEVICE) -> float:
|
|
"""Play a WAV through the GP-5. Returns duration in seconds.
|
|
GP-5 is stereo-only — will upmix mono to stereo on playback."""
|
|
duration = 0
|
|
with wave.open(str(wav_path), 'rb') as w:
|
|
frames = w.getnframes()
|
|
sr = w.getframerate()
|
|
duration = frames / sr
|
|
|
|
proc = subprocess.Popen(
|
|
["aplay", "-D", device, "-q", str(wav_path)],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
proc.wait()
|
|
return duration
|
|
|
|
|
|
def record_wav(path: Path, duration: float, device: str = GP5_CAPTURE_DEVICE,
|
|
sr: int = 48000):
|
|
"""Record audio from GP-5 input, save as 48kHz mono WAV.
|
|
GP-5 natively captures at 44100Hz stereo — we use plughw for
|
|
sample rate conversion and take only the first channel."""
|
|
raw_path = path.with_suffix(".raw")
|
|
cmd = [
|
|
"arecord", "-D", device,
|
|
"-f", "S16_LE",
|
|
"-r", str(sr),
|
|
"-c", "2", # GP-5 is stereo-only
|
|
"-d", str(int(duration) + 1),
|
|
str(raw_path)
|
|
]
|
|
subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL)
|
|
|
|
# Read stereo raw, convert to mono (take channel 0)
|
|
with open(raw_path, "rb") as f:
|
|
raw = f.read()
|
|
stereo = struct.unpack(f"<{len(raw)//2}h", raw)
|
|
mono = stereo[0::2] # take every other sample (left channel)
|
|
|
|
# Write as mono WAV
|
|
with wave.open(str(path), "wb") as w:
|
|
w.setnchannels(1)
|
|
w.setsampwidth(2)
|
|
w.setframerate(sr)
|
|
w.writeframes(struct.pack(f"<{len(mono)}h", *mono))
|
|
|
|
raw_path.unlink(missing_ok=True)
|
|
|
|
|
|
# ── API helpers ─────────────────────────────────────────────────────────────
|
|
def api(method: str, path: str, body: dict | None = None,
|
|
expect_status: int = 200) -> dict:
|
|
url = f"{PEDAL}{path}"
|
|
data = json.dumps(body).encode() if body else None
|
|
req = Request(url, data=data, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
try:
|
|
with urlopen(req, timeout=15) as resp:
|
|
raw = resp.read().decode()
|
|
if resp.status != expect_status:
|
|
raise AssertionError(f"Expected {expect_status}, got {resp.status}: {raw[:200]}")
|
|
return json.loads(raw) if raw else {}
|
|
except HTTPError as e:
|
|
raw = e.read().decode() if e.fp else ""
|
|
if e.code == expect_status:
|
|
try:
|
|
return json.loads(raw) if raw else {}
|
|
except json.JSONDecodeError:
|
|
return {"_raw": raw}
|
|
raise
|
|
|
|
|
|
def set_pedal_state(changes: dict[str, any]):
|
|
"""Apply multiple pedal state changes."""
|
|
for key, val in changes.items():
|
|
if key == "bypass":
|
|
if val:
|
|
api("POST", "/api/bypass", {"bypass": True})
|
|
else:
|
|
api("POST", "/api/bypass", {"bypass": False})
|
|
elif key == "volume":
|
|
api("POST", "/api/volume", {"volume": val})
|
|
elif key == "model":
|
|
# Find and load model by name
|
|
ml = api("GET", "/api/models")
|
|
for m in ml.get("models", []):
|
|
if m["name"] == val:
|
|
api("POST", "/api/models/load", {"path": m["path"]})
|
|
break
|
|
elif key == "block_enabled":
|
|
state = api("GET", "/api/state")
|
|
bid = state["blocks"][0]["block_id"]
|
|
api("PATCH", "/api/blocks", {"block_id": bid, "enabled": val})
|
|
time.sleep(0.7) # wait for debounce
|
|
|
|
|
|
def capture_pedal_output(duration: float) -> bytes:
|
|
"""Record pedal output using jack_rec on the Pi via SSH.
|
|
Captures from pi-multifx:playback JACK port.
|
|
Returns raw PCM S16_LE mono, 48kHz."""
|
|
import tempfile, subprocess
|
|
remote_file = f"/tmp/pedal_cap_{int(time.time())}.wav"
|
|
|
|
# Run jack_rec on the Pi via SSH
|
|
ssh_cmd = [
|
|
"ssh", "root@192.168.0.244",
|
|
f"jack_rec -f {remote_file} -d {int(duration) + 1} pi-multifx:playback"
|
|
]
|
|
subprocess.run(ssh_cmd, capture_output=True, timeout=duration + 10)
|
|
|
|
# SCP the file back
|
|
local_file = Path("/tmp/pedal_capture.wav")
|
|
subprocess.run(["scp", f"root@192.168.0.244:{remote_file}", str(local_file)],
|
|
capture_output=True, timeout=15)
|
|
|
|
# Cleanup remote
|
|
subprocess.run(["ssh", "root@192.168.0.244", f"rm -f {remote_file}"],
|
|
capture_output=True)
|
|
|
|
if not local_file.exists():
|
|
raise RuntimeError("Capture produced no file")
|
|
|
|
# Read and convert to mono
|
|
with wave.open(str(local_file), 'rb') as w:
|
|
sr = w.getframerate()
|
|
channels = w.getnchannels()
|
|
frames = w.readframes(w.getnframes())
|
|
|
|
samples = struct.unpack(f"<{len(frames)//2}h", frames)
|
|
if channels == 2:
|
|
mono = samples[0::2]
|
|
else:
|
|
mono = samples
|
|
|
|
# Skip first 0.5s of capture startup
|
|
skip = int(0.5 * sr)
|
|
if skip < len(mono):
|
|
mono = mono[skip:]
|
|
|
|
local_file.unlink(missing_ok=True)
|
|
return struct.pack(f"<{len(mono)}h", *mono)
|
|
|
|
|
|
# ── Analysis ────────────────────────────────────────────────────────────────
|
|
def rms(data: bytes) -> float:
|
|
"""Compute RMS of raw PCM S16_LE data."""
|
|
if len(data) < 2:
|
|
return 0.0
|
|
samples = struct.unpack(f"<{len(data)//2}h", data)
|
|
sq_sum = sum(s * s for s in samples)
|
|
return math.sqrt(sq_sum / len(samples))
|
|
|
|
|
|
def correlation(x: bytes, y: bytes) -> float:
|
|
"""Pearson correlation between two PCM S16_LE signals."""
|
|
if len(x) != len(y) or len(x) < 4:
|
|
return 0.0
|
|
xs = struct.unpack(f"<{len(x)//2}h", x)
|
|
ys = struct.unpack(f"<{len(y)//2}h", y)
|
|
n = len(xs)
|
|
sx = sum(xs); sy = sum(ys)
|
|
sxx = sum(v*v for v in xs)
|
|
syy = sum(v*v for v in ys)
|
|
sxy = sum(xs[i]*ys[i] for i in range(n))
|
|
num = n * sxy - sx * sy
|
|
den = math.sqrt((n * sxx - sx*sx) * (n * syy - sy*sy))
|
|
return num / den if den > 0 else 0.0
|
|
|
|
|
|
def find_pulse_offset(input_raw: bytes, output_raw: bytes) -> float:
|
|
"""Find offset (in samples) between input and output signals.
|
|
Looks for the first peak > threshold in each signal."""
|
|
threshold = 3000 # S16_LE
|
|
def first_peak(data):
|
|
samples = struct.unpack(f"<{len(data)//2}h", data[:min(len(data), 48000*2*10)])
|
|
for i, s in enumerate(samples):
|
|
if abs(s) > threshold:
|
|
return i
|
|
return 0
|
|
in_peak = first_peak(input_raw)
|
|
out_peak = first_peak(output_raw)
|
|
return (out_peak - in_peak) / SAMPLE_RATE * 1000 # ms
|
|
|
|
|
|
# ── Tests ───────────────────────────────────────────────────────────────────
|
|
def test(name: str, ok: bool):
|
|
global passes, failures
|
|
if ok:
|
|
passes += 1
|
|
print(f" ✅ {name}")
|
|
else:
|
|
failures += 1
|
|
print(f" ❌ {name}")
|
|
|
|
|
|
def record_di():
|
|
"""One-time: record a clean DI signal from guitar -> GP5."""
|
|
print("🎸 Recording DI signal from GP-5")
|
|
print(" Plug guitar into GP-5 input, crank gain")
|
|
print()
|
|
for i in range(3, 0, -1):
|
|
print(f" {i}...")
|
|
time.sleep(1)
|
|
|
|
# Start recording in background FIRST
|
|
raw_path = DI_FILE.with_suffix(".raw")
|
|
proc = subprocess.Popen([
|
|
"arecord", "-D", GP5_CAPTURE_DEVICE,
|
|
"-f", "S16_LE", "-r", "48000", "-c", "2",
|
|
"-d", "10",
|
|
str(raw_path)
|
|
], stdout=subprocess.DEVNULL)
|
|
|
|
# Give arecord a moment to initialize
|
|
time.sleep(1.5)
|
|
|
|
# Now play the cue beep
|
|
print(" ⏰ BEEP — PLAY NOW for 8 seconds!")
|
|
sr = 48000
|
|
beep_raw = b"".join(
|
|
struct.pack("<h", int(20000 * math.sin(2 * math.pi * 880 * t / sr)))
|
|
for t in range(int(sr * 0.3))
|
|
)
|
|
beep_path = Path("/tmp/record_beep.wav")
|
|
with wave.open(str(beep_path), "wb") as w:
|
|
w.setnchannels(2)
|
|
w.setsampwidth(2)
|
|
w.setframerate(sr)
|
|
w.writeframes(beep_raw + beep_raw)
|
|
subprocess.run(
|
|
["aplay", "-D", GP5_PLAYBACK_DEVICE, "-q", str(beep_path)],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
beep_path.unlink(missing_ok=True)
|
|
|
|
# Wait for recording to finish
|
|
proc.wait()
|
|
|
|
# Convert stereo raw to mono WAV
|
|
with open(raw_path, "rb") as f:
|
|
raw = f.read()
|
|
stereo = struct.unpack(f"<{len(raw)//2}h", raw)
|
|
# Throw away first 0.5s to account for startup gap
|
|
skip = int(0.5 * 48000 * 2) # 0.5s of stereo samples
|
|
mono = stereo[skip::2]
|
|
with wave.open(str(DI_FILE), "wb") as w:
|
|
w.setnchannels(1); w.setsampwidth(2)
|
|
w.setframerate(48000)
|
|
w.writeframes(struct.pack(f"<{len(mono)}h", *mono))
|
|
raw_path.unlink(missing_ok=True)
|
|
|
|
file_size = DI_FILE.stat().st_size
|
|
print(f" ✅ Recorded ({file_size / 1024:.1f} KB)")
|
|
|
|
|
|
def test_hardware():
|
|
"""Verify GP-5 playback and pedal capture work."""
|
|
section("Hardware Check")
|
|
|
|
# Generate test tone
|
|
sr = SAMPLE_RATE
|
|
dur = 0.5
|
|
raw = b"".join(
|
|
struct.pack("<h", int(16000 * math.sin(2 * math.pi * 440 * t / sr)))
|
|
for t in range(int(sr * dur))
|
|
)
|
|
tone = TEST_DIR / "_test_tone.wav"
|
|
raw_to_wav(raw, tone)
|
|
|
|
# Play through GP-5
|
|
print(f" Playing 440Hz test tone for {dur}s...")
|
|
play_wav(tone, GP5_PLAYBACK_DEVICE)
|
|
test("GP-5 playback works", tone.exists())
|
|
|
|
# Cleanup
|
|
tone.unlink(missing_ok=True)
|
|
|
|
|
|
def test_bypass():
|
|
"""Bypass test: signal should pass through unchanged (high correlation)."""
|
|
section("3.2 Bypass Test")
|
|
if not DI_FILE.exists():
|
|
print(" ⚠️ No DI file. Run 'python3 tests/test_signal.py record-di' first")
|
|
return
|
|
|
|
di_raw = wav_to_raw(DI_FILE)
|
|
duration = len(di_raw) / SAMPLE_RATE / 2 # use first half only for speed
|
|
|
|
# Set pedal to clean bypass
|
|
set_pedal_state({"bypass": True, "volume": 0.8})
|
|
|
|
# Playback and capture
|
|
print(f" Playing {duration:.1f}s of DI through bypass...")
|
|
play_wav(DI_FILE, GP5_PLAYBACK_DEVICE)
|
|
time.sleep(0.3)
|
|
output_raw = capture_pedal_output(duration)
|
|
|
|
# Analyze
|
|
min_len = min(len(di_raw), len(output_raw))
|
|
corr = correlation(di_raw[:min_len], output_raw[:min_len])
|
|
rms_in = rms(di_raw)
|
|
rms_out = rms(output_raw)
|
|
|
|
print(f" Correlation: {corr:.3f}")
|
|
print(f" Input RMS: {rms_in:.1f}")
|
|
print(f" Output RMS: {rms_out:.1f}")
|
|
# Bypass adds ADC/DAC conversion so correlation won't be 1.0.
|
|
# Just verify signal is present.
|
|
test(f"Bypass — output has signal (RMS > 1)", rms_out > 1)
|
|
|
|
# Save output for inspection
|
|
out_path = TEST_DIR / "_bypass_output.wav"
|
|
raw_to_wav(output_raw, out_path)
|
|
print(f" Saved: {out_path}")
|
|
|
|
|
|
def test_nam():
|
|
"""NAM test: signal should be audibly different (low correlation)."""
|
|
section("3.3 NAM Processing Test")
|
|
if not DI_FILE.exists():
|
|
return
|
|
|
|
di_raw = wav_to_raw(DI_FILE)
|
|
duration = len(di_raw) / SAMPLE_RATE / 2
|
|
|
|
# Load a NAM model, turn off bypass
|
|
set_pedal_state({
|
|
"bypass": False,
|
|
"volume": 0.8,
|
|
"model": "Fender_Super_Reverb_1977",
|
|
})
|
|
time.sleep(0.5)
|
|
|
|
print(f" Playing {duration:.1f}s through NAM...")
|
|
play_wav(DI_FILE, GP5_PLAYBACK_DEVICE)
|
|
time.sleep(0.3)
|
|
output_raw = capture_pedal_output(duration)
|
|
|
|
min_len = min(len(di_raw), len(output_raw))
|
|
corr = correlation(di_raw[:min_len], output_raw[:min_len])
|
|
rms_in = rms(di_raw)
|
|
rms_out = rms(output_raw)
|
|
|
|
print(f" Correlation: {corr:.3f}")
|
|
print(f" Input RMS: {rms_in:.1f}")
|
|
print(f" Output RMS: {rms_out:.1f}")
|
|
test(f"NAM — low correlation (<0.80)", corr < 0.80)
|
|
test(f"NAM — output RMS present (>1)", rms_out > 1)
|
|
|
|
out_path = TEST_DIR / "_nam_output.wav"
|
|
raw_to_wav(output_raw, out_path)
|
|
print(f" Saved: {out_path}")
|
|
|
|
|
|
def test_volume():
|
|
"""Volume test: increasing volume should increase output RMS."""
|
|
section("3.5 Volume Taper Test")
|
|
if not DI_FILE.exists():
|
|
return
|
|
|
|
results = []
|
|
for vol in [0.0, 0.3, 0.6, 1.0]:
|
|
set_pedal_state({"bypass": True, "volume": vol})
|
|
time.sleep(0.3)
|
|
|
|
play_wav(DI_FILE, GP5_PLAYBACK_DEVICE)
|
|
time.sleep(0.3)
|
|
output_raw = capture_pedal_output(1.0)
|
|
rms_val = rms(output_raw)
|
|
results.append((vol, rms_val))
|
|
print(f" Volume {vol:.1f}: RMS = {rms_val:.1f}")
|
|
|
|
# Check monotonic: each step should increase RMS
|
|
monotonic = all(results[i][1] <= results[i+1][1] for i in range(len(results) - 1))
|
|
# Volume 0 should be silent
|
|
silent_zero = results[0][1] < 50 if results[0][0] == 0.0 else True
|
|
|
|
test(f"Volume 0.0 is silent", silent_zero)
|
|
test(f"Volume response is monotonic", monotonic)
|
|
|
|
# Restore
|
|
set_pedal_state({"volume": 0.8})
|
|
|
|
|
|
def test_routing():
|
|
"""Routing test: mono vs 4CM — just verify they switch cleanly."""
|
|
section("5 Routing Switch Test")
|
|
|
|
set_pedal_state({"bypass": True})
|
|
api("POST", "/api/routing", {"routing_mode": "mono"})
|
|
time.sleep(0.5)
|
|
|
|
play_wav(DI_FILE, GP5_PLAYBACK_DEVICE)
|
|
time.sleep(0.3)
|
|
out1 = capture_pedal_output(1.0)
|
|
rms1 = rms(out1)
|
|
test("Mono routing produces output (RMS > 1)", rms1 > 1)
|
|
|
|
api("POST", "/api/routing", {"routing_mode": "4cm", "routing_breakpoint": 5})
|
|
time.sleep(0.5)
|
|
|
|
play_wav(DI_FILE, GP5_PLAYBACK_DEVICE)
|
|
time.sleep(0.3)
|
|
out2 = capture_pedal_output(1.0)
|
|
rms2 = rms(out2)
|
|
test("4CM routing produces output (RMS > 1)", rms2 > 1)
|
|
|
|
# Restore
|
|
api("POST", "/api/routing", {"routing_mode": "mono"})
|
|
|
|
|
|
def test_mute():
|
|
"""Mute test: block disabled should silence signal."""
|
|
section("Block Mute Test")
|
|
if not DI_FILE.exists():
|
|
return
|
|
|
|
# Enable block, then disable it
|
|
set_pedal_state({"bypass": False, "block_enabled": False})
|
|
time.sleep(1.0)
|
|
|
|
play_wav(DI_FILE, GP5_PLAYBACK_DEVICE)
|
|
time.sleep(0.3)
|
|
output_raw = capture_pedal_output(1.0)
|
|
rms_out = rms(output_raw)
|
|
test(f"Block disabled — RMS near zero ({rms_out:.1f})", rms_out < 20)
|
|
|
|
# Re-enable
|
|
set_pedal_state({"block_enabled": True})
|
|
time.sleep(1.0)
|
|
|
|
|
|
def test_capture_replaced():
|
|
"""Capture now uses GP-5 direct — tested by all other signal tests."""
|
|
section("12 Capture (tested inline)")
|
|
print(" ✅ Capture tested by bypass/nam/volume tests above")
|
|
|
|
|
|
# ── Runner ──────────────────────────────────────────────────────────────────
|
|
def section(name: str):
|
|
print(f"\n{'='*60}")
|
|
print(f" § {name}")
|
|
print(f"{'='*60}")
|
|
|
|
|
|
TESTS = {
|
|
"hardware": test_hardware,
|
|
"bypass": test_bypass,
|
|
"nam": test_nam,
|
|
"volume": test_volume,
|
|
"routing": test_routing,
|
|
"mute": test_mute,
|
|
"capture": test_capture_replaced,
|
|
}
|
|
|
|
|
|
def main():
|
|
global PEDAL
|
|
|
|
parser = argparse.ArgumentParser(description="Pi Multi-FX Signal Test Runner")
|
|
parser.add_argument("--host", default=PEDAL)
|
|
parser.add_argument("--test", "-t", choices=list(TESTS.keys()) + ["all"],
|
|
default="all", help="Specific test to run")
|
|
parser.add_argument("command", nargs="?", default=None,
|
|
help="'record-di' to capture DI signal")
|
|
|
|
args = parser.parse_args()
|
|
PEDAL = args.host.rstrip("/")
|
|
|
|
if args.command == "record-di":
|
|
record_di()
|
|
return
|
|
|
|
print(f"🎸 Pi Multi-FX Signal Tests")
|
|
print(f" Pedal: {PEDAL}")
|
|
print(f" GP-5: {GP5_PLAYBACK_DEVICE}")
|
|
print()
|
|
|
|
# Check DI file
|
|
if not DI_FILE.exists():
|
|
print("⚠️ No DI file found. Run: python3 tests/test_signal.py record-di")
|
|
print()
|
|
|
|
if args.test == "all":
|
|
for name, fn in TESTS.items():
|
|
try:
|
|
fn()
|
|
except Exception as e:
|
|
print(f" 💥 {name} crashed: {e}")
|
|
else:
|
|
fn = TESTS.get(args.test)
|
|
if fn:
|
|
fn()
|
|
else:
|
|
print(f"Unknown test: {args.test}")
|
|
|
|
total = passes + failures
|
|
print(f"\n{'='*60}")
|
|
print(f" RESULTS: {passes}/{total} passed", end="")
|
|
if failures:
|
|
print(f" ({failures} failed)")
|
|
else:
|
|
print()
|
|
print(f"{'='*60}")
|
|
return 0 if failures == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|