david
8cd796e3dd
- 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)
122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
#!/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()
|