#!/usr/bin/env python3 """Measure out-ring glitches while GUI bridge runs with /frame + /preset load. Feeds sine into in ring (realtime pace), reads out ring continuously, detects discontinuities (sample jumps) in the output signal.""" 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 FRAME = sys.argv[3] if len(sys.argv) > 3 else "on" # "on" = poll /frame 30fps 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, "frame:", FRAME, 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] 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 — aborting", 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 — rc=", p2.poll(), flush=True); return url = "http://127.0.0.1:%d" % port print("gui port:", port, flush=True) # Feed sine at realtime pace; read out ring continuously; detect glitches. phase = [0.0] def feed(): 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[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) if FRAME == "on": threading.Thread(target=frame_loop, daemon=True).start() def reader(): last = h.out_write prev_samp = 0.0 prev_ok = False expected_next = None while not stop.is_set(): w = h.out_write while last < w: slot = last & (FXRT_OUT_SLOTS - 1) off = OUT_L_OFF + slot * FXRT_BLOCK blk = npbuf[off // 4:(off // 4) + FXRT_BLOCK].copy() # detect discontinuity: jump vs previous sample if prev_ok: for i in range(FXRT_BLOCK): if abs(blk[i] - prev_samp) > 0.25: glitches.append((time.time(), i)) 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() # /preset poll like frontend 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 t=%.2f idx=%d" % (g[0] - (time.time() - t0), 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()