uno-q-audio-synth/scripts/diagnostics/test-pipeline.py

#!/usr/bin/env python3
"""Full pipeline diagnostic: test MIDI → viz → router → MCU → LED matrix.

Usage:
  python3 scripts/diagnostics/test-pipeline.py [--host HOST]

Tests:
  1. Router Unix socket presence and connectivity
  2. MCU RPC: get_status, clear, draw_frame
  3. Viz UDP reception and frame sending
  4. Mid-bridge service health
  5. Full end-to-end: send MIDI → verify viz stats → verify MCU state
  6. Router buffer drain test: send 100 rapid frames, verify MCU stays responsive
"""
import socket, msgpack, time, json, os, sys, subprocess

HOST = sys.argv[1] if len(sys.argv) > 1 else "192.168.9.167"
ROUTER_SOCK = "/var/run/arduino-router.sock"
VIZ_PORT = 8082

passed = 0
failed = 0

def check(name, ok, detail=""):
    global passed, failed
    if ok:
        passed += 1
        print(f"  ✓ {name}")
    else:
        failed += 1
        print(f"  ✗ {name}")
        if detail:
            print(f"      {detail}")

def ssh(cmd):
    return subprocess.run(
        ["ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null",
         "-i", os.path.expanduser("~/.ssh/unoq_deploy_key"),
         f"arduino@{HOST}", cmd],
        capture_output=True, text=True, timeout=15
    )

def run_python(code):
    r = ssh(f"python3 -c {sh_quote(code)}")
    return r.stdout, r.stderr, r.returncode

def sh_quote(s):
    return f"'{s.replace(chr(39), chr(39)+chr(92)+chr(39)+chr(39))}'"

# ── Test 1: Router connectivity ──
print("\n=== Router ===")
r = ssh("test -S /var/run/arduino-router.sock && echo 'exists'")
check("Router socket exists", "exists" in r.stdout)

r = ssh("python3 -c 'import socket; s=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM); s.settimeout(3); s.connect(\"/var/run/arduino-router.sock\"); print(\"ok\"); s.close()'")
check("Router connectable", "ok" in r.stdout, r.stderr)

# ── Test 2: MCU RPC ──
print("\n=== MCU RPC ===")
out, err, rc = run_python("""
import socket, msgpack
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(5)
s.connect('/var/run/arduino-router.sock')
p = msgpack.Packer()
req = p.pack([0, 1, 'get_status', ['']])
s.sendall(req)
d = s.recv(4096)
obj = msgpack.unpackb(d)
print(f"status={obj[3]}")
req = p.pack([0, 2, 'clear', ['']])
s.sendall(req)
d = s.recv(4096)
obj = msgpack.unpackb(d)
print(f"clear={obj[3]}")
req = p.pack([0, 3, 'draw_frame', ['-1','-1','-1','-1']])
s.sendall(req)
d = s.recv(4096)
obj = msgpack.unpackb(d)
print(f"draw_frame={obj[3]}")
s.close()
""")
check("MCU get_status", "status=" in out, err)
check("MCU clear", "clear=" in out and "'OK'" in out, err)
check("MCU draw_frame", "draw_frame=" in out and "'OK'" in out, err)

# ── Test 3: Router buffer drain ──
print("\n=== Router buffer drain ===")
out, err, rc = run_python("""
import socket, msgpack, time
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(5)
s.connect('/var/run/arduino-router.sock')
p = msgpack.Packer()
# Send 100 draw_frame REQUESTs and drain each response
for i in range(100):
    req = p.pack([0, i, 'draw_frame', ['-1','-1','-1','-1']])
    s.sendall(req)
    s.recv(4096)
# Verify MCU still responsive
req = p.pack([0, 200, 'get_status', ['']])
s.sendall(req)
d = s.recv(4096)
obj = msgpack.unpackb(d)
print(f"ok={obj[3]}")
s.close()
""")
check("MCU responsive after 100 frames", "ok=" in out and "frame" in out, err)

# ── Test 4: Viz UDP reception ──
print("\n=== Viz pipeline ===")
out, err, rc = run_python("""
import socket, time, json
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.sendto(b'midi 144 60 100;', ('127.0.0.1', 8082))
s.sendto(b'midi 144 64 100;', ('127.0.0.1', 8082))
s.sendto(b'midi 144 67 100;', ('127.0.0.1', 8082))
time.sleep(3)
with open('/tmp/viz-stats.json') as f:
    d = json.load(f)
print(f"sent={d['frames_sent']} dropped={d['frames_dropped']} active={d['active_notes']}")
""")
check("Viz received MIDI", "active" in out, err)
check("Viz zero drops", "dropped=0" in out, err)

# ── Test 5: Viz router drain ──
print("\n=== Viz + router coexistence ===")
out, err, rc = run_python("""
import socket, msgpack, time, json

# Send MIDI via viz
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
for note in range(60, 80):
    s.sendto(b'midi 144 %d 100;' % note, ('127.0.0.1', 8082))
    time.sleep(0.05)
for note in range(60, 80):
    s.sendto(b'midi 128 %d 0;' % note, ('127.0.0.1', 8082))
    time.sleep(0.02)
time.sleep(2)

# Check viz stats
with open('/tmp/viz-stats.json') as f:
    d = json.load(f)
print(f"viz_sent={d['frames_sent']} viz_dropped={d['frames_dropped']}")

# MCU still responsive?
s2 = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s2.settimeout(5)
s2.connect('/var/run/arduino-router.sock')
p = msgpack.Packer()
req = p.pack([0, 1, 'get_status', ['']])
s2.sendall(req)
d2 = s2.recv(4096)
obj = msgpack.unpackb(d2)
print(f"mcu_status={obj[3]}")
s2.close()
""")
check("Viz zero drops after burst", "viz_dropped=0" in out, err)
check("MCU responsive after viz burst", "mcu_status=" in out and "'frame'" in out, err)

# ── Summary ──
print(f"\n{'='*40}")
print(f"  {passed} passed, {failed} failed")
print(f"{'='*40}")
sys.exit(0 if failed == 0 else 1)