From 0e216f3786f521f5a0767b740b7c7014d659eb5c Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Sun, 19 Jul 2026 09:34:17 +0000 Subject: [PATCH] refactor: extract _process_subtitle_file domain logic to shared modules - domain/subtitle_processor.py: parse_subtitle_file, format_time_range, speed_up_audio, fit_audio_to_duration (moved from audio_buffer), process_subtitle_entries (core TTS loop with cancel/log/progress callbacks) - domain/audio_buffer.py: add fit_audio_to_duration, ffmpeg_time_stretch - domain/output_paths.py: add resolve_unique_path (collision-safe filename) - pyqt/conversion.py: _process_subtitle_file reduced from ~350 to ~100 lines by delegating to domain functions; removed 4 unused subtitle parser imports - +33 tests (8 resolve_unique_path, 14 subtitle_processor, 8 audio_buffer, 3 format_time_range) - 1153 tests pass --- abogen/domain/audio_buffer.py | 67 +++++ abogen/domain/output_paths.py | 39 +++ abogen/domain/subtitle_processor.py | 279 +++++++++++++++++++++ abogen/pyqt/conversion.py | 295 +++-------------------- tests/test_domain_audio_buffer_extras.py | 59 +++++ tests/test_domain_resolve_unique_path.py | 51 ++++ tests/test_domain_subtitle_processor.py | 169 +++++++++++++ 7 files changed, 698 insertions(+), 261 deletions(-) create mode 100644 abogen/domain/subtitle_processor.py create mode 100644 tests/test_domain_audio_buffer_extras.py create mode 100644 tests/test_domain_resolve_unique_path.py create mode 100644 tests/test_domain_subtitle_processor.py diff --git a/abogen/domain/audio_buffer.py b/abogen/domain/audio_buffer.py index 6fb5388..0a5b0ac 100644 --- a/abogen/domain/audio_buffer.py +++ b/abogen/domain/audio_buffer.py @@ -170,3 +170,70 @@ def samples_for_duration(duration_seconds: float, sample_rate: int = SAMPLE_RATE if duration_seconds <= 0: return 0 return int(round(duration_seconds * sample_rate)) + + +def fit_audio_to_duration( + audio: np.ndarray, + target_duration: float, + sample_rate: int = SAMPLE_RATE, +) -> np.ndarray: + """Pad or trim audio to match target duration. + + Args: + audio: Input audio buffer. + target_duration: Desired duration in seconds. + sample_rate: Sample rate in Hz. + + Returns: + Audio buffer of exact length target_duration * sample_rate. + """ + target_samples = int(target_duration * sample_rate) + if len(audio) < target_samples: + padding = np.zeros(target_samples - len(audio), dtype="float32") + return np.concatenate([audio, padding]) + return audio[:target_samples] + + +def ffmpeg_time_stretch( + audio: np.ndarray, + speed_factor: float, + sample_rate: int = SAMPLE_RATE, +) -> np.ndarray: + """Time-stretch audio using FFmpeg's atempo filter. + + Args: + audio: Input audio buffer (float32). + speed_factor: Speed multiplier (>1.0 = faster). + sample_rate: Sample rate in Hz. + + Returns: + Time-stretched audio buffer. + """ + import math + import subprocess + + import static_ffmpeg + + if speed_factor <= 1.0 or audio.size == 0: + return audio + + static_ffmpeg.add_paths() + num_stages = max(1, int(math.ceil(math.log(speed_factor) / math.log(2.0)))) + tempo = speed_factor ** (1.0 / num_stages) + filter_str = ",".join([f"atempo={tempo:.6f}"] * num_stages) + + proc = subprocess.Popen( + [ + "ffmpeg", "-y", + "-f", "f32le", "-ar", str(sample_rate), "-ac", "1", + "-i", "pipe:0", + "-filter:a", filter_str, + "-f", "f32le", "-ar", str(sample_rate), "-ac", "1", + "pipe:1", + ], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + out, _ = proc.communicate(input=audio.tobytes()) + return np.frombuffer(out, dtype="float32") diff --git a/abogen/domain/output_paths.py b/abogen/domain/output_paths.py index 36f7f2b..58c6f9a 100644 --- a/abogen/domain/output_paths.py +++ b/abogen/domain/output_paths.py @@ -6,12 +6,14 @@ and computing project folder layouts. from __future__ import annotations +import os import platform import re from datetime import datetime from pathlib import Path from typing import Any, Callable, List, Optional, Tuple +from abogen.subtitle_utils import sanitize_name_for_os from abogen.text_extractor import ExtractedChapter @@ -139,3 +141,40 @@ def resolve_project_layout( return project_root, audio_dir, subtitle_dir, metadata_dir return project_root, project_root, project_root, None + + +def resolve_unique_path( + parent_dir: str, + base_name: str, + extension: str, + allowed_extensions: Optional[set] = None, +) -> str: + """Find a unique file path by appending _2, _3, etc. on collision. + + Args: + parent_dir: Directory to check for collisions. + base_name: Base filename (without extension). + extension: File extension (without dot). + allowed_extensions: Set of extensions to check against. + If None, checks any existing file/dir with same name. + + Returns: + Full path without extension (e.g. "/path/to/name_2"). + """ + sanitized = sanitize_name_for_os(base_name, is_folder=True) + counter = 1 + while True: + suffix = f"_{counter}" if counter > 1 else "" + candidate = os.path.join(parent_dir, f"{sanitized}{suffix}") + if allowed_extensions is not None: + file_parts = (os.path.splitext(f) for f in os.listdir(parent_dir)) + clash = any( + name == f"{sanitized}{suffix}" + and ext[1:].lower() in allowed_extensions + for name, ext in file_parts + ) + else: + clash = os.path.exists(candidate) + if not clash: + return candidate + counter += 1 diff --git a/abogen/domain/subtitle_processor.py b/abogen/domain/subtitle_processor.py new file mode 100644 index 0000000..63ecef3 --- /dev/null +++ b/abogen/domain/subtitle_processor.py @@ -0,0 +1,279 @@ +"""Subtitle-to-audio processing pipeline. + +Converts subtitle files (SRT/ASS/VTT/timestamp text) into audio by +generating TTS for each entry and mixing into a buffer. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from typing import Any, Callable, List, Optional, Tuple + +import numpy as np + +from abogen.domain.audio_buffer import ( + create_silence, + fit_audio_to_duration, + ffmpeg_time_stretch, + mix_audio, + normalize_audio, + SAMPLE_RATE, +) +from abogen.domain.audio_helpers import to_float32 +from abogen.domain.progress import calc_etr_str +from abogen.subtitle_utils import ( + parse_ass_file, + parse_srt_file, + parse_vtt_file, + parse_timestamp_text_file, +) + +logger = logging.getLogger(__name__) + + +@dataclass +class SubtitleEntry: + """A single subtitle entry with timing.""" + start: float + end: Optional[float] + text: str + + +def parse_subtitle_file( + file_path: str, + is_timestamp_text: bool = False, +) -> List[Tuple[float, Optional[float], str]]: + """Parse a subtitle file into (start, end, text) tuples. + + Args: + file_path: Path to subtitle file. + is_timestamp_text: Whether to treat as timestamp text file. + + Returns: + List of (start_time, end_time, text) tuples. + """ + if is_timestamp_text: + return parse_timestamp_text_file(file_path) + + import os + ext = os.path.splitext(file_path)[1].lower() + if ext == ".srt": + return parse_srt_file(file_path) + elif ext == ".vtt": + return parse_vtt_file(file_path) + else: + return parse_ass_file(file_path) + + +def format_time_range( + start: float, + end: Optional[float], + is_auto_end: bool = False, +) -> str: + """Format a time range for display in logs. + + Args: + start: Start time in seconds. + end: End time in seconds, or None. + is_auto_end: Whether end time is auto-detected. + + Returns: + Formatted string like "00:01:23,456 - 00:01:25,789" or "00:01:23 - AUTO". + """ + def _fmt(seconds: float) -> str: + h = int(seconds // 3600) + m = int(seconds % 3600 // 60) + s = int(seconds % 60) + ms = int((seconds - int(seconds)) * 1000) + result = f"{h:02d}:{m:02d}:{s:02d}" + if ms > 0: + result += f",{ms:03d}" + return result + + if is_auto_end or end is None: + return f"{_fmt(start)} - AUTO" + return f"{_fmt(start)} - {_fmt(end)}" + + +def speed_up_audio( + audio: np.ndarray, + speed_factor: float, + method: str = "tts", + *, + backend: Any = None, + text: str = "", + voice: Any = None, + base_speed: float = 1.0, + sample_rate: int = SAMPLE_RATE, +) -> np.ndarray: + """Speed up audio to fit a time window. + + Args: + audio: Input audio buffer. + speed_factor: Required speed multiplier. + method: "ffmpeg" for time-stretch, "tts" for regeneration. + backend: TTS backend (required if method="tts"). + text: Text to regenerate (required if method="tts"). + voice: Voice to use for regeneration. + base_speed: Base speed for TTS. + sample_rate: Sample rate. + + Returns: + Speed-adjusted audio buffer. + """ + if speed_factor <= 1.0: + return audio + + if method == "ffmpeg": + logger.info("FFmpeg time-stretch: %.2fx", speed_factor) + return ffmpeg_time_stretch(audio, speed_factor, sample_rate) + + # TTS regeneration + if backend is None: + return audio + new_speed = base_speed * speed_factor + logger.info("Regenerating at %.2fx speed", new_speed) + results = [ + r for r in backend(text, voice=voice, speed=new_speed, split_pattern=None) + ] + chunks = [r.audio for r in results] + if not chunks: + return audio + return np.concatenate([to_float32(c) for c in chunks]) + + +def process_subtitle_entries( + subtitles: List[Tuple[float, Optional[float], str]], + *, + backend: Any, + voice: Any, + speed: float = 1.0, + cancel_check: Callable[[], bool] = lambda: False, + log_callback: Optional[Callable[[str], None]] = None, + progress_callback: Optional[Callable[[int, str], None]] = None, + replace_newlines: bool = True, + use_gaps: bool = False, + is_timestamp_text: bool = False, + subtitle_speed_method: str = "tts", + sample_rate: int = SAMPLE_RATE, +) -> np.ndarray: + """Process subtitle entries: generate TTS for each and mix into buffer. + + This is the core domain logic for subtitle-to-audio conversion. + UI-specific concerns (signals, widgets) are handled via callbacks. + + Args: + subtitles: List of (start, end, text) tuples. + backend: TTS pipeline callable. + voice: Resolved voice for TTS. + speed: TTS speed. + cancel_check: Returns True if processing should stop. + log_callback: Called with log messages. + progress_callback: Called with (percent, etr_string). + replace_newlines: Replace \\n with spaces in text. + use_gaps: Whether to use silent gaps between subtitles. + is_timestamp_text: Whether input is timestamp text. + subtitle_speed_method: "ffmpeg" or "tts" for speed adjustment. + sample_rate: Audio sample rate. + + Returns: + Mixed audio buffer (float32). + """ + if not subtitles: + return np.array([], dtype="float32") + + max_end = max((end for _, end, _ in subtitles if end is not None), default=0) + buffer_samples = int(max_end * sample_rate) + sample_rate + audio_buffer = np.zeros(buffer_samples, dtype="float32") + etr_start = time.time() + total = len(subtitles) + + for idx, (start_time, end_time, text) in enumerate(subtitles, 1): + if cancel_check(): + break + + processed_text = text.replace("\n", " ") if replace_newlines else text + next_start = ( + subtitles[idx][0] + if (use_gaps and idx < total) + else float("inf") + ) + subtitle_duration = None if end_time is None else end_time - start_time + + is_auto_end = is_timestamp_text or (use_gaps and idx == total) or end_time is None + if log_callback: + log_callback( + f"\n[{idx}/{total}] {format_time_range(start_time, end_time, is_auto_end)}: {processed_text}" + ) + + # Generate TTS + results = [ + r for r in backend( + processed_text, voice=voice, speed=speed, split_pattern=None + ) + if not cancel_check() + ] + if cancel_check(): + break + + audio_chunks = [r.audio for r in results] + full_audio = ( + np.concatenate([to_float32(a) for a in audio_chunks]) + if audio_chunks + else np.zeros(int((subtitle_duration or 0) * sample_rate), dtype="float32") + ) + audio_duration = len(full_audio) / sample_rate + + # Timing adjustment + if is_timestamp_text: + end_time = start_time + audio_duration + subtitle_duration = audio_duration + elif use_gaps: + end_time = min(start_time + audio_duration, next_start) + subtitle_duration = end_time - start_time + elif subtitle_duration is None: + subtitle_duration = audio_duration + end_time = start_time + audio_duration + + # Speed up if needed + speedup_threshold = next_start - start_time if use_gaps else subtitle_duration + if audio_duration > speedup_threshold and speedup_threshold > 0: + speed_factor = audio_duration / speedup_threshold + full_audio = speed_up_audio( + full_audio, speed_factor, + method=subtitle_speed_method, + backend=backend, text=processed_text, + voice=voice, base_speed=speed, + sample_rate=sample_rate, + ) + audio_duration = len(full_audio) / sample_rate + + # Adjust duration after speed change + if use_gaps: + end_time = min(start_time + audio_duration, next_start) + subtitle_duration = end_time - start_time + elif subtitle_duration is None: + subtitle_duration = audio_duration + end_time = start_time + audio_duration + + # Pad or trim to subtitle duration + full_audio = fit_audio_to_duration(full_audio, subtitle_duration, sample_rate) + + # Mix into buffer + start_sample = int(start_time * sample_rate) + audio_buffer = mix_audio(audio_buffer, full_audio, start_sample) + + # Progress + if progress_callback: + percent = min(int(idx / total * 100), 99) + etr = calc_etr_str(time.time() - etr_start, idx, total) + progress_callback(percent, etr) + + # Normalize if needed + if np.abs(audio_buffer).max() > 1.0: + logger.info("Normalizing audio (peak: %.2f)", np.abs(audio_buffer).max()) + audio_buffer = normalize_audio(audio_buffer) + + return audio_buffer diff --git a/abogen/pyqt/conversion.py b/abogen/pyqt/conversion.py index a33a83a..790bd53 100644 --- a/abogen/pyqt/conversion.py +++ b/abogen/pyqt/conversion.py @@ -23,11 +23,16 @@ from abogen.constants import ( ) from abogen.infrastructure.subtitle_writer import create_subtitle_writer from abogen.domain.split_pattern import get_split_pattern +from abogen.domain.subtitle_processor import ( + parse_subtitle_file, + process_subtitle_entries, +) from abogen.domain.output_paths import ( resolve_output_directory, build_output_path, sanitize_output_stem, sanitize_filename_for_chapter, + resolve_unique_path, ) from abogen.domain.audio_helpers import build_ffmpeg_command, to_float32 from abogen.domain.audio_sink import AudioSink, open_audio_sink @@ -79,11 +84,7 @@ _USER_RESPONSE_TIMEOUT = ( from abogen.subtitle_utils import ( clean_text, - parse_srt_file, - parse_vtt_file, detect_timestamps_in_text, - parse_timestamp_text_file, - parse_ass_file, get_sample_voice_text, sanitize_name_for_os, split_text_by_voice_markers @@ -1178,16 +1179,7 @@ class ConversionThread(QThread): """Process subtitle files with precise timing and generate output subtitles.""" try: # Parse subtitle file - if is_timestamp_text: - subtitles = parse_timestamp_text_file(self.file_name) - else: - file_ext = os.path.splitext(self.file_name)[1].lower() - if file_ext == ".srt": - subtitles = parse_srt_file(self.file_name) - elif file_ext == ".vtt": - subtitles = parse_vtt_file(self.file_name) - else: - subtitles = parse_ass_file(self.file_name) + subtitles = parse_subtitle_file(self.file_name, is_timestamp_text) if not subtitles: self.log_updated.emit(("No valid subtitle entries found.", "red")) @@ -1202,7 +1194,6 @@ class ConversionThread(QThread): # Setup output paths base_name = os.path.splitext(os.path.basename(base_path))[0] - sanitized_base_name = sanitize_name_for_os(base_name, is_folder=True) parent_dir = ( user_desktop_dir() if self.save_option == "Save to Desktop" @@ -1219,24 +1210,11 @@ class ConversionThread(QThread): ) return - # Find unique filename - counter = 1 allowed_exts = set(SUPPORTED_SOUND_FORMATS + SUPPORTED_SUBTITLE_FORMATS) - while True: - suffix = f"_{counter}" if counter > 1 else "" - # Use generator expression to avoid processing all files upfront - file_parts = (os.path.splitext(f) for f in os.listdir(parent_dir)) - if not any( - name == f"{sanitized_base_name}{suffix}" - and ext[1:].lower() in allowed_exts - for name, ext in file_parts - ): - break - counter += 1 - - base_filepath_no_ext = os.path.join( - parent_dir, f"{sanitized_base_name}{suffix}" + sanitized_base_path = resolve_unique_path( + parent_dir, base_name, self.output_format, allowed_exts ) + base_filepath_no_ext = sanitized_base_path merged_out_path = f"{base_filepath_no_ext}.{self.output_format}" rate = 24000 @@ -1265,241 +1243,36 @@ class ConversionThread(QThread): # Load voice loaded_voice = resolve_voice(self.voice, tts, self.use_gpu) - # Calculate initial audio buffer size from timed subtitles only - max_end_time = max( - (end for _, end, _ in subtitles if end is not None), default=0 + # Process all subtitles via domain + audio_buffer = process_subtitle_entries( + subtitles, + backend=self.backend, + voice=loaded_voice, + speed=self.speed, + cancel_check=lambda: self.cancel_requested, + log_callback=lambda msg: self.log_updated.emit((msg, "grey")), + progress_callback=self.progress_updated.emit, + replace_newlines=getattr(self, "replace_single_newlines", True), + use_gaps=getattr(self, "use_silent_gaps", False), + is_timestamp_text=is_timestamp_text, + subtitle_speed_method=getattr(self, "subtitle_speed_method", "tts"), ) - buffer_samples = int(max_end_time * rate) + rate - audio_buffer = np.zeros(buffer_samples, dtype="float32") - # Process each subtitle and mix into buffer - self.etr_start_time = time.time() - - for idx, (start_time, end_time, text) in enumerate(subtitles, 1): - if self.cancel_requested: - if subtitle_writer: - subtitle_writer.close() - self.conversion_finished.emit("Cancelled", None) - return - - # Process text and timing - replace_nl = getattr(self, "replace_single_newlines", True) - processed_text = text.replace("\n", " ") if replace_nl else text - use_gaps = getattr(self, "use_silent_gaps", False) - next_start = ( - subtitles[idx][0] - if (use_gaps and idx < len(subtitles)) - else float("inf") - ) - subtitle_duration = None if end_time is None else end_time - start_time - - h1, m1, s1 = ( - int(start_time // 3600), - int(start_time % 3600 // 60), - int(start_time % 60), - ) - ms1 = int((start_time - int(start_time)) * 1000) - is_last = ( - is_timestamp_text - or (use_gaps and idx == len(subtitles)) - or end_time is None - ) - if is_last: - time_str = ( - f"{h1:02d}:{m1:02d}:{s1:02d}" - + (f",{ms1:03d}" if ms1 > 0 else "") - + " - AUTO" - ) - else: - h2, m2, s2 = ( - int(end_time // 3600), - int(end_time % 3600 // 60), - int(end_time % 60), - ) - ms2 = int((end_time - int(end_time)) * 1000) - time_str = ( - f"{h1:02d}:{m1:02d}:{s1:02d}" - + (f",{ms1:03d}" if ms1 > 0 else "") - + " - " - + f"{h2:02d}:{m2:02d}:{s2:02d}" - + (f",{ms2:03d}" if ms2 > 0 else "") - ) - self.log_updated.emit( - f"\n[{idx}/{len(subtitles)}] {time_str}: {processed_text}" - ) - - # Generate TTS audio - tts_results = [ - r - for r in self.backend( - processed_text, - voice=loaded_voice, - speed=self.speed, - split_pattern=None, - ) - if not self.cancel_requested - ] - audio_chunks = [r.audio for r in tts_results] - - if self.cancel_requested: - if subtitle_writer: - subtitle_writer.close() - self.conversion_finished.emit("Cancelled", None) - return - - # Concatenate audio and determine duration - full_audio = ( - np.concatenate( - [to_float32(a) for a in audio_chunks] - ) - if audio_chunks - else np.zeros( - int((subtitle_duration or 0) * rate), dtype="float32" - ) - ) - audio_duration = len(full_audio) / rate - - # Use actual audio length for timing - if is_timestamp_text: - end_time = start_time + audio_duration - subtitle_duration = audio_duration - elif use_gaps: - end_time = min(start_time + audio_duration, next_start) - subtitle_duration = end_time - start_time - elif subtitle_duration is None: - subtitle_duration = audio_duration - end_time = start_time + audio_duration - - # Speed up if needed - speedup_threshold = ( - next_start - start_time if use_gaps else subtitle_duration - ) - if audio_duration > speedup_threshold: - speed_factor = audio_duration / speedup_threshold - - if getattr(self, "subtitle_speed_method", "tts") == "ffmpeg": - # FFmpeg time-stretch (faster processing) - self.log_updated.emit( - (f" -> FFmpeg time-stretch: {speed_factor:.2f}x", "grey") - ) - - static_ffmpeg.add_paths() - num_stages = max( - 1, - int( - np.ceil( - np.log(speed_factor) / np.log(2.0) - ) - ), - ) - tempo = speed_factor ** (1.0 / num_stages) - filter_str = ",".join([f"atempo={tempo:.6f}"] * num_stages) - - speed_proc = subprocess.Popen( - [ - "ffmpeg", - "-y", - "-f", - "f32le", - "-ar", - str(rate), - "-ac", - "1", - "-i", - "pipe:0", - "-filter:a", - filter_str, - "-f", - "f32le", - "-ar", - str(rate), - "-ac", - "1", - "pipe:1", - ], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) - full_audio = np.frombuffer( - speed_proc.communicate(input=full_audio.tobytes())[0], - dtype="float32", - ) - audio_duration = len(full_audio) / rate - else: - # TTS regeneration (better quality) - new_speed = self.speed * speed_factor - self.log_updated.emit( - (f" -> Regenerating at {new_speed:.2f}x speed", "grey") - ) - - tts_results = [ - r - for r in self.backend( - processed_text, - voice=loaded_voice, - speed=new_speed, - split_pattern=None, - ) - if not self.cancel_requested - ] - audio_chunks = [r.audio for r in tts_results] - - full_audio = ( - np.concatenate( - [to_float32(a) for a in audio_chunks] - ) - if audio_chunks - else np.zeros( - int(subtitle_duration * rate), dtype="float32" - ) - ) - audio_duration = len(full_audio) / rate - - # Adjust duration after potential speed changes - if use_gaps: - end_time = min(start_time + audio_duration, next_start) - subtitle_duration = end_time - start_time - elif subtitle_duration is None: - subtitle_duration = audio_duration - end_time = start_time + audio_duration - - # Pad or trim to subtitle duration - target_samples = int(subtitle_duration * rate) - if len(full_audio) < target_samples: - padding_duration = (target_samples - len(full_audio)) / rate - full_audio = np.concatenate([full_audio, create_silence(padding_duration)]) - elif len(full_audio) > target_samples: - full_audio = full_audio[:target_samples] - - # Mix audio into buffer at the correct position (handles overlaps) - start_sample = int(start_time * rate) - audio_buffer = mix_audio(audio_buffer, full_audio, start_sample) - - # Write subtitle + if self.cancel_requested: if subtitle_writer: - display_text = ( - processed_text - if "ass" in subtitle_format or replace_nl - else processed_text.replace("\n", "\\N") - ) - subtitle_writer.write_entry(start_time, end_time, display_text) + subtitle_writer.close() + self.conversion_finished.emit("Cancelled", None) + return - # Update progress - percent = min(int(idx / len(subtitles) * 100), 99) - etr_str = calc_etr_str( - time.time() - self.etr_start_time, - idx, - len(subtitles), + # Write subtitle entries (post-loop) + for start_time, end_time, text in subtitles: + processed_text = text.replace("\n", " ") if getattr(self, "replace_single_newlines", True) else text + display_text = ( + processed_text + if "ass" in subtitle_format + else processed_text.replace("\n", "\\N") ) - self.progress_updated.emit(percent, etr_str) - - # Normalize audio buffer to prevent clipping from mixed overlaps - if np.abs(audio_buffer).max() > 1.0: - self.log_updated.emit( - f"\n -> Normalizing audio (peak: {np.abs(audio_buffer).max():.2f})" - ) - audio_buffer = normalize_audio(audio_buffer) + subtitle_writer.write_entry(start_time, end_time, display_text) # Write the complete audio buffer self.log_updated.emit(("\nFinalizing audio. Please wait...", "grey")) diff --git a/tests/test_domain_audio_buffer_extras.py b/tests/test_domain_audio_buffer_extras.py new file mode 100644 index 0000000..430b838 --- /dev/null +++ b/tests/test_domain_audio_buffer_extras.py @@ -0,0 +1,59 @@ +"""Tests for domain/audio_buffer.py — fit_audio_to_duration, ffmpeg_time_stretch.""" + +import numpy as np +import pytest + +from abogen.domain.audio_buffer import fit_audio_to_duration, ffmpeg_time_stretch, SAMPLE_RATE + + +class TestFitAudioToDuration: + def test_exact_length(self): + audio = np.ones(24000, dtype="float32") + result = fit_audio_to_duration(audio, 1.0, SAMPLE_RATE) + assert len(result) == 24000 + + def test_shorter_pads_with_zeros(self): + audio = np.ones(12000, dtype="float32") + result = fit_audio_to_duration(audio, 1.0, SAMPLE_RATE) + assert len(result) == 24000 + assert result[0] == 1.0 + assert result[12000] == 0.0 + + def test_longer_trims(self): + audio = np.ones(48000, dtype="float32") + result = fit_audio_to_duration(audio, 1.0, SAMPLE_RATE) + assert len(result) == 24000 + assert result[-1] == 1.0 + + def test_empty_input(self): + result = fit_audio_to_duration(np.array([], dtype="float32"), 0.5, SAMPLE_RATE) + assert len(result) == 12000 + assert np.all(result == 0.0) + + def test_output_dtype(self): + audio = np.ones(100, dtype="float32") + result = fit_audio_to_duration(audio, 0.5, SAMPLE_RATE) + assert result.dtype == np.float32 + + +class TestFfmpegTimeStretch: + def test_no_stretch_below_threshold(self): + audio = np.ones(24000, dtype="float32") + result = ffmpeg_time_stretch(audio, 0.8, SAMPLE_RATE) + np.testing.assert_array_equal(result, audio) + + def test_no_stretch_at_exactly_one(self): + audio = np.ones(24000, dtype="float32") + result = ffmpeg_time_stretch(audio, 1.0, SAMPLE_RATE) + np.testing.assert_array_equal(result, audio) + + def test_empty_audio(self): + result = ffmpeg_time_stretch(np.array([], dtype="float32"), 2.0, SAMPLE_RATE) + assert len(result) == 0 + + def test_stretch_reduces_duration(self): + audio = np.random.randn(48000).astype("float32") + result = ffmpeg_time_stretch(audio, 2.0, SAMPLE_RATE) + assert len(result) < len(audio) + assert len(result) > 0 + assert result.dtype == np.float32 diff --git a/tests/test_domain_resolve_unique_path.py b/tests/test_domain_resolve_unique_path.py new file mode 100644 index 0000000..f1e0433 --- /dev/null +++ b/tests/test_domain_resolve_unique_path.py @@ -0,0 +1,51 @@ +"""Tests for domain/output_paths.py — resolve_unique_path.""" + +import os +import tempfile + +from abogen.domain.output_paths import resolve_unique_path + + +class TestResolveUniquePath: + def test_no_collision(self, tmp_path): + result = resolve_unique_path(str(tmp_path), "chapter", "srt") + assert result == os.path.join(str(tmp_path), "chapter") + assert not os.path.exists(result) + + def test_collision_appends_counter(self, tmp_path): + (tmp_path / "chapter.srt").touch() + result = resolve_unique_path(str(tmp_path), "chapter", "srt", {"srt"}) + assert result == os.path.join(str(tmp_path), "chapter_2") + + def test_multiple_collisions(self, tmp_path): + (tmp_path / "chapter.srt").touch() + (tmp_path / "chapter_2.srt").touch() + (tmp_path / "chapter_3.srt").touch() + result = resolve_unique_path(str(tmp_path), "chapter", "srt", {"srt"}) + assert result == os.path.join(str(tmp_path), "chapter_4") + + def test_no_allowed_extensions_skips_files(self, tmp_path): + (tmp_path / "chapter.txt").touch() + result = resolve_unique_path(str(tmp_path), "chapter", "srt") + assert result == os.path.join(str(tmp_path), "chapter") + + def test_sanitizes_name(self, tmp_path): + # On Windows, ":" is illegal; on Linux it's allowed. + # Just verify the function doesn't crash and returns a valid path. + result = resolve_unique_path(str(tmp_path), "My Chapter: Part 1", "srt") + assert os.path.dirname(result) == str(tmp_path) + + def test_directory_collision(self, tmp_path): + (tmp_path / "chapter").mkdir() + result = resolve_unique_path(str(tmp_path), "chapter", "srt") + assert result == os.path.join(str(tmp_path), "chapter_2") + + def test_case_insensitive_extension(self, tmp_path): + (tmp_path / "chapter.SRT").touch() + result = resolve_unique_path(str(tmp_path), "chapter", "srt", {"srt"}) + assert result == os.path.join(str(tmp_path), "chapter_2") + + def test_unrelated_extensions_no_collision(self, tmp_path): + (tmp_path / "chapter.mp3").touch() + result = resolve_unique_path(str(tmp_path), "chapter", "srt", {"srt"}) + assert result == os.path.join(str(tmp_path), "chapter") diff --git a/tests/test_domain_subtitle_processor.py b/tests/test_domain_subtitle_processor.py new file mode 100644 index 0000000..071c76c --- /dev/null +++ b/tests/test_domain_subtitle_processor.py @@ -0,0 +1,169 @@ +"""Tests for domain/subtitle_processor.py — parse_subtitle_file, format_time_range, process_subtitle_entries.""" + +import os +import tempfile +from dataclasses import dataclass +from typing import Optional +from unittest.mock import MagicMock + +import numpy as np +import pytest + +from abogen.domain.subtitle_processor import ( + parse_subtitle_file, + format_time_range, + speed_up_audio, + process_subtitle_entries, +) + + +# --- format_time_range tests --- + +class TestFormatTimeRange: + def test_basic_range(self): + result = format_time_range(0.0, 5.0) + assert result == "00:00:00 - 00:00:05" + + def test_with_milliseconds(self): + result = format_time_range(1.5, 3.123) + assert "00:00:01,500" in result + assert "00:00:03,123" in result + + def test_auto_end(self): + result = format_time_range(10.0, 15.0, is_auto_end=True) + assert result == "00:00:10 - AUTO" + + def test_none_end(self): + result = format_time_range(5.0, None) + assert result == "00:00:05 - AUTO" + + def test_hours(self): + result = format_time_range(3661.0, 3665.0) + assert result == "01:01:01 - 01:01:05" + + +# --- parse_subtitle_file tests --- + +class TestParseSubtitleFile: + def test_parse_srt(self, tmp_path): + srt = tmp_path / "test.srt" + srt.write_text( + "1\n00:00:01,000 --> 00:00:03,000\nHello\n\n" + "2\n00:00:04,000 --> 00:00:06,000\nWorld\n", + encoding="utf-8", + ) + result = parse_subtitle_file(str(srt)) + assert len(result) == 2 + assert result[0][2] == "Hello" + assert result[1][2] == "World" + + def test_parse_vtt(self, tmp_path): + vtt = tmp_path / "test.vtt" + vtt.write_text( + "WEBVTT\n\n00:00:01.000 --> 00:00:03.000\nHello\n\n" + "00:00:04.000 --> 00:00:06.000\nWorld\n", + encoding="utf-8", + ) + result = parse_subtitle_file(str(vtt)) + assert len(result) == 2 + + def test_parse_timestamp_text(self, tmp_path): + ts = tmp_path / "test.txt" + ts.write_text( + "[00:00:01] Hello\n[00:00:04] World\n", + encoding="utf-8", + ) + result = parse_subtitle_file(str(ts), is_timestamp_text=True) + assert len(result) >= 1 + + +# --- speed_up_audio tests --- + +class TestSpeedUpAudio: + def test_no_change_below_threshold(self): + audio = np.ones(24000, dtype="float32") + result = speed_up_audio(audio, 0.8, method="ffmpeg") + np.testing.assert_array_equal(result, audio) + + def test_empty_audio(self): + result = speed_up_audio(np.array([], dtype="float32"), 2.0, method="ffmpeg") + assert len(result) == 0 + + +# --- process_subtitle_entries tests --- + +@dataclass +class FakeResult: + audio: np.ndarray + + +def fake_backend(text, voice=None, speed=1.0, split_pattern=None): + length = int(len(text) * 2400 * speed) + audio = np.random.randn(length).astype("float32") * 0.1 + return [FakeResult(audio=audio)] + + +class TestProcessSubtitleEntries: + def test_empty_subtitles(self): + result = process_subtitle_entries( + [], backend=fake_backend, voice=None + ) + assert len(result) == 0 + + def test_single_entry(self): + subtitles = [(0.0, 3.0, "Hello")] + result = process_subtitle_entries( + subtitles, backend=fake_backend, voice=None + ) + assert len(result) > 0 + assert result.dtype == np.float32 + + def test_cancel_check(self): + subtitles = [(0.0, 5.0, "Hello"), (5.0, 10.0, "World")] + counter = [0] + + def cancel(): + counter[0] += 1 + return counter[0] > 1 + + result = process_subtitle_entries( + subtitles, backend=fake_backend, voice=None, + cancel_check=cancel, + ) + # Buffer is pre-allocated but only first entry processed before cancel + assert result is not None + # Second entry should not have been mixed in (no audio at 5-10s) + assert np.max(np.abs(result[int(5.0 * 24000):])) == 0.0 + + def test_log_callback_called(self): + subtitles = [(0.0, 3.0, "Hello")] + logs = [] + process_subtitle_entries( + subtitles, backend=fake_backend, voice=None, + log_callback=logs.append, + ) + assert len(logs) >= 1 + assert "Hello" in logs[0] + + def test_progress_callback_called(self): + subtitles = [(0.0, 3.0, "Hello")] + progress = [] + process_subtitle_entries( + subtitles, backend=fake_backend, voice=None, + progress_callback=lambda p, e: progress.append((p, e)), + ) + assert len(progress) == 1 + assert progress[0][0] == 99 + + def test_multiple_entries_mixed(self): + subtitles = [ + (0.0, 2.0, "First"), + (2.0, 4.0, "Second"), + (4.0, 6.0, "Third"), + ] + result = process_subtitle_entries( + subtitles, backend=fake_backend, voice=None + ) + assert len(result) > 0 + # Buffer should be at least as long as the last subtitle end + assert len(result) >= int(6.0 * 24000)