307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""Audio helpers: conversion, trim, normalize, loudness, auto-trim."""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import subprocess
|
|
import uuid
|
|
import wave
|
|
from pathlib import Path
|
|
|
|
from pydub import AudioSegment
|
|
|
|
from core.constants import _VOICE_TARGET_DBFS, _VOICE_PEAK_DBFS
|
|
from core.registry import TEMP_DIR
|
|
|
|
|
|
def _to_wav_24k(src: Path) -> Path:
|
|
out = TEMP_DIR / f"{src.stem}_24k.wav"
|
|
seg = AudioSegment.from_file(str(src))
|
|
seg = seg.set_frame_rate(24000).set_channels(1).set_sample_width(2)
|
|
seg.export(str(out), format="wav")
|
|
return out
|
|
|
|
|
|
def _to_wav_16k(src: Path) -> Path:
|
|
"""16kHz mono WAV — required by Whisper/WhisperX VAD and wav2vec2 alignment."""
|
|
out = TEMP_DIR / f"{src.stem}_16k.wav"
|
|
seg = AudioSegment.from_file(str(src))
|
|
seg = seg.set_frame_rate(16000).set_channels(1).set_sample_width(2)
|
|
seg.export(str(out), format="wav")
|
|
return out
|
|
|
|
|
|
def _change_tempo(wav_bytes: bytes, speed: float) -> bytes:
|
|
"""Time-stretch WAV audio without shifting pitch, via ffmpeg's atempo
|
|
filter — used by App Routing's per-route playback-speed setting. A naive
|
|
frame-rate change (or pydub's speedup(), which only accelerates and uses
|
|
a much cruder splice technique) shifts pitch along with speed, which
|
|
reads as a chipmunk/slow-motion effect rather than someone just talking
|
|
faster or slower. atempo only accepts 0.5-2.0 per instance; the caller
|
|
(core/routing.py's _normalize_route) already clamps to that range.
|
|
"""
|
|
speed = max(0.5, min(2.0, speed))
|
|
in_path = TEMP_DIR / f"{uuid.uuid4().hex}_tempo_in.wav"
|
|
out_path = TEMP_DIR / f"{uuid.uuid4().hex}_tempo_out.wav"
|
|
in_path.write_bytes(wav_bytes)
|
|
try:
|
|
result = subprocess.run(
|
|
["ffmpeg", "-y", "-i", str(in_path), "-filter:a", f"atempo={speed}", str(out_path)],
|
|
capture_output=True, timeout=30,
|
|
)
|
|
if result.returncode != 0 or not out_path.exists():
|
|
raise RuntimeError(result.stderr.decode(errors="replace").strip() or "ffmpeg atempo failed")
|
|
return out_path.read_bytes()
|
|
finally:
|
|
in_path.unlink(missing_ok=True)
|
|
out_path.unlink(missing_ok=True)
|
|
|
|
|
|
def _trim(src: Path, start_s: float, end_s: float) -> Path:
|
|
seg = AudioSegment.from_file(str(src))
|
|
trimmed = seg[int(start_s * 1000):int(end_s * 1000)]
|
|
trimmed, _ = _normalize_segment(trimmed)
|
|
out = TEMP_DIR / f"{uuid.uuid4().hex}_trimmed.wav"
|
|
trimmed.export(str(out), format="wav")
|
|
return out
|
|
|
|
|
|
def _duration(path: Path) -> float:
|
|
if path.suffix.lower() == ".wav":
|
|
try:
|
|
with wave.open(str(path), "rb") as wf:
|
|
frames = wf.getnframes()
|
|
rate = wf.getframerate()
|
|
if rate:
|
|
return frames / float(rate)
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
seg = AudioSegment.from_file(str(path))
|
|
return len(seg) / 1000.0
|
|
except Exception:
|
|
pass
|
|
|
|
probe = subprocess.run(
|
|
[
|
|
"ffprobe", "-v", "error",
|
|
"-show_entries", "format=duration",
|
|
"-of", "default=noprint_wrappers=1:nokey=1",
|
|
str(path),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
if probe.returncode != 0:
|
|
raise RuntimeError(probe.stderr.strip() or "ffprobe failed")
|
|
return float(probe.stdout.strip())
|
|
|
|
|
|
def _normalize_segment(seg: AudioSegment, target_dbfs: float = _VOICE_TARGET_DBFS,
|
|
peak_dbfs: float = _VOICE_PEAK_DBFS) -> tuple[AudioSegment, dict]:
|
|
before_dbfs = seg.dBFS if seg.dBFS != float("-inf") else None
|
|
before_peak = seg.max_dBFS if seg.max_dBFS != float("-inf") else None
|
|
if before_dbfs is None or before_peak is None:
|
|
return seg, {"before_dbfs": before_dbfs, "after_dbfs": before_dbfs, "gain_db": 0.0, "peak_dbfs": before_peak}
|
|
|
|
gain = target_dbfs - before_dbfs
|
|
if before_peak + gain > peak_dbfs:
|
|
gain = peak_dbfs - before_peak
|
|
normalized = seg.apply_gain(gain)
|
|
after_dbfs = normalized.dBFS if normalized.dBFS != float("-inf") else None
|
|
after_peak = normalized.max_dBFS if normalized.max_dBFS != float("-inf") else None
|
|
return normalized, {
|
|
"before_dbfs": round(before_dbfs, 2),
|
|
"after_dbfs": round(after_dbfs, 2) if after_dbfs is not None else None,
|
|
"gain_db": round(gain, 2),
|
|
"peak_dbfs": round(after_peak, 2) if after_peak is not None else None,
|
|
}
|
|
|
|
|
|
def _export_normalized_wav(src: Path, dest: Path, target_dbfs: float = _VOICE_TARGET_DBFS) -> dict:
|
|
seg = AudioSegment.from_file(str(src))
|
|
seg = seg.set_frame_rate(24000).set_channels(1).set_sample_width(2)
|
|
seg, info = _normalize_segment(seg, target_dbfs=target_dbfs)
|
|
seg.export(str(dest), format="wav")
|
|
return info
|
|
|
|
|
|
def _loudness_info(path: Path) -> dict:
|
|
seg = AudioSegment.from_file(str(path))
|
|
dbfs = seg.dBFS if seg.dBFS != float("-inf") else None
|
|
peak = seg.max_dBFS if seg.max_dBFS != float("-inf") else None
|
|
return {
|
|
"dbfs": round(dbfs, 2) if dbfs is not None else None,
|
|
"peak_dbfs": round(peak, 2) if peak is not None else None,
|
|
"target_dbfs": _VOICE_TARGET_DBFS,
|
|
}
|
|
|
|
|
|
# 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)
|
|
if dur_ms <= 20_000:
|
|
return {
|
|
"start": 0.0,
|
|
"end": dur_ms / 1000.0,
|
|
"duration": dur_ms / 1000.0,
|
|
"reason": "Audio is already short enough.",
|
|
}
|
|
|
|
chunk_ms = 250
|
|
chunks = []
|
|
overall_db = seg.dBFS if seg.dBFS != float("-inf") else -60.0
|
|
speech_floor = max(overall_db - 18.0, -45.0)
|
|
|
|
for pos in range(0, dur_ms, chunk_ms):
|
|
ch = seg[pos:pos + chunk_ms]
|
|
db = ch.dBFS if ch.dBFS != float("-inf") else -80.0
|
|
max_db = ch.max_dBFS if ch.max_dBFS != float("-inf") else -80.0
|
|
chunks.append({"db": db, "speech": db >= speech_floor, "clipped": max_db > -1.0})
|
|
|
|
def score_window(start_ms: int, length_ms: int) -> tuple[float, dict]:
|
|
first = max(0, start_ms // chunk_ms)
|
|
last = min(len(chunks), (start_ms + length_ms + chunk_ms - 1) // chunk_ms)
|
|
win = chunks[first:last]
|
|
if not win:
|
|
return -9999.0, {}
|
|
speech_ratio = sum(1 for c in win if c["speech"]) / len(win)
|
|
silence_ratio = 1.0 - speech_ratio
|
|
clip_ratio = sum(1 for c in win if c["clipped"]) / len(win)
|
|
speech_dbs = [c["db"] for c in win if c["speech"]]
|
|
avg_db = sum(speech_dbs) / len(speech_dbs) if speech_dbs else -80.0
|
|
variance = sum((x - avg_db) ** 2 for x in speech_dbs) / len(speech_dbs) if speech_dbs else 100.0
|
|
loudness_penalty = abs(avg_db - (-20.0)) * 1.7
|
|
steadiness_penalty = min(18.0, variance ** 0.5 * 1.4)
|
|
duration_s = length_ms / 1000.0
|
|
duration_penalty = abs(duration_s - 12.0) * 0.9
|
|
score = (
|
|
speech_ratio * 100.0
|
|
- silence_ratio * 55.0
|
|
- clip_ratio * 85.0
|
|
- loudness_penalty
|
|
- steadiness_penalty
|
|
- duration_penalty
|
|
)
|
|
return score, {
|
|
"speech_ratio": speech_ratio,
|
|
"silence_ratio": silence_ratio,
|
|
"clip_ratio": clip_ratio,
|
|
"avg_db": avg_db,
|
|
}
|
|
|
|
best = None
|
|
window_lengths = [8_000, 10_000, 12_000, 15_000, 18_000]
|
|
for length_ms in window_lengths:
|
|
if length_ms > dur_ms:
|
|
continue
|
|
for start_ms in range(0, dur_ms - length_ms + 1, 500):
|
|
score, metrics = score_window(start_ms, length_ms)
|
|
if best is None or score > best["score"]:
|
|
best = {"start_ms": start_ms, "length_ms": length_ms, "score": score, "metrics": metrics}
|
|
|
|
if best is None:
|
|
end_ms = min(dur_ms, 12_000)
|
|
return {"start": 0.0, "end": end_ms / 1000.0, "duration": end_ms / 1000.0,
|
|
"reason": "Using the beginning because no stable speech window was found."}
|
|
|
|
start_s = best["start_ms"] / 1000.0
|
|
end_s = (best["start_ms"] + best["length_ms"]) / 1000.0
|
|
m = best["metrics"]
|
|
return {
|
|
"start": round(start_s, 2),
|
|
"end": round(end_s, 2),
|
|
"duration": round(end_s - start_s, 2),
|
|
"score": round(best["score"], 2),
|
|
"reason": (
|
|
f"Selected {end_s - start_s:.1f}s with "
|
|
f"{m.get('speech_ratio', 0) * 100:.0f}% speech, "
|
|
f"{m.get('silence_ratio', 0) * 100:.0f}% silence, "
|
|
f"avg {m.get('avg_db', -80):.1f} dBFS."
|
|
),
|
|
}
|
|
|
|
|
|
def _audio_segment_from_wav(audio: bytes) -> AudioSegment:
|
|
segment = AudioSegment.from_file(io.BytesIO(audio), format="wav")
|
|
return segment.set_channels(1).set_sample_width(2).set_frame_rate(24000)
|
|
|
|
|
|
def _wav_bytes_from_segment(segment: AudioSegment) -> bytes:
|
|
out = io.BytesIO()
|
|
segment.export(out, format="wav")
|
|
return out.getvalue()
|