SonicForgeStudio/fxrt_probe.py

# fxrt_probe.py — đo bridge realtime với VST thật (tạm, chẩn đoán crackle).
# Chạy: cd repo && python fxrt_probe.py [--secs 20]
#   - tạo session realtime (bắt stderr bridge ra fxrt_bridge_<tag>.log)
#   - producer: publish batch 4 block mỗi 21.333ms (timeBeginPeriod(1))
#   - consumer: drain output, ghi thời điểm từng block tới
#   - báo cáo: engine drops, bridge perf, output gap
import ctypes
import os
import sys
import threading
import time

import numpy as np

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from app.core import fx_realtime  # noqa: E402

SR = 48000
BLOCK = 256
BATCH = 4
BATCH_PERIOD = BATCH * BLOCK / SR  # 21.333ms
BLOCK_PERIOD = BLOCK / SR          # 5.333ms

IZ = r"C:\Program Files\Common Files\VST3\iZotope"
CHAIN_OZONE_MASTER = [
    {"type": "vst3", "path": IZ + r"\Ozone 11 Equalizer.vst3", "active": True},
    {"type": "vst3", "path": IZ + r"\Ozone 11 Dynamics.vst3", "active": True},
    {"type": "vst3", "path": IZ + r"\Ozone 11 Maximizer.vst3", "active": True},
]
CHAIN_OZONE_1 = [
    {"type": "vst3", "path": IZ + r"\Ozone 11 Maximizer.vst3", "active": True},
]
CHAIN_RX_DENOISE = [
    {"type": "vst3", "path": IZ + r"\RX 11 Spectral De-noise.vst3", "active": True},
]
CHAIN_EMPTY = []


def set_timer_res(ms=1):
    try:
        ctypes.windll.winmm.timeBeginPeriod(ms)
    except Exception:
        pass


def run_test(tag, chain, secs):
    log_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                            "fxrt_bridge_%s.log" % tag)
    real_popen = fx_realtime.subprocess.Popen

    def capped_popen(cmd, **kw):
        kw.pop("stderr", None)
        f = open(log_path, "ab", buffering=0)
        return real_popen(cmd, stderr=f, **kw)

    fx_realtime.subprocess.Popen = capped_popen
    try:
        sess = fx_realtime.start_session(chain, SR)
    finally:
        fx_realtime.subprocess.Popen = real_popen

    print("== %s: session=%s chain=%d slot(s)" % (tag, sess.name, len(chain)))

    stop = threading.Event()
    arr_times = []   # (block_idx, perf_counter) khi consumer đọc ra
    arr_lock = threading.Lock()

    def producer():
        # 1 block = 256 sample stereo = 512 float; gửi 4 block liên tiếp
        # (mô phỏng worklet burst), engine gom 4 → publish 1 batch.
        t = np.arange(BLOCK) / SR
        sig = 0.2 * np.sin(2 * np.pi * 220 * t)
        inter = np.empty(BLOCK * 2, dtype=np.float32)
        inter[0::2] = sig
        inter[1::2] = 0.9 * sig
        data = inter.tobytes()
        start = time.perf_counter()
        k = 0
        while not stop.is_set():
            for _ in range(BATCH):
                sess.write_input(data)
            k += 1
            nxt = start + k * BATCH_PERIOD
            d = nxt - time.perf_counter()
            if d > 0:
                time.sleep(d)

    def consumer():
        idx = 0
        while not stop.is_set():
            out = sess.read_output()
            if not out:
                time.sleep(0.001)
                continue
            now = time.perf_counter()
            with arr_lock:
                for _ in out:
                    arr_times.append((idx, now))
                    idx += 1

    pt = threading.Thread(target=producer)
    ct = threading.Thread(target=consumer)
    t0 = time.perf_counter()
    pt.start(); ct.start()
    time.sleep(secs)
    stop.set()
    pt.join(timeout=5); ct.join(timeout=5)

    stats = dict(sess.stats)
    sess.close()
    time.sleep(0.3)

    with arr_lock:
        times = list(arr_times)
    if len(times) > 1:
        gaps = [b - a for (i, a), (j, b) in zip(times[:-1], times[1:]) if b > a]
        dur = time.perf_counter() - t0
        rate = len(times) / dur
        exp = 1.0 / (BATCH / BLOCK_PERIOD)  # batch period
        print("  engine stats: %s" % stats)
        print("  output: %d block / %.1fs = %.1f blk/s (expect %.1f)"
              % (len(times), dur, rate, BLOCK / BLOCK_PERIOD))
        print("  gap: n=%d avg=%.3fms p95=%.3fms max=%.3fms"
              % (len(gaps), np.mean(gaps) * 1e3, np.percentile(gaps, 95) * 1e3,
                 max(gaps) * 1e3))
        over = [g * 1e3 for g in gaps if g > 1.35 * BLOCK_PERIOD]
        print("  gaps>%.1fms (underrun candidate): %d (%.2f%%)"
              % (1.35 * BLOCK_PERIOD * 1e3, len(over),
                 100.0 * len(over) / len(gaps) if gaps else 0.0))
    else:
        print("  NO OUTPUT (engine stats: %s)" % stats)

    print("  bridge log tail (%s):" % log_path)
    try:
        with open(log_path, "r", encoding="utf-8", errors="replace") as f:
            lines = f.readlines()
        for ln in lines[-12:]:
            print("    " + ln.rstrip())
    except Exception as e:
        print("    (no log: %s)" % e)


if __name__ == "__main__":
    set_timer_res(1)
    secs = 20
    if "--secs" in sys.argv:
        secs = int(sys.argv[sys.argv.index("--secs") + 1])
    tests = []
    if "--test" in sys.argv:
        t = sys.argv[sys.argv.index("--test") + 1]
        tests = [(t, {"ozone1": CHAIN_OZONE_1,
                      "ozone3": CHAIN_OZONE_MASTER,
                      "rx": CHAIN_RX_DENOISE,
                      "empty": CHAIN_EMPTY}[t])]
    else:
        tests = [("empty", CHAIN_EMPTY),
                 ("ozone1", CHAIN_OZONE_1),
                 ("ozone3", CHAIN_OZONE_MASTER),
                 ("rx", CHAIN_RX_DENOISE)]
    for tag, chain in tests:
        try:
            run_test(tag, chain, secs)
        except Exception as e:
            print("== %s FAILED: %r" % (tag, e))
            import traceback; traceback.print_exc()