root
ef9573c294
Per-worker OpenVINO CPU runs the MetricGAN BLSTM mask; SpeechBrain keeps STFT/ISTFT. ENHANCE_INFER=auto with torch fallback. Hop is 0.1 s after isolated last_ms ~6 ms (model+ov+beamform).
124 lines
3.5 KiB
Python
124 lines
3.5 KiB
Python
import inspect
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from audio_cleanup import AudioCleaner
|
|
|
|
|
|
def test_enhance_batch_without_lengths_typeerror():
|
|
c = AudioCleaner(mode="force")
|
|
assert "model" in c.backend
|
|
import torch
|
|
|
|
w = torch.zeros(1, 8000)
|
|
with pytest.raises(TypeError):
|
|
c._enhancer.enhance_batch(w)
|
|
|
|
|
|
def test_process_uses_relative_lengths_one():
|
|
src = inspect.getsource(AudioCleaner._process_inner)
|
|
assert "tensor([1.0])" in src
|
|
|
|
|
|
def test_process_preserves_length_and_int16():
|
|
c = AudioCleaner(mode="force")
|
|
n = 16000
|
|
pcm = (np.sin(2 * np.pi * 440 * np.arange(n) / 16000) * 8000).astype(np.int16)
|
|
out = c.process(pcm)
|
|
assert out.dtype == np.int16
|
|
assert out.shape == pcm.shape
|
|
|
|
|
|
def test_overlap_process_emits_hop_length():
|
|
c = AudioCleaner(mode="force", chunk_s=1.0, hop_s=0.2)
|
|
hop = 3200
|
|
pcm = (np.random.randn(hop) * 500).astype(np.int16)
|
|
out = c.process(pcm)
|
|
assert out.shape == (hop,)
|
|
|
|
|
|
def test_overlap_runs_model_after_enough_context():
|
|
c = AudioCleaner(mode="force", chunk_s=0.5, hop_s=0.25)
|
|
hop = 4000
|
|
for _ in range(4):
|
|
pcm = (np.sin(2 * np.pi * 440 * np.arange(hop) / 16000) * 8000).astype(
|
|
np.int16
|
|
)
|
|
out = c.process(pcm)
|
|
assert out.shape == (hop,)
|
|
assert out.dtype == np.int16
|
|
|
|
|
|
def test_beamform_loads_delaysum_module():
|
|
c = AudioCleaner(mode="off", beamform="force")
|
|
assert c._beamformer is not None
|
|
assert type(c._beamformer).__name__ == "DelaySum_Beamformer"
|
|
assert "beamform" in c.backend
|
|
|
|
|
|
def test_beamform_skips_normalizer_at_native_rate():
|
|
c = AudioCleaner(mode="off", beamform="force", sample_rate=16000)
|
|
assert c._normalizer is None
|
|
|
|
|
|
def test_tdoa_reused_across_hops():
|
|
c = AudioCleaner(mode="off", beamform="force", tdoa_every=5)
|
|
assert c._gccphat is not None
|
|
calls = {"n": 0}
|
|
orig = c._gccphat
|
|
|
|
def wrapped(*a, **k):
|
|
calls["n"] += 1
|
|
return orig(*a, **k)
|
|
|
|
c._gccphat = wrapped
|
|
hop = 3200
|
|
t = np.arange(hop)
|
|
ch0 = (np.sin(2 * np.pi * 440 * t / 16000) * 8000).astype(np.int16)
|
|
stereo = np.stack([ch0, np.roll(ch0, 2)], axis=1)
|
|
for _ in range(3):
|
|
out = c.process(stereo)
|
|
assert out.shape == (hop,)
|
|
assert calls["n"] == 1
|
|
|
|
|
|
def test_process_records_last_dt_ms():
|
|
c = AudioCleaner(mode="off", beamform="off")
|
|
hop = 3200
|
|
pcm = (np.random.randn(hop) * 500).astype(np.int16)
|
|
c.process(pcm)
|
|
assert c.last_dt_ms >= 0.0
|
|
|
|
|
|
def test_stereo_beamform_then_model_preserves_hop():
|
|
c = AudioCleaner(
|
|
mode="force", beamform="force", chunk_s=0.5, hop_s=0.25)
|
|
assert "model" in c.backend
|
|
assert "beamform" in c.backend
|
|
hop = 4000
|
|
t = np.arange(hop)
|
|
ch0 = (np.sin(2 * np.pi * 440 * t / 16000) * 8000).astype(np.int16)
|
|
ch1 = np.roll(ch0, 3)
|
|
stereo = np.stack([ch0, ch1], axis=1)
|
|
for _ in range(4):
|
|
out = c.process(stereo)
|
|
assert out.dtype == np.int16
|
|
assert out.shape == (hop,)
|
|
|
|
|
|
def test_openvino_cpu_infer_loads_and_preserves_hop():
|
|
c = AudioCleaner(
|
|
mode="force", beamform="off", enhance_infer="openvino",
|
|
chunk_s=0.5, hop_s=0.25)
|
|
assert c.infer_backend == "openvino"
|
|
assert c._ov_compiled is not None
|
|
assert "ov" in c.backend
|
|
hop = 4000
|
|
pcm = (np.sin(2 * np.pi * 440 * np.arange(hop) / 16000) * 8000).astype(
|
|
np.int16)
|
|
for _ in range(4):
|
|
out = c.process(pcm)
|
|
assert out.dtype == np.int16
|
|
assert out.shape == (hop,)
|