diff --git a/nodes/vad_profiler.py b/nodes/vad_profiler.py new file mode 100644 index 0000000..a81de4b --- /dev/null +++ b/nodes/vad_profiler.py @@ -0,0 +1,311 @@ +"""每视频自适应 VAD 调参(信号分析 + 片段网格验证)。 + +背景(实测 CJOD-255):全程 BGM 覆盖的视频几乎没有纯静音,固定 VAD +参数会把音乐当语音,导致 whisper 全段解码 → 碎片化(80%)、漏句(32%)、 +敏感段丢失。不同视频声学差异大,需**每视频独立分析后再定 VAD 参数**。 + +方案(用户确认): +1. 信号分析(秒级,无模型)缩窄参数空间:1s 能量分布 → 静音比例、 + BGM 底噪、语音疏密,推出 threshold/min_silence/speech_pad 候选; +2. 片段小网格验证(90s 代表片段,3~5 组参数跑 whisper): + - **不用参考字幕**(实际部署无参考):纯转录质量启发式评分; + - **常出现的幻觉词作为扣分项**:幻觉越多 → VAD 把非语音当语音 → + 该参数组合越差。 +3. 全片用选出的 vad_parameters 跑一次。 + +门控:whisper 节点 detected 未显式传 vad_parameters 且 vad_filter=true 时 +自动调用本模块;WOV_AUTO_VAD=0 可关闭。 +""" + +from __future__ import annotations + +import math +import os +import re +from dataclasses import dataclass, field +from pathlib import Path + +import wave # 仅读 WAV 头/样本 + +# 幻觉词扣分集合:VAD 若把音乐/静音当语音,whisper 常在这些段输出套话。 +# 每条命中扣 1 分;重复次数越高 → 该参数组合越差。 +HALLUCINATION_TOKENS = ( + "ご視聴ありがとうございました", # 感谢观看 + "ご視聴ありがとうございました。", + "おやすみなさい", # 晚安 + "音楽", # 音乐 + "また見てね", # 下次再见 + "Goodbye", + "ありがとうございました", # 感谢 + "エンドロール", + "お楽しみに", +) + +# 单字/语气词(碎片化判定):纯这些字符的条目视为"碎声"。 +_FRAGMENT_RE = re.compile(r"^[ぁ-んァ-ヶあいうえおっーんすよわ\s、。]+$") + +# 默认候选网格(片段验证时用;由信号分析缩小范围)。 +_THRESHOLDS = (0.3, 0.5, 0.7) +_SILENCES_MS = (300, 1000, 2000) +_PADS_MS = (0, 200, 400) + + +@dataclass +class AudioProfile: + """音频信号分析结果。""" + + sample_rate: int = 16000 + duration_seconds: float = 0.0 + rms_bins: list[float] = field(default_factory=list) # 每 1s RMS + silence_ratio: float = 0.0 # RMS<300(静音级)占比 + lowish_ratio: float = 0.0 # RMS<900(音乐/低语)占比 + voice_ratio: float = 0.0 # 高能(语音)占比 + bgm_heavy: bool = False # 是否 BGM 覆盖广(低能量占比高但非纯静音) + long_silence: bool = False # 是否长停顿常见 + + @property + def median_rms(self) -> float: + if not self.rms_bins: + return 0.0 + ordered = sorted(self.rms_bins) + return ordered[len(ordered) // 2] + + +def _read_wav_samples(audio_path: Path, sample_rate: int) -> tuple[list[int], int]: + """读取单声道 WAV 的全部样本(16-bit 小端)。返回 (样本列表, 实际采样率)。""" + with wave.open(str(audio_path), "rb") as wav: + rate = wav.getframerate() + n = wav.getnframes() + data = wav.readframes(n) + import array + + samples = array.array("h") + samples.frombytes(data) + return samples, rate + + +def profile_audio(audio_path: Path, sample_rate: int = 16000) -> AudioProfile: + """做 1s 网格能量分析,返回 AudioProfile。 + + 每秒计算 RMS;据此得出静音/音乐/语音占比,判断是否 BGM 覆盖、有无长静音。 + """ + samples, rate = _read_wav_samples(audio_path, sample_rate) + if rate != sample_rate: + # 采样率不一致时用 wave 实际 rate 近似(不重采样,仅用于比例判断)。 + rate = rate or sample_rate + bin_frames = rate # 1 秒一个 bin + bins: list[float] = [] + n_bins = len(samples) // bin_frames + for i in range(n_bins): + seg = samples[i * bin_frames : (i + 1) * bin_frames] + rms = math.sqrt(sum(s * s for s in seg) / len(seg)) if seg else 0 + bins.append(rms) + + if not bins: + return AudioProfile(sample_rate=sample_rate, duration_seconds=0.0) + + silence = sum(1 for r in bins if r < 300) / len(bins) + lowish = sum(1 for r in bins if r < 900) / len(bins) + voice = sum(1 for r in bins if r >= 900) / len(bins) + + # BGM 覆盖:低能量占比高(音频不静也不高亢),说明有音乐铺底。 + bgm_heavy = lowish > 0.4 and silence < 0.4 + # 长静音:存在连续 >=5 个静音 bin(5s 以上停顿)。 + long_silence = False + run = 0 + for r in bins: + if r < 300: + run += 1 + if run >= 5: + long_silence = True + break + else: + run = 0 + + return AudioProfile( + sample_rate=rate, + duration_seconds=len(samples) / rate, + rms_bins=bins, + silence_ratio=silence, + lowish_ratio=lowish, + voice_ratio=voice, + bgm_heavy=bgm_heavy, + long_silence=long_silence, + ) + + +def suggest_vad_parameters(profile: AudioProfile) -> dict: + """由信号分析推出候选 VAD 参数(不跑模型,秒级)。 + + 返回 dict 含 threshold/min_silence_duration_ms/speech_pad_ms 候选。 + 规则见 docs/adaptive_vad.md。 + """ + if profile.bgm_heavy: + # BGM 覆盖广:降低 threshold 增强人声敏感、减 min_silence 防并句、 + # 减 speech_pad 防时间偏移。 + return { + "threshold": 0.3, + "min_silence_duration_ms": 300, + "speech_pad_ms": 0, + } + if profile.long_silence: + # 长停顿常见:speech_pad 减小防时间轴压缩漂移,threshold 正常。 + return { + "threshold": 0.5, + "min_silence_duration_ms": 1000, + "speech_pad_ms": 200, + } + if profile.silence_ratio > 0.3: + # 静音占比高(正常带静音的音频):threhold 略高剔除虚警。 + return { + "threshold": 0.6, + "min_silence_duration_ms": 2000, + "speech_pad_ms": 400, + } + # 常规。 + return { + "threshold": 0.5, + "min_silence_duration_ms": 1000, + "speech_pad_ms": 400, + } + + +def _hits_hallucination(text: str) -> int: + """统计一条文本命中幻觉词的次数(多个套话可叠加)。""" + hits = 0 + for token in HALLUCINATION_TOKENS: + if token in text: + hits += 1 + return hits + + +def score_transcript(segments: list) -> float: + """启发式评分转录质量(不用参考字幕)。 + + 分数越高越好: + - 碎片率(纯单字/语气词条目占比)越低越好:penalty += 碎片数 * 2 + - 幻觉词(感谢观看/晚安/音乐)越多越差:penalty += 幻觉命中次数 * 3 + - 平均每条字数适中(不碎不并):penalty += max(0, 平均字数-30) * 0.5 + 返回 100 - penalty。 + """ + if not segments: + return 0.0 + n = len(segments) + frag = 0 + hall = 0 + total_chars = 0 + for seg in segments: + text = getattr(seg, "text", "") + if not text: + continue + total_chars += len(text) + if _FRAGMENT_RE.match(text): + frag += 1 + hall += _hits_hallucination(text) + avg_len = total_chars / n if n else 0 + penalty = frag * 2.0 + hall * 3.0 + max(0.0, (avg_len - 20)) * 0.3 + return max(0.0, 100.0 - penalty) + + +def vad_parameters_for_audio( + audio_path: Path, + sample_rate: int = 16000, + whisper_invoke=None, + run_dir: Path | None = None, + chunk_seconds: int = 60, +) -> dict: + """为单个音频决定最优 VAD 参数(信号分析 + 片段网格验证)。 + + 若不提供 whisper_invoke(无法跑片段),退化为仅信号分析启发式。 + """ + profile = profile_audio(audio_path, sample_rate) + suggested = suggest_vad_parameters(profile) + + if whisper_invoke is None or not profile.rms_bins: + # 无推理能力:直接用信号分析结论,不再网格验证。 + return suggested + + # 片段网格:取 1 段 90s 代表性片段(选能量中等段,避免纯音乐/纯静音)。 + frag_start = _pick_representative_start(profile, 90) + fragment_params = _grid_search_vad( + whisper_invoke, audio_path, run_dir, chunk_seconds, frag_start, suggested + ) + return fragment_params + + +def _pick_representative_start(profile: AudioProfile, window: int) -> int: + """从 1s 能量里选一段 90s 窗口(平均能量接近中位、不纯静音)。 + + 返回窗口起始秒;音频不足 window 秒时返回 0。 + """ + bins = profile.rms_bins + n = len(bins) + if n <= window: + return 0 + median = profile.median_rms + best = 0 + best_score = float("inf") + for start in range(0, n - window, max(1, window // 2)): + window_bins = bins[start : start + window] + avg = sum(window_bins) / len(window_bins) + # 分数 = 与中位能量差的绝对值 + 静音惩罚(窗口全静音不好)。 + silence = sum(1 for r in window_bins if r < 300) / len(window_bins) + score = abs(avg - median) + (silence * 1000 if silence > 0.6 else 0) + if score < best_score: + best_score = score + best = start + return best + + +def _grid_search_vad( + whisper_invoke, + audio_path: Path, + run_dir: Path | None, + chunk_seconds: int, + frag_start: int, + suggested: dict, +) -> dict: + """对候选参数网格各跑一次片段 whisper,按启发式评分选最优。 + + 候选 = 信号分析建议值 + 少量邻域组合(_candidate_params), + 每组合在 60s 片段上转写,按 score_transcript 选最高分。 + 若无法运行 whisper(如缺模型)返回 suggested。 + """ + window_seconds = 60 + cand = _candidate_params(suggested) + best = suggested + best_score = -1.0 + for params in cand: + try: + segs = whisper_invoke( + audio_path=audio_path, + start_seconds=frag_start, + window_seconds=window_seconds, + vad_parameters=params, + output_dir=str(run_dir / "vad_probe") if run_dir else None, + chunk_seconds=chunk_seconds, + ) + except Exception: + # 单组参数调用失败不致命,跳过继续评估其它组合。 + continue + if not isinstance(segs, list): + continue + score = score_transcript(segs) + if score > best_score: + best_score = score + best = params + return best + + +def _candidate_params(suggested: dict) -> list[dict]: + """由建议值扩展出候选网格(附近组合,避免过深搜索)。""" + base = dict(suggested) + thres = base.get("threshold", 0.5) + sil = base.get("min_silence_duration_ms", 1000) + pad = base.get("speech_pad_ms", 400) + cand = [] + for t in (thres, round(thres - 0.1, 1), round(thres + 0.1, 1)): + for s in (sil, sil // 2, sil * 2): + for p in (0, pad // 2, pad): + cand.append({"threshold": t, "min_silence_duration_ms": s, "speech_pad_ms": p}) + return cand \ No newline at end of file diff --git a/tests/test_vad_profiler.py b/tests/test_vad_profiler.py new file mode 100644 index 0000000..c91244c --- /dev/null +++ b/tests/test_vad_profiler.py @@ -0,0 +1,252 @@ +"""每视频自适应 VAD 调参测试(先红后绿)。 + +验证信号分析 profile_audio、启发式参数建议 suggest_vad_parameters、 +转录质量评分 score_transcript(幻觉词扣分)与 vad_parameters_for_audio +(信号分析 + 片段网格验证)。 + +评分**不依赖参考字幕**(实际部署无参考),用转录质量 + 幻觉词扣分。 +""" + +from __future__ import annotations + +import wave +from pathlib import Path + +import pytest + +from nodes.vad_profiler import ( + AudioProfile, + HALLUCINATION_TOKENS, + profile_audio, + score_transcript, + suggest_vad_parameters, + vad_parameters_for_audio, + _pick_representative_start, +) + + +class _Seg: + """模拟 whisper segment:仅含 text。""" + + def __init__(self, text): + self.text = text + + +def _make_wav(path: Path, silence_seconds: int, voice_seconds: int) -> None: + """生成 [静音 N 秒 + 语音 N 秒] 的 16kHz 单声道 WAV。 + + silence 用 0 样本(静音),voice 用较大振幅样本(语音)。 + """ + import array + + rate = 16000 + silence = array.array("h", [0] * rate * silence_seconds) + voice = array.array("h", [8000] * rate * voice_seconds) + samples = silence + voice + with wave.open(str(path), "wb") as f: + f.setnchannels(1) + f.setsampwidth(2) + f.setframerate(rate) + f.writeframes(samples.tobytes()) + + +def test_profile_audio_detects_bgm_heavy() -> None: + """纯静音+语音的视频:silence_ratio 高、非 BGM 覆盖;mediam_rms 合理。""" + wav = Path("/tmp/test_vad_plain.wav") + _make_wav(wav, silence_seconds=40, voice_seconds=10) + p = profile_audio(wav, 16000) + assert p.silence_ratio > 0.5 + assert p.bgm_heavy is False + assert p.duration_seconds == pytest.approx(50.0, abs=1) + + +def test_suggest_vad_parameters_plain_silence() -> None: + """静音占比高且无长停顿 -> 建议 threshold 偏高(静音权重分支)。""" + wav = Path("/tmp/test_vad_plain2.wav") + # 交替短静音避免触发 long_silence(<5s 连续静音)。 + _make_wav(wav, silence_seconds=3, voice_seconds=1) + _make_wav(wav, silence_seconds=3, voice_seconds=1) + p = profile_audio(wav, 16000) + assert p.long_silence is False + params = suggest_vad_parameters(p) + assert params["threshold"] >= 0.5 + + +def test_suggest_vad_parameters_bgm() -> None: + """BGM 覆盖(低能量占比高但静音少)-> threshold 降低、min_silence 减小。""" + p = AudioProfile( + silence_ratio=0.1, # 几乎无静音 + lowish_ratio=0.6, # 大量低能量(音乐) + bgm_heavy=True, # 直接标记 BGM 覆盖 + long_silence=False, + rms_bins=[500] * 100, + ) + params = suggest_vad_parameters(p) + assert params["threshold"] < 0.5 # 降低 + assert params["min_silence_duration_ms"] < 1000 # 减小 + + +def test_score_transcript_penalizes_hallucination() -> None: + """幻觉词(感谢观看/晚安/音乐)多 -> 评分低。""" + good = [_Seg("ありがとうございます本日は"), _Seg("かしこまりました")] + bad = [_Seg("ご視聴ありがとうございました"), _Seg("おやすみなさい"), _Seg("音楽")] + assert score_transcript(good) > score_transcript(bad) + + +def test_score_transcript_penalizes_fragments() -> None: + """碎片(纯单字/语气词)多 -> 评分低。""" + clean = [_Seg("今日はとても暑いですね"), _Seg("それでは始めましょう")] + frag = [_Seg("あ"), _Seg("うん"), _Seg("はい"), _Seg("あっ")] + assert score_transcript(clean) > score_transcript(frag) + + +def test_vad_parameters_for_audio_uses_grid_when_provider() -> None: + """提供 whisper 回调时:从候选网格选评分最高的参数(threshold 最小者)。""" + wav = Path("/tmp/test_vad_grid.wav") + _make_wav(wav, silence_seconds=10, voice_seconds=5) + + def fake_whisper(**kwargs): + # 假设最优 = threshold 最小的候选(suggest 0.5 时候选含 0.4)。 + t = kwargs.get("vad_parameters", {}).get("threshold", 0.5) + if t <= 0.4: + # 最优组合:无幻觉的正常长句(分数最高)。 + return [_Seg("今日はお客様のために精神整備を務めさせていただきます")] + # 其它组合:幻觉套话(分数低)。 + return [_Seg("ご視聴ありがとうございました"), _Seg("おやすみなさい")] + + params = vad_parameters_for_audio( + wav, sample_rate=16000, whisper_invoke=fake_whisper, run_dir=Path("/tmp") + ) + # 网格应从候选里选出评分最高的 threshold=0.4 的组合。 + assert params["threshold"] == pytest.approx(0.4, abs=0.1) + + +def test_vad_parameters_for_audio_fallback_without_provider() -> None: + """无 whisper 回调:退化为信号分析建议,不报错。""" + wav = Path("/tmp/test_vad_noprov.wav") + _make_wav(wav, silence_seconds=10, voice_seconds=5) + params = vad_parameters_for_audio(wav, sample_rate=16000, whisper_invoke=None) + assert "threshold" in params + assert "min_silence_duration_ms" in params + + +def test_hallucination_tokens_included() -> None: + """幻觉词集合应含常用套话(感谢观看/晚安/音乐)。""" + all_str = " ".join(HALLUCINATION_TOKENS) + assert "ご視聴ありがとうございました" in all_str + assert "おやすみなさい" in all_str + assert "音楽" in all_str + +def test_suggest_vad_parameters_long_silence() -> None: + """长停顿常见 -> 建议 threshold 0.5、speech_pad 200(防时间轴压缩漂移)。""" + p = AudioProfile( + silence_ratio=0.2, lowish_ratio=0.3, bgm_heavy=False, long_silence=True, + rms_bins=[500] * 100, + ) + params = suggest_vad_parameters(p) + assert params["threshold"] == 0.5 + assert params["speech_pad_ms"] == 200 + + +def test_suggest_vad_parameters_regular() -> None: + """常规音频(无特殊标记)-> 默认 0.5/1000/400。""" + p = AudioProfile( + silence_ratio=0.2, lowish_ratio=0.3, bgm_heavy=False, long_silence=False, + rms_bins=[1500] * 100, + ) + params = suggest_vad_parameters(p) + assert params["threshold"] == 0.5 + assert params["min_silence_duration_ms"] == 1000 + assert params["speech_pad_ms"] == 400 + + +def test_pick_representative_start_all_silence() -> None: + """全静音音频:窗口恐怖沉默用 score>0.55 惩罚,仍返回起始 0。""" + p = AudioProfile( + rms_bins=[50] * 100, silence_ratio=0.9, lowish_ratio=0.9, + bgm_heavy=False, long_silence=False, + ) + start = _pick_representative_start(p, 90) + assert start >= 0 and start < 10 + + +def test_grid_search_exception_skipped() -> None: + """网格搜索某组参数调用抛异常:跳过该组,不崩溃。""" + called = {"n": 0} + + def failing_whisper(**kwargs): + called["n"] += 1 + raise RuntimeError("mock download fail") + + from nodes.vad_profiler import _grid_search_vad + + params = _grid_search_vad( + failing_whisper, Path("/tmp/x.wav"), Path("/tmp"), 60, 0, + {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400}, + ) + # 全部失败时回退 suggested。 + assert params == {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400} + assert called["n"] > 0 + + +def test_candidate_params_expands_grid() -> None: + """候选网格应含建议值邻域(threshold±0.1、silence 倍、pad 组合)。""" + from nodes.vad_profiler import _candidate_params + + cands = _candidate_params( + {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400} + ) + assert len(cands) == 27 + assert {"threshold": 0.4, "min_silence_duration_ms": 500, "speech_pad_ms": 0} in cands + assert {"threshold": 0.6, "min_silence_duration_ms": 2000, "speech_pad_ms": 400} in cands + + +def test_median_rms_empty_returns_zero() -> None: + """rms_bins 为空时 median_rms 返回 0。""" + p = AudioProfile(rms_bins=[]) + assert p.median_rms == 0.0 + + +def test_profile_audio_empty_wav() -> None: + """空的 WAV(无样本)-> 返回 duration 0 的空 profile,不崩溃。""" + import array + wav = Path("/tmp/test_vad_empty.wav") + with wave.open(str(wav), "wb") as f: + f.setnchannels(1); f.setsampwidth(2); f.setframerate(16000) + f.writeframes(array.array("h", []).tobytes()) + p = profile_audio(wav, 16000) + assert p.duration_seconds == 0.0 + + +def test_grid_search_ignores_non_list() -> None: + """网格搜索回调返回非 list(如 None)应被跳过,回退建议值。""" + from nodes.vad_profiler import _grid_search_vad + + params = _grid_search_vad( + lambda **kw: None, Path("/tmp/x.wav"), Path("/tmp"), 60, 0, + {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400}, + ) + assert params == {"threshold": 0.5, "min_silence_duration_ms": 1000, "speech_pad_ms": 400} + + +def test_score_transcript_empty_list() -> None: + """空列表评分返回 0。""" + assert score_transcript([]) == 0.0 + + +def test_score_transcript_ignores_empty_text_seg() -> None: + """含空文本的 segment 被跳过,不报错。""" + assert score_transcript([_Seg(""), _Seg("今日は暑いです")]) > 0 + + +def test_profile_audio_different_sample_rate() -> None: + """采样率与请求不一致时用 wave 实际 rate 近似,不崩溃。""" + import array + wav = Path("/tmp/test_vad_rate.wav") + rate = 8000 + samples = array.array("h", [5000] * rate * 2) # 2s 语音 + with wave.open(str(wav), "wb") as f: + f.setnchannels(1); f.setsampwidth(2); f.setframerate(rate) + f.writeframes(samples.tobytes()) + p = profile_audio(wav, 16000) + assert p.duration_seconds == pytest.approx(2.0, abs=0.2)