File size: 11,329 Bytes
a9536c4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 | import importlib.util
import re
import sys
import unittest
from pathlib import Path
import numpy as np
REPO_ROOT = Path(__file__).resolve().parents[1]
def _load_quality_policy_module():
module_path = REPO_ROOT / "infer" / "quality_policy.py"
spec = importlib.util.spec_from_file_location("quality_policy", module_path)
if spec is None or spec.loader is None:
raise ImportError(f"Unable to load module from {module_path}")
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
quality_policy = _load_quality_policy_module()
build_conservative_crepe_fill_mask = quality_policy.build_conservative_crepe_fill_mask
build_conservative_harvest_fill_mask = quality_policy.build_conservative_harvest_fill_mask
compute_chunk_crossfade_samples = quality_policy.compute_chunk_crossfade_samples
compute_active_source_replace = quality_policy.compute_active_source_replace
compute_breath_preserving_energy_gates = quality_policy.compute_breath_preserving_energy_gates
compute_source_cleanup_budget = quality_policy.compute_source_cleanup_budget
resolve_cover_f0_policy = quality_policy.resolve_cover_f0_policy
class F0RoutingPolicyTests(unittest.TestCase):
def test_hybrid_routes_to_conservative_rmvpe_fallback(self):
policy = resolve_cover_f0_policy("hybrid", "off")
self.assertEqual(policy.requested_method, "hybrid")
self.assertEqual(policy.vc_method, "rmvpe")
self.assertEqual(policy.hybrid_mode, "fallback")
self.assertEqual(policy.gate_method, "rmvpe")
class ConservativeCrepeFillTests(unittest.TestCase):
def test_only_short_internal_gaps_are_filled(self):
f0_rmvpe = np.array([0.0, 0.0, 120.0, 120.0, 0.0, 0.0, 120.0, 120.0, 0.0, 0.0], dtype=np.float32)
f0_crepe = np.full_like(f0_rmvpe, 121.0)
confidence = np.full_like(f0_rmvpe, 0.95)
fill_mask = build_conservative_crepe_fill_mask(
f0_rmvpe,
f0_crepe,
confidence,
confidence_threshold=0.6,
max_ratio=0.5,
max_frames=4,
context_radius=2,
)
expected = np.array([False, False, False, False, True, True, False, False, False, False])
np.testing.assert_array_equal(fill_mask, expected)
def test_harvest_fill_rejects_long_gap_and_accepts_short_consistent_gap(self):
reference_f0 = np.array(
[120.0, 121.0, 122.0, 0.0, 0.0, 121.5, 122.0, 121.0, 0.0, 0.0, 0.0, 0.0, 122.0, 121.0],
dtype=np.float32,
)
fallback_f0 = np.array(
[0.0, 0.0, 0.0, 121.0, 121.5, 0.0, 0.0, 0.0, 122.0, 122.0, 122.0, 122.0, 0.0, 0.0],
dtype=np.float32,
)
dropout_mask = reference_f0 <= 0
fill_mask = build_conservative_harvest_fill_mask(
reference_f0=reference_f0,
fallback_f0=fallback_f0,
dropout_mask=dropout_mask,
max_run=3,
local_radius=2,
max_semitones=2.0,
)
expected = np.array(
[False, False, False, True, True, False, False, False, False, False, False, False, False, False],
dtype=bool,
)
np.testing.assert_array_equal(fill_mask, expected)
def test_chunk_crossfade_scales_above_legacy_floor_for_multi_segment_audio(self):
self.assertEqual(
compute_chunk_crossfade_samples(tgt_sr=48000, t_pad_tgt=144000, segment_count=1),
0,
)
self.assertEqual(
compute_chunk_crossfade_samples(tgt_sr=48000, t_pad_tgt=144000, segment_count=2),
864,
)
self.assertEqual(
compute_chunk_crossfade_samples(tgt_sr=48000, t_pad_tgt=144000, segment_count=5),
1152,
)
class SourceConstraintPolicyTests(unittest.TestCase):
def test_active_echo_frames_keep_nonzero_replace_pressure(self):
activity = np.array([1.0, 1.0, 0.0], dtype=np.float32)
soft_mask = np.array([[0.10, 0.55, 0.10]], dtype=np.float32)
echo_ratio = np.array([[0.90, 0.85, 0.40]], dtype=np.float32)
direct_ratio = np.array([0.10, 0.25, 0.15], dtype=np.float32)
replace = compute_active_source_replace(activity, soft_mask, echo_ratio, direct_ratio)
self.assertGreater(replace[0, 0], 0.40)
self.assertGreater(replace[0, 1], 0.10)
self.assertLess(replace[0, 1], replace[0, 0])
self.assertLessEqual(float(np.max(replace)), 0.82)
def test_cleanup_budget_caps_active_boost_below_two_x(self):
energy_guard = np.array([0.0, 0.5, 1.0], dtype=np.float32)
phrase_activity = np.array([0.0, 0.5, 1.0], dtype=np.float32)
allowed_boost, cleanup_floor = compute_source_cleanup_budget(
energy_guard,
phrase_activity,
)
np.testing.assert_allclose(
allowed_boost,
np.array([0.35, 0.85, 1.35], dtype=np.float32),
atol=1e-6,
)
np.testing.assert_allclose(
cleanup_floor,
np.array([0.62, 0.70, 0.78], dtype=np.float32),
atol=1e-6,
)
self.assertLess(float(np.max(allowed_boost)), 1.5)
class BreathEnergyGateTests(unittest.TestCase):
def test_marginal_quiet_unvoiced_frames_keep_more_feature_than_pitch(self):
energy_db = np.array([-70.0, -60.0, -47.0, -38.0], dtype=np.float32)
unvoiced_mask = np.array([True, True, True, False], dtype=bool)
feature_gate, pitch_gate = compute_breath_preserving_energy_gates(
energy_db=energy_db,
ref_db=-12.0,
unvoiced_mask=unvoiced_mask,
quiet_floor=0.05,
breath_floor=0.28,
breath_active_margin_db=52.0,
transition_width_db=6.0,
)
self.assertAlmostEqual(float(feature_gate[0]), 0.05, places=5)
self.assertAlmostEqual(float(pitch_gate[0]), 0.05, places=5)
self.assertGreater(float(feature_gate[1]), float(pitch_gate[1]))
self.assertGreater(float(feature_gate[1]), 0.13)
self.assertLess(float(pitch_gate[1]), 0.13)
self.assertGreater(float(feature_gate[1]), float(pitch_gate[1]))
self.assertAlmostEqual(float(feature_gate[2]), float(pitch_gate[2]), places=5)
self.assertAlmostEqual(float(feature_gate[3]), float(pitch_gate[3]), places=5)
class SourceRegressionTests(unittest.TestCase):
def test_cover_pipeline_uses_single_conservative_cleanup_budget(self):
source = (REPO_ROOT / "infer" / "cover_pipeline.py").read_text(encoding="utf-8")
self.assertNotIn("allowed_boost = 0.50 + 1.50 * energy_guard", source)
self.assertNotRegex(
source,
r"cleanup_gain\s*=\s*np\.clip\(\s*frame_budget\s*/\s*\(constrained_frame_rms \+ eps\),\s*0\.75 \+ 0\.20 \* phrase_activity",
)
def test_cover_pipeline_gain_parameters_match_runtime_clip(self):
source = (REPO_ROOT / "infer" / "cover_pipeline.py").read_text(encoding="utf-8")
self.assertIn("min_gain=0.85", source)
self.assertIn("max_gain=1.12", source)
self.assertNotIn("min_gain=0.95", source)
self.assertNotIn("max_gain=1.30", source)
self.assertIn("reduction_ratio", source)
self.assertIn("strict_deecho_plus", source)
self.assertIn("_apply_source_breath_cleanup", source)
self.assertIn("_apply_source_transition_cleanup", source)
self.assertIn("Source breath cleanup:", source)
self.assertIn("Source transition cleanup:", source)
self.assertIn("_record_quality_debug", source)
self.assertIn("quality_debug.json", source)
self.assertIn("debug_clips", source)
self.assertIn("VC backend: upstream_official_raw + current postprocess", source)
self.assertIn("convert_vocals_official_upstream(", source)
self.assertNotIn('log.detail("使用当前项目官方封装VC进行转换")\n convert_vocals_official(', source)
def test_official_vc_pipeline_no_longer_forces_hybrid_to_aggressive_crepe(self):
source = (REPO_ROOT / "infer" / "modules" / "vc" / "pipeline.py").read_text(encoding="utf-8")
self.assertNotIn('self.f0_hybrid_mode = "rmvpe+crepe"', source)
def test_config_defaults_match_conservative_hybrid_policy(self):
source = (REPO_ROOT / "configs" / "config.json").read_text(encoding="utf-8")
self.assertRegex(source, r'"f0_hybrid_mode"\s*:\s*"fallback"')
self.assertNotRegex(source, r'"crepe_force_ratio"\s*:\s*0\.0')
def test_official_vc_pipeline_uses_breath_preserving_energy_gates(self):
source = (REPO_ROOT / "infer" / "modules" / "vc" / "pipeline.py").read_text(encoding="utf-8")
self.assertIn("compute_breath_preserving_energy_gates", source)
self.assertIn("self.unvoiced_feature_gate_floor", source)
self.assertIn("compute_chunk_crossfade_samples", source)
self.assertIn("build_conservative_harvest_fill_mask", source)
self.assertIn("分段边界(秒)", source)
def test_config_includes_breath_gate_defaults(self):
source = (REPO_ROOT / "configs" / "config.json").read_text(encoding="utf-8")
self.assertRegex(source, r'"unvoiced_feature_gate_floor"\s*:\s*0\.28')
self.assertRegex(source, r'"breath_active_margin_db"\s*:\s*52\.0')
def test_ui_only_exposes_strict_sota_vc_preprocess_modes(self):
ui_source = (REPO_ROOT / "ui" / "app.py").read_text(encoding="utf-8")
i18n_source = (REPO_ROOT / "i18n" / "zh_CN.json").read_text(encoding="utf-8")
cover_source = (REPO_ROOT / "infer" / "cover_pipeline.py").read_text(encoding="utf-8")
dereverb_source = (REPO_ROOT / "infer" / "advanced_dereverb.py").read_text(encoding="utf-8")
self.assertIn('t("vc_preprocess_auto", "cover"): "auto"', ui_source)
self.assertIn('t("vc_preprocess_uvr_deecho", "cover"): "uvr_deecho"', ui_source)
self.assertNotIn('t("vc_preprocess_direct", "cover"): "direct"', ui_source)
self.assertNotIn('t("vc_preprocess_legacy", "cover"): "legacy"', ui_source)
self.assertIn('vc_preprocess_mode not in {"auto", "uvr_deecho"}', ui_source)
self.assertIn("不可用时停止，不降级", i18n_source)
self.assertNotIn('"vc_preprocess_direct"', i18n_source)
self.assertNotIn('"vc_preprocess_legacy"', i18n_source)
self.assertNotIn("主唱直通", i18n_source)
self.assertNotIn("回退到算法", i18n_source)
self.assertNotIn('self._last_vc_preprocess_mode == "direct"', cover_source)
self.assertNotIn("apply_reverb_to_converted", dereverb_source)
def test_upstream_official_adapter_routes_hybrid_to_conservative_vc_method(self):
source = (REPO_ROOT / "infer" / "official_adapter.py").read_text(encoding="utf-8")
self.assertIn("effective_f0_method = f0_policy.vc_method", source)
self.assertIn("官方F0路由解析", source)
if __name__ == "__main__":
unittest.main()
|