276 lines
12 KiB
Python
276 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Review DAW-standard — TASKS_DAW_A.md §49-57 (R1-R8), cài đặt thực tế.
|
||
|
||
Mọi test chạy qua DAW THẬT: PythonRenderEngine.render_project (RENDER_ENGINE=
|
||
bridge mặc định) + native_bridge fx_vst_bridge.exe --render-fx + project JSON
|
||
đầy đủ (AUDIO_ITEM track, master, mastering_settings.chain). API smoke qua
|
||
TestClient(app) + /api/v1/plugins/render (auth admin) — chính là luồng frontend
|
||
Export.
|
||
|
||
- R1: track insert chain [eq, vst3-adelay, limiter] — impulse → peak tại
|
||
44100 (delay 1s), im lặng trước: vst3 ở giữa chain, không gom cuối.
|
||
- R2: gain/pan sau track chain — rebalance mid-only (giết side) + pan 0.5 →
|
||
output L≠R với tỉ lệ constant-power: pan phải chạy SAU chain.
|
||
- R3: master chain serial trên mix tổng — mastering_settings.chain
|
||
[eq, vst3-adelay, limiter] → impulse peak tại 44100.
|
||
- R4: master fader cuối (đã có test_55_fader_after_master_chain).
|
||
- R5: live=export (đã có test_53_live_equals_export).
|
||
- R6: PDC (đã có test_56_pdc_aligns_tracks).
|
||
- R7: gain staging — track fader -6 / master fader -6 / chain +6 / fader -6
|
||
net 0: dB cộng dồn đúng qua từng stage.
|
||
- R8: mono/stereo — mono source pad L==R giữ qua chain vst3; stereo source
|
||
giữ separation qua chain builtin.
|
||
- R-API: TestClient + POST /api/v1/plugins/render project thật (track chain +
|
||
master chain) → WAV peak tại 44100.
|
||
"""
|
||
import os
|
||
import sys
|
||
import wave
|
||
|
||
import numpy as np
|
||
import pytest
|
||
|
||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
sys.path.insert(0, REPO)
|
||
sys.path.insert(0, os.path.dirname(__file__))
|
||
import test_daw_phase5 as p5 # noqa: E402 (dùng chung helper/fixture)
|
||
|
||
SR = p5.SR
|
||
_write_wav = p5._write_wav
|
||
_read_wav = p5._read_wav
|
||
_snr_db = p5._snr_db
|
||
ADELAY = p5.ADELAY
|
||
FX_BRIDGE = p5.FX_BRIDGE
|
||
|
||
|
||
@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
|
||
|
||
|
||
@pytest.fixture
|
||
def bridge_env(fx_bridge, monkeypatch):
|
||
"""Engine render_project → native_render.render_fx_chain → fx_vst_bridge.exe
|
||
(tìm qua SF_FX_BRIDGE_PATH)."""
|
||
monkeypatch.setenv("SF_FX_BRIDGE_PATH", fx_bridge)
|
||
return fx_bridge
|
||
|
||
|
||
# ── helpers ────────────────────────────────────────────────────────────────
|
||
def _audio_track(wav_path, vol=0.0, pan=0.0, fx_chain=None, track_id="t1"):
|
||
return {
|
||
"id": track_id, "type": "AUDIO",
|
||
"volume_db": vol, "pan": pan, "mute": False,
|
||
"fx_chain": fx_chain or [], "vst_fx_chain": [], "vst_fx_bypass": False,
|
||
"fx_active": True,
|
||
"items": [{
|
||
"type": "AUDIO_ITEM", "start_bar": 0.0, "duration_bars": 1.0,
|
||
"clip_start_offset_bars": 0.0,
|
||
"source_data": {"audio_file_url": wav_path},
|
||
}],
|
||
}
|
||
|
||
|
||
def _project(tracks, master_vol=0.0, master_chain=None, ms_chain=None,
|
||
length_bars=1.0):
|
||
master = {"bypass": False, "volume_db": master_vol}
|
||
if master_chain is not None:
|
||
master["fx_chain"] = master_chain
|
||
proj = {
|
||
"metadata": {"bpm": 120.0, "time_signature_numerator": 4},
|
||
"main_session": {"length_bars": length_bars, "tracks": tracks, "master": master},
|
||
"section_store": {},
|
||
}
|
||
if ms_chain is not None:
|
||
proj["mastering_settings"] = {"chain": ms_chain}
|
||
return proj
|
||
|
||
|
||
def _impulse_wav(path, amp=1.0, n=2 * SR):
|
||
audio = np.zeros((2, n), dtype=np.float32)
|
||
audio[0, 0] = amp
|
||
audio[1, 0] = amp
|
||
_write_wav(path, audio, SR)
|
||
|
||
|
||
def _render_project(project, out):
|
||
from app.core.render_engine import PythonRenderEngine
|
||
engine = PythonRenderEngine()
|
||
engine.render_project(project, out)
|
||
return _read_wav(out)[0]
|
||
|
||
|
||
def _peak_idx(data):
|
||
return int(np.unravel_index(np.argmax(np.abs(data)), data.shape)[1])
|
||
|
||
|
||
# ── R1 Track insert chain serial ───────────────────────────────────────────
|
||
def test_r1_track_chain_order_impulse(bridge_env, adelay, tmp_path):
|
||
"""Track chain [eq, vst3-adelay, limiter] qua render_project thật — impulse
|
||
peak tại ≈44100, im lặng trước: vst3 ở giữa chain (sau EQ, trước limiter)."""
|
||
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}},
|
||
]
|
||
wav = str(tmp_path / "imp.wav")
|
||
_impulse_wav(wav)
|
||
out = str(tmp_path / "r1.wav")
|
||
data = _render_project(_project([_audio_track(wav, fx_chain=chain)]), out)
|
||
|
||
idx = _peak_idx(data)
|
||
assert abs(idx - SR) < 200, f"peak tại {idx} — mong đợi ~{SR} (delay 1s)"
|
||
assert float(np.max(np.abs(data[:, :int(SR * 0.9)]))) == 0.0, "có tín hiệu trước delay"
|
||
|
||
|
||
# ── R2 Gain/pan sau track chain ────────────────────────────────────────────
|
||
def test_r2_gain_pan_after_track_chain(bridge_env, tmp_path):
|
||
"""Chain rebalance mid-only (giết side) → L==R. Pan 0.5 SAU chain tạo lại
|
||
side (L≠R, constant-power R/L=sin/cos≈2.414). Nếu pan chạy trước chain,
|
||
rebalance sẽ mono lại → L==R (fail)."""
|
||
chain = [{"type": "builtin", "id": "rebalance",
|
||
"params": {"mid": 0.0, "side": -120.0}}]
|
||
wav = str(tmp_path / "imp.wav")
|
||
_impulse_wav(wav, amp=0.8)
|
||
|
||
d0 = _render_project(_project([_audio_track(wav, pan=0.0, fx_chain=chain)]),
|
||
str(tmp_path / "pan0.wav"))
|
||
assert _snr_db(d0[0:1], d0[1:2]) >= 60.0, "rebalance mid-only phải ra L==R"
|
||
|
||
d5 = _render_project(_project([_audio_track(wav, pan=0.5, fx_chain=chain)]),
|
||
str(tmp_path / "pan5.wav"))
|
||
lr = float(np.max(np.abs(d5[1]))) / float(np.max(np.abs(d5[0])))
|
||
assert 1.8 < lr < 3.2, f"pan không SAU chain: R/L={lr:.3f} (mong đợi 2.414)"
|
||
|
||
# gain cũng sau chain: chain rebalance mid-only không đổi amp → pan=0,
|
||
# vol -6 → đỉnh = base/2 (gain scale đúng ở stage riêng)
|
||
d6 = _render_project(_project([_audio_track(wav, pan=0.0, vol=-6.0, fx_chain=chain)]),
|
||
str(tmp_path / "vol6.wav"))
|
||
assert abs(float(np.max(np.abs(d6))) - 0.4) < 0.03, "vol_db -6 sau chain sai"
|
||
|
||
|
||
# ── R3 Master chain serial trên mix tổng ───────────────────────────────────
|
||
def test_r3_master_chain_serial_on_mix(bridge_env, adelay, tmp_path):
|
||
"""mastering_settings.chain [eq, vst3-adelay, limiter] trên mix tổng —
|
||
impulse peak tại ≈44100 (vst3 delay giữa chain master, không gom cuối)."""
|
||
chain = [
|
||
{"type": "eq", "params": {"g1": 3.0, "g2": -2.0, "g3": 1.0, "g4": 0.5}},
|
||
{"type": "vst3", "path": adelay},
|
||
{"type": "limiter", "params": {"ceiling": -3.0}},
|
||
]
|
||
wav = str(tmp_path / "imp.wav")
|
||
_impulse_wav(wav)
|
||
out = str(tmp_path / "r3.wav")
|
||
data = _render_project(_project([_audio_track(wav)], ms_chain=chain), out)
|
||
|
||
idx = _peak_idx(data)
|
||
assert abs(idx - SR) < 200, f"peak tại {idx} — mong đợi ~{SR}"
|
||
assert float(np.max(np.abs(data[:, :int(SR * 0.9)]))) == 0.0, "có tín hiệu trước delay"
|
||
|
||
|
||
# ── R7 Gain staging ────────────────────────────────────────────────────────
|
||
def test_r7_gain_staging(bridge_env, tmp_path):
|
||
"""dB cộng dồn qua từng stage: track fader -6 / master fader -6 / master
|
||
chain +6 / fader -6 — net 0 → bằng base; net -6 → base/2."""
|
||
t = np.arange(SR // 2) / SR
|
||
sig = (0.3 * np.sin(2 * np.pi * 440 * t)).astype(np.float32)
|
||
wav = str(tmp_path / "sig.wav")
|
||
_write_wav(wav, np.stack([sig, sig]), SR)
|
||
GAIN6 = [{"type": "builtin", "id": "gain", "params": {"db": 6.0}}]
|
||
|
||
def peak(vol=0.0, mvol=0.0, mchain=None):
|
||
out = str(tmp_path / f"g_{vol}_{mvol}_{bool(mchain)}.wav")
|
||
d = _render_project(_project([_audio_track(wav, vol=vol)],
|
||
master_vol=mvol, master_chain=mchain), out)
|
||
return float(np.max(np.abs(d)))
|
||
|
||
base = peak()
|
||
assert base > 0.1
|
||
# net -6 dB → base/2
|
||
for name, v in [("track fader -6", peak(vol=-6.0)),
|
||
("master fader -6", peak(mvol=-6.0)),
|
||
("track-6 + chain+6 + fader-6", peak(vol=-6.0, mvol=-6.0, mchain=GAIN6))]:
|
||
assert abs(v - base / 2) < 0.06 * base, \
|
||
f"{name} staging sai: {v:.4f} vs base/2 {base / 2:.4f}"
|
||
# net 0 dB → base
|
||
for name, v in [("track-6 + chain+6", peak(vol=-6.0, mchain=GAIN6)),
|
||
("chain+6 + fader-6", peak(mvol=-6.0, mchain=GAIN6))]:
|
||
assert abs(v - base) < 0.06 * base, f"{name} staging sai: {v:.4f} vs base {base:.4f}"
|
||
|
||
|
||
# ── R8 Mono/Stereo ─────────────────────────────────────────────────────────
|
||
def test_r8_mono_stereo(bridge_env, adelay, tmp_path):
|
||
"""Mono source → pad L==R, giữ qua chain vst3 (delay 1s). Stereo source
|
||
L≠R → giữ separation qua chain builtin (xử lý per-channel đúng)."""
|
||
mono = np.zeros(2 * SR, dtype=np.float32)
|
||
mono[0] = 1.0
|
||
wp = str(tmp_path / "mono.wav")
|
||
with wave.open(wp, "wb") as w:
|
||
w.setnchannels(1)
|
||
w.setsampwidth(2)
|
||
w.setframerate(SR)
|
||
w.writeframes((mono * 32767).astype(np.int16).tobytes())
|
||
|
||
d = _render_project(_project([_audio_track(wp, fx_chain=[{"type": "vst3", "path": adelay}])]),
|
||
str(tmp_path / "mono_out.wav"))
|
||
assert _snr_db(d[0:1], d[1:2]) >= 60.0, "mono pad L==R không giữ qua chain vst3"
|
||
assert abs(_peak_idx(d) - SR) < 200, "mono delay qua chain sai vị trí"
|
||
|
||
st = np.zeros((2, 2 * SR), dtype=np.float32)
|
||
st[0, 0] = 1.0
|
||
st[1, 0] = 0.4
|
||
wp2 = str(tmp_path / "st.wav")
|
||
_write_wav(wp2, st, SR)
|
||
eq = [{"type": "builtin", "id": "eq", "params": {"g1": 3.0}}]
|
||
d2 = _render_project(_project([_audio_track(wp2, fx_chain=eq)]),
|
||
str(tmp_path / "st_out.wav"))
|
||
ratio = float(np.max(np.abs(d2[0]))) / float(np.max(np.abs(d2[1])))
|
||
assert 1.8 < ratio < 3.2, f"stereo separation không giữ qua chain: L/R={ratio:.3f}"
|
||
|
||
|
||
# ── R-API: cài đặt thực tế — TestClient + /render ──────────────────────────
|
||
def test_r_api_render_real_project(bridge_env, adelay, tmp_path):
|
||
"""Luồng Export thật: TestClient(app), login admin, GET /capabilities, POST
|
||
/api/v1/plugins/render với project thật (track chain + master chain đều có
|
||
vst3 adelay, project 2 bars) → impulse qua 2 delay 1s → peak tại ≈88200
|
||
(2×SR) — 2 chain nối tiếp không mất tín hiệu."""
|
||
from test_plugin_api import client, get_admin_token
|
||
|
||
token = get_admin_token()
|
||
assert token, "admin login fail"
|
||
h = {"Authorization": f"Bearer {token}"}
|
||
|
||
r = client.get("/api/v1/system/capabilities")
|
||
assert r.status_code == 200, r.text[:300]
|
||
|
||
wav = str(tmp_path / "imp.wav")
|
||
_impulse_wav(wav, n=3 * SR)
|
||
chain = [
|
||
{"type": "builtin", "id": "eq", "params": {"g1": 3.0}},
|
||
{"type": "vst3", "path": adelay},
|
||
{"type": "builtin", "id": "limiter", "params": {"ceiling": -3.0}},
|
||
]
|
||
proj = _project([_audio_track(wav, fx_chain=chain)], ms_chain=chain,
|
||
length_bars=2.0)
|
||
r = client.post("/api/v1/plugins/render", headers=h,
|
||
json={"project_json": proj, "output_filename": "review_api_out.wav"})
|
||
assert r.status_code == 200, f"render fail: {r.text[:400]}"
|
||
body = r.json()
|
||
path = body.get("path")
|
||
assert path and os.path.isfile(path), f"no output wav: {body}"
|
||
|
||
data = _read_wav(path)[0]
|
||
idx = _peak_idx(data)
|
||
assert abs(idx - 2 * SR) < 300, f"peak tại {idx} — mong đợi ~{2 * SR} (track 1s + master 1s)"
|
||
assert float(np.max(np.abs(data[:, :int(0.9 * SR)]))) == 0.0, "có tín hiệu trước delay"
|