mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-22 07:10:28 +02:00
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
This commit is contained in:
Artem Akymenko
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
+34
-261
@@ -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"))
|
||||
|
||||
@@ -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
|
||||
@@ -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")
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user