# app/core/fx_realtime.py """VST FX SHM realtime (cloud-FX loop): engine tạo mapping FxRealtimeIPC, spawn fx_vst_bridge --realtime-fx, đẩy/kéo block audio qua SHM ring. Layout mirror chính xác native_bridge/include/FxRealtimeIPC.h (mọi field u32/float 4-byte, không padding): header 12 u32 (48B): magic state sampleRate blockSize running heartbeat inWrite inRead inSlots outWrite outRead outSlots inL[4][256] inR[4][256] outL[8][256] outR[8][256] Luồng dữ liệu: browser → WS → write_input() → [bridge xử lý] → read_output() → WS → browser. Block = FXRT_BLOCK=256 stereo float32 = 2048 bytes. """ import ctypes import json import os import subprocess import tempfile import threading import time import uuid import numpy as np from app.config import settings from app.core.native_render import find_fx_bridge_exe FXRT_MAGIC = 0x46585254 FXRT_BLOCK = 256 FXRT_IN_SLOTS = 4 FXRT_OUT_SLOTS = 8 FXRT_STATE_STARTING = 0 FXRT_STATE_READY = 1 FXRT_STATE_ERROR = 2 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), ] HEADER_SIZE = ctypes.sizeof(FxRTHeader) # 48 IN_L_OFF = HEADER_SIZE # 48 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 class FxRealtimeSession: """Một session realtime: SHM + subprocess fx_vst_bridge + I/O block.""" def __init__(self, fx_chain, sample_rate, block_size=FXRT_BLOCK): from multiprocessing import shared_memory self.fx_chain = fx_chain or [] self.sample_rate = int(sample_rate) self.block_size = int(block_size or FXRT_BLOCK) if self.block_size != FXRT_BLOCK: raise ValueError("block_size phải = 256 (khớp C++ FxRealtimeIPC)") self.name = "SonicForge_FXRT_" + uuid.uuid4().hex[:12] self.shm = shared_memory.SharedMemory(name=self.name, create=True, size=TOTAL_SIZE) self.shm.buf[:TOTAL_SIZE] = b"\x00" * TOTAL_SIZE self.h = FxRTHeader.from_buffer(self.shm.buf) self.h.magic = FXRT_MAGIC self.h.state = FXRT_STATE_STARTING self.h.sample_rate = self.sample_rate self.h.block_size = self.block_size self.h.running = 1 self.h.in_slots = FXRT_IN_SLOTS self.h.out_slots = FXRT_OUT_SLOTS self._np = np.frombuffer(self.shm.buf, dtype=np.float32) self._lock = threading.Lock() self.proc = None self._job_path = "" self._started_at = time.time() self._spawn() # ── process ───────────────────────────────────────────────────────────── def _chain_json_for_bridge(self): """JSON mảng {path,bypass,preset_b64} — RealtimeFxChain::buildChain đọc đúng các field này (type/name bỏ qua).""" out = [] for s in self.fx_chain: if not s or not s.get("path"): continue out.append({ "path": s["path"], "bypass": bool(s.get("bypass")), "preset_b64": s.get("preset_b64") or "", }) return out def _spawn(self): exe = find_fx_bridge_exe() if not exe: raise RuntimeError("Không tìm thấy fx_vst_bridge.exe") job = { "sample_rate": self.sample_rate, "block_size": self.block_size, "fx_chain": self._chain_json_for_bridge(), } fd, self._job_path = tempfile.mkstemp(suffix=".json", prefix="fxrt_") with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(job, f) cmd = [exe, "--realtime-fx", self._job_path, "--shm", self.name, "--parent", str(os.getpid())] self.proc = subprocess.Popen( cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) def wait_ready(self, timeout=30.0): deadline = time.time() + timeout while time.time() < deadline: if self.proc.poll() is not None: raise RuntimeError( f"fx_vst_bridge thoát sớm (rc={self.proc.returncode})") if self.h.state == FXRT_STATE_READY: return True time.sleep(0.02) raise RuntimeError("fx_vst_bridge không sẵn sàng trong %ss" % timeout) def alive(self): return self.proc is not None and self.proc.poll() is None def close(self): try: self.h.running = 0 except Exception: pass if self.proc is not None: try: self.proc.wait(timeout=3) except Exception: try: self.proc.kill() except Exception: pass self.proc = None try: self.shm.close() self.shm.unlink() except Exception: pass if self._job_path and os.path.exists(self._job_path): try: os.remove(self._job_path) except Exception: pass # ── audio I/O ─────────────────────────────────────────────────────────── def write_input(self, interleaved): """Ghi 1 block stereo float32 (2048 bytes) vào input ring. Ring đầy → drop block cũ nhất (giữ latency thấp).""" arr = np.frombuffer(interleaved, dtype=np.float32) if arr.size != self.block_size * 2: return arr = arr.reshape(self.block_size, 2) with self._lock: h = self.h while True: if h.in_write - h.in_read < h.in_slots: slot = h.in_write & (h.in_slots - 1) bl = IN_L_OFF // 4 + slot * self.block_size br = IN_R_OFF // 4 + slot * self.block_size self._np[bl:bl + self.block_size] = arr[:, 0] self._np[br:br + self.block_size] = arr[:, 1] h.in_write += 1 return h.in_read += 1 # full → drop oldest input def read_output(self): """Drain output ring → list bytes (mỗi block 2048 bytes interleaved).""" out = [] with self._lock: h = self.h while h.out_read < h.out_write: slot = h.out_read & (h.out_slots - 1) bl = OUT_L_OFF // 4 + slot * self.block_size br = OUT_R_OFF // 4 + slot * self.block_size inter = np.empty(self.block_size * 2, dtype=np.float32) inter[0::2] = self._np[bl:bl + self.block_size] inter[1::2] = self._np[br:br + self.block_size] out.append(inter.tobytes()) h.out_read += 1 return out def heartbeat_age(self): return time.time() - self._started_at # placeholder: engine đọc h.heartbeat # Registry toàn tiến trình (desktop app 1 user). _SESSIONS = {} _SESSIONS_LOCK = threading.Lock() def start_session(fx_chain, sample_rate, block_size=FXRT_BLOCK): sess = FxRealtimeSession(fx_chain, sample_rate, block_size) try: sess.wait_ready(timeout=30) except Exception: sess.close() raise with _SESSIONS_LOCK: _SESSIONS[sess.name] = sess return sess def get_session(session_id): with _SESSIONS_LOCK: return _SESSIONS.get(session_id) def stop_session(session_id): with _SESSIONS_LOCK: sess = _SESSIONS.pop(session_id, None) if sess: sess.close() return sess is not None def stop_all(): with _SESSIONS_LOCK: ids = list(_SESSIONS.keys()) for sid in ids: stop_session(sid)