From ae9ab70421118cd5def3cc9292e0b694bbd89913 Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Wed, 15 Jul 2026 12:15:10 +0000 Subject: [PATCH] refactor: extract audio helpers to domain/audio_helpers.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- abogen/domain/audio_helpers.py | 118 +++++++++++++++++++++++++++ abogen/webui/conversion_runner.py | 106 ++----------------------- tests/test_audio_helpers.py | 127 ++++++++++++++++++++++++++++++ 3 files changed, 253 insertions(+), 98 deletions(-) create mode 100644 abogen/domain/audio_helpers.py create mode 100644 tests/test_audio_helpers.py diff --git a/abogen/domain/audio_helpers.py b/abogen/domain/audio_helpers.py new file mode 100644 index 0000000..9cef77f --- /dev/null +++ b/abogen/domain/audio_helpers.py @@ -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 diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 8c8c6ae..84987c4 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -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 diff --git a/tests/test_audio_helpers.py b/tests/test_audio_helpers.py new file mode 100644 index 0000000..3ce6bfc --- /dev/null +++ b/tests/test_audio_helpers.py @@ -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"}] + )