diff --git a/CHANGELOG.md b/CHANGELOG.md index 12657c3..247f6e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,9 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi --- ## [Unreleased] +### Added +- **Voice health: timestamped audio problems.** `core.audio.audio_problems` flags long interior silence, clipping and sudden level jumps (`problems` list with `kind`, `start_s`, `end_s`, `detail`, plus one warning per kind) in the route-test voice health; existing keys unchanged. Computed on demand and cached by file mtime, not in the voice list. Shown in the routing test result. Tests: `tests/test_audio_problems.py`. + ### Changed - **Magazine interviews: journalist/guest voices.** Interview labels such as `c't:` (also `c’t`, `ct`, OCR `Cc't`), `Redaktion`, `Interviewer`, `Frage` are the interviewing side and are read in the narrator voice; every other recurring label is a guest and gets the second voice. Unlabelled blocks continue the current speaker's turn; headings, decks and bylines stay narrator. Detection stays strict (label starts a block and recurs at least twice). Audio-timing records now store each segment's voice and speaker. Tests: `tests/test_magazine.py` (`InterviewJournalistTests`). diff --git a/Manual.md b/Manual.md index 6c3c1c0..59d6bb6 100644 --- a/Manual.md +++ b/Manual.md @@ -8,6 +8,13 @@ Living manual. Every feature updates this file in the same commit as the code. --- +## Voice health: audio problems + +**Purpose:** the routing test (Settings, Routing, Test) checks the routed voice's reference audio and lists problems with timestamps: `silence` (over 1.5 s of silence inside the clip), `clipping` (peaks at full scale for 0.1 s or more), `level jump` (15 dB or more change between adjacent half-second windows). +**Usage:** run the route test; the result line shows "Audio problems: clipping at 2.0s–2.4s (peak at 0.0 dBFS); ..." (first 8). Each kind also adds a warning. The text label carries the meaning, not colour. +**Errors:** unreadable audio yields no problems. Only the first 120 s are scanned; results are cached per file modification time. The voice list does not run this check (it returns `problems: []`) to stay fast. +**Privacy/offline:** local analysis only, nothing is sent anywhere. + ## Audiobook casting (Read Aloud → Cast as Audiobook) ### Purpose diff --git a/core/audio.py b/core/audio.py index 786b8b7..2e7a44f 100644 --- a/core/audio.py +++ b/core/audio.py @@ -138,6 +138,78 @@ def _loudness_info(path: Path) -> dict: } +# Audio problem detection (voice health). Thresholds are module constants. +PROBLEM_CHUNK_MS = 50 +PROBLEM_MAX_S = 120 # bounded cost: only the first 2 minutes are scanned +PROBLEM_SILENCE_DB = -50.0 +PROBLEM_SILENCE_MIN_S = 1.5 +PROBLEM_CLIP_DB = -0.1 +PROBLEM_CLIP_MIN_S = 0.1 +PROBLEM_JUMP_DB = 15.0 +PROBLEM_JUMP_WIN = 10 # chunks per level window (0.5 s) + + +def audio_problems(src) -> list[dict]: + """Timestamped problems: long interior silence, clipping, sudden level jumps. + + `src` is a path or an AudioSegment. Returns [{kind, start_s, end_s, detail}]. + """ + seg = src if isinstance(src, AudioSegment) else AudioSegment.from_file(str(src)) + seg = seg.set_channels(1)[: PROBLEM_MAX_S * 1000] + step = PROBLEM_CHUNK_MS + rms, peak = [], [] + for pos in range(0, len(seg), step): + ch = seg[pos:pos + step] + d, m = ch.dBFS, ch.max_dBFS + rms.append(d if d != float("-inf") else -90.0) + peak.append(m if m != float("-inf") else -90.0) + + def runs(flags, min_chunks): + out, i = [], 0 + while i < len(flags): + if flags[i]: + j = i + while j < len(flags) and flags[j]: + j += 1 + if j - i >= min_chunks: + out.append((i, j)) + i = j + else: + i += 1 + return out + + t = lambda i: round(i * step / 1000.0, 2) + problems: list[dict] = [] + n = len(rms) + speech = [i for i, d in enumerate(rms) if d > PROBLEM_SILENCE_DB] + first, last = (speech[0], speech[-1]) if speech else (0, -1) + for a, b in runs([d <= PROBLEM_SILENCE_DB for d in rms], int(PROBLEM_SILENCE_MIN_S * 1000 / step)): + if a > first and b <= last: # leading/trailing silence is normal + problems.append({"kind": "silence", "start_s": t(a), "end_s": t(b), + "detail": f"{(b - a) * step / 1000:.1f}s of silence"}) + for a, b in runs([m >= PROBLEM_CLIP_DB for m in peak], int(PROBLEM_CLIP_MIN_S * 1000 / step)): + problems.append({"kind": "clipping", "start_s": t(a), "end_s": t(b), + "detail": f"peak at {max(peak[a:b]):.1f} dBFS"}) + w = PROBLEM_JUMP_WIN + mean = lambda xs: sum(xs) / len(xs) + cand = [] # (index, delta) where adjacent windows differ enough; keep the strongest per cluster + for i in range(w, n - w + 1): + before, after = rms[i - w:i], rms[i:i + w] + if min(before) > PROBLEM_SILENCE_DB and min(after) > PROBLEM_SILENCE_DB: + delta = mean(after) - mean(before) + if abs(delta) >= PROBLEM_JUMP_DB: + if cand and i - cand[-1][0] <= w and abs(delta) <= abs(cand[-1][1]): + continue + if cand and i - cand[-1][0] <= w: + cand.pop() + cand.append((i, delta)) + for i, delta in cand: + problems.append({"kind": "level_jump", "start_s": t(i), "end_s": t(i), + "detail": f"{delta:+.0f} dB level change"}) + problems.sort(key=lambda p: p["start_s"]) + return problems + + def _auto_trim_bounds(path: Path) -> dict: seg = AudioSegment.from_file(str(path)).set_channels(1) dur_ms = len(seg) diff --git a/core/voice.py b/core/voice.py index 8069f33..4c69bbf 100644 --- a/core/voice.py +++ b/core/voice.py @@ -19,7 +19,7 @@ from core.constants import ( _VOICE_TARGET_DBFS, _VOICE_PEAK_DBFS, _BENCHMARK_SENTENCES, _MAX_TTS_OUTPUT_SECONDS, ) -from core.audio import _duration, _export_normalized_wav +from core.audio import _duration, _export_normalized_wav, audio_problems # ── File type constants ─────────────────────────────────────────────────────── @@ -334,12 +334,35 @@ def _read_reference_text(audio: Path) -> tuple[bool, str]: # ── Voice health and entry ──────────────────────────────────────────────────── +_PROBLEM_CACHE: dict[tuple[str, float], list[dict]] = {} +_PROBLEM_LABEL = {"silence": "Long silence in reference audio", "clipping": "Clipping in reference audio", + "level_jump": "Sudden level jump in reference audio"} + + +def _voice_problems(p: Path) -> list[dict]: + """Cached by file mtime; decoding happens only on a cache miss.""" + try: + key = (str(p), p.stat().st_mtime) + except OSError: + return [] + if key not in _PROBLEM_CACHE: + try: + found = audio_problems(p) + except Exception: + found = [] + if len(_PROBLEM_CACHE) > 512: + _PROBLEM_CACHE.clear() + _PROBLEM_CACHE[key] = found + return _PROBLEM_CACHE[key] + + def _voice_health( p: Path, *, meta: dict | None = None, duration: float | None = None, transcript: str | None = None, + analyze: bool = False, ) -> dict: if transcript is None: _has_ref, transcript = _read_reference_text(p) @@ -367,9 +390,14 @@ def _voice_health( peak = loudness.get("peak_dbfs") if peak is not None and float(peak) > -0.1: warnings.append("Reference audio is clipping or too loud") + # Decoding audio is too slow for the voice list; only single-voice checks opt in. + problems = _voice_problems(p) if analyze else [] + for kind in dict.fromkeys(q["kind"] for q in problems): + warnings.append(_PROBLEM_LABEL[kind]) return { "ok": not warnings, "warnings": warnings, + "problems": problems, "duration": round(duration, 2) if duration else None, "word_count": word_count, "words_per_sec": round(words_per_sec, 2) if words_per_sec else 0, diff --git a/routes/tts.py b/routes/tts.py index fd78891..61da9bd 100644 --- a/routes/tts.py +++ b/routes/tts.py @@ -1013,7 +1013,7 @@ async def tts_route_test(request: Request): "detected_language": str(route.get("detected_language", _detect_text_language(text))) if route else _detect_text_language(text), "matched": bool(route), "route": route, - "voice_health": _voice_health(routed_audio) if routed_audio else None, + "voice_health": _voice_health(routed_audio, analyze=True) if routed_audio else None, "sounds": sound_status, } _routing_log_add( diff --git a/static/js/routing.js b/static/js/routing.js index b7b9b72..2645d83 100644 --- a/static/js/routing.js +++ b/static/js/routing.js @@ -394,6 +394,12 @@ function renderRouteTestResult(d) { if (warnings.length) { parts.push('Warning: ' + warnings.map(escHtml).join('; ')); } + const problems = Array.isArray(health.problems) ? health.problems : []; + if (problems.length) { + parts.push('Audio problems: ' + problems.slice(0, 8).map(q => + `${escHtml(String(q.kind).replace('_', ' '))} at ${escHtml(String(q.start_s))}s` + + (q.end_s > q.start_s ? `–${escHtml(String(q.end_s))}s` : '') + ` (${escHtml(q.detail)})`).join('; ')); + } const sounds = d.sounds || {}; for (const [key, sound] of Object.entries(sounds)) { const label = key === 'before_sound' ? 'before sound' : 'after sound'; diff --git a/tests/test_audio_problems.py b/tests/test_audio_problems.py new file mode 100644 index 0000000..b46bd8f --- /dev/null +++ b/tests/test_audio_problems.py @@ -0,0 +1,53 @@ +"""audio_problems: silence gap, clipped burst, level jump, clean signal.""" +import sys +from pathlib import Path +import unittest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from pydub import AudioSegment +from pydub.generators import Sine, Square + +from core.audio import audio_problems + +SR = 24000 + + +def tone(ms, gain=-20): + return Sine(300, sample_rate=SR).to_audio_segment(ms).apply_gain(gain) + + +def kinds(seg): + return {p["kind"] for p in audio_problems(seg)} + + +class AudioProblemTests(unittest.TestCase): + def test_clean(self): + self.assertEqual(audio_problems(tone(4000, -18).set_frame_rate(SR)), []) + + def test_silence_gap(self): + seg = tone(2000) + AudioSegment.silent(2000, frame_rate=SR) + tone(2000) + p = [x for x in audio_problems(seg) if x["kind"] == "silence"] + self.assertEqual(len(p), 1) + self.assertAlmostEqual(p[0]["start_s"], 2.0, delta=0.1) + + def test_edge_silence_ignored(self): + seg = AudioSegment.silent(3000, frame_rate=SR) + tone(2000) + AudioSegment.silent(3000, frame_rate=SR) + self.assertNotIn("silence", kinds(seg)) + + def test_clipping(self): + burst = Square(300, sample_rate=SR).to_audio_segment(400).apply_gain(+10) + seg = tone(2000) + burst + tone(2000) + p = [x for x in audio_problems(seg) if x["kind"] == "clipping"] + self.assertTrue(p) + self.assertAlmostEqual(p[0]["start_s"], 2.0, delta=0.1) + + def test_level_jump(self): + seg = tone(2000, -40) + tone(2000, -10) + p = [x for x in audio_problems(seg) if x["kind"] == "level_jump"] + self.assertEqual(len(p), 1) + self.assertAlmostEqual(p[0]["start_s"], 2.0, delta=0.1) + + +if __name__ == "__main__": + unittest.main()