Initial commit: Arduino Uno Q USB Audio Synth project

- USB gadget (UAC1 + MIDI) with boot persistence - USB host mode with auto-detection - Pure Data synth patch with ALSA + UDP netreceive MIDI input - MIDI rawmidi-to-UDP bridge (midi-bridge.py) - LED matrix visualization (led-matrix-viz.py via arduino-router RPC) - MCU sketch for LED matrix control (Arduino_RouterBridge + ArduinoGraphics) - Conftest/OPA Rego policy validation suite for .pd files - gen_synth_pd.py: programmatic .pd file generator - System services: midi-bridge, led-matrix-viz, pd-synth-auto, usb-role-detect - Justfile with deploy, enable, diagnostic recipes - Diagnostic script (diag.sh)
2026-06-22 22:46:36 -07:00
commit 8cd796e3dd
38 changed files with 2681 additions and 0 deletions
@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Validate .pd files using Conftest/OPA Rego policies."""
+import json, sys, subprocess, tempfile, os, glob
+from pathlib import Path
+from pd2json import parse_pd
+
+def main():
+    policy_dir = Path(__file__).parent
+
+    # Find conftest/opa (check mise install dirs first)
+    mise_dir = os.path.expanduser("~/.local/share/mise/installs")
+    for tool in ["conftest", "opa"]:
+        for ver_dir in sorted(glob.glob(os.path.join(mise_dir, tool, "*")), reverse=True):
+            for p in [os.path.join(ver_dir, tool), os.path.join(ver_dir, "bin", tool)]:
+                if os.path.isfile(p):
+                    os.environ["PATH"] = os.path.dirname(p) + ":" + os.environ.get("PATH", "")
+                    break
+            else:
+                continue
+            break
+
+    runner = None
+    for cmd in ["conftest", "opa"]:
+        try:
+            subprocess.run([cmd, "--version"], capture_output=True, timeout=5)
+            runner = cmd
+            break
+        except (FileNotFoundError, subprocess.TimeoutExpired):
+            continue
+
+    if not runner:
+        print("ERROR: neither 'conftest' nor 'opa' found in PATH.")
+        print("Install: pip install conftest  or  https://www.conftest.dev/")
+        sys.exit(1)
+
+    pd_files = sys.argv[1:] or list(Path(".").glob("*.pd"))
+    if not pd_files:
+        print("Usage: validate.py <file.pd> [file2.pd ...]")
+        sys.exit(1)
+
+    all_passed = True
+    for pf in pd_files:
+        pf = Path(pf)
+        data = parse_pd(pf)
+
+        if runner == "conftest":
+            result = run_conftest(data, policy_dir)
+        else:
+            result = run_opa(data, policy_dir)
+
+        if result["passed"]:
+            print(f"  ✓ {pf.name} — all checks passed")
+        else:
+            all_passed = False
+            print(f"  ✗ {pf.name} — {len(result['violations'])} violation(s):")
+            for v in result["violations"]:
+                print(f"      • {v}")
+
+    sys.exit(0 if all_passed else 1)
+
+
+def run_conftest(data, policy_dir):
+    with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
+        json.dump(data, f)
+        json_path = f.name
+
+    try:
+        result = subprocess.run(
+            ["conftest", "test", "--policy", str(policy_dir), json_path],
+            capture_output=True, text=True, timeout=30
+        )
+        out = result.stdout + result.stderr
+        violations = []
+        for l in out.split("\n"):
+            if "FAIL" in l or "failed" in l.lower():
+                violations.append(l.strip())
+        # Also parse structured output if json
+        if result.stdout and not violations:
+            try:
+                import json as j
+                report = j.loads(result.stdout)
+                for fname, checks in report.get("failures", {}).items():
+                    for c in checks:
+                        violations.append(c)
+            except:
+                pass
+        return {"passed": result.returncode == 0, "violations": violations}
+    finally:
+        os.unlink(json_path)
+
+
+def run_opa(data, policy_dir):
+    pkg = "main"
+    query = "data.main.violations"
+
+    with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
+        json.dump(data, f)
+        json_path = f.name
+
+    try:
+        result = subprocess.run(
+            ["opa", "eval", "-d", str(policy_dir), "-d", json_path, "--format", "json", query],
+            capture_output=True, text=True, timeout=30
+        )
+        if result.returncode != 0:
+            return {"passed": False, "violations": [result.stderr.strip()]}
+
+        out = json.loads(result.stdout)
+        violations = []
+        for expr in out.get("result", []):
+            for binding in expr.get("expressions", []):
+                vals = binding.get("value", [])
+                if isinstance(vals, list):
+                    violations.extend(vals)
+        return {"passed": len(violations) == 0, "violations": violations}
+    finally:
+        os.unlink(json_path)
+
+
+if __name__ == "__main__":
+    main()