diff options
| -rw-r--r-- | README.md | 1 | ||||
| -rw-r--r-- | converter/audio.py | 82 | ||||
| -rw-r--r-- | converter/chunking.py | 47 | ||||
| -rw-r--r-- | converter/config.py | 29 | ||||
| -rw-r--r-- | converter/tts.py | 142 | ||||
| -rw-r--r-- | tests/test_audio.py | 67 | ||||
| -rw-r--r-- | tests/test_chunking.py | 70 | ||||
| -rw-r--r-- | tests/test_tts.py | 155 |
8 files changed, 540 insertions, 53 deletions
@@ -193,7 +193,6 @@ If you're cloning one language and outputting another language, `--no-transcript Even tiny amounts of pause between phrases in the sample audio can have a big impact. Try increasing or decreasing them. - ## License MIT diff --git a/converter/audio.py b/converter/audio.py index 6e5a910..3e163be 100644 --- a/converter/audio.py +++ b/converter/audio.py @@ -5,6 +5,7 @@ import re import shutil import subprocess import traceback +import wave from pathlib import Path from typing import Dict, List, NamedTuple, Optional, Tuple @@ -50,6 +51,87 @@ def _concat_escape(path: str) -> str: return path.replace("'", "'\\''") +def _concat_wav_files(sources: List[Path], destination: Path) -> bool: + """Concatenate WAV files with matching parameters using the wave module. + + Returns False (touching nothing) when any input is not a readable WAV + or the parameters differ, so the caller can fall back to ffmpeg. + """ + opened = [] + try: + parameters = None + for source in sources: + wav_file = wave.open(str(source), "rb") + opened.append(wav_file) + current = (wav_file.getnchannels(), wav_file.getsampwidth(), + wav_file.getframerate()) + if parameters is None: + parameters = current + elif current != parameters: + return False + if parameters is None or min(parameters) < 1: + return False + with wave.open(str(destination), "wb") as output: + output.setnchannels(parameters[0]) + output.setsampwidth(parameters[1]) + output.setframerate(parameters[2]) + for wav_file in opened: + output.writeframes(wav_file.readframes(wav_file.getnframes())) + return True + except (wave.Error, EOFError, OSError): + return False + finally: + for wav_file in opened: + try: + wav_file.close() + except Exception: + pass + + +def _concat_with_ffmpeg(sources: List[Path], destination: Path) -> None: + """Concatenate audio files with ffmpeg's concat demuxer, re-encoding to + 16-bit PCM WAV (handles inputs the wave module cannot).""" + if shutil.which("ffmpeg") is None: + raise RuntimeError( + "ffmpeg is required to concatenate audio parts in non-WAV formats " + "(install ffmpeg and try again)" + ) + list_path = destination.with_name(destination.stem + "_parts.txt") + try: + with open(list_path, "w", encoding="utf-8") as list_file: + for source in sources: + list_file.write(f"file '{_concat_escape(str(source))}'\n") + command = [ + "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", + "-f", "concat", "-safe", "0", "-i", str(list_path), + "-c:a", "pcm_s16le", str(destination), + ] + proc = subprocess.run(command, capture_output=True, text=True) + if proc.returncode != 0: + raise RuntimeError( + f"ffmpeg failed to concatenate audio parts: {proc.stderr[-500:]}") + finally: + try: + list_path.unlink(missing_ok=True) + except OSError: + pass + + +def concat_audio_files(sources: List[Path], destination: Path) -> None: + """Concatenate audio files into one file at ``destination``. + + Joins the audio returned by several TTS sub-requests for a single + chunk. Uses the stdlib wave module when every input is a WAV with + matching parameters (lossless, no external tools); otherwise falls + back to ffmpeg's concat demuxer with re-encoding. + """ + if not sources: + raise ValueError("No audio files to concatenate") + if _concat_wav_files(sources, destination): + return + _concat_with_ffmpeg(sources, destination) + + def _encode_args(output_format: str) -> List[str]: """Return ffmpeg output codec/bitrate args for the requested container.""" if output_format == "m4b": diff --git a/converter/chunking.py b/converter/chunking.py index 0c85adf..425765f 100644 --- a/converter/chunking.py +++ b/converter/chunking.py @@ -1,21 +1,40 @@ """Split extracted book text into TTS-sized chunks.""" +import logging import re from typing import List from . import config +logger = logging.getLogger(__name__) + def split_into_chunks(text: str, max_words: int = config.CHUNK_SIZE_WORDS) -> List[str]: """Split text into chunks of at most ``max_words`` words. - Splits on sentence boundaries. Sentences longer than the limit are split - further at clause punctuation (which is kept attached for TTS prosody). - Clause splits only happen at whitespace after punctuation, so tokens like - "1,000,000" or "12:30" are never broken apart or re-joined with added - spaces. A single sentence with no usable split point longer than the - limit is kept intact as one oversized chunk. + ``max_words`` is clamped to ``config.MAX_REQUEST_WORDS``: requests + beyond that ceiling are silently truncated by the TTS servers (no + error is reported), so chunks larger than the ceiling are never + produced regardless of configuration. + + Splits on sentence boundaries. Sentences longer than the limit are + split further at clause punctuation (which is kept attached for TTS + prosody). Clause splits only happen at whitespace after punctuation, + so tokens like "1,000,000" or "12:30" are never broken apart. A piece + with no usable punctuation split point longer than the limit is split + at word boundaries as a last resort: individual tokens stay intact, + but whitespace between them is normalized. """ + if max_words > config.MAX_REQUEST_WORDS: + logger.warning( + "Requested chunk size of %d words exceeds the %d-word request ceiling; " + "larger requests are silently truncated by the TTS servers, so the " + "size is clamped to %d words (see MAX_REQUEST_WORDS in converter/config.py)", + max_words, config.MAX_REQUEST_WORDS, config.MAX_REQUEST_WORDS) + max_words = config.MAX_REQUEST_WORDS + if max_words < 1: + max_words = 1 + if not text.strip(): return [] @@ -35,11 +54,23 @@ def split_into_chunks(text: str, max_words: int = config.CHUNK_SIZE_WORDS) -> Li # Split long sentences at clause boundaries, keeping punctuation. # Only split where whitespace already follows the punctuation so - # the reassembled text is byte-identical to the input (no spaces - # injected into "1,000,000" or "12:30"). + # tokens are never broken apart or re-joined with added spaces + # (no spaces are injected into "1,000,000" or "12:30"). parts = re.split(r"(?<=[,;:])\s+", sentence) for part in parts: part_words = len(part.split()) + if part_words > max_words: + # Last resort: no punctuation split point is available, + # so split at word boundaries. Tokens themselves (and + # therefore numbers like "1,000,000") stay intact. + if current_chunk: + chunks.append(current_chunk.strip()) + current_chunk = "" + current_words = 0 + words = part.split() + for start in range(0, len(words), max_words): + chunks.append(" ".join(words[start:start + max_words])) + continue if current_words + part_words <= max_words: current_chunk += part + " " current_words += part_words diff --git a/converter/config.py b/converter/config.py index eb01462..a19ac63 100644 --- a/converter/config.py +++ b/converter/config.py @@ -13,14 +13,35 @@ QWEN_API_URL = "http://127.0.0.1:7860" # CustomVoice demo VOICE_CLONE_API_URL = "http://127.0.0.1:7861" # Base-model demo FASTER_TTS_API_URL = "http://127.0.0.1:8000" # faster-qwen3-tts server -# Words per TTS generation request. Each API call is ONE model generation: -# long generations lose prosody and can degrade into garbled audio. -CHUNK_SIZE_WORDS = 40 +# Words per TTS generation request. Each request is ONE model generation: +# the voice is re-sampled per request (every chunk boundary can drift +# slightly), while over-long generations lose prosody and can turn garbled. +# ~250 words is ~1.5-2 minutes of speech: few voice boundaries while +# staying inside both servers' generation caps. +CHUNK_SIZE_WORDS = 250 + +# Hard ceiling on words per request, regardless of CHUNK_SIZE_WORDS. Both +# TTS servers silently truncate audio when a single generation exceeds its +# cap (~2.5 min for the faster backend's static KV cache, ~11 min for the +# Gradio demo) without reporting any error, so larger requests are always +# split client-side. Keep a margin below ~300 words to survive slow +# narration on the faster backend. +MAX_REQUEST_WORDS = 250 + +# Duration sanity check: a response whose audio is far shorter than its +# word count implies is treated as silently truncated, fails the request, +# and goes through the normal retry logic. 150 wpm is a typical spoken +# pace; the ratio is set low (0.5) so only gross truncation trips it. +ESTIMATED_WORDS_PER_MINUTE = 150 +MIN_AUDIO_DURATION_RATIO = 0.5 +MIN_WORDS_FOR_DURATION_CHECK = 10 + VOICE_CLONE_MAX_CHUNK_CHARS = 200 # Server-side re-chunking limit for clone requests VOICE_CLONE_CHUNK_GAP = 0 # Pause (seconds) between server-side clone chunks + MIN_DELAY_BETWEEN_CHUNKS = 0 # Pause between API calls (rate-limit protection; local servers need none) -API_TIMEOUT = 300 # Seconds before an API call times out +API_TIMEOUT = 600 # Seconds before an API call times out (a ~250-word request can take minutes on the Gradio demo) MAX_RETRIES = 3 # Attempts per chunk request HEARTBEAT_INTERVAL_SECONDS = 30 # Print "still working" this often during a chunk diff --git a/converter/tts.py b/converter/tts.py index 9930558..8a1667a 100644 --- a/converter/tts.py +++ b/converter/tts.py @@ -12,6 +12,7 @@ import json import logging import shutil import sys +import tempfile import threading import time import urllib.error @@ -21,6 +22,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from . import config +from .audio import concat_audio_files, probe_duration_ms from .chunking import split_into_chunks logger = logging.getLogger(__name__) @@ -151,6 +153,49 @@ def transcribe_reference_audio(audio_path: str, model_name: str = "base") -> Opt return None +def check_for_truncation(text: str, actual_seconds: Optional[float], label: str) -> None: + """Raise RuntimeError when audio is far shorter than its text implies. + + Both backends silently truncate audio when a single generation hits an + internal cap (no error is reported to the client), so grossly short + audio must be detected client-side: failing the request lets the retry + logic re-run it, and persistent failures surface as failed chunks + instead of a "successful" run with missing audio. ``actual_seconds`` + is None when the duration could not be determined, in which case the + check is skipped. Requests shorter than + ``config.MIN_WORDS_FOR_DURATION_CHECK`` words are not checked (their + duration estimates are too noisy). + """ + words = len(text.split()) + if actual_seconds is None or words < config.MIN_WORDS_FOR_DURATION_CHECK: + return + expected_seconds = 60.0 * words / config.ESTIMATED_WORDS_PER_MINUTE + if actual_seconds < expected_seconds * config.MIN_AUDIO_DURATION_RATIO: + raise RuntimeError( + f"{label}: audio is far shorter than the text implies " + f"({actual_seconds:.1f}s of audio for {words} words, expected at " + f"least {expected_seconds * config.MIN_AUDIO_DURATION_RATIO:.0f}s); " + "the TTS server likely truncated the generation silently" + ) + + +def _audio_duration_seconds(path: Path) -> Optional[float]: + """Return an audio file's duration in seconds, or None when unknown.""" + try: + with wave.open(str(path), "rb") as wav_file: + framerate = wav_file.getframerate() + if framerate > 0: + return wav_file.getnframes() / float(framerate) + except (wave.Error, EOFError, OSError): + pass + if shutil.which("ffprobe") is None: + return None + milliseconds = probe_duration_ms(path) + if milliseconds <= 0: + return None + return milliseconds / 1000.0 + + class _BaseTTSClient: """Shared chunk retry logic, heartbeat, and chunk file bookkeeping.""" @@ -355,39 +400,76 @@ class QwenTTSClient(_BaseTTSClient): # ------------------------------------------------------------------ def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: - """Generate one audio chunk; returns its path in the chunks folder.""" + """Generate one audio chunk; returns its path in the chunks folder. + + The text is split into sub-requests of at most + ``config.MAX_REQUEST_WORDS`` words each (the book-level chunker + normally guarantees this already; the split is defense in depth + against pathological input such as a punctuation-free run of + text), and the audio files returned for the sub-requests are + concatenated into one chunk file. + """ try: - if self.voice_mode == VOICE_MODE_CUSTOM: - with self._chunk_heartbeat(chunk_num): - result = self._generate_custom_voice(text) - elif self.voice_mode == VOICE_MODE_CLONE: - with self._chunk_heartbeat(chunk_num): - result = self._generate_voice_clone(text) - else: - raise ValueError(f"Unknown voice mode: {self.voice_mode}") - - if not isinstance(result, (tuple, list)) or not result: - raise RuntimeError("Qwen API returned an invalid result") - - audio_path = result[0] # First element is the audio file path - if not isinstance(audio_path, (str, Path)) or not audio_path: - raise RuntimeError("Qwen API did not return an audio file path") - - source = Path(audio_path) - if not source.exists(): - raise RuntimeError(f"Generated audio file not found: {audio_path}") - - suffix = source.suffix or ".wav" - output_path = self._chunk_path(chunk_num, suffix) - shutil.copy2(source, output_path) + sub_texts = split_into_chunks(text, max_words=config.MAX_REQUEST_WORDS) + if not sub_texts: + raise RuntimeError("No text to synthesize") - logger.debug("Chunk %d generated successfully", chunk_num) + output_path: Optional[Path] = None + with tempfile.TemporaryDirectory(prefix="tts_parts_") as parts_dir, \ + self._chunk_heartbeat(chunk_num): + part_paths = [ + self._generate_sub_request(sub_text, parts_dir, sub_num, + len(sub_texts), chunk_num) + for sub_num, sub_text in enumerate(sub_texts, 1) + ] + if len(part_paths) == 1: + suffix = part_paths[0].suffix or ".wav" + output_path = self._chunk_path(chunk_num, suffix) + shutil.copy2(part_paths[0], output_path) + else: + output_path = self._chunk_path(chunk_num, ".wav") + concat_audio_files(part_paths, output_path) + + logger.debug("Chunk %d generated successfully (%d sub-request(s))", + chunk_num, len(sub_texts)) return str(output_path) except Exception as exc: logger.error("Qwen chunk processing failed for chunk %d: %s", chunk_num, exc) return None + def _generate_sub_request(self, text: str, parts_dir: str, sub_num: int, + sub_total: int, chunk_num: int) -> Path: + """Run one API generation for ``text``; returns the downloaded audio.""" + if sub_total > 1: + logger.info("Chunk %d: oversized input split into %d requests " + "(sub-request %d/%d)", chunk_num, sub_total, sub_num, sub_total) + if self.voice_mode == VOICE_MODE_CUSTOM: + result = self._generate_custom_voice(text) + elif self.voice_mode == VOICE_MODE_CLONE: + result = self._generate_voice_clone(text) + else: + raise ValueError(f"Unknown voice mode: {self.voice_mode}") + + if not isinstance(result, (tuple, list)) or not result: + raise RuntimeError("Qwen API returned an invalid result") + + audio_path = result[0] # First element is the audio file path + if not isinstance(audio_path, (str, Path)) or not audio_path: + raise RuntimeError("Qwen API did not return an audio file path") + + source = Path(audio_path) + if not source.exists(): + raise RuntimeError(f"Generated audio file not found: {audio_path}") + + destination = Path(parts_dir) / f"part_{sub_num:02d}{source.suffix or '.wav'}" + shutil.copy2(source, destination) + + check_for_truncation( + text, _audio_duration_seconds(destination), + f"Chunk {chunk_num} sub-request {sub_num}/{sub_total}") + return destination + # ------------------------------------------------------------------ # API payloads # ------------------------------------------------------------------ @@ -556,15 +638,19 @@ class FasterTTSClient(_BaseTTSClient): def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: """Generate one audio chunk; returns its path in the chunks folder.""" try: - sub_chunks = split_into_chunks(text, max_words=config.CHUNK_SIZE_WORDS) + sub_chunks = split_into_chunks(text, max_words=config.MAX_REQUEST_WORDS) if not sub_chunks: raise RuntimeError("No text to synthesize") pcm_parts: List[bytes] = [] with self._chunk_heartbeat(chunk_num): for sub_num, sub_text in enumerate(sub_chunks, 1): - pcm_parts.append(self._request_pcm_with_retry( - sub_text, chunk_num, sub_num, len(sub_chunks))) + pcm = self._request_pcm_with_retry( + sub_text, chunk_num, sub_num, len(sub_chunks)) + check_for_truncation( + sub_text, len(pcm) / (2 * SAMPLE_RATE), + f"Chunk {chunk_num} sub-chunk {sub_num}/{len(sub_chunks)}") + pcm_parts.append(pcm) output_path = self._chunk_path(chunk_num, ".wav") with wave.open(str(output_path), "wb") as wav_file: diff --git a/tests/test_audio.py b/tests/test_audio.py index fb448f8..c043766 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -1,9 +1,11 @@ """Tests for audio helpers: speed parameters, chunk cleanup, encoding, -command construction, and duration verification.""" +command construction, duration verification, and audio concatenation.""" import tempfile import unittest +import wave from pathlib import Path +from unittest.mock import patch from converter import audio from converter import config @@ -17,6 +19,7 @@ from converter.audio import ( build_ffmetadata, build_m4b_chapters_command, cleanup_chunks, + concat_audio_files, speed_export_params, verify_output_duration, ) @@ -422,5 +425,67 @@ class BuildM4bChaptersCommandMetaTests(unittest.TestCase): self.assertNotIn("-metadata", cmd) +class ConcatAudioFilesTests(unittest.TestCase): + """Concatenation of sub-request audio into one chunk file.""" + + @staticmethod + def _write_wav(path: Path, frames: bytes, framerate: int = 24000) -> Path: + with wave.open(str(path), "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(framerate) + wav_file.writeframes(frames) + return path + + def test_wav_files_are_merged_in_order(self): + with tempfile.TemporaryDirectory() as tmp: + first = self._write_wav(Path(tmp) / "a.wav", b"\x01\x00" * 10) + second = self._write_wav(Path(tmp) / "b.wav", b"\x02\x00" * 20) + destination = Path(tmp) / "out.wav" + concat_audio_files([first, second], destination) + with wave.open(str(destination), "rb") as wav_file: + self.assertEqual(wav_file.getframerate(), 24000) + self.assertEqual(wav_file.getnchannels(), 1) + self.assertEqual(wav_file.getsampwidth(), 2) + frames = wav_file.readframes(wav_file.getnframes()) + self.assertEqual(frames, b"\x01\x00" * 10 + b"\x02\x00" * 20) + + def test_single_wav_file_is_copied(self): + with tempfile.TemporaryDirectory() as tmp: + source = self._write_wav(Path(tmp) / "a.wav", b"\x03\x00" * 15) + destination = Path(tmp) / "out.wav" + concat_audio_files([source], destination) + with wave.open(str(destination), "rb") as wav_file: + self.assertEqual(wav_file.readframes(wav_file.getnframes()), + b"\x03\x00" * 15) + + def test_empty_source_list_raises(self): + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(ValueError): + concat_audio_files([], Path(tmp) / "out.wav") + + def test_mismatched_wav_parameters_fall_back_to_ffmpeg(self): + with tempfile.TemporaryDirectory() as tmp: + first = self._write_wav(Path(tmp) / "a.wav", b"\x01\x00" * 10, framerate=24000) + second = self._write_wav(Path(tmp) / "b.wav", b"\x02\x00" * 10, framerate=16000) + destination = Path(tmp) / "out.wav" + with patch("converter.audio.shutil.which", return_value=None), \ + self.assertRaises(RuntimeError) as ctx: + concat_audio_files([first, second], destination) + self.assertIn("ffmpeg", str(ctx.exception)) + # The wave-module path must not have written a partial output. + self.assertFalse(destination.exists()) + + def test_non_wav_input_falls_back_to_ffmpeg(self): + with tempfile.TemporaryDirectory() as tmp: + source = Path(tmp) / "part.mp3" + source.write_bytes(b"not a wav file") + destination = Path(tmp) / "out.wav" + with patch("converter.audio.shutil.which", return_value=None), \ + self.assertRaises(RuntimeError) as ctx: + concat_audio_files([source], destination) + self.assertIn("ffmpeg", str(ctx.exception)) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_chunking.py b/tests/test_chunking.py index d1d95bd..2062b4e 100644 --- a/tests/test_chunking.py +++ b/tests/test_chunking.py @@ -1,18 +1,45 @@ """Tests for text chunking.""" import unittest +from unittest.mock import patch from converter import config from converter.chunking import split_into_chunks class ChunkSizeDefaultTests(unittest.TestCase): - """Guard the default chunk size: each API call is one model generation, - and long single generations lose prosody, can turn garbled, and are - truncated at the model's token limit (text past it is never spoken).""" + """Guard the request-size settings: each API call is one model + generation, and the servers silently truncate audio past their caps + (~2.5 min faster backend, ~11 min Gradio demo), so both the default + chunk size and the hard ceiling must stay well inside that budget.""" - def test_default_chunk_size_within_single_generation_budget(self): - self.assertLessEqual(config.CHUNK_SIZE_WORDS, 60) + def test_default_chunk_size_within_request_ceiling(self): + self.assertLessEqual(config.CHUNK_SIZE_WORDS, config.MAX_REQUEST_WORDS) + + def test_request_ceiling_within_single_generation_budget(self): + self.assertLessEqual(config.MAX_REQUEST_WORDS, 300) + + def test_sizes_are_positive(self): + self.assertGreaterEqual(config.CHUNK_SIZE_WORDS, 1) + self.assertGreaterEqual(config.MAX_REQUEST_WORDS, 1) + + +class RequestCeilingClampTests(unittest.TestCase): + def test_oversized_chunk_size_is_clamped_with_warning(self): + text = " ".join(f"word{i}" for i in range(30)) + "." + with patch.object(config, "MAX_REQUEST_WORDS", 10), \ + self.assertLogs("converter.chunking", level="WARNING") as logs: + chunks = split_into_chunks(text, max_words=5000) + self.assertTrue(all(len(chunk.split()) <= 10 for chunk in chunks)) + self.assertIn("clamped", " ".join(logs.output)) + + def test_default_ceiling_clamps_realistic_configuration(self): + sentences = " ".join( + f"S{i} " + " ".join(["word"] * 8) + "." for i in range(60)) + chunks = split_into_chunks(sentences, max_words=5000) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(chunk.split()) <= config.MAX_REQUEST_WORDS + for chunk in chunks)) class SplitIntoChunksTests(unittest.TestCase): @@ -54,16 +81,39 @@ class SplitIntoChunksTests(unittest.TestCase): self.assertNotIn("12: 30", joined) def test_clause_split_requires_whitespace_after_punctuation(self): - # Run-on clauses without spaces after commas have no split point and - # must stay byte-identical rather than being re-joined with spaces. + # Run-on clauses without spaces after commas have no clause split + # point, so the last-resort word-boundary split fires instead. + # Tokens themselves (and numbers like "1,000,000") stay intact. sentence = ",".join([" ".join(["w"] * 5) for _ in range(10)]) + "." chunks = split_into_chunks(sentence, max_words=12) - self.assertEqual(chunks, [sentence]) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(chunk.split()) <= 12 for chunk in chunks)) + tokens = sentence.replace(",", " , ").split() + rejoined = " ".join(chunks).replace(",", " , ").split() + self.assertEqual(rejoined, tokens) - def test_single_oversized_sentence_stays_intact(self): + def test_single_oversized_sentence_is_word_split(self): + # A punctuation-free sentence longer than the limit is split at word + # boundaries: the request-size ceiling is a hard limit because the + # TTS servers silently truncate oversized generations. sentence = " ".join(["word"] * 30) + "." chunks = split_into_chunks(sentence, max_words=10) - self.assertEqual(chunks, [sentence]) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(len(chunk.split()) <= 10 for chunk in chunks)) + self.assertEqual(sum(len(chunk.split()) for chunk in chunks), 30) + + def test_word_split_never_breaks_number_tokens(self): + # Numbers and other punctuation-bearing tokens are single words and + # must never be broken apart by the last-resort word split. + sentence = ("There were exactly 1,000,000 soldiers marching at 12:30 " + "and " + "they kept marching onward " * 20) + "endlessly." + chunks = split_into_chunks(sentence, max_words=10) + self.assertGreater(len(chunks), 1) + joined = " ".join(chunks) + self.assertIn("1,000,000", joined) + self.assertIn("12:30", joined) + self.assertNotIn("1, 000", joined) + self.assertNotIn("12: 30", joined) if __name__ == "__main__": diff --git a/tests/test_tts.py b/tests/test_tts.py index 47a8309..26f663e 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -241,13 +241,25 @@ class FasterTTSClientGenerateTests(unittest.TestCase): sentences = [" ".join(f"word{i}" for i in range(6)) + "." for _ in range(3)] text = " ".join(sentences) pcm_parts = [b"\x01\x00" * 10, b"\x02\x00" * 20, b"\x03\x00" * 30] - with patch.object(config, "CHUNK_SIZE_WORDS", 10), \ + with patch.object(config, "MAX_REQUEST_WORDS", 10), \ patch.object(client, "_request_pcm", side_effect=pcm_parts) as mock_pcm: result = client.generate_chunk(text, 1) self.assertEqual(mock_pcm.call_count, 3) _, _, _, frames = self._read_wav(Path(result)) self.assertEqual(frames, b"".join(pcm_parts)) + def test_subchunk_size_is_clamped_to_request_ceiling(self): + client = self._make_client() + text = " ".join(f"word{i}" for i in range(8)) + pcm = b"\x01\x00" * 10 + with patch.object(config, "CHUNK_SIZE_WORDS", 4), \ + patch.object(client, "_request_pcm", return_value=pcm) as mock_pcm: + result = client.generate_chunk(text, 1) + # CHUNK_SIZE_WORDS no longer drives request size: the hard ceiling + # does, so the whole (8-word) text is one request here. + self.assertEqual(mock_pcm.call_count, 1) + self.assertIsNotNone(result) + def test_stale_chunk_files_are_removed(self): stale = Path(self._tmp.name) / "chunk_0001.mp3" stale.write_bytes(b"old") @@ -316,6 +328,147 @@ class FasterTTSClientGenerateTests(unittest.TestCase): self.assertEqual(payload["response_format"], "pcm") +class TruncationDetectionTests(unittest.TestCase): + """check_for_truncation: servers cut audio silently past their caps, so + grossly short audio must fail the request (then retry / fail visibly).""" + + def test_duration_below_ratio_raises(self): + with self.assertRaises(RuntimeError) as ctx: + tts.check_for_truncation(" ".join(["w"] * 150), 25.0, "Chunk 1") + self.assertIn("truncated", str(ctx.exception)) + + def test_duration_at_ratio_passes(self): + tts.check_for_truncation(" ".join(["w"] * 150), 30.0, "Chunk 1") + + def test_unknown_duration_skips_check(self): + tts.check_for_truncation(" ".join(["w"] * 150), None, "Chunk 1") + + def test_short_requests_skip_check(self): + tts.check_for_truncation(" ".join(["w"] * 9), 0.0, "Chunk 1") + + def test_zero_duration_fails_checked_requests(self): + with self.assertRaises(RuntimeError): + tts.check_for_truncation(" ".join(["w"] * 150), 0.0, "Chunk 1") + + +class FasterTTSClientTruncationTests(unittest.TestCase): + """Audio far shorter than its text implies fails the faster request.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) + self._chunks.start() + + def tearDown(self): + self._chunks.stop() + self._tmp.cleanup() + + def _make_client(self): + client = FasterTTSClient.__new__(FasterTTSClient) + client.voice = "default" + client.api_url = "http://127.0.0.1:8000" + return client + + def test_truncated_pcm_fails_the_chunk(self): + client = self._make_client() + text = " ".join(f"word{i}" for i in range(12)) + with patch.object(client, "_request_pcm", return_value=b"\x01\x00" * 24), \ + self.assertLogs("converter.tts", level="ERROR") as logs: + result = client.generate_chunk(text, 1) + self.assertIsNone(result) + self.assertTrue(any("truncated" in line for line in logs.output)) + + def test_full_length_pcm_passes(self): + client = self._make_client() + text = " ".join(f"word{i}" for i in range(12)) + # 12 words -> expected 4.8s, half is 2.4s -> 2.5s of audio passes. + pcm = b"\x01\x00" * int(2.5 * tts.SAMPLE_RATE) + with patch.object(client, "_request_pcm", return_value=pcm): + result = client.generate_chunk(text, 1) + self.assertIsNotNone(result) + + +class QwenTTSClientGenerateTests(unittest.TestCase): + """Qwen chunk generation: sub-request splitting and concatenation.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) + self._chunks.start() + + def tearDown(self): + self._chunks.stop() + self._tmp.cleanup() + + def _make_client(self): + client = QwenTTSClient.__new__(QwenTTSClient) + client.voice_mode = tts.VOICE_MODE_CUSTOM + return client + + @staticmethod + def _write_wav(path: Path, frames: bytes) -> Path: + with wave.open(str(path), "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(tts.SAMPLE_RATE) + wav_file.writeframes(frames) + return path + + def _read_wav_frames(self, path: Path) -> bytes: + with wave.open(str(path), "rb") as wav_file: + return wav_file.readframes(wav_file.getnframes()) + + def test_single_request_copies_audio(self): + client = self._make_client() + source = self._write_wav(Path(self._tmp.name) / "server.wav", b"\x01\x00" * 50) + with patch.object(client, "_generate_custom_voice", + return_value=(str(source),)) as mock_generate: + result = client.generate_chunk("Hello world.", 1) + mock_generate.assert_called_once_with("Hello world.") + path = Path(result) + self.assertEqual(path.name, "chunk_0001.wav") + self.assertEqual(self._read_wav_frames(path), b"\x01\x00" * 50) + + def test_oversized_input_is_split_and_concatenated_in_order(self): + client = self._make_client() + first = self._write_wav(Path(self._tmp.name) / "one.wav", b"\x01\x00" * 10) + second = self._write_wav(Path(self._tmp.name) / "two.wav", b"\x02\x00" * 20) + text = " ".join(f"word{i}" for i in range(12)) + with patch.object(config, "MAX_REQUEST_WORDS", 5), \ + patch.object(client, "_generate_custom_voice", + side_effect=[(str(first),), (str(second),), + (str(first),)]) as mock_generate: + result = client.generate_chunk(text, 1) + self.assertEqual(mock_generate.call_count, 3) + path = Path(result) + self.assertEqual(path.name, "chunk_0001.wav") + self.assertEqual(self._read_wav_frames(path), + b"\x01\x00" * 10 + b"\x02\x00" * 20 + b"\x01\x00" * 10) + for call in mock_generate.call_args_list: + self.assertLessEqual(len(call[0][0].split()), 5) + + def test_truncated_sub_request_fails_the_chunk(self): + client = self._make_client() + # 0.1s of audio for 12 words (expected >= 2.4s). + source = self._write_wav(Path(self._tmp.name) / "short.wav", + b"\x01\x00" * int(0.1 * tts.SAMPLE_RATE)) + text = " ".join(f"word{i}" for i in range(12)) + with patch.object(client, "_generate_custom_voice", + return_value=(str(source),)) as mock_generate, \ + self.assertLogs("converter.tts", level="ERROR") as logs: + result = client.generate_chunk(text, 1) + self.assertIsNone(result) + self.assertEqual(mock_generate.call_count, 1) + self.assertTrue(any("truncated" in line for line in logs.output)) + + def test_empty_text_fails_the_chunk(self): + client = self._make_client() + with patch.object(client, "_generate_custom_voice") as mock_generate: + result = client.generate_chunk(" ", 1) + self.assertIsNone(result) + mock_generate.assert_not_called() + + class FasterModeWiringTests(unittest.TestCase): """AudiobookConverter wiring for the --faster backend.""" |