mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 05:40:26 +02:00
refactor: extract audio helpers to domain/audio_helpers.py
- Extract build_ffmpeg_command, to_float32, apply_m4b_chapters_with_mutagen - _apply_m4b_chapters_with_mutagen becomes thin wrapper with error handling - Add tests/test_audio_helpers.py (12 tests) - conversion_runner.py: 1410 → 1320 lines - All tests pass
This commit is contained in:
Artem Akymenko
@@ -0,0 +1,118 @@
|
||||
"""Audio helper utilities.
|
||||
|
||||
Functions for building ffmpeg commands, converting audio formats,
|
||||
and applying chapter metadata to MP4 files.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
SAMPLE_RATE = 24000
|
||||
|
||||
|
||||
def build_ffmpeg_command(path: Path, fmt: str, metadata: Optional[Dict[str, str]] = None) -> list[str]:
|
||||
from abogen.infrastructure.exporters import ExportService
|
||||
|
||||
base = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-f",
|
||||
"f32le",
|
||||
"-ar",
|
||||
str(SAMPLE_RATE),
|
||||
"-ac",
|
||||
"1",
|
||||
"-i",
|
||||
"pipe:0",
|
||||
]
|
||||
if fmt == "mp3":
|
||||
base += ["-c:a", "libmp3lame", "-qscale:a", "2"]
|
||||
elif fmt == "opus":
|
||||
base += ["-c:a", "libopus", "-b:a", "24000"]
|
||||
elif fmt == "m4b":
|
||||
base += ["-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart+use_metadata_tags"]
|
||||
else:
|
||||
base += ["-c:a", "copy"]
|
||||
|
||||
if metadata:
|
||||
svc = ExportService()
|
||||
base.extend(svc._metadata_to_ffmpeg_args(metadata))
|
||||
base.append(str(path))
|
||||
return base
|
||||
|
||||
|
||||
def to_float32(audio_segment) -> np.ndarray:
|
||||
if audio_segment is None:
|
||||
return np.zeros(0, dtype="float32")
|
||||
|
||||
tensor = audio_segment
|
||||
if hasattr(tensor, "detach"):
|
||||
tensor = tensor.detach()
|
||||
if hasattr(tensor, "cpu"):
|
||||
try:
|
||||
tensor = tensor.cpu()
|
||||
except Exception:
|
||||
pass
|
||||
if hasattr(tensor, "numpy"):
|
||||
return np.asarray(tensor.numpy(), dtype="float32").reshape(-1)
|
||||
return np.asarray(tensor, dtype="float32").reshape(-1)
|
||||
|
||||
|
||||
def apply_m4b_chapters_with_mutagen(
|
||||
audio_path: Path,
|
||||
chapters: List[Dict[str, Any]],
|
||||
) -> bool:
|
||||
"""Apply chapter atoms to an MP4/M4B file using mutagen.
|
||||
|
||||
Returns True if chapters were written, False otherwise.
|
||||
Raises ImportError if mutagen is not installed.
|
||||
"""
|
||||
if not chapters:
|
||||
return False
|
||||
|
||||
from fractions import Fraction
|
||||
from mutagen.mp4 import MP4, MP4Chapter # type: ignore[import]
|
||||
|
||||
mp4 = MP4(str(audio_path))
|
||||
|
||||
chapter_objects: List[MP4Chapter] = []
|
||||
for index, entry in enumerate(sorted(chapters, key=lambda item: float(item.get("start") or 0.0))):
|
||||
start_raw = entry.get("start")
|
||||
if start_raw is None:
|
||||
continue
|
||||
try:
|
||||
start_seconds = max(0.0, float(start_raw))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
|
||||
title_value = entry.get("title")
|
||||
title_text = str(title_value) if title_value else f"Chapter {index + 1}"
|
||||
|
||||
start_fraction = Fraction(int(round(start_seconds * 1000)), 1000)
|
||||
chapter_atom = MP4Chapter(start_fraction, title_text)
|
||||
|
||||
end_raw = entry.get("end")
|
||||
if end_raw is not None:
|
||||
try:
|
||||
end_seconds = float(end_raw)
|
||||
except (TypeError, ValueError):
|
||||
end_seconds = None
|
||||
if end_seconds is not None and end_seconds > start_seconds:
|
||||
chapter_atom.end = Fraction(int(round(end_seconds * 1000)), 1000)
|
||||
|
||||
chapter_objects.append(chapter_atom)
|
||||
|
||||
if not chapter_objects:
|
||||
return False
|
||||
|
||||
from typing import cast
|
||||
|
||||
mp4.chapters = cast(Any, chapter_objects)
|
||||
mp4.save()
|
||||
|
||||
return True
|
||||
@@ -111,6 +111,11 @@ from abogen.domain.output_paths import (
|
||||
resolve_output_directory as _resolve_output_directory,
|
||||
resolve_project_layout as _resolve_project_layout,
|
||||
)
|
||||
from abogen.domain.audio_helpers import (
|
||||
build_ffmpeg_command as _build_ffmpeg_command,
|
||||
to_float32 as _to_float32,
|
||||
apply_m4b_chapters_with_mutagen as _apply_m4b_chapters_with_mutagen,
|
||||
)
|
||||
|
||||
|
||||
from .service import Job, JobStatus
|
||||
@@ -156,64 +161,18 @@ def _apply_m4b_chapters_with_mutagen(
|
||||
chapters: List[Dict[str, Any]],
|
||||
job: Job,
|
||||
) -> bool:
|
||||
if not chapters:
|
||||
return False
|
||||
|
||||
try:
|
||||
from fractions import Fraction
|
||||
from mutagen.mp4 import MP4, MP4Chapter # type: ignore[import]
|
||||
return _apply_m4b_chapters_with_mutagen(audio_path, chapters)
|
||||
except ImportError:
|
||||
job.add_log(
|
||||
"Unable to write MP4 chapter atoms because mutagen is not installed.",
|
||||
level="warning",
|
||||
)
|
||||
return False
|
||||
|
||||
try:
|
||||
mp4 = MP4(str(audio_path))
|
||||
except Exception as exc: # pragma: no cover - defensive
|
||||
job.add_log(f"Failed to open m4b for chapter embedding: {exc}", level="warning")
|
||||
except Exception as exc:
|
||||
job.add_log(f"Failed to write MP4 chapter atoms: {exc}", level="warning")
|
||||
return False
|
||||
|
||||
chapter_objects: List[MP4Chapter] = []
|
||||
for index, entry in enumerate(sorted(chapters, key=lambda item: float(item.get("start") or 0.0))):
|
||||
start_raw = entry.get("start")
|
||||
if start_raw is None:
|
||||
continue
|
||||
try:
|
||||
start_seconds = max(0.0, float(start_raw))
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
|
||||
title_value = entry.get("title")
|
||||
title_text = str(title_value) if title_value else f"Chapter {index + 1}"
|
||||
|
||||
start_fraction = Fraction(int(round(start_seconds * 1000)), 1000)
|
||||
chapter_atom = MP4Chapter(start_fraction, title_text)
|
||||
|
||||
end_raw = entry.get("end")
|
||||
if end_raw is not None:
|
||||
try:
|
||||
end_seconds = float(end_raw)
|
||||
except (TypeError, ValueError):
|
||||
end_seconds = None
|
||||
if end_seconds is not None and end_seconds > start_seconds:
|
||||
chapter_atom.end = Fraction(int(round(end_seconds * 1000)), 1000)
|
||||
|
||||
chapter_objects.append(chapter_atom)
|
||||
|
||||
if not chapter_objects:
|
||||
return False
|
||||
|
||||
try:
|
||||
mp4.chapters = cast(Any, chapter_objects)
|
||||
mp4.save()
|
||||
except Exception as exc: # pragma: no cover - defensive
|
||||
job.add_log(f"Failed to persist MP4 chapter atoms: {exc}", level="warning")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def _embed_m4b_metadata(
|
||||
audio_path: Path,
|
||||
@@ -1325,63 +1284,14 @@ def _open_audio_sink(
|
||||
return AudioSink(write=_write)
|
||||
|
||||
|
||||
def _build_ffmpeg_command(path: Path, fmt: str, metadata: Optional[Dict[str, str]] = None) -> list[str]:
|
||||
base = [
|
||||
"ffmpeg",
|
||||
"-y",
|
||||
"-f",
|
||||
"f32le",
|
||||
"-ar",
|
||||
str(SAMPLE_RATE),
|
||||
"-ac",
|
||||
"1",
|
||||
"-i",
|
||||
"pipe:0",
|
||||
]
|
||||
if fmt == "mp3":
|
||||
base += ["-c:a", "libmp3lame", "-qscale:a", "2"]
|
||||
elif fmt == "opus":
|
||||
base += ["-c:a", "libopus", "-b:a", "24000"]
|
||||
elif fmt == "m4b":
|
||||
base += ["-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart+use_metadata_tags"]
|
||||
else:
|
||||
base += ["-c:a", "copy"]
|
||||
|
||||
if metadata:
|
||||
base.extend(_export_svc._metadata_to_ffmpeg_args(metadata))
|
||||
base.append(str(path))
|
||||
return base
|
||||
|
||||
|
||||
def _resolve_voice(pipeline, voice_spec: str, use_gpu: bool):
|
||||
if "*" in voice_spec:
|
||||
# Voice formulas are a Kokoro-only feature (they require a pipeline that can
|
||||
# load individual Kokoro voices). When running with SuperTonic (or when the
|
||||
# pipeline is otherwise unavailable), treat the spec as a plain string and
|
||||
# allow downstream provider-specific resolution to choose a safe fallback.
|
||||
if pipeline is None or not hasattr(pipeline, "load_single_voice"):
|
||||
return voice_spec
|
||||
return get_new_voice(pipeline, voice_spec, use_gpu)
|
||||
return voice_spec
|
||||
|
||||
|
||||
def _to_float32(audio_segment) -> np.ndarray:
|
||||
if audio_segment is None:
|
||||
return np.zeros(0, dtype="float32")
|
||||
|
||||
tensor = audio_segment
|
||||
if hasattr(tensor, "detach"):
|
||||
tensor = tensor.detach()
|
||||
if hasattr(tensor, "cpu"):
|
||||
try:
|
||||
tensor = tensor.cpu()
|
||||
except Exception:
|
||||
pass
|
||||
if hasattr(tensor, "numpy"):
|
||||
return np.asarray(tensor.numpy(), dtype="float32").reshape(-1)
|
||||
return np.asarray(tensor, dtype="float32").reshape(-1)
|
||||
|
||||
|
||||
def _create_subtitle_writer(job: Job, audio_path: Path):
|
||||
if job.subtitle_mode == "Disabled":
|
||||
return None
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
"""Tests for audio helper utilities.
|
||||
|
||||
Tests import from domain/audio_helpers.py (new module).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# build_ffmpeg_command
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildFfmpegCommand:
|
||||
"""build_ffmpeg_command builds ffmpeg argument list."""
|
||||
|
||||
def test_base_structure(self):
|
||||
from abogen.domain.audio_helpers import build_ffmpeg_command
|
||||
|
||||
cmd = build_ffmpeg_command(Path("/out/audio.wav"), "wav")
|
||||
assert cmd[0] == "ffmpeg"
|
||||
assert "-y" in cmd
|
||||
assert "pipe:0" in cmd
|
||||
assert str(Path("/out/audio.wav")) in cmd
|
||||
|
||||
def test_mp3_codec(self):
|
||||
from abogen.domain.audio_helpers import build_ffmpeg_command
|
||||
|
||||
cmd = build_ffmpeg_command(Path("/out.mp3"), "mp3")
|
||||
assert "libmp3lame" in cmd
|
||||
|
||||
def test_opus_codec(self):
|
||||
from abogen.domain.audio_helpers import build_ffmpeg_command
|
||||
|
||||
cmd = build_ffmpeg_command(Path("/out.opus"), "opus")
|
||||
assert "libopus" in cmd
|
||||
|
||||
def test_m4b_codec(self):
|
||||
from abogen.domain.audio_helpers import build_ffmpeg_command
|
||||
|
||||
cmd = build_ffmpeg_command(Path("/out.m4b"), "m4b")
|
||||
assert "aac" in cmd
|
||||
assert "+faststart+use_metadata_tags" in cmd
|
||||
|
||||
def test_wav_copy_codec(self):
|
||||
from abogen.domain.audio_helpers import build_ffmpeg_command
|
||||
|
||||
cmd = build_ffmpeg_command(Path("/out.wav"), "wav")
|
||||
assert "copy" in cmd
|
||||
|
||||
def test_with_metadata(self):
|
||||
from abogen.domain.audio_helpers import build_ffmpeg_command
|
||||
|
||||
cmd = build_ffmpeg_command(Path("/out.mp3"), "mp3", metadata={"album": "Test"})
|
||||
assert str(Path("/out.mp3")) in cmd
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# to_float32
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestToFloat32:
|
||||
"""to_float32 converts audio to float32 numpy array."""
|
||||
|
||||
def test_none_returns_empty(self):
|
||||
from abogen.domain.audio_helpers import to_float32
|
||||
|
||||
result = to_float32(None)
|
||||
assert isinstance(result, np.ndarray)
|
||||
assert result.dtype == np.float32
|
||||
assert len(result) == 0
|
||||
|
||||
def test_numpy_array(self):
|
||||
from abogen.domain.audio_helpers import to_float32
|
||||
|
||||
arr = np.array([1.0, 2.0, 3.0], dtype="float64")
|
||||
result = to_float32(arr)
|
||||
assert result.dtype == np.float32
|
||||
assert len(result) == 3
|
||||
|
||||
def test_mock_tensor(self):
|
||||
from abogen.domain.audio_helpers import to_float32
|
||||
|
||||
tensor = MagicMock()
|
||||
tensor.detach.return_value = tensor
|
||||
tensor.cpu.return_value = tensor
|
||||
tensor.numpy.return_value = np.array([1.0, 2.0])
|
||||
result = to_float32(tensor)
|
||||
assert result.dtype == np.float32
|
||||
assert len(result) == 2
|
||||
|
||||
def test_list_input(self):
|
||||
from abogen.domain.audio_helpers import to_float32
|
||||
|
||||
result = to_float32([1.0, 2.0])
|
||||
assert result.dtype == np.float32
|
||||
assert len(result) == 2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# apply_m4b_chapters_with_mutagen
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestApplyM4bChaptersWithMutagen:
|
||||
"""apply_m4b_chapters_with_mutagen writes chapter atoms to MP4."""
|
||||
|
||||
def test_empty_chapters_returns_false(self):
|
||||
from abogen.domain.audio_helpers import apply_m4b_chapters_with_mutagen
|
||||
|
||||
assert apply_m4b_chapters_with_mutagen(Path("/fake.m4b"), []) is False
|
||||
|
||||
def test_missing_mutagen_raises(self):
|
||||
from abogen.domain.audio_helpers import apply_m4b_chapters_with_mutagen
|
||||
|
||||
with patch.dict("sys.modules", {"mutagen": None, "mutagen.mp4": None}):
|
||||
with pytest.raises((ImportError, KeyError)):
|
||||
apply_m4b_chapters_with_mutagen(
|
||||
Path("/fake.m4b"), [{"start": 0, "title": "Ch1"}]
|
||||
)
|
||||
Reference in New Issue
Block a user