#!/usr/bin/env python3 """Diagnostic harness: spawn realtime + GUI bridge, feed audio, poll /preset. Checks: (1) blob drift over time (would trigger chainKey change -> set_chain -> crackle), (2) /preset latency (HTTP worker block -> dead clicks).""" import ctypes, json, os, subprocess, tempfile, threading, time, uuid, urllib.request, sys import numpy as np from multiprocessing import shared_memory FXRT_MAGIC = 0x46585254 FXRT_BLOCK = 256 FXRT_IN_SLOTS = 4 FXRT_OUT_SLOTS = 8 FXRT_CTRL_SLOTS = 8 FXRT_LAT_SLOTS = 8 FXRT_STATE_STARTING = 0 FXRT_STATE_READY = 1 class FxRTHeader(ctypes.Structure): _fields_ = [("magic", ctypes.c_uint32), ("state", ctypes.c_uint32), ("sample_rate", ctypes.c_uint32), ("block_size", ctypes.c_uint32), ("running", ctypes.c_uint32), ("heartbeat", ctypes.c_uint32), ("in_write", ctypes.c_uint32), ("in_read", ctypes.c_uint32), ("in_slots", ctypes.c_uint32), ("out_write", ctypes.c_uint32), ("out_read", ctypes.c_uint32), ("out_slots", ctypes.c_uint32), ("ctrl_write", ctypes.c_uint32), ("ctrl_read", ctypes.c_uint32), ("ctrl_slots", ctypes.c_uint32), ("lat_write", ctypes.c_uint32), ("lat_read", ctypes.c_uint32), ("lat_slots", ctypes.c_uint32)] HEADER_SIZE = ctypes.sizeof(FxRTHeader) IN_L_OFF = HEADER_SIZE IN_R_OFF = IN_L_OFF + FXRT_IN_SLOTS * FXRT_BLOCK * 4 OUT_L_OFF = IN_R_OFF + FXRT_IN_SLOTS * FXRT_BLOCK * 4 OUT_R_OFF = OUT_L_OFF + FXRT_OUT_SLOTS * FXRT_BLOCK * 4 TOTAL_SIZE = OUT_R_OFF + FXRT_OUT_SLOTS * FXRT_BLOCK * 4 EXE = r"C:\Users\locpham\SonicForgeStudio\src-tauri\binaries\fx_vst_bridge.exe" PLUGIN = sys.argv[1] if len(sys.argv) > 1 else r"C:\Program Files\Common Files\VST3\WaveObserver.vst3" DURATION = float(sys.argv[2]) if len(sys.argv) > 2 else 30.0 def http_get(url, timeout=10): t0 = time.time() try: with urllib.request.urlopen(url, timeout=timeout) as r: body = r.read().decode() return time.time() - t0, body except Exception as e: return time.time() - t0, "ERR:" + str(e) def main(): name = "SonicForge_FXRT_" + uuid.uuid4().hex[:12] print("plugin:", PLUGIN, "shm:", name, flush=True) shm = shared_memory.SharedMemory(name=name, create=True, size=TOTAL_SIZE) shm.buf[:TOTAL_SIZE] = b"\x00" * TOTAL_SIZE h = FxRTHeader.from_buffer(shm.buf) h.magic = FXRT_MAGIC; h.state = FXRT_STATE_STARTING h.sample_rate = 44100; h.block_size = FXRT_BLOCK; h.running = 1 h.in_slots = FXRT_IN_SLOTS; h.out_slots = FXRT_OUT_SLOTS h.ctrl_slots = FXRT_CTRL_SLOTS; h.lat_slots = FXRT_LAT_SLOTS npbuf = np.frombuffer(shm.buf, dtype=np.float32) stop = threading.Event() procs = [] try: job1 = {"sample_rate": 44100, "block_size": 256, "seq": 0, "fx_chain": [{"type": "vst3", "path": PLUGIN, "name": "probe", "preset_b64": "", "bypass": False}]} fd, jp1 = tempfile.mkstemp(suffix=".json", prefix="fxrt_probe_") with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(job1, f) p1 = subprocess.Popen([EXE, "--realtime-fx", jp1, "--shm", name, "--parent", str(os.getpid())], creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) procs.append(p1) t0 = time.time() while time.time() - t0 < 30 and h.state != FXRT_STATE_READY: time.sleep(0.05) if p1.poll() is not None: print("REALTIME BRIDGE EXITED rc=", p1.returncode, flush=True); return print("realtime state:", h.state, "took %.2fs" % (time.time() - t0), flush=True) if h.state != FXRT_STATE_READY: print("realtime not ready — aborting", flush=True); return job2 = {"path": PLUGIN, "name": "probe", "shm": name, "preset_b64": ""} fd, jp2 = tempfile.mkstemp(suffix=".json", prefix="fxgui_probe_") with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(job2, f) p2 = subprocess.Popen([EXE, "--fx-gui", jp2], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) procs.append(p2) port = None t0 = time.time() while time.time() - t0 < 20: line = p2.stdout.readline() if not line: if p2.poll() is not None: break continue line = line.strip() if line.startswith("SF_FXGUI_PORT="): port = int(line.split("=", 1)[1]); break print("[gui]", line, flush=True) if not port: print("GUI bridge: no port — rc=", p2.poll(), flush=True); return url = "http://127.0.0.1:%d" % port print("gui port:", port, flush=True) def feed(): phase = 0.0 while not stop.is_set(): if h.in_write - h.in_read < FXRT_IN_SLOTS: slot = h.in_write & (FXRT_IN_SLOTS - 1) off = IN_L_OFF + slot * FXRT_BLOCK t = np.arange(FXRT_BLOCK) / 44100.0 + phase sine = 0.1 * np.sin(2 * np.pi * 440 * t).astype(np.float32) npbuf[off:off + FXRT_BLOCK] = sine npbuf[off + FXRT_IN_SLOTS * FXRT_BLOCK:off + FXRT_IN_SLOTS * FXRT_BLOCK + FXRT_BLOCK] = sine h.in_write += 1 phase += FXRT_BLOCK / 44100.0 time.sleep(0.003) th = threading.Thread(target=feed, daemon=True) th.start() # Poll /preset every 1.5s — same as frontend — measure drift + latency. last = None last16 = None drift_changes = 0 t0 = time.time() n = 0 while time.time() - t0 < DURATION: time.sleep(1.5) lat, body = http_get(url + "/preset") n += 1 import json as _j try: d = _j.loads(body) b64 = d.get("preset_b64", "") except Exception: print(" poll %d: bad json (%s)" % (n, body[:80]), flush=True) continue b16 = b64[:16] first_diff = (last16 is not None and b16 != last16) full_diff = (last is not None and b64 != last) if first_diff: drift_changes += 1 print(" poll %d lat=%.0fms blob_len=%d first16=%s %s%s%s" % ( n, lat * 1000, len(b64), b16, "FIRST16_CHANGED" if first_diff else "", " full_changed" if (full_diff and not first_diff) else "", " (NEW)" if last is None else ""), flush=True) last = b64; last16 = b16 print("RESULT: %d polls, %d first16 changes (would trigger set_chain -> crackle)" % (n, drift_changes), flush=True) finally: stop.set() for p in procs: try: p.terminate() except Exception: pass time.sleep(0.5) for p in procs: try: p.kill() except Exception: pass shm.close() try: shm.unlink() except Exception: pass if __name__ == "__main__": main()