| 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() | |