feat(voices): timestamped audio problem flags in voice health
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
2594a236c4
commit
19283a53a4
@ -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`).
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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';
|
||||
|
||||
53
tests/test_audio_problems.py
Normal file
53
tests/test_audio_problems.py
Normal file
@ -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()
|
||||
Loading…
Reference in New Issue
Block a user