audio.py

transcription_pipeline/audio.py

1. """ffmpeg: any audio or video in, 16 kHz mono out.
3. Uses the system ffmpeg when there is one, else the static binary that the
4. imageio-ffmpeg package ships, so nothing needs installing system-wide.
5. """
7. from __future__ import annotations
9. import shutil
10. import subprocess
11. import wave
12. from pathlib import Path
14. AUDIO_EXTS = {".mp3", ".m4a", ".wav", ".aac", ".flac", ".ogg", ".opus", ".wma", ".aiff", ".aif",
15. ".mp4", ".mov", ".m4v", ".mkv", ".webm", ".avi"}
18. class AudioError(RuntimeError):
19. pass
22. def ffmpeg_path() -> str:
23. found = shutil.which("ffmpeg")
24. if found:
25. return found
26. try:
27. import imageio_ffmpeg
28. return imageio_ffmpeg.get_ffmpeg_exe()
29. except Exception as e: # pragma: no cover - depends on the machine
30. raise AudioError("ffmpeg not found: install it, or `pip install imageio-ffmpeg`.") from e
33. def _run(args: list[str]) -> None:
34. cmd = [ffmpeg_path(), "-nostdin", "-hide_banner", "-loglevel", "error", "-y", *args]
35. p = subprocess.run(cmd, capture_output=True, text=True)
36. if p.returncode != 0:
37. msg = (p.stderr or "").strip().splitlines()
38. raise AudioError(f"ffmpeg could not read this file: {msg[-1] if msg else 'unknown error'}")
41. def to_wav16k(src: Path, dst: Path) -> Path:
42. """Decode to 16 kHz mono 16-bit WAV: what the local models expect."""
43. _run(["-i", str(src), "-vn", "-ac", "1", "-ar", "16000", "-c:a", "pcm_s16le", str(dst)])
44. return dst
47. def to_flac16k(src: Path, dst: Path) -> Path:
48. """16 kHz mono FLAC for upload: lossless, and about half the size of WAV."""
49. _run(["-i", str(src), "-vn", "-ac", "1", "-ar", "16000", "-c:a", "flac", str(dst)])
50. return dst
53. def wav_duration(path: Path) -> float:
54. with wave.open(str(path), "rb") as w:
55. return w.getnframes() / float(w.getframerate())
58. def hms(seconds: float) -> str:
59. s = int(round(seconds))
60. return f"{s // 3600}:{s // 60 % 60:02d}:{s % 60:02d}"
63. def load_wav16k(path: Path):
64. """The 16 kHz mono WAV as float32 samples in [-1, 1], the array WhisperX takes.
66. Read directly rather than through whisperx.load_audio, which shells out to
67. an `ffmpeg` on PATH and so would bring back the system dependency.
68. """
69. import numpy as np
70. with wave.open(str(path), "rb") as w:
71. if w.getframerate() != 16000 or w.getnchannels() != 1 or w.getsampwidth() != 2:
72. raise AudioError(f"{path.name}: expected 16 kHz mono 16-bit WAV")
73. frames = w.readframes(w.getnframes())
74. return np.frombuffer(frames, dtype="<i2").astype(np.float32) / 32768.0