#!/usr/bin/env python3 """Measure out-ring glitches while set_chain fires every 3s (preset apply), with GUI bridge + frame + preset poll running. set_chain is the only path that reloads a VST3 in the realtime process.""" 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 40.0 def http_get(url, timeout=5): try: with urllib.request.urlopen(url, timeout=timeout) as r: return r.read() except Exception as e: return b"ERR:" + str(e).encode() 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 = [] glitches = [] out_total = [0] jp1 = None 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 if h.state != FXRT_STATE_READY: print("realtime not ready", flush=True); return print("realtime ready in %.2fs" % (time.time() - t0), flush=True) 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 if not port: print("GUI bridge: no port", flush=True); return url = "http://127.0.0.1:%d" % port print("gui port:", port, flush=True) phase = [0.0] def feed(): while not stop.is_set(): if h.in_write - h.in_read < FXRT_IN_SLOTS: s = h.in_write & (FXRT_IN_SLOTS - 1) off = IN_L_OFF + s * FXRT_BLOCK t = np.arange(FXRT_BLOCK) / 44100.0 + phase[0] sine = (0.1 * np.sin(2 * np.pi * 440 * t)).astype(np.float32) npbuf[off // 4:(off // 4) + FXRT_BLOCK] = sine npbuf[(off + FXRT_IN_SLOTS * FXRT_BLOCK) // 4:(off + FXRT_IN_SLOTS * FXRT_BLOCK) // 4 + FXRT_BLOCK] = sine h.in_write += 1 phase[0] += FXRT_BLOCK / 44100.0 time.sleep(0.003) threading.Thread(target=feed, daemon=True).start() def frame_loop(): while not stop.is_set(): http_get(url + "/frame") time.sleep(1.0 / 30.0) threading.Thread(target=frame_loop, daemon=True).start() seq = [0] def setchain_loop(): while not stop.is_set(): time.sleep(3.0) try: with urllib.request.urlopen(url + "/preset", timeout=5) as r: d = json.loads(r.read().decode()) pb = d.get("preset_b64", "") except Exception: pb = "" seq[0] += 1 job = {"sample_rate": 44100, "block_size": 256, "seq": seq[0], "fx_chain": [{"type": "vst3", "path": PLUGIN, "name": "probe", "preset_b64": pb, "bypass": False}]} with open(jp1, "w", encoding="utf-8") as f: json.dump(job, f) print(" set_chain seq=%d preset_len=%d at %.2fs" % (seq[0], len(pb), time.time() - t0), flush=True) threading.Thread(target=setchain_loop, daemon=True).start() def reader(): last = h.out_write prev_samp = 0.0 prev_ok = False while not stop.is_set(): w = h.out_write while last < w: s = last & (FXRT_OUT_SLOTS - 1) off = OUT_L_OFF + s * FXRT_BLOCK blk = npbuf[off // 4:(off // 4) + FXRT_BLOCK].copy() if prev_ok: for i in range(FXRT_BLOCK): if abs(blk[i] - prev_samp) > 0.25: glitches.append((time.time(), i, last)) break prev_samp = blk[-1] prev_ok = True out_total[0] += FXRT_BLOCK last += 1 time.sleep(0.001) threading.Thread(target=reader, daemon=True).start() t0 = time.time() n = 0 while time.time() - t0 < DURATION: time.sleep(1.5) http_get(url + "/preset") n += 1 stop.set() time.sleep(0.3) print("RESULT: %d polls, out blocks=%d, glitches=%d" % (n, out_total[0], len(glitches)), flush=True) for g in glitches[:10]: print(" glitch block=%d idx=%d" % (g[2], g[1]), 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() if __name__ == "__main__": main()