# fxrt_toggle_probe.py — đo toggle bypass dry->wet với VST instance cache. # Kịch bản THẬT: start với slot ACTIVE (load + cache) -> toggle bypass=true # (chain đổi, KHÔNG unload) -> toggle bypass=false (cache hit, không reload) # Đo: thời gian set_chain -> latency report (chain applied) mỗi toggle. # python fxrt_toggle_probe.py [--port 8012] import asyncio import ctypes import json import os import subprocess import sys import tempfile import time import numpy as np REPO = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, REPO) SR = 48000 BLOCK = 256 BATCH = 4 BATCH_PERIOD = BATCH * BLOCK / SR # 21.333ms ADELAY = os.path.join(REPO, "native_bridge", "build", "VST3", "Release", "adelay.vst3") OZONE_MAX = r"C:\Program Files\Common Files\VST3\iZotope\Ozone 11 Maximizer.vst3" PLUGIN = os.environ.get("FXRT_PROBE_PLUGIN", OZONE_MAX) # mặc định Ozone (latency thật 747smp) def make_block(): t = np.arange(BLOCK) / SR amp = float(os.environ.get("FXRT_PROBE_AMP", "1.0")) sig = amp * (0.2 * np.sin(2 * np.pi * 220 * t) + 0.1 * np.sin(2 * np.pi * 331 * t)) inter = np.empty(BLOCK * 2, dtype=np.float32) inter[0::2] = sig inter[1::2] = 0.85 * sig return inter.tobytes() def chain_with(bypass): return [{"type": "vst3", "path": PLUGIN, "name": os.path.basename(PLUGIN), "preset_b64": "", "bypass": bypass}] async def probe(port): import websockets import urllib.request base = "http://127.0.0.1:%d" % port block_bytes = make_block() def post(path, obj): req = urllib.request.Request( base + path, data=json.dumps(obj).encode(), headers={"Content-Type": "application/json"}, method="POST") with urllib.request.urlopen(req, timeout=60) as r: return json.loads(r.read().decode()) st = post("/api/v1/plugins/fx-realtime/start", {"fx_chain": chain_with(False), "sample_rate": SR}) sid = st["session_id"] print("session=%s started slot ACTIVE (load + cache)" % sid) ws_url = "ws://127.0.0.1:%d/api/v1/plugins/ws/fx-realtime/%s" % (port, sid) lat_events = [] # (time, seq) stop = asyncio.Event() started = asyncio.Event() async def producer(): next_t = time.perf_counter() + 0.3 while not stop.is_set(): await asyncio.to_thread(time.sleep, max(0, next_t - time.perf_counter())) if started.is_set(): await ws.send(block_bytes) next_t += BATCH_PERIOD async def consumer(): while not stop.is_set(): try: msg = await ws.recv() except Exception: break if not isinstance(msg, bytes): try: d = json.loads(msg) except Exception: continue if isinstance(d, dict) and d.get("cmd") == "latency": lat_events.append((time.perf_counter(), d.get("total"))) async with websockets.connect(ws_url) as ws: tasks = [asyncio.create_task(producer()), asyncio.create_task(consumer())] await asyncio.sleep(0.5) started.set() await asyncio.sleep(1.5) # để start-load + latency report đầu n0 = len(lat_events) print("baseline latency reports:", lat_events[n0 - 3:n0]) # toggle 1: bypass=true (chain đổi, instance GIỮ trong cache) t0 = time.perf_counter() post("/api/v1/plugins/fx-realtime/set-chain", {"session_id": sid, "fx_chain": chain_with(True)}) await asyncio.sleep(0.6) t1 = time.perf_counter() new = [x for x in lat_events if x[0] >= t0] print("TOGGLE1 bypass=true -> chain applied in %s (%.0fms)" % ("%.0fms" % ((new[0][0] - t0) * 1e3) if new else "NOT SEEN", 0)) # toggle 2: bypass=false (CACHE HIT — không reload) t2 = time.perf_counter() post("/api/v1/plugins/fx-realtime/set-chain", {"session_id": sid, "fx_chain": chain_with(False)}) await asyncio.sleep(0.6) t3 = time.perf_counter() new2 = [x for x in lat_events if x[0] >= t2] print("TOGGLE2 bypass=false -> chain applied in %s" % ("%.0fms" % ((new2[0][0] - t2) * 1e3) if new2 else "NOT SEEN")) # toggle 3: bypass=true lần 2 (cache hit) t4 = time.perf_counter() post("/api/v1/plugins/fx-realtime/set-chain", {"session_id": sid, "fx_chain": chain_with(True)}) await asyncio.sleep(0.6) new3 = [x for x in lat_events if x[0] >= t4] print("TOGGLE3 bypass=true -> chain applied in %s" % ("%.0fms" % ((new3[0][0] - t4) * 1e3) if new3 else "NOT SEEN")) stop.set() await asyncio.sleep(0.3) for t in tasks: t.cancel() try: post("/api/v1/plugins/fx-realtime/stop", {"session_id": sid}) except Exception: pass print("all latency events:", ["%.2fs/%.0f" % (x[0] - lat_events[0][0], x[1] or 0) for x in lat_events]) def main(): try: ctypes.windll.winmm.timeBeginPeriod(1) except Exception: pass port = 8012 if "--port" in sys.argv: port = int(sys.argv[sys.argv.index("--port") + 1]) storage = tempfile.mkdtemp(prefix="sf_toggleprobe_") env = dict(os.environ) env["SONICFORGE_STORAGE_DIR"] = storage env["FXRT_BRIDGE_LOG"] = os.path.join(REPO, "fxrt_toggle_bridge_%d.log" % port) logf = open(os.path.join(REPO, "fxrt_toggle_engine_%d.log" % port), "ab", buffering=0) boot = ("import ctypes; ctypes.windll.winmm.timeBeginPeriod(1); " "import uvicorn, app.main; " "uvicorn.run(app.main.app, host='127.0.0.1', port=%d, log_level='warning')" % port) engine = subprocess.Popen( [sys.executable, "-c", boot], cwd=REPO, env=env, stdout=logf, stderr=subprocess.STDOUT, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) try: import urllib.request deadline = time.time() + 60 while time.time() < deadline: try: with urllib.request.urlopen("http://127.0.0.1:%d/health" % port, timeout=2) as r: if r.status == 200: break except Exception: time.sleep(0.5) else: print("engine not up; log tail:") print(open(logf.name, "rb").read().decode(errors="replace")[-2000:]) return print("engine up on :%d" % port) asyncio.run(probe(port)) finally: engine.terminate() try: engine.wait(timeout=5) except Exception: engine.kill() logf.close() if __name__ == "__main__": main()