diff options
| -rw-r--r-- | converter/tts.py | 62 | ||||
| -rw-r--r-- | tests/test_tts.py | 62 |
2 files changed, 6 insertions, 118 deletions
diff --git a/converter/tts.py b/converter/tts.py index 0d867a2..0803bb9 100644 --- a/converter/tts.py +++ b/converter/tts.py @@ -29,7 +29,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 .audio import concat_audio_files from .chunking import split_into_chunks logger = logging.getLogger(__name__) @@ -267,56 +267,9 @@ def whisper_backend_available() -> Optional[str]: return None -# 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. +# 150 wpm is a typical spoken pace; used only to size the HTTP request +# timeout for long audio.cpp generations (not as a correctness check). _ESTIMATED_WORDS_PER_MINUTE = 150 -_MIN_AUDIO_DURATION_RATIO = 0.5 -_MIN_WORDS_FOR_DURATION_CHECK = 10 - - -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 - ``_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 < _MIN_WORDS_FOR_DURATION_CHECK: - return - expected_seconds = 60.0 * words / _ESTIMATED_WORDS_PER_MINUTE - if actual_seconds < expected_seconds * _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 * _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: @@ -596,9 +549,6 @@ class QwenTTSClient(_BaseTTSClient): 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 # ------------------------------------------------------------------ @@ -776,9 +726,6 @@ class FasterTTSClient(_BaseTTSClient): for sub_num, sub_text in enumerate(sub_chunks, 1): 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") @@ -1185,9 +1132,6 @@ class AudioCppTTSClient(_BaseTTSClient): sub_text, chunk_num, sub_num, len(sub_texts)) destination = Path(parts_dir) / f"part_{sub_num:02d}.wav" destination.write_bytes(wav) - check_for_truncation( - sub_text, _audio_duration_seconds(destination), - f"Chunk {chunk_num} sub-request {sub_num}/{len(sub_texts)}") part_paths.append(destination) if len(part_paths) == 1: output_path = self._chunk_path(chunk_num, ".wav") diff --git a/tests/test_tts.py b/tests/test_tts.py index 0b6da02..a0828de 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -385,31 +385,8 @@ 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.""" +class FasterTTSClientGenerateTests(unittest.TestCase): + """Faster chunk generation: full-length audio produces a chunk file.""" def setUp(self): self._tmp = tempfile.TemporaryDirectory() @@ -426,15 +403,6 @@ class FasterTTSClientTruncationTests(unittest.TestCase): 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)) @@ -504,20 +472,6 @@ class QwenTTSClientGenerateTests(unittest.TestCase): 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: @@ -861,7 +815,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_whole_text_sent_as_one_request_without_client_chunking(self): client = self._make_client(chunk_text=False) # 9 words with CHUNK_SIZE=5 would split in two if client chunking - # were on; kept under the 10-word truncation-check threshold. + # were on. text = " ".join(f"word{i}" for i in range(9)) with patch.object(config, "CHUNK_SIZE", 5), \ patch.object(client, "_request_wav", @@ -1116,16 +1070,6 @@ class AudioCppTTSClientTruncationTests(unittest.TestCase): wav_file.writeframes(frames) return buffer.getvalue() - def test_truncated_wav_fails_the_chunk(self): - client = self._make_client() - text = " ".join(f"word{i}" for i in range(12)) - wav = self._wav_bytes(b"\x01\x00" * 24) # 0.001s for ~4.8s of speech - with patch.object(client, "_request_wav", return_value=wav), \ - 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_wav_passes(self): client = self._make_client() text = " ".join(f"word{i}" for i in range(12)) |