From acdff5eb9b13ffc04fedfdda366481786d607048 Mon Sep 17 00:00:00 2001 From: Shawn Date: Fri, 19 Jun 2026 06:32:10 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20signal-based=20test=20suite=20with=20ja?= =?UTF-8?q?ck=5Frec=20capture=20=E2=80=94=209/9=20passing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_signal.py | 573 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 573 insertions(+) create mode 100644 tests/test_signal.py diff --git a/tests/test_signal.py b/tests/test_signal.py new file mode 100644 index 0000000..c988748 --- /dev/null +++ b/tests/test_signal.py @@ -0,0 +1,573 @@ +#!/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(" 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())