67 lines
3.0 KiB
Python
67 lines
3.0 KiB
Python
# Sweep ALL system VST3: gui_probe <plugin> bridge_two (exact bridge arch:
|
|
# 2 worker threads, own STA, main pump), classify load/attach/hang/timeout.
|
|
# Usage: python sweep_all_gui.py [variant] [secs] [jobs]
|
|
import subprocess, sys, os, time, concurrent.futures, re, json
|
|
|
|
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
|
|
VST = r"C:\Program Files\Common Files\VST3"
|
|
PROBE = r"C:\Users\locpham\SonicForgeStudio\native_bridge\build\Release\gui_probe.exe"
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
variant = sys.argv[1] if len(sys.argv) > 1 else "bridge_two"
|
|
secs = int(sys.argv[2]) if len(sys.argv) > 2 else 8
|
|
jobs = int(sys.argv[3]) if len(sys.argv) > 3 else 4
|
|
|
|
# top-level bundles only (dedupe inner Contents/x86_64-win/*.vst3)
|
|
plugins = []
|
|
for root, dirs, files in os.walk(VST):
|
|
if "/Contents/" in root or "\\Contents\\" in root:
|
|
continue
|
|
for f in files:
|
|
if f.lower().endswith(".vst3"):
|
|
plugins.append(os.path.join(root, f))
|
|
plugins.sort()
|
|
print(f"sweep {len(plugins)} plugins, variant={variant} secs={secs} jobs={jobs}", flush=True)
|
|
|
|
def run_one(p):
|
|
t0 = time.time()
|
|
try:
|
|
r = subprocess.run([PROBE, p, variant, str(secs)],
|
|
capture_output=True, text=True, timeout=secs + 20,
|
|
creationflags=subprocess.CREATE_NO_WINDOW)
|
|
out = (r.stdout or "") + (r.stderr or "")
|
|
code = r.returncode
|
|
except subprocess.TimeoutExpired:
|
|
out = "HARD-TIMEOUT"
|
|
code = -1
|
|
dt = time.time() - t0
|
|
m = re.search(r"RESULT: (.*)", out)
|
|
res = m.group(1).strip() if m else ""
|
|
if "WATCHDOG-KILL" in out: verdict = "HANG"
|
|
elif code == 0: verdict = "PASS"
|
|
elif code == 3: verdict = "FAIL"
|
|
elif "load=" in out and re.search(r"load=0", out): verdict = "LOAD-FAIL"
|
|
else: verdict = f"EXIT{code}"
|
|
# extract individual attach results if present
|
|
a = re.search(r"workerA openGUI inst1=(\d)", out)
|
|
b = re.search(r"workerB openGUI inst2=(\d)", out)
|
|
detail = f"a={a.group(1) if a else '?'} b={b.group(1) if b else '?'}"
|
|
return {"plugin": p, "verdict": verdict, "detail": detail, "secs": round(dt, 1), "res": res}
|
|
|
|
results = []
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as ex:
|
|
futs = {ex.submit(run_one, p): p for p in plugins}
|
|
done = 0
|
|
for f in concurrent.futures.as_completed(futs):
|
|
results.append(f.result())
|
|
done += 1
|
|
print(f"[{done}/{len(plugins)}] {results[-1]['verdict']:10s} {results[-1]['detail']} {results[-1]['secs']}s {os.path.basename(results[-1]['plugin'])}", flush=True)
|
|
|
|
results.sort(key=lambda x: (x["verdict"], x["plugin"]))
|
|
print("\n==== SUMMARY ====", flush=True)
|
|
for r in results:
|
|
print(f"{r['verdict']:10s} {r['detail']:8s} {r['secs']:5.1f}s {r['plugin']}", flush=True)
|
|
from collections import Counter
|
|
print("counts:", dict(Counter(r["verdict"] for r in results)), flush=True)
|
|
with open(os.path.join(HERE, f"sweep_{variant}_{secs}s.json"), "w", encoding="utf-8") as f:
|
|
json.dump(results, f, ensure_ascii=False, indent=2)
|