659 lines
28 KiB
Python
659 lines
28 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Phase 5 verify — TASKS_DAW_A.md §39-47 (chuẩn DAW checklist).
|
||
|
||
5.3 Golden live=export: realtime SHM loop vs offline --render-fx cùng chain.
|
||
5.4 Order: chain [eq, vst3-delay, limiter] giữ thứ tự (contract) + impulse
|
||
qua adelay (VST delay áp giữa chain, không gom cuối, không bỏ).
|
||
5.5 Fader: master chain active + fader -6dB → đỉnh giảm 2x, chain vẫn áp.
|
||
5.6 PDC: slot latency_samples=1024 + audio bị delay 1024 → bù track → xuyên
|
||
pha hết (sum = 2×B); control không PDC → lệch pha.
|
||
5.7 Smoke: /capabilities + render master imager+maximizer+vst3.
|
||
|
||
Skip nếu thiếu fx_vst_bridge.exe (native_bridge/build/Release) — ngoại trừ
|
||
các test contract thuần (không spawn bridge).
|
||
"""
|
||
import io
|
||
import json
|
||
import math
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import wave
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
sys.path.insert(0, REPO)
|
||
|
||
FX_BRIDGE = os.path.join(REPO, "native_bridge", "build", "Release", "fx_vst_bridge.exe")
|
||
DAW_BRIDGE = os.path.join(REPO, "native_bridge", "build", "Release", "daw_vst_bridge.exe")
|
||
ADELAY = os.path.join(REPO, "native_bridge", "build", "VST3", "Release", "adelay.vst3")
|
||
SR = 44100
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def fx_bridge():
|
||
if not os.path.isfile(FX_BRIDGE):
|
||
pytest.skip("fx_vst_bridge.exe chưa build (native_bridge/build/Release)")
|
||
return FX_BRIDGE
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def adelay():
|
||
if not os.path.isfile(os.path.join(ADELAY, "Contents", "x86_64-win", "adelay.vst3")):
|
||
pytest.skip("adelay.vst3 sample chưa build (native_bridge/build/VST3/Release)")
|
||
return ADELAY
|
||
|
||
|
||
def _write_wav(path, data, sr):
|
||
pcm = (np.clip(data, -1.0, 1.0).T * 32767).astype(np.int16)
|
||
with wave.open(path, "wb") as w:
|
||
w.setnchannels(2)
|
||
w.setsampwidth(2)
|
||
w.setframerate(sr)
|
||
w.writeframes(pcm.tobytes())
|
||
|
||
|
||
def _read_wav(path):
|
||
import soundfile as sf
|
||
data, sr = sf.read(path, dtype="float32", always_2d=True)
|
||
return data.T, sr
|
||
|
||
|
||
def _render_bridge(bridge, wav_in, sr, chain, block=256):
|
||
job = {"sample_rate": sr, "block_size": block, "fx_chain": chain}
|
||
with tempfile.TemporaryDirectory() as td:
|
||
job_path = os.path.join(td, "job.json")
|
||
out_path = os.path.join(td, "out.wav")
|
||
with io.open(job_path, "w", encoding="utf-8") as f:
|
||
json.dump(job, f)
|
||
r = subprocess.run([bridge, "--render-fx", job_path, "--in", wav_in,
|
||
"--out", out_path],
|
||
capture_output=True, text=True, timeout=180)
|
||
assert r.returncode == 0, f"bridge rc={r.returncode}: {r.stdout[-500:]} {r.stderr[-500:]}"
|
||
assert os.path.isfile(out_path), f"no out wav: {r.stdout[-500:]} {r.stderr[-500:]}"
|
||
return _read_wav(out_path)[0]
|
||
|
||
|
||
def _snr_db(ref, got):
|
||
ref = ref.astype(np.float64)
|
||
got = got.astype(np.float64)
|
||
n = min(ref.shape[1], got.shape[1])
|
||
if n == 0:
|
||
return -float("inf")
|
||
ref, got = ref[:, :n], got[:, :n]
|
||
denom = float(np.sum((ref - got) ** 2))
|
||
if denom < 1e-12:
|
||
return float("inf")
|
||
return 10.0 * math.log10(float(np.sum(ref ** 2)) / denom)
|
||
|
||
|
||
def _find_delay(live, ref, max_lag=2048):
|
||
a = live[0].astype(np.float64)
|
||
b = ref[0].astype(np.float64)
|
||
n = min(a.shape[0], b.shape[0])
|
||
a, b = a[:n], b[:n]
|
||
corr = np.correlate(a, b, mode="same")
|
||
lag = int(np.argmax(np.abs(corr))) - n // 2
|
||
if abs(lag) > max_lag:
|
||
return 0
|
||
return lag
|
||
|
||
|
||
def _midi_track():
|
||
return {
|
||
"id": "t1",
|
||
"type": "MIDI",
|
||
"instrument_id": "__no_such_plugin__",
|
||
"instrument_source": "vsti",
|
||
"fx_type": None,
|
||
"volume_db": 0.0,
|
||
"pan": 0.0,
|
||
"mute": False,
|
||
"items": [{
|
||
"type": "MIDI_ITEM",
|
||
"start_bar": 0.0,
|
||
"duration_bars": 1.0,
|
||
"clip_start_offset_bars": 0.0,
|
||
"source_data": {"notes": [
|
||
{"pitch": 60, "velocity": 0.8, "start_beat": 0.0, "duration_beats": 1.0}
|
||
]},
|
||
}],
|
||
}
|
||
|
||
|
||
def _side_rms(path):
|
||
import soundfile as sf
|
||
data, _ = sf.read(path, dtype="float32", always_2d=True)
|
||
side = data[:, 0] - data[:, 1]
|
||
return float(np.sqrt(np.mean(side ** 2)))
|
||
|
||
|
||
# ── 5.3 Golden live=export ────────────────────────────────────────────────
|
||
def test_53_live_equals_export(fx_bridge, monkeypatch):
|
||
"""Realtime SHM loop vs offline --render-fx, cùng chain builtin, block 256
|
||
→ SNR ≥ 30 dB (cùng engine C++, live = export)."""
|
||
monkeypatch.setenv("SF_FX_BRIDGE_PATH", fx_bridge)
|
||
from app.core import fx_realtime
|
||
|
||
chain = [
|
||
{"type": "builtin", "id": "eq",
|
||
"params": {"g1": 3.0, "g2": -2.0, "g3": 1.0, "g4": 0.5}},
|
||
{"type": "builtin", "id": "compressor",
|
||
"params": {"threshold": -18.0, "ratio": 3.0, "makeup": 2.0}},
|
||
{"type": "builtin", "id": "limiter", "params": {"ceiling": -3.0}},
|
||
]
|
||
n = 172 * 256 # bội của block 256
|
||
t = np.arange(n) / SR
|
||
rng = np.random.default_rng(7)
|
||
sig = (0.2 * np.sin(2 * np.pi * 220 * t) + 0.1 * np.sin(2 * np.pi * 1100 * t)
|
||
+ 0.02 * rng.standard_normal(n))
|
||
audio = np.stack([sig, 0.9 * sig]).astype(np.float32)
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
wav_in = os.path.join(td, "in.wav")
|
||
_write_wav(wav_in, audio, SR)
|
||
offline = _render_bridge(fx_bridge, wav_in, SR, chain, block=256)
|
||
|
||
sess = fx_realtime.start_session(chain, SR)
|
||
try:
|
||
out = bytearray()
|
||
nb = n // 256
|
||
for i in range(nb):
|
||
blk = audio[:, i * 256:(i + 1) * 256]
|
||
inter = np.empty(512, dtype=np.float32)
|
||
inter[0::2] = blk[0]
|
||
inter[1::2] = blk[1]
|
||
sess.write_input(inter.tobytes())
|
||
time.sleep(0.006)
|
||
for b in sess.read_output():
|
||
out += b
|
||
time.sleep(0.05)
|
||
for b in sess.read_output():
|
||
out += b
|
||
finally:
|
||
sess.close()
|
||
arr = np.frombuffer(bytes(out), dtype=np.float32)
|
||
assert arr.size > 0, "realtime loop không ra output"
|
||
live = arr.reshape(-1, 2).T
|
||
assert live.shape[1] >= n // 2, f"live quá ngắn: {live.shape}"
|
||
|
||
lag = _find_delay(live, offline)
|
||
if lag >= 0:
|
||
got = live[:, lag:lag + offline.shape[1]]
|
||
else:
|
||
got = live[:, :offline.shape[1] + lag]
|
||
n_cmp = min(offline.shape[1], got.shape[1]) - 8
|
||
assert n_cmp > 1000, "đoạn so sánh quá ngắn"
|
||
snr = _snr_db(offline[:, :n_cmp], got[:, :n_cmp])
|
||
assert snr >= 30.0, f"live≠export SNR {snr:.1f} dB (lag {lag})"
|
||
|
||
|
||
# ── 5.4 Order ─────────────────────────────────────────────────────────────
|
||
def test_54_chain_order_contract():
|
||
"""Chain [eq, vst3-delay, limiter] giữ NGUYÊN thứ tự qua merge offline +
|
||
live JSON — không gom VST cuối (regression cho "không gom cuối")."""
|
||
from app.core import fx_realtime
|
||
from app.core.render_engine import _merge_track_fx_chain
|
||
|
||
track = {
|
||
"id": "t1",
|
||
"fx_chain": [
|
||
{"type": "builtin", "id": "eq", "params": {"g1": 3.0}},
|
||
{"type": "vst3", "path": ADELAY, "active": True},
|
||
{"type": "builtin", "id": "limiter", "params": {"ceiling": -3.0}},
|
||
],
|
||
"vst_fx_chain": [],
|
||
"vst_fx_bypass": False,
|
||
}
|
||
merged = _merge_track_fx_chain(track)
|
||
assert [s.get("id") or s.get("type") for s in merged] == ["eq", "vst3", "limiter"]
|
||
|
||
s = object.__new__(fx_realtime.FxRealtimeSession)
|
||
s.chain = list(merged)
|
||
s._lock = None
|
||
out = s._chain_json_for_bridge()
|
||
assert [o["type"] for o in out] == ["builtin", "vst3", "builtin"]
|
||
assert out[1]["path"] == ADELAY
|
||
|
||
|
||
def test_54_vst3_delay_in_middle_audio(fx_bridge, adelay):
|
||
"""Impulse qua [eq, adelay(default 1s), limiter] → peak tại ≈44100, im lặng
|
||
trước đó; SNR vs reference Python (eq → shift 44100 → limiter) ≥ 18 dB —
|
||
(EQ Python vs C++ impulse response lệch ~22 dB; check order thật là vị trí
|
||
peak + pre-silence) — VST delay áp giữa chain (sau EQ, trước limiter)."""
|
||
from app.core.render_engine import _apply_builtin_fx_chain
|
||
|
||
chain = [
|
||
{"type": "builtin", "id": "eq",
|
||
"params": {"g1": 3.0, "g2": -2.0, "g3": 1.0, "g4": 0.5}},
|
||
{"type": "vst3", "path": adelay},
|
||
{"type": "builtin", "id": "limiter", "params": {"ceiling": -3.0}},
|
||
]
|
||
n = int(SR * 1.5)
|
||
audio = np.zeros((2, n), dtype=np.float32)
|
||
audio[0, 0] = 1.0
|
||
audio[1, 0] = 1.0
|
||
with tempfile.TemporaryDirectory() as td:
|
||
wav_in = os.path.join(td, "in.wav")
|
||
_write_wav(wav_in, audio, SR)
|
||
got = _render_bridge(fx_bridge, wav_in, SR, chain, block=256)
|
||
|
||
idx = int(np.unravel_index(np.argmax(np.abs(got)), got.shape)[1])
|
||
assert abs(idx - SR) < 200, f"peak tại {idx} — mong đợi ~{SR} (delay 1s)"
|
||
assert float(np.max(np.abs(got[:, :int(SR * 0.9)]))) == 0.0, "có tín hiệu trước delay"
|
||
|
||
ref = _apply_builtin_fx_chain(
|
||
audio.copy(),
|
||
[{"type": "eq", "params": {"g1": 3.0, "g2": -2.0, "g3": 1.0, "g4": 0.5}}], SR)
|
||
ref = np.roll(ref, SR, axis=1)
|
||
ref = _apply_builtin_fx_chain(ref, [{"type": "limiter", "params": {"ceiling": -3.0}}], SR)
|
||
n_cmp = min(ref.shape[1], got.shape[1]) - 8
|
||
snr = _snr_db(ref[:, :n_cmp], got[:, :n_cmp])
|
||
assert snr >= 18.0, f"SNR {snr:.1f} dB"
|
||
|
||
|
||
# ── 5.5 Fader sau master chain ────────────────────────────────────────────
|
||
def test_55_fader_after_master_chain(monkeypatch, tmp_path, fx_bridge):
|
||
"""Master chain active (gain +6dB) + fader -6dB → đỉnh giảm 2x, chain vẫn
|
||
áp (đỉnh fader-6 ≈ base, đỉnh fader0 ≈ 2×base) — fader SAU chain."""
|
||
monkeypatch.setenv("SF_FX_BRIDGE_PATH", fx_bridge)
|
||
if not os.path.isfile(DAW_BRIDGE):
|
||
pytest.skip("daw_vst_bridge.exe chưa build")
|
||
monkeypatch.setenv("SF_BRIDGE_PATH", DAW_BRIDGE)
|
||
from app.api.v1 import plugins as pl
|
||
monkeypatch.setattr(pl.settings, "RENDER_ENGINE", "bridge")
|
||
from app.core.render_engine import PythonRenderEngine
|
||
engine = PythonRenderEngine()
|
||
|
||
def project(master_vol, chain):
|
||
track = _midi_track()
|
||
track["volume_db"] = -20.0 # tránh normalize (>1.0)
|
||
return {
|
||
"metadata": {"bpm": 120.0, "time_signature_numerator": 4},
|
||
"main_session": {
|
||
"length_bars": 1.0,
|
||
"tracks": [track],
|
||
"master": {
|
||
"fx_chain": chain,
|
||
"bypass": False,
|
||
"volume_db": master_vol,
|
||
},
|
||
},
|
||
"section_store": {},
|
||
}
|
||
|
||
GAIN = [{"type": "builtin", "id": "gain", "params": {"db": 6.0}}]
|
||
|
||
def peak(name):
|
||
chain = {"base.wav": [], "0.wav": GAIN, "6.wav": GAIN}[name]
|
||
vol = {"base.wav": 0.0, "0.wav": 0.0, "6.wav": -6.0}[name]
|
||
out = str(tmp_path / name)
|
||
engine.render_project(project(vol, chain), out)
|
||
data, _ = _read_wav(out)
|
||
return float(np.max(np.abs(data)))
|
||
|
||
base = peak("base.wav")
|
||
p0 = peak("0.wav")
|
||
p6 = peak("6.wav")
|
||
assert p0 > 1.7 * base, f"chain +6dB không áp: {p0:.4f} vs base {base:.4f}"
|
||
assert 0.4 < p6 / p0 < 0.6, f"fader -6dB phải ~1/2 đỉnh: {p6:.4f}/{p0:.4f}"
|
||
assert 0.8 < p6 / base < 1.2, f"fader-6 phải ≈ base: {p6:.4f} vs {base:.4f}"
|
||
|
||
# ── 5.8 Master L/R volume + LINK + MUTE ─────────────────────────────────
|
||
def test_58_master_lr_link_mute(monkeypatch, tmp_path, fx_bridge):
|
||
"""Master strip L/R: volumeL_db/volumeR_db tách kênh, link_lr → R theo L,
|
||
muted → zero output. Fallback volume_db (project cũ) vẫn chạy."""
|
||
monkeypatch.setenv("SF_FX_BRIDGE_PATH", fx_bridge)
|
||
if not os.path.isfile(DAW_BRIDGE):
|
||
pytest.skip("daw_vst_bridge.exe chưa build")
|
||
monkeypatch.setenv("SF_BRIDGE_PATH", DAW_BRIDGE)
|
||
from app.api.v1 import plugins as pl
|
||
monkeypatch.setattr(pl.settings, "RENDER_ENGINE", "bridge")
|
||
from app.core.render_engine import PythonRenderEngine
|
||
engine = PythonRenderEngine()
|
||
|
||
def project(master):
|
||
track = _midi_track()
|
||
track["volume_db"] = -20.0 # tránh normalize (>1.0)
|
||
return {
|
||
"metadata": {"bpm": 120.0, "time_signature_numerator": 4},
|
||
"main_session": {
|
||
"length_bars": 1.0,
|
||
"tracks": [track],
|
||
"master": dict({"fx_chain": [], "bypass": False}, **master),
|
||
},
|
||
"section_store": {},
|
||
}
|
||
|
||
def render(name, master):
|
||
out = str(tmp_path / name)
|
||
engine.render_project(project(master), out)
|
||
data, _ = _read_wav(out)
|
||
return data
|
||
|
||
base = render("base.wav", {"volumeL_db": 0.0, "volumeR_db": 0.0, "link_lr": True})
|
||
baseL = float(np.max(np.abs(base[0])))
|
||
baseR = float(np.max(np.abs(base[1])))
|
||
assert baseL > 0
|
||
|
||
lr = render("lr.wav", {"volumeL_db": -6.0, "volumeR_db": 0.0, "link_lr": False})
|
||
assert 0.4 < float(np.max(np.abs(lr[0]))) / baseL < 0.6, f"L -6dB phải ~1/2 đỉnh: {float(np.max(np.abs(lr[0]))):.4f}/{baseL:.4f}"
|
||
assert abs(float(np.max(np.abs(lr[1]))) - baseR) / baseR < 0.15, "R phải giữ nguyên khi link_lr=false"
|
||
|
||
link = render("link.wav", {"volumeL_db": -6.0, "volumeR_db": -3.0, "link_lr": True})
|
||
ratio = float(np.max(np.abs(link[1]))) / float(np.max(np.abs(link[0])))
|
||
assert 0.9 < ratio < 1.1, f"link_lr: R phải theo L ({ratio:.3f})"
|
||
|
||
muted = render("muted.wav", {"volumeL_db": 0.0, "volumeR_db": 0.0,
|
||
"link_lr": True, "muted": True})
|
||
assert float(np.max(np.abs(muted))) < 1e-6, "MUTE master phải zero output"
|
||
|
||
legacy = render("legacy.wav", {"volume_db": -6.0})
|
||
assert 0.4 < float(np.max(np.abs(legacy[0]))) / baseL < 0.6 and 0.4 < float(np.max(np.abs(legacy[1]))) / baseR < 0.6, "fallback volume_db (project cũ) phải áp cả 2 kênh"
|
||
|
||
|
||
# ── 5.6 PDC ───────────────────────────────────────────────────────────────
|
||
def test_56_pdc_aligns_tracks(monkeypatch, tmp_path, fx_bridge):
|
||
"""Plugin latency 1024 (slot latency_samples) + audio thật bị delay 1024 →
|
||
PDC bù track ngắn → sum = 2×B (xuyên pha hết). Control latency=0 → sum
|
||
lệch pha (SNR thấp)."""
|
||
monkeypatch.setenv("SF_FX_BRIDGE_PATH", fx_bridge)
|
||
from app.api.v1 import plugins as pl
|
||
monkeypatch.setattr(pl.settings, "RENDER_ENGINE", "bridge")
|
||
from app.core.render_engine import PythonRenderEngine
|
||
engine = PythonRenderEngine()
|
||
|
||
n = SR * 2 # 1 bar @120bpm 4/4 = 88200
|
||
t = np.arange(n) / SR
|
||
w = (0.2 * np.sin(2 * np.pi * 220 * t) + 0.1 * np.sin(2 * np.pi * 1100 * t)).astype(np.float32)
|
||
ws = np.zeros_like(w)
|
||
ws[1024:] = w[:n - 1024]
|
||
fa = str(tmp_path / "a.wav")
|
||
fb = str(tmp_path / "b.wav")
|
||
_write_wav(fa, np.stack([w, w]), SR)
|
||
_write_wav(fb, np.stack([ws, ws]), SR)
|
||
|
||
def project(lat):
|
||
def track(path, chain):
|
||
return {
|
||
"id": path,
|
||
"type": "AUDIO",
|
||
"volume_db": 0.0,
|
||
"pan": 0.0,
|
||
"fx_chain": chain,
|
||
"vst_fx_chain": [],
|
||
"vst_fx_bypass": False,
|
||
"items": [{
|
||
"type": "AUDIO_ITEM",
|
||
"start_bar": 0.0,
|
||
"duration_bars": 1.0,
|
||
"clip_start_offset_bars": 0.0,
|
||
"source_data": {"audio_file_url": path},
|
||
}],
|
||
}
|
||
chain_b = []
|
||
if lat:
|
||
# bypass passthrough — chỉ khai latency cho PDC (spec §3 field)
|
||
chain_b = [{"type": "vst3", "path": "", "bypass": True,
|
||
"active": True, "latency_samples": lat}]
|
||
return {
|
||
"metadata": {"bpm": 120.0, "time_signature_numerator": 4},
|
||
"main_session": {
|
||
"length_bars": 1.0,
|
||
"tracks": [track(fa, []), track(fb, chain_b)],
|
||
},
|
||
"section_store": {},
|
||
}
|
||
|
||
out_pdc = str(tmp_path / "pdc.wav")
|
||
out_nopdc = str(tmp_path / "nopdc.wav")
|
||
engine.render_project(project(1024), out_pdc)
|
||
engine.render_project(project(0), out_nopdc)
|
||
d_pdc, _ = _read_wav(out_pdc)
|
||
d_nopdc, _ = _read_wav(out_nopdc)
|
||
|
||
expected = np.stack([ws, ws]) * 2.0 # PDC delay track A → cùng pha B
|
||
n_cmp = n - 16
|
||
snr_pdc = _snr_db(expected[:, :n_cmp], d_pdc[:, :n_cmp])
|
||
snr_nopdc = _snr_db(expected[:, :n_cmp], d_nopdc[:, :n_cmp])
|
||
assert snr_pdc >= 30.0, f"PDC sum lệch: SNR {snr_pdc:.1f} dB"
|
||
assert snr_nopdc < 15.0, f"control không lệch pha: SNR {snr_nopdc:.1f} dB"
|
||
|
||
|
||
# ── 5.7 Smoke ─────────────────────────────────────────────────────────────
|
||
def test_57_capabilities_smoke():
|
||
"""Server /capabilities phản hồi OK (smoke API)."""
|
||
from fastapi.testclient import TestClient
|
||
from app.main import app
|
||
r = TestClient(app).get("/api/v1/system/capabilities")
|
||
assert r.status_code == 200, r.text[:300]
|
||
assert isinstance(r.json(), dict)
|
||
|
||
|
||
def test_57_master_chain_imager_maximizer_vst3(monkeypatch, tmp_path, fx_bridge, adelay):
|
||
"""Render master imager+maximizer+vst3 (adelay) qua bridge — không crash,
|
||
non-silent; imager vẫn tác dụng khi chain có vst3."""
|
||
monkeypatch.setenv("SF_FX_BRIDGE_PATH", fx_bridge)
|
||
if not os.path.isfile(DAW_BRIDGE):
|
||
pytest.skip("daw_vst_bridge.exe chưa build")
|
||
monkeypatch.setenv("SF_BRIDGE_PATH", DAW_BRIDGE)
|
||
from app.api.v1 import plugins as pl
|
||
monkeypatch.setattr(pl.settings, "RENDER_ENGINE", "bridge")
|
||
from app.core.render_engine import PythonRenderEngine
|
||
engine = PythonRenderEngine()
|
||
|
||
def render(chain, name):
|
||
track = _midi_track()
|
||
track["volume_db"] = -20.0
|
||
track["pan"] = 0.6
|
||
proj = {
|
||
"metadata": {"bpm": 120.0, "time_signature_numerator": 4},
|
||
"main_session": {"length_bars": 1.0, "tracks": [track]},
|
||
"section_store": {},
|
||
}
|
||
if chain is not None:
|
||
proj["mastering_settings"] = {
|
||
"chain": chain,
|
||
"w1": 0, "w2": 0, "w3": 0, "w4": 0,
|
||
"maxGain": 0.0, "maxSoftClip": False, "maxUpward": False,
|
||
"ceiling": -1.0,
|
||
}
|
||
out = str(tmp_path / name)
|
||
engine.render_project(proj, out)
|
||
return out
|
||
|
||
base = _side_rms(render(None, "base.wav"))
|
||
s0 = _side_rms(render([
|
||
{"id": "imager", "type": "imager", "name": "Imager", "active": True},
|
||
{"type": "vst3", "path": adelay},
|
||
{"id": "maximizer", "type": "maximizer", "name": "Max", "active": True},
|
||
], "im0.wav"))
|
||
s2 = _side_rms(render([
|
||
{"id": "imager", "type": "imager", "name": "Imager", "active": True,
|
||
"params": {"w1": 200, "w2": 200, "w3": 200, "w4": 200}},
|
||
{"type": "vst3", "path": adelay},
|
||
{"id": "maximizer", "type": "maximizer", "name": "Max", "active": True},
|
||
], "im200.wav"))
|
||
assert base > 0.001, "pan tạo side content"
|
||
assert s0 < 0.05 * base, f"imager w=0 không giết side: {s0:.5f} vs {base:.5f}"
|
||
assert s2 > 1.5 * base, f"imager w=200 không tăng side: {s2:.5f} vs {base:.5f}"
|
||
|
||
def test_59_limiter_brickwall_oversample(fx_bridge):
|
||
"""D2: LimiterFx mode brickwall — lookahead + oversample 4x + hard clip.
|
||
Impulse 0dB qua ceiling -1dB: peak <= ceiling + 0.005, length giu nguyen;
|
||
soft mode giu tanh (peak ~1.0); low-level qua brickwall ~ unity."""
|
||
n = 4096
|
||
imp = np.zeros(n, dtype=np.float32)
|
||
imp[512] = 1.0
|
||
audio = np.stack([imp, imp * 0.8]).astype(np.float32)
|
||
ceil_lin = 10.0 ** (-1.0 / 20.0)
|
||
|
||
brick = [{"type": "builtin", "id": "limiter",
|
||
"params": {"ceiling": -1.0, "mode": 1, "lookahead_ms": 2.0}}]
|
||
brick_str = [{"type": "builtin", "id": "limiter",
|
||
"params": {"ceiling": -1.0, "mode": "brickwall", "lookahead_ms": 3.0}}]
|
||
soft = [{"type": "builtin", "id": "limiter",
|
||
"params": {"ceiling": -1.0}}]
|
||
|
||
with tempfile.TemporaryDirectory() as td:
|
||
wav_in = os.path.join(td, "in.wav")
|
||
_write_wav(wav_in, audio, SR)
|
||
out_brick = _render_bridge(fx_bridge, wav_in, SR, brick, block=512)
|
||
out_brick_str = _render_bridge(fx_bridge, wav_in, SR, brick_str, block=512)
|
||
out_soft = _render_bridge(fx_bridge, wav_in, SR, soft, block=512)
|
||
|
||
assert out_brick.shape == audio.shape, "brickwall length giu nguyen"
|
||
pk = float(np.max(np.abs(out_brick)))
|
||
assert pk <= ceil_lin + 0.005, f"brickwall peak {pk:.4f} > ceiling {ceil_lin:.4f}"
|
||
assert pk >= 0.70, f"brickwall khong giam dinh? peak {pk:.4f}"
|
||
|
||
pk_str = float(np.max(np.abs(out_brick_str)))
|
||
assert pk_str <= ceil_lin + 0.005, f"mode string peak {pk_str:.4f}"
|
||
|
||
pk_soft = float(np.max(np.abs(out_soft)))
|
||
assert pk_soft > 0.95, f"soft peak {pk_soft:.4f} (tanh normalized ~1.0)"
|
||
assert pk_soft <= 1.0005, f"soft peak {pk_soft:.4f} vuot 1.0"
|
||
|
||
imp2 = np.zeros(n, dtype=np.float32)
|
||
imp2[2048] = 0.1
|
||
audio2 = np.stack([imp2, imp2]).astype(np.float32)
|
||
with tempfile.TemporaryDirectory() as td:
|
||
wav_in2 = os.path.join(td, "in2.wav")
|
||
_write_wav(wav_in2, audio2, SR)
|
||
out_low = _render_bridge(fx_bridge, wav_in2, SR, brick, block=512)
|
||
pk_low = float(np.max(np.abs(out_low)))
|
||
assert 0.05 <= pk_low <= 0.15, f"low-level peak {pk_low:.4f} lech unity"
|
||
|
||
def test_61_maximizer_oversample(fx_bridge):
|
||
"""Gap 4: MaximizerFx oversample 4x — impulse delay 8 mẫu (2×FIR33 group
|
||
delay 16@4x / 4) chứng minh oversample active; sine -20dB boost 0 trong
|
||
suốt (unity); 5kHz soft_clip 100: harmonic 3 (15k) in-band giữ > -40dBFS
|
||
còn alias spur 9.1k (fold cua harmonic 7 = 35k) bi FIR33 anti-alias chặn
|
||
< -50dBFS; hot sine bám ceiling -1dB."""
|
||
|
||
def _bin_db(x, f):
|
||
x0 = x[0].astype(np.float64)
|
||
n = len(x0)
|
||
win = np.hanning(n)
|
||
X = np.fft.rfft((x0 - np.mean(x0)) * win)
|
||
freq = np.fft.rfftfreq(n, 1.0 / SR)
|
||
i = int(np.argmin(np.abs(freq - f)))
|
||
amp = np.abs(X[i]) * 2.0 / np.sum(win)
|
||
return 20.0 * np.log10(amp + 1e-12)
|
||
|
||
mx = [{"type": "builtin", "id": "maximizer",
|
||
"params": {"boost_db": 0.0, "soft_clip": 0.0, "upward": 0.0,
|
||
"ceiling_db": -1.0}}]
|
||
n = 8192
|
||
t = np.arange(n) / SR
|
||
|
||
# 1) impulse → delay 8 (oversample FIR chain active), gain ~1
|
||
imp = np.zeros(n, dtype=np.float32)
|
||
imp[512] = 1.0
|
||
audio = np.stack([imp, imp]).astype(np.float32)
|
||
with tempfile.TemporaryDirectory() as td:
|
||
wav_in = os.path.join(td, "in.wav")
|
||
_write_wav(wav_in, audio, SR)
|
||
out = _render_bridge(fx_bridge, wav_in, SR, mx, block=512)
|
||
argmax = int(np.argmax(np.abs(out[0])))
|
||
assert argmax == 512 + 8, f"impulse delay {argmax - 512} (expect 8)"
|
||
pk = float(np.max(np.abs(out[0])))
|
||
assert 0.8 <= pk <= 1.2, f"impulse peak {pk:.4f} (expect ~1.0)"
|
||
|
||
# 2) trong suốt: -20dBFS 1kHz qua maximizer gain 0 → unity
|
||
sine = (0.1 * np.sin(2 * np.pi * 1000.0 * t)).astype(np.float32)
|
||
audio = np.stack([sine, sine])
|
||
with tempfile.TemporaryDirectory() as td:
|
||
wav_in = os.path.join(td, "in.wav")
|
||
_write_wav(wav_in, audio, SR)
|
||
out2 = _render_bridge(fx_bridge, wav_in, SR, mx, block=512)
|
||
pk2 = float(np.max(np.abs(out2[0])))
|
||
assert 0.09 <= pk2 <= 0.11, f"transparent peak {pk2:.4f} (expect 0.1)"
|
||
|
||
# 3) soft_clip 100: 15k in-band giữ, alias 9.1k bị chặn
|
||
mx_hot = [{"type": "builtin", "id": "maximizer",
|
||
"params": {"boost_db": 0.0, "soft_clip": 100.0, "upward": 0.0,
|
||
"ceiling_db": -1.0}}]
|
||
sine5 = (0.7 * np.sin(2 * np.pi * 5000.0 * t)).astype(np.float32)
|
||
audio = np.stack([sine5, sine5])
|
||
with tempfile.TemporaryDirectory() as td:
|
||
wav_in = os.path.join(td, "in.wav")
|
||
_write_wav(wav_in, audio, SR)
|
||
out3 = _render_bridge(fx_bridge, wav_in, SR, mx_hot, block=512)
|
||
assert out3.shape == audio.shape, "length giu nguyen"
|
||
h3 = _bin_db(out3, 15000.0)
|
||
alias = _bin_db(out3, 9100.0)
|
||
print(f"maximizer: h3@15k={h3:.1f}dBFS alias@9.1k={alias:.1f}dBFS")
|
||
assert h3 > -40.0, f"3rd harmonic 15k {h3:.1f}dBFS qua thap"
|
||
assert alias < -50.0, f"alias spur 9.1k {alias:.1f}dBFS (oversample phai chan fold 35k)"
|
||
|
||
# 4) hot sine bám ceiling -1dB
|
||
ceil_lin = 10.0 ** (-1.0 / 20.0)
|
||
sine_hot = (0.9 * np.sin(2 * np.pi * 1000.0 * t)).astype(np.float32)
|
||
audio = np.stack([sine_hot, sine_hot])
|
||
mx_boost = [{"type": "builtin", "id": "maximizer",
|
||
"params": {"boost_db": 18.0, "soft_clip": 100.0, "upward": 0.0,
|
||
"ceiling_db": -1.0}}]
|
||
with tempfile.TemporaryDirectory() as td:
|
||
wav_in = os.path.join(td, "in.wav")
|
||
_write_wav(wav_in, audio, SR)
|
||
out4 = _render_bridge(fx_bridge, wav_in, SR, mx_boost, block=512)
|
||
pk4 = float(np.max(np.abs(out4[0])))
|
||
assert pk4 <= ceil_lin + 0.005, f"ceiling peak {pk4:.4f} > {ceil_lin:.4f}"
|
||
assert pk4 >= 0.7, f"ceiling peak {pk4:.4f} qua thap"
|
||
|
||
|
||
# ── 5.C Realtime VST3 setParam live (PLAN_MASTER_ENHANCE C) ───────────────
|
||
def test_60_vst3_setparam_live(fx_bridge, adelay, monkeypatch):
|
||
"""set_param runtime cho VST3: chain [vst3-delay, builtin limiter], đổi
|
||
param 'delay' giữa stream 1.0 → 0.25 → echo dịch ~1s → ~0.25s NGAY trên
|
||
session hiện tại (restart → plugin default 1s → echo vẫn ở 1s)."""
|
||
monkeypatch.setenv("SF_FX_BRIDGE_PATH", fx_bridge)
|
||
from app.core import fx_realtime
|
||
|
||
chain = [
|
||
{"type": "vst3", "path": adelay},
|
||
{"type": "builtin", "id": "limiter", "params": {"ceiling": -3.0}},
|
||
]
|
||
sess = fx_realtime.start_session(chain, SR)
|
||
try:
|
||
out = bytearray()
|
||
nblk = 280 # 1.24s — đủ phủ vùng echo cũ (1s) để chứng minh không còn
|
||
for i in range(nblk):
|
||
blk = np.zeros((2, 256), dtype=np.float32)
|
||
if i == 103: # impulse2 tại 0.6s (26368)
|
||
blk[0, 0] = 1.0
|
||
blk[1, 0] = 1.0
|
||
if i == 86: # 0.5s — setParam VST3 'delay' 1.0 -> 0.25 (normalized)
|
||
sess.set_param(0, "delay", 0.25)
|
||
inter = np.empty(512, dtype=np.float32)
|
||
inter[0::2] = blk[0]
|
||
inter[1::2] = blk[1]
|
||
sess.write_input(inter.tobytes())
|
||
time.sleep(0.006)
|
||
for b in sess.read_output():
|
||
out += b
|
||
time.sleep(0.05)
|
||
for b in sess.read_output():
|
||
out += b
|
||
finally:
|
||
sess.close()
|
||
arr = np.frombuffer(bytes(out), dtype=np.float32)
|
||
assert arr.size > 0, "realtime loop không ra output"
|
||
got = arr.reshape(-1, 2).T
|
||
|
||
def _peak(lo, hi):
|
||
if lo >= got.shape[1]:
|
||
return 0.0
|
||
return float(np.max(np.abs(got[:, lo:hi])))
|
||
|
||
e2 = 103 * 256 + int(0.25 * SR) # echo delay 0.25s ≈ 37393
|
||
assert _peak(e2 - 400, e2 + 400) > 0.5, f"echo delay 0.25s không thấy tại {e2} (peak {_peak(e2-400, e2+400):.3f})"
|
||
e_old = 103 * 256 + SR # echo delay cũ 1s ≈ 70468 (restart/no-apply)
|
||
assert _peak(e_old - 400, e_old + 400) < 0.3, "param không áp live — echo vẫn ở delay 1s"
|
||
assert _peak(0, e2 - 700) < 0.3, "có tín hiệu lạ trước echo 0.25s"
|