test(wave119): 抽离noise_reduction_config领域模型 + 39单测 #1014

Closed
xiaoxia wants to merge 1 commits from test/wave119-noise-reduction-config into develop
3 changed files with 509 additions and 177 deletions
@@ -2,126 +2,26 @@
支持对音频进行背景噪音消除、人声增强,适用于语音录制、采访等场景。
使用方式:
config = NoiseReductionConfig(level="medium")
engine = NoiseReductionEngine(config)
filter_str = engine.build_filter(input_label, output_label)
# 结果: [0:a]afftdn=nf=-25[out]
降级策略:
- 参数越界自动钳制
- FFmpeg 不支持 afftdn 时,调用方可捕获异常并跳过
领域模型已抽离至 packages/domain/noise_reduction_config.py,本模块保留薄包装以维持向后兼容。
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from packages.domain.noise_reduction_config import ( # noqa: F401 — 向后兼容
NoiseReductionLevel,
NoiseReductionConfig,
build_afftdn_filter as _build_afftdn_filter_base,
build_arnndn_filter as _build_arnndn_filter_base,
apply_noise_reduction_if_needed as _apply_noise_reduction_if_needed_base,
)
logger = logging.getLogger(__name__)
# ── 降噪等级 ──────────────────────────────────────────────────────────────────
class NoiseReductionLevel(str, Enum):
"""降噪等级预设。"""
LOW = "low" # 轻度降噪,保留细节,适合轻微背景噪音
MEDIUM = "medium" # 中度降噪,平衡效果和音质
HIGH = "high" # 高度降噪,适合嘈杂环境,可能轻微影响音质
CUSTOM = "custom" # 自定义参数
# 各等级对应的降噪参数(afftdn 的 noise floor,单位 dB
# 值越大(越接近 0),降噪越强;值越小(越负),降噪越弱
_LEVEL_PARAMS = {
NoiseReductionLevel.LOW: {
"nf": -35, # 噪音阈值(dB),越负越保守
"tn": -10, # 噪音频谱平滑度
"tr": 50, # 时间分辨率(ms
},
NoiseReductionLevel.MEDIUM: {
"nf": -25,
"tn": -10,
"tr": 50,
},
NoiseReductionLevel.HIGH: {
"nf": -15,
"tn": -5,
"tr": 30,
},
}
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class NoiseReductionConfig:
"""音频降噪配置。
Attributes:
enabled: 是否启用降噪
level: 降噪等级 low/medium/high/custom
noise_floor: 自定义噪音阈值(dB),仅 level=custom 时有效,范围 -60 ~ -5
voice_enhance: 是否启用人声增强
output_format: 输出格式描述(内部使用)
"""
enabled: bool = False
level: NoiseReductionLevel = NoiseReductionLevel.MEDIUM
noise_floor: float = -25.0 # dB
voice_enhance: bool = False
@classmethod
def from_dict(cls, data: dict | None) -> "NoiseReductionConfig":
"""从字典解析配置,参数越界自动钳制。"""
if not data or not data.get("enabled", False):
return cls(enabled=False)
level_str = str(data.get("level", "medium")).lower()
try:
level = NoiseReductionLevel(level_str)
except ValueError:
level = NoiseReductionLevel.MEDIUM
try:
noise_floor = float(data.get("noise_floor", -25.0))
except (TypeError, ValueError):
noise_floor = -25.0
voice_enhance = bool(data.get("voice_enhance", False))
# 钳制到合法范围
noise_floor = max(-60.0, min(-5.0, noise_floor))
return cls(
enabled=True,
level=level,
noise_floor=noise_floor,
voice_enhance=voice_enhance,
)
def has_effect(self) -> bool:
"""判断是否有实际降噪效果。"""
return self.enabled
def get_effective_noise_floor(self) -> float:
"""获取实际生效的噪音阈值(dB)。"""
if self.level == NoiseReductionLevel.CUSTOM:
return self.noise_floor
params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[NoiseReductionLevel.MEDIUM])
return float(params["nf"])
# ── 引擎实现 ──────────────────────────────────────────────────────────────────
class NoiseReductionEngine:
"""音频降噪引擎
"""音频降噪引擎 — 薄包装,实际逻辑在 domain.noise_reduction_config.
基于 FFmpeg afftdnAudio FFt Denoiser)滤镜实现:
- 使用短时傅里叶变换分析音频频谱
@@ -133,97 +33,40 @@ class NoiseReductionEngine:
self.config = config
def build_filter(self, input_label: str, output_label: str) -> str:
"""构建音频降噪滤镜字符串
"""构建音频降噪滤镜字符串.
Args:
input_label: 输入标签,如 "[0:a]""[a0]"
output_label: 输出标签,如 "[nr0]"
Returns:
FFmpeg 滤镜字符串,如 "[a0]afftdn=nf=-25:tn=-10:tr=50[nr0]"
Raises:
ValueError: 配置无效时抛出(调用方应捕获并降级)
FFmpeg 滤镜字符串
"""
if not self.config.has_effect():
return f"{input_label}anull{output_label}"
# 获取参数
if self.config.level == NoiseReductionLevel.CUSTOM:
nf = self.config.noise_floor
tn = -10 # 默认频谱平滑度
tr = 50 # 默认时间分辨率
else:
params = _LEVEL_PARAMS.get(
self.config.level,
_LEVEL_PARAMS[NoiseReductionLevel.MEDIUM],
)
nf = float(params["nf"])
tn = float(params["tn"])
tr = float(params["tr"])
# 构建 afftdn 滤镜
# nf: noise floor (dB)
# tn: temporal noise floor smoothing (dB)
# tr: time resolution (ms)
filter_parts = [f"afftdn=nf={nf}:tn={tn}:tr={tr}"]
# 人声增强:通过 highpass + 轻微压缩实现
if self.config.voice_enhance:
# 1. 高通滤波,去除低频噪音
filter_parts.append("highpass=f=80")
# 2. 轻微压缩,提升人声清晰度
filter_parts.append("acompressor=threshold=-20:ratio=2:attack=5:release=50")
# 3. 响度归一化
filter_parts.append("loudnorm=I=-16:TP=-1.5:LRA=11")
filter_str = f"{input_label}{','.join(filter_parts)}{output_label}"
return filter_str
return _build_afftdn_filter_base(self.config, input_label, output_label)
def build_filter_arnndn(self, input_label: str, output_label: str, model_file: str) -> str:
"""使用 RNN 降噪滤镜(arnndn,效果更好但需要模型文件)
注意:需要额外下载 RNNNoise 模型文件,默认使用 afftdn(无需额外依赖)。
"""使用 RNN 降噪滤镜(arnndn,效果更好但需要模型文件).
Args:
input_label: 输入标签
output_label: 输出标签
model_file: RNNNoise 模型文件路径.rnnn 格式)
model_file: RNNNoise 模型文件路径
Returns:
FFmpeg 滤镜字符串
"""
if not self.config.has_effect():
return f"{input_label}anull{output_label}"
return f"{input_label}arnndn=m={model_file}{output_label}"
return _build_arnndn_filter_base(self.config, input_label, output_label, model_file)
def apply_noise_reduction_if_needed(
config_data: dict | None,
input_label: str,
output_label: str,
) -> Optional[str]:
"""便捷函数:根据配置判断是否需要应用音频降噪。
def apply_noise_reduction_if_needed(config_data, input_label: str, output_label: str):
"""便捷函数:根据配置判断是否需要应用音频降噪.
Args:
config_data: 降噪配置字典(从 plan.config.audio_noise_reduction 或 clip.config.noise_reduction 读取)
config_data: 降噪配置字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要降噪时返回 None
"""
if not config_data:
return None
try:
config = NoiseReductionConfig.from_dict(config_data)
if not config.has_effect():
return None
engine = NoiseReductionEngine(config)
return engine.build_filter(input_label, output_label)
except Exception as e:
logger.warning("[noise-reduction] 应用降噪失败,跳过: %s", e)
return None
return _apply_noise_reduction_if_needed_base(config_data, input_label, output_label)
+231
View File
@@ -0,0 +1,231 @@
"""音频降噪配置领域模型 — 纯逻辑,无FFmpeg依赖.
抽离自 noise_reduction_engine.py,包含:
- NoiseReductionLevel 枚举(low/medium/high/custom
- NoiseReductionConfig 数据类(解析/钳制/效果判断)
- 等级预设参数
- afftdn / arnndn 滤镜构建
- 便捷函数(apply_noise_reduction_if_needed
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
# ── 降噪等级 ──────────────────────────────────────────────────────────────────
class NoiseReductionLevel(str, Enum):
"""降噪等级预设."""
LOW = "low" # 轻度降噪,保留细节,适合轻微背景噪音
MEDIUM = "medium" # 中度降噪,平衡效果和音质
HIGH = "high" # 高度降噪,适合嘈杂环境,可能轻微影响音质
CUSTOM = "custom" # 自定义参数
# 各等级对应的降噪参数(afftdn 的 noise floor,单位 dB
# 值越大(越接近 0),降噪越强;值越小(越负),降噪越弱
_LEVEL_PARAMS: dict[NoiseReductionLevel, dict[str, float]] = {
NoiseReductionLevel.LOW: {
"nf": -35, # 噪音阈值(dB),越负越保守
"tn": -10, # 噪音频谱平滑度
"tr": 50, # 时间分辨率(ms
},
NoiseReductionLevel.MEDIUM: {
"nf": -25,
"tn": -10,
"tr": 50,
},
NoiseReductionLevel.HIGH: {
"nf": -15,
"tn": -5,
"tr": 30,
},
}
# 参数范围
MIN_NOISE_FLOOR = -60.0
MAX_NOISE_FLOOR = -5.0
# 默认值
DEFAULT_LEVEL = NoiseReductionLevel.MEDIUM
DEFAULT_NOISE_FLOOR = -25.0
# ── 配置模型 ──────────────────────────────────────────────────────────────────
@dataclass
class NoiseReductionConfig:
"""音频降噪配置.
Attributes:
enabled: 是否启用降噪
level: 降噪等级 low/medium/high/custom
noise_floor: 自定义噪音阈值(dB),仅 level=custom 时有效,范围 -60 ~ -5
voice_enhance: 是否启用人声增强
"""
enabled: bool = False
level: NoiseReductionLevel = DEFAULT_LEVEL
noise_floor: float = DEFAULT_NOISE_FLOOR # dB
voice_enhance: bool = False
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> NoiseReductionConfig:
"""从字典解析配置,参数越界自动钳制."""
if not data or not data.get("enabled", False):
return cls(enabled=False)
level_str = str(data.get("level", "medium")).lower()
try:
level = NoiseReductionLevel(level_str)
except ValueError:
level = DEFAULT_LEVEL
try:
noise_floor = float(data.get("noise_floor", DEFAULT_NOISE_FLOOR))
except (TypeError, ValueError):
noise_floor = DEFAULT_NOISE_FLOOR
voice_enhance = bool(data.get("voice_enhance", False))
# 钳制到合法范围
noise_floor = max(MIN_NOISE_FLOOR, min(MAX_NOISE_FLOOR, noise_floor))
return cls(
enabled=True,
level=level,
noise_floor=noise_floor,
voice_enhance=voice_enhance,
)
def has_effect(self) -> bool:
"""判断是否有实际降噪效果."""
return self.enabled
def get_effective_noise_floor(self) -> float:
"""获取实际生效的噪音阈值(dB."""
if self.level == NoiseReductionLevel.CUSTOM:
return self.noise_floor
params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[DEFAULT_LEVEL])
return float(params["nf"])
def get_level_params(self) -> dict[str, float]:
"""获取当前等级的完整参数字典."""
if self.level == NoiseReductionLevel.CUSTOM:
return {
"nf": self.noise_floor,
"tn": -10.0,
"tr": 50.0,
}
params = _LEVEL_PARAMS.get(self.level, _LEVEL_PARAMS[DEFAULT_LEVEL])
return {k: float(v) for k, v in params.items()}
def validate(self) -> tuple[bool, str]:
"""校验配置是否有效."""
if not self.enabled:
return True, ""
if not (MIN_NOISE_FLOOR <= self.noise_floor <= MAX_NOISE_FLOOR):
return False, f"noise_floor 必须在 {MIN_NOISE_FLOOR}~{MAX_NOISE_FLOOR} dB 之间"
return True, ""
# ── 滤镜构建 ────────────────────────────────────────────────────────────────
def build_afftdn_filter(
config: NoiseReductionConfig,
input_label: str,
output_label: str,
) -> str:
"""构建 afftdn 音频降噪滤镜字符串.
Args:
config: 降噪配置
input_label: 输入标签,如 "[0:a]""[a0]"
output_label: 输出标签,如 "[nr0]"
Returns:
FFmpeg 滤镜字符串
"""
if not config.has_effect():
return f"{input_label}anull{output_label}"
params = config.get_level_params()
nf = params["nf"]
tn = params["tn"]
tr = params["tr"]
# 构建 afftdn 滤镜
filter_parts = [f"afftdn=nf={nf}:tn={tn}:tr={tr}"]
# 人声增强:通过 highpass + 压缩 + 响度归一化实现
if config.voice_enhance:
filter_parts.append("highpass=f=80")
filter_parts.append("acompressor=threshold=-20:ratio=2:attack=5:release=50")
filter_parts.append("loudnorm=I=-16:TP=-1.5:LRA=11")
return f"{input_label}{','.join(filter_parts)}{output_label}"
def build_arnndn_filter(
config: NoiseReductionConfig,
input_label: str,
output_label: str,
model_file: str,
) -> str:
"""使用 RNN 降噪滤镜(arnndn,效果更好但需要模型文件).
注意:需要额外下载 RNNNoise 模型文件,默认使用 afftdn(无需额外依赖)。
"""
if not config.has_effect():
return f"{input_label}anull{output_label}"
return f"{input_label}arnndn=m={model_file}{output_label}"
# ── 便捷函数 ────────────────────────────────────────────────────────────────
def apply_noise_reduction_if_needed(
config_data: dict[str, Any] | None,
input_label: str,
output_label: str,
) -> str | None:
"""便捷函数:根据配置判断是否需要应用音频降噪.
Args:
config_data: 降噪配置字典
input_label: 输入标签
output_label: 输出标签
Returns:
滤镜字符串,不需要降噪时返回 None
"""
if not config_data:
return None
try:
config = NoiseReductionConfig.from_dict(config_data)
if not config.has_effect():
return None
return build_afftdn_filter(config, input_label, output_label)
except Exception as e:
logger.warning("[noise-reduction] 应用降噪失败,跳过: %s", e)
return None
def get_level_names() -> list[str]:
"""获取所有降噪等级名称列表."""
return [level.value for level in NoiseReductionLevel]
+258
View File
@@ -0,0 +1,258 @@
"""noise_reduction_config 领域模型单测."""
from __future__ import annotations
import pytest
from packages.domain.noise_reduction_config import (
DEFAULT_LEVEL,
DEFAULT_NOISE_FLOOR,
MAX_NOISE_FLOOR,
MIN_NOISE_FLOOR,
NoiseReductionConfig,
NoiseReductionLevel,
apply_noise_reduction_if_needed,
build_afftdn_filter,
build_arnndn_filter,
get_level_names,
)
# ── NoiseReductionLevel 枚举测试 ───────────────────────────────────────────
class TestNoiseReductionLevel:
def test_four_levels(self):
assert len(NoiseReductionLevel) == 4
def test_level_values(self):
assert NoiseReductionLevel.LOW.value == "low"
assert NoiseReductionLevel.MEDIUM.value == "medium"
assert NoiseReductionLevel.HIGH.value == "high"
assert NoiseReductionLevel.CUSTOM.value == "custom"
def test_from_string(self):
assert NoiseReductionLevel("low") == NoiseReductionLevel.LOW
assert NoiseReductionLevel("medium") == NoiseReductionLevel.MEDIUM
assert NoiseReductionLevel("high") == NoiseReductionLevel.HIGH
assert NoiseReductionLevel("custom") == NoiseReductionLevel.CUSTOM
# ── NoiseReductionConfig.from_dict 测试 ───────────────────────────────────
class TestNoiseReductionConfigFromDict:
def test_none_returns_disabled(self):
cfg = NoiseReductionConfig.from_dict(None)
assert cfg.enabled is False
def test_empty_dict_returns_disabled(self):
cfg = NoiseReductionConfig.from_dict({})
assert cfg.enabled is False
def test_disabled_returns_disabled(self):
cfg = NoiseReductionConfig.from_dict({"enabled": False})
assert cfg.enabled is False
def test_enabled_default_params(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True})
assert cfg.enabled is True
assert cfg.level == NoiseReductionLevel.MEDIUM
assert cfg.noise_floor == DEFAULT_NOISE_FLOOR
assert cfg.voice_enhance is False
def test_custom_level(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -30.0})
assert cfg.level == NoiseReductionLevel.CUSTOM
assert cfg.noise_floor == -30.0
def test_invalid_level_defaults_medium(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "invalid"})
assert cfg.level == NoiseReductionLevel.MEDIUM
def test_noise_floor_clamped_low(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": -100.0})
assert cfg.noise_floor == MIN_NOISE_FLOOR
def test_noise_floor_clamped_high(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": 0.0})
assert cfg.noise_floor == MAX_NOISE_FLOOR
def test_invalid_noise_floor_type_uses_default(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "custom", "noise_floor": "not_a_number"})
assert cfg.noise_floor == DEFAULT_NOISE_FLOOR
def test_voice_enhance_true(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "voice_enhance": True})
assert cfg.voice_enhance is True
def test_case_insensitive_level(self):
cfg = NoiseReductionConfig.from_dict({"enabled": True, "level": "HIGH"})
assert cfg.level == NoiseReductionLevel.HIGH
# ── has_effect / get_effective_noise_floor 测试 ───────────────────────────
class TestConfigProperties:
def test_disabled_no_effect(self):
cfg = NoiseReductionConfig(enabled=False)
assert cfg.has_effect() is False
def test_enabled_has_effect(self):
cfg = NoiseReductionConfig(enabled=True)
assert cfg.has_effect() is True
def test_effective_noise_floor_low(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.LOW)
assert cfg.get_effective_noise_floor() == -35.0
def test_effective_noise_floor_medium(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
assert cfg.get_effective_noise_floor() == -25.0
def test_effective_noise_floor_high(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.HIGH)
assert cfg.get_effective_noise_floor() == -15.0
def test_effective_noise_floor_custom(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-40.0)
assert cfg.get_effective_noise_floor() == -40.0
def test_get_level_params_medium(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
params = cfg.get_level_params()
assert params["nf"] == -25.0
assert params["tn"] == -10.0
assert params["tr"] == 50.0
def test_get_level_params_custom(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-30.0)
params = cfg.get_level_params()
assert params["nf"] == -30.0
assert "tn" in params
assert "tr" in params
# ── validate 测试 ─────────────────────────────────────────────────────────
class TestValidate:
def test_disabled_valid(self):
cfg = NoiseReductionConfig(enabled=False)
ok, msg = cfg.validate()
assert ok is True
assert msg == ""
def test_enabled_valid(self):
cfg = NoiseReductionConfig(enabled=True, noise_floor=-25.0)
ok, msg = cfg.validate()
assert ok is True
def test_noise_floor_out_of_range(self):
cfg = NoiseReductionConfig(enabled=True, noise_floor=-100.0)
ok, msg = cfg.validate()
assert ok is False
assert "noise_floor" in msg
# ── build_afftdn_filter 测试 ───────────────────────────────────────────────
class TestBuildAfftdnFilter:
def test_disabled_returns_anull(self):
cfg = NoiseReductionConfig(enabled=False)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "anull" in result
assert "[in]" in result
assert "[out]" in result
def test_medium_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.MEDIUM)
result = build_afftdn_filter(cfg, "[a]", "[nr]")
assert "afftdn=" in result
assert "nf=-25.0" in result or "nf=-25" in result
assert "[a]" in result
assert "[nr]" in result
def test_high_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.HIGH)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "afftdn=" in result
assert "nf=-15.0" in result or "nf=-15" in result
def test_low_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.LOW)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "afftdn=" in result
assert "nf=-35.0" in result or "nf=-35" in result
def test_custom_level(self):
cfg = NoiseReductionConfig(enabled=True, level=NoiseReductionLevel.CUSTOM, noise_floor=-40.0)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "afftdn=" in result
assert "nf=-40.0" in result or "nf=-40" in result
def test_voice_enhance_adds_filters(self):
cfg = NoiseReductionConfig(enabled=True, voice_enhance=True)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "highpass" in result
assert "acompressor" in result
assert "loudnorm" in result
def test_no_voice_enhance_no_extra_filters(self):
cfg = NoiseReductionConfig(enabled=True, voice_enhance=False)
result = build_afftdn_filter(cfg, "[in]", "[out]")
assert "highpass" not in result
assert "acompressor" not in result
# ── build_arnndn_filter 测试 ───────────────────────────────────────────────
class TestBuildArnndnFilter:
def test_disabled_returns_anull(self):
cfg = NoiseReductionConfig(enabled=False)
result = build_arnndn_filter(cfg, "[in]", "[out]", "model.rnnn")
assert "anull" in result
def test_enabled_returns_arnndn(self):
cfg = NoiseReductionConfig(enabled=True)
result = build_arnndn_filter(cfg, "[a]", "[nr]", "/path/to/model.rnnn")
assert "arnndn=" in result
assert "m=/path/to/model.rnnn" in result
assert "[a]" in result
assert "[nr]" in result
# ── apply_noise_reduction_if_needed 测试 ─────────────────────────────────
class TestApplyNoiseReductionIfNeeded:
def test_none_config_returns_none(self):
assert apply_noise_reduction_if_needed(None, "[in]", "[out]") is None
def test_disabled_returns_none(self):
assert apply_noise_reduction_if_needed({"enabled": False}, "[in]", "[out]") is None
def test_enabled_returns_filter(self):
result = apply_noise_reduction_if_needed({"enabled": True, "level": "medium"}, "[in]", "[out]")
assert result is not None
assert "afftdn" in result
def test_invalid_config_handles_exception(self):
# 异常情况应该返回 None 而不是抛出
result = apply_noise_reduction_if_needed("invalid", "[in]", "[out]")
assert result is None
# ── 工具函数测试 ───────────────────────────────────────────────────────────
class TestUtils:
def test_get_level_names_returns_four(self):
names = get_level_names()
assert len(names) == 4
assert "low" in names
assert "medium" in names
assert "high" in names
assert "custom" in names