479 lines
17 KiB
Python
479 lines
17 KiB
Python
"""
|
|
Unit tests cho DSP Engine - SonicForge Studio.
|
|
Kiểm nghiệm thuật toán Zero-Crossing, Waveform, Cut/Loop/Fade, Multitrack Mixdown.
|
|
"""
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import numpy as np
|
|
import soundfile as sf
|
|
import pytest
|
|
|
|
# Thêm project root vào path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
from app.core.dsp_utils import (
|
|
find_zero_crossing,
|
|
find_nearest_zero_crossing_file,
|
|
apply_micro_fade,
|
|
generate_peak_waveform,
|
|
generate_rms_waveform,
|
|
)
|
|
from app.core.audio_editor import (
|
|
cut_and_loop_segment,
|
|
mix_multitrack_session,
|
|
export_audio,
|
|
)
|
|
from app.core.analyzer import (
|
|
analyze_audio,
|
|
analyze_audio_advanced,
|
|
_estimate_structure_heuristic,
|
|
)
|
|
|
|
|
|
# ── Fixtures ─────────────────────────────────────────────────
|
|
|
|
def _create_test_wav(duration_sec: float = 2.0, sr: int = 44100,
|
|
freq: float = 440.0) -> str:
|
|
"""Tạo file WAV dạng sine wave cho testing."""
|
|
t = np.linspace(0, duration_sec, int(sr * duration_sec), endpoint=False)
|
|
y = np.sin(2 * np.pi * freq * t).astype(np.float32)
|
|
|
|
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
sf.write(tmp.name, y, sr)
|
|
tmp.close()
|
|
return tmp.name
|
|
|
|
|
|
def _create_test_wav_stereo(duration_sec: float = 2.0, sr: int = 44100) -> str:
|
|
"""Tạo file WAV stereo cho testing."""
|
|
t = np.linspace(0, duration_sec, int(sr * duration_sec), endpoint=False)
|
|
left = np.sin(2 * np.pi * 440.0 * t).astype(np.float32)
|
|
right = np.sin(2 * np.pi * 880.0 * t).astype(np.float32)
|
|
stereo = np.column_stack([left, right])
|
|
|
|
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
sf.write(tmp.name, stereo, sr)
|
|
tmp.close()
|
|
return tmp.name
|
|
|
|
|
|
# ── Test Zero-Crossing (Week 2) ──────────────────────────────
|
|
|
|
class TestZeroCrossing:
|
|
"""Kiểm nghiệm thuật toán Zero-Crossing (triệt tiêu click/pop)."""
|
|
|
|
def test_find_zero_crossing_known_sine(self):
|
|
"""
|
|
Sine wave 440Hz có zero-crossing tại bội số của 1/(2*440).
|
|
Kiểm tra điểm tìm được phải nằm rất gần điểm đổi dấu thật.
|
|
"""
|
|
sr = 44100
|
|
duration = 1.0
|
|
freq = 440.0
|
|
t = np.linspace(0, duration, int(sr * duration), endpoint=False)
|
|
y = np.sin(2 * np.pi * freq * t).astype(np.float32)
|
|
|
|
# Target tại 0.5s
|
|
result = find_zero_crossing(y, sr, target_time=0.5, window_seconds=0.04)
|
|
|
|
# Kết quả phải nằm trong khoảng ±40ms từ target
|
|
assert abs(result - 0.5) <= 0.04
|
|
|
|
# Xác minh biên độ tại điểm zero-crossing rất nhỏ
|
|
result_sample = int(result * sr)
|
|
if result_sample < len(y) - 1:
|
|
# Kiểm tra đổi dấu
|
|
assert y[result_sample] * y[result_sample + 1] <= 0 or abs(y[result_sample]) < 0.01
|
|
|
|
def test_find_zero_crossing_returns_target_when_no_crossing(self):
|
|
"""Với tín hiệu DC (không có zero-crossing), trả về vị trí gốc."""
|
|
sr = 44100
|
|
y = np.ones(sr, dtype=np.float32) # DC signal, no crossing
|
|
|
|
result = find_zero_crossing(y, sr, target_time=0.5)
|
|
assert result == 0.5
|
|
|
|
def test_find_zero_crossing_edge_start(self):
|
|
"""Zero-crossing gần đầu mảng."""
|
|
sr = 44100
|
|
y = np.sin(2 * np.pi * 440.0 * np.linspace(0, 1, sr, endpoint=False)).astype(np.float32)
|
|
|
|
result = find_zero_crossing(y, sr, target_time=0.0)
|
|
assert result >= 0.0
|
|
assert result <= 0.04 # Phải nằm trong window
|
|
|
|
def test_find_zero_crossing_edge_end(self):
|
|
"""Zero-crossing gần cuối mảng."""
|
|
sr = 44100
|
|
y = np.sin(2 * np.pi * 440.0 * np.linspace(0, 1, sr, endpoint=False)).astype(np.float32)
|
|
|
|
result = find_zero_crossing(y, sr, target_time=0.99)
|
|
assert result >= 0.95
|
|
assert result <= 1.0
|
|
|
|
def test_find_zero_crossing_negative_time(self):
|
|
"""Handle target_time âm."""
|
|
sr = 44100
|
|
y = np.sin(2 * np.pi * 440.0 * np.linspace(0, 1, sr, endpoint=False)).astype(np.float32)
|
|
|
|
result = find_zero_crossing(y, sr, target_time=-1.0)
|
|
assert result == -1.0
|
|
|
|
def test_find_zero_crossing_file_wrapper(self):
|
|
"""Test find_nearest_zero_crossing_file wrapper."""
|
|
wav_path = _create_test_wav(duration_sec=1.0)
|
|
try:
|
|
result = find_nearest_zero_crossing_file(wav_path, 0.5)
|
|
assert isinstance(result, float)
|
|
assert abs(result - 0.5) <= 0.05
|
|
finally:
|
|
os.unlink(wav_path)
|
|
|
|
def test_zero_crossing_precision_sync(self):
|
|
"""
|
|
Sai số định vị mẫu âm học phải tiến về 0.
|
|
Kiểm tra sai số <= 1 sample.
|
|
"""
|
|
sr = 44100
|
|
freq = 440.0
|
|
t = np.linspace(0, 1.0, sr, endpoint=False)
|
|
y = np.sin(2 * np.pi * freq * t).astype(np.float32)
|
|
|
|
# Tìm zero-crossing đầu tiên thực tế
|
|
signs = np.sign(y)
|
|
true_crossings = np.where(np.diff(signs) != 0)[0]
|
|
|
|
if len(true_crossings) > 5:
|
|
# Nhắm vào zero-crossing thứ 5
|
|
true_time = float(true_crossings[5]) / sr
|
|
found_time = find_zero_crossing(y, sr, true_time, window_seconds=0.04)
|
|
|
|
# Sai số tối đa: 1 sample = 1/44100 ≈ 0.0000227s
|
|
sample_error = abs(found_time * sr - true_crossings[5])
|
|
assert sample_error <= 1.0, f"Sai số {sample_error} samples vượt ngưỡng 1 sample"
|
|
|
|
|
|
# ── Test Peak Waveform (Week 2) ──────────────────────────────
|
|
|
|
class TestPeakWaveform:
|
|
"""Kiểm nghiệm Peak Waveform generation."""
|
|
|
|
def test_generate_peak_waveform(self):
|
|
wav_path = _create_test_wav(duration_sec=1.0)
|
|
try:
|
|
result = generate_peak_waveform(wav_path, num_peaks=100)
|
|
|
|
assert "peaks" in result
|
|
assert "duration" in result
|
|
assert "sample_rate" in result
|
|
assert len(result["peaks"]) == 100
|
|
assert result["duration"] > 0.9
|
|
assert all(0 <= p <= 1.0 for p in result["peaks"])
|
|
finally:
|
|
os.unlink(wav_path)
|
|
|
|
def test_generate_rms_waveform(self):
|
|
wav_path = _create_test_wav(duration_sec=1.0)
|
|
try:
|
|
result = generate_rms_waveform(wav_path, num_points=50)
|
|
|
|
assert "rms" in result
|
|
assert len(result["rms"]) == 50
|
|
assert all(v >= 0 for v in result["rms"])
|
|
finally:
|
|
os.unlink(wav_path)
|
|
|
|
def test_peak_waveform_empty_audio(self):
|
|
"""Test với file WAV rất ngắn."""
|
|
sr = 44100
|
|
y = np.zeros(100, dtype=np.float32)
|
|
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
sf.write(tmp.name, y, sr)
|
|
tmp.close()
|
|
|
|
try:
|
|
result = generate_peak_waveform(tmp.name, num_peaks=10)
|
|
assert "peaks" in result
|
|
# Với tín hiệu zero, tất cả peaks phải = 0
|
|
assert all(p == 0 for p in result["peaks"])
|
|
finally:
|
|
os.unlink(tmp.name)
|
|
|
|
|
|
# ── Test Micro-Fade ──────────────────────────────────────────
|
|
|
|
class TestMicroFade:
|
|
def test_apply_micro_fade(self):
|
|
from pydub import AudioSegment
|
|
from pydub.generators import Sine
|
|
|
|
# Tạo 1-second sine tone
|
|
tone = Sine(440).to_audio_segment(duration=1000)
|
|
faded = apply_micro_fade(tone, fade_duration_ms=50)
|
|
|
|
# Độ dài không đổi
|
|
assert len(faded) == len(tone)
|
|
|
|
def test_apply_micro_fade_short_segment(self):
|
|
from pydub import AudioSegment
|
|
from pydub.generators import Sine
|
|
|
|
# Segment ngắn hơn 2x fade duration
|
|
tone = Sine(440).to_audio_segment(duration=80)
|
|
faded = apply_micro_fade(tone, fade_duration_ms=50)
|
|
assert len(faded) == len(tone)
|
|
|
|
|
|
# ── Test Cut & Loop (Week 2-3) ───────────────────────────────
|
|
|
|
class TestCutAndLoop:
|
|
def test_cut_segment(self):
|
|
wav_path = _create_test_wav(duration_sec=5.0)
|
|
try:
|
|
result = cut_and_loop_segment(
|
|
file_path=wav_path,
|
|
start_sec=1.0,
|
|
end_sec=3.0,
|
|
loop_count=1,
|
|
fade_in_ms=50,
|
|
fade_out_ms=50
|
|
)
|
|
|
|
# 2 giây = 2000ms (±tolerance cho fade)
|
|
assert abs(len(result) - 2000) < 50
|
|
finally:
|
|
os.unlink(wav_path)
|
|
|
|
def test_cut_and_loop(self):
|
|
wav_path = _create_test_wav(duration_sec=5.0)
|
|
try:
|
|
result = cut_and_loop_segment(
|
|
file_path=wav_path,
|
|
start_sec=1.0,
|
|
end_sec=2.0,
|
|
loop_count=3,
|
|
fade_in_ms=0,
|
|
fade_out_ms=0
|
|
)
|
|
|
|
# 1 giây * 3 lần = 3000ms
|
|
assert abs(len(result) - 3000) < 50
|
|
finally:
|
|
os.unlink(wav_path)
|
|
|
|
def test_cut_with_volume_change(self):
|
|
wav_path = _create_test_wav(duration_sec=2.0)
|
|
try:
|
|
result = cut_and_loop_segment(
|
|
file_path=wav_path,
|
|
start_sec=0.0,
|
|
end_sec=1.0,
|
|
volume_db_change=-6.0
|
|
)
|
|
|
|
assert len(result) > 0
|
|
finally:
|
|
os.unlink(wav_path)
|
|
|
|
|
|
# ── Test Multitrack Mixdown (Week 3) ─────────────────────────
|
|
|
|
class TestMultitrackMixdown:
|
|
def test_mix_two_tracks(self):
|
|
wav1 = _create_test_wav(duration_sec=2.0, freq=440.0)
|
|
wav2 = _create_test_wav(duration_sec=2.0, freq=880.0)
|
|
|
|
tmp_output = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
tmp_output.close()
|
|
|
|
try:
|
|
result = mix_multitrack_session(
|
|
tracks_meta=[
|
|
{"file_path": wav1, "volume": 0.8, "muted": False},
|
|
{"file_path": wav2, "volume": 0.6, "muted": False},
|
|
],
|
|
output_path=tmp_output.name,
|
|
sample_rate=44100,
|
|
bit_depth=16
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["tracks_processed"] == 2
|
|
assert os.path.exists(tmp_output.name)
|
|
|
|
# Verify output is valid WAV
|
|
y, sr = sf.read(tmp_output.name)
|
|
assert sr == 44100
|
|
assert len(y) > 0
|
|
finally:
|
|
os.unlink(wav1)
|
|
os.unlink(wav2)
|
|
if os.path.exists(tmp_output.name):
|
|
os.unlink(tmp_output.name)
|
|
|
|
def test_mix_with_muted_track(self):
|
|
wav1 = _create_test_wav(duration_sec=1.0, freq=440.0)
|
|
wav2 = _create_test_wav(duration_sec=1.0, freq=880.0)
|
|
|
|
tmp_output = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
tmp_output.close()
|
|
|
|
try:
|
|
result = mix_multitrack_session(
|
|
tracks_meta=[
|
|
{"file_path": wav1, "volume": 1.0, "muted": False},
|
|
{"file_path": wav2, "volume": 1.0, "muted": True},
|
|
],
|
|
output_path=tmp_output.name
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["tracks_processed"] == 1
|
|
finally:
|
|
os.unlink(wav1)
|
|
os.unlink(wav2)
|
|
if os.path.exists(tmp_output.name):
|
|
os.unlink(tmp_output.name)
|
|
|
|
def test_mix_all_muted(self):
|
|
"""Khi tất cả tracks đều muted, trả về lỗi."""
|
|
result = mix_multitrack_session(
|
|
tracks_meta=[
|
|
{"file_path": "/dummy", "volume": 1.0, "muted": True},
|
|
],
|
|
output_path="/tmp/kilo/test_output.wav"
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
def test_mix_gain_no_clipping(self):
|
|
"""
|
|
Đảm bảo tăng giảm âm lượng không gây méo tiếng (Clipping distortion).
|
|
Volume 0.5 => gain_db ≈ -6.02 dB
|
|
"""
|
|
wav_path = _create_test_wav(duration_sec=1.0, freq=440.0)
|
|
tmp_output = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
tmp_output.close()
|
|
|
|
try:
|
|
result = mix_multitrack_session(
|
|
tracks_meta=[
|
|
{"file_path": wav_path, "volume": 0.5, "muted": False},
|
|
],
|
|
output_path=tmp_output.name,
|
|
sample_rate=44100,
|
|
bit_depth=16
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
# Kiểm tra output: max amplitude phải < 1.0 (no clipping)
|
|
y, sr = sf.read(tmp_output.name)
|
|
max_amp = np.max(np.abs(y))
|
|
assert max_amp <= 1.0, f"Clipping detected: max amplitude = {max_amp}"
|
|
finally:
|
|
os.unlink(wav_path)
|
|
if os.path.exists(tmp_output.name):
|
|
os.unlink(tmp_output.name)
|
|
|
|
|
|
# ── Test Multi-Format Export (Week 1 / 5) ────────────────────
|
|
|
|
class TestExport:
|
|
def test_export_wav_16bit(self):
|
|
wav_path = _create_test_wav(duration_sec=1.0)
|
|
tmp_output = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
tmp_output.close()
|
|
|
|
try:
|
|
result = export_audio(wav_path, tmp_output.name, format="wav",
|
|
sample_rate=44100, bit_depth=16)
|
|
assert result["success"] is True
|
|
assert os.path.exists(tmp_output.name)
|
|
|
|
info = sf.info(tmp_output.name)
|
|
assert info.subtype == "PCM_16"
|
|
finally:
|
|
os.unlink(wav_path)
|
|
if os.path.exists(tmp_output.name):
|
|
os.unlink(tmp_output.name)
|
|
|
|
def test_export_wav_24bit(self):
|
|
wav_path = _create_test_wav(duration_sec=1.0)
|
|
tmp_output = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
tmp_output.close()
|
|
|
|
try:
|
|
result = export_audio(wav_path, tmp_output.name, format="wav",
|
|
sample_rate=44100, bit_depth=24)
|
|
assert result["success"] is True
|
|
|
|
info = sf.info(tmp_output.name)
|
|
assert info.subtype == "PCM_24"
|
|
finally:
|
|
os.unlink(wav_path)
|
|
if os.path.exists(tmp_output.name):
|
|
os.unlink(tmp_output.name)
|
|
|
|
|
|
# ── Test Audio Analysis (Week 4) ─────────────────────────────
|
|
|
|
class TestAnalyzer:
|
|
def test_analyze_audio_basic(self):
|
|
wav_path = _create_test_wav(duration_sec=5.0)
|
|
try:
|
|
result = analyze_audio(wav_path)
|
|
|
|
assert "bpm" in result
|
|
assert "beats" in result
|
|
assert "bars" in result
|
|
assert "duration" in result
|
|
assert result["duration"] > 4.5
|
|
assert isinstance(result["bpm"], float)
|
|
finally:
|
|
os.unlink(wav_path)
|
|
|
|
def test_analyze_audio_advanced(self):
|
|
wav_path = _create_test_wav(duration_sec=5.0)
|
|
try:
|
|
result = analyze_audio_advanced(wav_path)
|
|
|
|
assert "spectral_centroid_avg" in result
|
|
assert "rms_energy_avg" in result
|
|
assert "zero_crossing_rate_avg" in result
|
|
assert "sample_rate" in result
|
|
finally:
|
|
os.unlink(wav_path)
|
|
|
|
def test_structure_heuristic(self):
|
|
"""Test heuristic structure estimation."""
|
|
analysis = {
|
|
"duration": 120.0,
|
|
"bars": [0.0, 2.0, 4.0, 6.0, 8.0, 10.0, 12.0, 14.0,
|
|
16.0, 18.0, 20.0, 22.0, 24.0, 26.0, 28.0, 30.0,
|
|
32.0, 34.0, 36.0, 38.0, 40.0, 42.0, 44.0, 46.0,
|
|
48.0, 50.0]
|
|
}
|
|
|
|
result = _estimate_structure_heuristic(analysis)
|
|
|
|
assert "sections" in result
|
|
assert len(result["sections"]) > 0
|
|
|
|
# Kiểm tra có Intro
|
|
section_names = [s["name"] for s in result["sections"]]
|
|
assert "Intro" in section_names
|
|
|
|
def test_structure_heuristic_short(self):
|
|
"""Test heuristic với audio quá ngắn."""
|
|
analysis = {"duration": 5.0, "bars": []}
|
|
result = _estimate_structure_heuristic(analysis)
|
|
|
|
assert len(result["sections"]) == 1
|
|
assert result["sections"][0]["name"] == "Full"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|