diff options
| -rw-r--r-- | README.md | 16 | ||||
| -rw-r--r-- | converter/audio.py | 14 | ||||
| -rw-r--r-- | converter/converter.py | 40 | ||||
| -rw-r--r-- | converter/tts.py | 16 | ||||
| -rw-r--r-- | tests/test_audio.py | 52 | ||||
| -rw-r--r-- | tests/test_converter.py | 70 | ||||
| -rw-r--r-- | tests/test_make_audiocpp_server_json.py | 129 | ||||
| -rw-r--r-- | tests/test_tts.py | 65 | ||||
| -rwxr-xr-x | tools/make_audiocpp_server_json.py | 88 |
9 files changed, 372 insertions, 118 deletions
@@ -19,10 +19,11 @@ The converter sends text extracted from your books to a locally running Qwen3-TT - Python 3.12 - ffmpeg -- Enough VRAM to run the 1.7B model (~6GB) ## Installation +Create a python 3.12 environment, clone the repo, and install the requirements. + ```bash conda create -n qwen3-tts python=3.12 -y conda activate qwen3-tts @@ -33,14 +34,13 @@ pip install -r requirements.txt Put your book files (epub, etc.) in the `input/` directory. The output goes to `output/`. -You will also need to install one of the following backends (see below for installation/usage) +You need to install one of the following backends (see below for installation/usage) -| Backend | Description | -| -------------------------------------------------------------------- | ------------------------------------------------- | -| [Qwen-TTS](https://pypi.org/project/qwen-tts/) | Gradio server released by Qwen | -| [Faster-Qwen-TTS](https://github.com/andimarafioti/faster-qwen3-tts) | Server with 2-8x faster inference for NVidia GPUs | -| [audio.cpp](https://github.com/0xShug0/audio.cpp) (Qwen) | Newer C++ TTS backend that supports Qwen-TTS | -| [audio.cpp](https://github.com/0xShug0/audio.cpp) (other families) | Same backend hosting larger/higher-quality models | +| Backend | Description | +| -------------------------------------------------------------------- | ------------------------------------------------------ | +| [Qwen-TTS](https://pypi.org/project/qwen-tts/) | Gradio server released by Qwen | +| [Faster-Qwen-TTS](https://github.com/andimarafioti/faster-qwen3-tts) | Qwen server with 2-8x faster inference for NVidia GPUs | +| [audio.cpp](https://github.com/0xShug0/audio.cpp) | Newer C++ TTS backend that supports many recent models | ## Options diff --git a/converter/audio.py b/converter/audio.py index 01f78ab..d1dc28b 100644 --- a/converter/audio.py +++ b/converter/audio.py @@ -425,11 +425,19 @@ def combine_chunks(total_chunks: int, output_path: Path, if intermediate: logger.info("Chapter audio saved (intermediate): %s (%d/%d chunks)", output_path, len(chunk_files), total_chunks) - print(f"[INFO] Saved chapter audio (intermediate): {output_path.name} " - f"({len(chunk_files)}/{total_chunks} chunks)") + if len(chunk_files) == 1 and total_chunks == 1: + print(f"[INFO] Saved chapter audio (intermediate): " + f"{output_path.name}") + else: + print(f"[INFO] Saved chapter audio (intermediate): {output_path.name} " + f"({len(chunk_files)}/{total_chunks} chunks)") else: logger.info("Audiobook saved: %s (%d/%d chunks)", output_path, len(chunk_files), total_chunks) - print(f"[INFO] Saved audiobook: {output_path.name} ({len(chunk_files)}/{total_chunks} chunks)") + if len(chunk_files) == 1 and total_chunks == 1: + print(f"[INFO] Saved audiobook: {output_path.name}") + else: + print(f"[INFO] Saved audiobook: {output_path.name} " + f"({len(chunk_files)}/{total_chunks} chunks)") if speed_path is not None: logger.info("Saved speed-adjusted audiobook (%gx): %s", speed, speed_path) diff --git a/converter/converter.py b/converter/converter.py index de302ad..0ae1506 100644 --- a/converter/converter.py +++ b/converter/converter.py @@ -410,9 +410,10 @@ class AudiobookConverter: also dumped there, and every request/response is logged. """ total_chunks = len(chunks) - print(f"\n{'=' * 50}") - print(f"PROCESSING {total_chunks} CHUNKS") - print(f"{'=' * 50}") + if self.client_chunks: + print(f"\n{'=' * 50}") + print(f"PROCESSING {total_chunks} CHUNKS") + print(f"{'=' * 50}") results: Dict[int, Optional[Path]] = {} for chunk_num, chunk_text in enumerate(chunks, 1): @@ -433,7 +434,8 @@ class AudiobookConverter: destination = f" -> {copied.name}" if copied else "" logger.debug("Chunk %d/%d response in %.1fs%s", chunk_num, total_chunks, elapsed, destination) - print(f"[OK] Chunk {chunk_num:3d}/{total_chunks} completed") + if self.client_chunks: + print(f"[OK] Chunk {chunk_num:3d}/{total_chunks} completed") logger.info("+ Chunk %d/%d completed", chunk_num, total_chunks) else: logger.error("Chunk %d/%d failed", chunk_num, total_chunks) @@ -443,10 +445,11 @@ class AudiobookConverter: logger.error("Chunk %d/%d error: %s", chunk_num, total_chunks, exc) successful_chunks = sum(1 for path in results.values() if path) - print(f"\n{'=' * 50}") - print("CHUNK PROCESSING COMPLETE") - print(f"Successful: {successful_chunks}/{total_chunks}") - print(f"{'=' * 50}") + if self.client_chunks: + print(f"\n{'=' * 50}") + print("CHUNK PROCESSING COMPLETE") + print(f"Successful: {successful_chunks}/{total_chunks}") + print(f"{'=' * 50}") logger.info("Chunk processing completed: %d/%d chunks", successful_chunks, total_chunks) return results @@ -508,7 +511,19 @@ class AudiobookConverter: BACKEND_AUDIOCPP: "audio.cpp server", } backend = backend_labels.get(self.backend, "Qwen API") - print(f"[INFO] Processing {total_chunks} chunks via {backend}...") + if self.client_chunks: + print(f"[INFO] Processing {total_chunks} chunks via {backend}...") + else: + # The whole request is sent at once and the server does its + # own long-form chunking, so the chunk vocabulary does not + # apply; warn that this one request can take a very long time. + subject = (f"chapter {chapter[0]}/{chapter[1]}" + if chapter is not None else "text") + print(f"[INFO] Sending the {subject} to the {backend} as a " + "single request...") + print("[NOTE] It is expected for this to take a very long " + "time: the server synthesizes the entire request before " + "returning any audio.") results = self._synthesize_chunks(chunks, debug_dir=debug_dir) successful_chunks = sum(1 for path in results.values() if path) @@ -534,8 +549,11 @@ class AudiobookConverter: logger.info("Chapter %d/%d converted in %dm %ds (%d/%d chunks)", chapter[0], chapter[1], minutes, seconds, successful_chunks, total_chunks) - print(f"[INFO] Chapter {chapter[0]}/{chapter[1]} converted " - f"({successful_chunks}/{total_chunks} chunks)") + if self.client_chunks: + print(f"[INFO] Chapter {chapter[0]}/{chapter[1]} converted " + f"({successful_chunks}/{total_chunks} chunks)") + else: + print(f"[INFO] Chapter {chapter[0]}/{chapter[1]} converted") else: logger.info("Conversion completed in %dm %ds: %s", minutes, seconds, output_path) else: diff --git a/converter/tts.py b/converter/tts.py index 1a41643..0d867a2 100644 --- a/converter/tts.py +++ b/converter/tts.py @@ -363,15 +363,21 @@ class _BaseTTSClient: return None @contextlib.contextmanager - def _chunk_heartbeat(self, chunk_num: int): - """Print a periodic "still working" message while a chunk generates.""" + def _chunk_heartbeat(self, chunk_num: int, label: Optional[str] = None): + """Print a periodic "still working" message while a request generates. + + ``label`` overrides the default "Chunk {chunk_num}" subject, for + backends that send one request per chapter without client-side + chunking (the audio.cpp default) where "chunk" would be misleading. + """ stop = threading.Event() + subject = label if label is not None else f"Chunk {chunk_num}" def _beat(): start = time.time() while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS): elapsed = time.time() - start - print(f"[...] Chunk {chunk_num} still generating — " + print(f"[...] {subject} still generating — " f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed", flush=True) thread = threading.Thread(target=_beat, daemon=True) @@ -1170,7 +1176,9 @@ class AudioCppTTSClient(_BaseTTSClient): output_path: Optional[Path] = None with tempfile.TemporaryDirectory(prefix="tts_parts_") as parts_dir, \ - self._chunk_heartbeat(chunk_num): + self._chunk_heartbeat( + chunk_num, + label=None if self.chunk_text else "Request"): part_paths = [] for sub_num, sub_text in enumerate(sub_texts, 1): wav = self._request_wav_with_retry( diff --git a/tests/test_audio.py b/tests/test_audio.py index c127311..ef5e92a 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -1,11 +1,13 @@ """Tests for audio helpers: speed parameters, chunk cleanup, encoding, command construction, duration verification, and audio concatenation.""" +import io import tempfile import unittest import wave +from contextlib import redirect_stdout from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch from converter import audio from converter import config @@ -470,5 +472,53 @@ class ConcatAudioFilesTests(unittest.TestCase): self.assertIn("ffmpeg", str(ctx.exception)) +class CombineChunksPrintTests(unittest.TestCase): + """Single-request runs (audiocpp whole-chapter) omit the chunks suffix.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self._chunks = patch.object(audio, "CHUNKS_FOLDER", Path(self._tmp.name)) + self._chunks.start() + self.addCleanup(self._chunks.stop) + + def _combine(self, total_chunks, chunk_results, intermediate=False): + buf = io.StringIO() + with patch.object(audio.shutil, "which", return_value="/usr/bin/ffmpeg"), \ + patch.object(audio, "atempo_filters", return_value=False), \ + patch.object(audio, "build_concat_command", + return_value=["ffmpeg"]), \ + patch.object(audio.subprocess, "run", + return_value=MagicMock(returncode=0)), \ + patch.object(audio, "probe_duration_ms", return_value=1000), \ + patch.object(audio, "verify_output_duration", + return_value=True), \ + redirect_stdout(buf): + ok = audio.combine_chunks( + total_chunks, Path("out.m4b"), chunk_results, + output_format="m4b", intermediate=intermediate) + self.assertTrue(ok) + return buf.getvalue() + + def test_single_chunk_omits_chunks_suffix(self): + chunk = Path(self._tmp.name) / "chunk_0001.wav" + chunk.write_bytes(b"x") + out = self._combine(1, {1: chunk}) + self.assertEqual(out.strip(), "[INFO] Saved audiobook: out.m4b") + + def test_multi_chunk_keeps_chunks_suffix(self): + chunk = Path(self._tmp.name) / "chunk_0001.wav" + chunk.write_bytes(b"x") + out = self._combine(1, {1: chunk}, intermediate=True) + self.assertEqual(out.strip(), + "[INFO] Saved chapter audio (intermediate): out.m4b") + + def test_partial_chunk_run_keeps_chunks_suffix(self): + chunk = Path(self._tmp.name) / "chunk_0001.wav" + chunk.write_bytes(b"x") + out = self._combine(2, {1: chunk, 2: chunk}) + self.assertEqual(out.strip(), + "[INFO] Saved audiobook: out.m4b (2/2 chunks)") + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_converter.py b/tests/test_converter.py index 8402b30..d525c18 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -1,8 +1,11 @@ """Tests for the audiobook converter orchestration helpers.""" +import io import logging import tempfile +import time import unittest +from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch @@ -181,6 +184,7 @@ class DebugDumpTests(unittest.TestCase): self._debug_folder.start() self.debug_root = Path(self._tmp.name) self.converter = AudiobookConverter.__new__(AudiobookConverter) + self.converter.client_chunks = True self.converter.tts = MagicMock() def tearDown(self): @@ -323,6 +327,7 @@ class SynthesizeChunkLoggingTests(unittest.TestCase): def setUp(self): self.converter = AudiobookConverter.__new__(AudiobookConverter) + self.converter.client_chunks = True self.converter.tts = MagicMock() def test_failed_chunk_logs_single_error(self): @@ -342,6 +347,71 @@ class SynthesizeChunkLoggingTests(unittest.TestCase): self.assertIn("Chunk 1/1 error: boom", logs.output[0]) +class ServerSideChunkingOutputTests(unittest.TestCase): + """With client-side chunking off (audiocpp default), the console skips + the chunk vocabulary because the whole request is one server call.""" + + def _converter(self, client_chunks: bool): + converter = AudiobookConverter.__new__(AudiobookConverter) + converter.client_chunks = client_chunks + converter.backend = tts.BACKEND_AUDIOCPP + converter.speed = 1.0 + converter.output_format = "mp3" + converter.tts = MagicMock() + converter.tts.process_chunk_with_retry.return_value = "chunk.wav" + return converter + + def test_client_chunking_prints_chunk_progress(self): + buf = io.StringIO() + with redirect_stdout(buf): + self._converter(client_chunks=True)._synthesize_chunks(["Hello."]) + out = buf.getvalue() + self.assertIn("PROCESSING 1 CHUNKS", out) + self.assertIn("Chunk 1/1 completed", out) + self.assertIn("Successful: 1/1", out) + + def test_server_side_chunking_suppresses_chunk_output(self): + buf = io.StringIO() + with redirect_stdout(buf): + self._converter(client_chunks=False)._synthesize_chunks(["Hello."]) + self.assertEqual(buf.getvalue(), "") + + def test_server_side_chunking_suppresses_chapter_chunk_suffix(self): + buf = io.StringIO() + with patch.object(converter_mod.audio, "combine_chunks", return_value=True), \ + redirect_stdout(buf): + ok = self._converter(client_chunks=False)._convert_text( + "Hello world.", Path("out.mp3"), time.time(), chapter=(2, 5)) + self.assertTrue(ok) + out = buf.getvalue() + self.assertIn("Chapter 2/5 converted", out) + self.assertNotIn("chunk", out.lower()) + + def test_single_request_run_notes_long_wait(self): + buf = io.StringIO() + with patch.object(converter_mod.audio, "combine_chunks", return_value=True), \ + redirect_stdout(buf): + ok = self._converter(client_chunks=False)._convert_text( + "Hello world.", Path("out.mp3"), time.time(), chapter=(2, 5)) + self.assertTrue(ok) + out = buf.getvalue() + self.assertIn("Sending the chapter 2/5 to the audio.cpp server as a " + "single request", out) + self.assertIn("expected for this to take a very long time", out) + + def test_client_chunking_run_keeps_chunk_phrasing(self): + buf = io.StringIO() + with patch.object(converter_mod.audio, "combine_chunks", return_value=True), \ + redirect_stdout(buf): + ok = self._converter(client_chunks=True)._convert_text( + "Hello world.", Path("out.mp3"), time.time(), chapter=(2, 5)) + self.assertTrue(ok) + out = buf.getvalue() + self.assertIn("Processing 1 chunks via audio.cpp server", out) + self.assertNotIn("single request", out) + self.assertIn("Chapter 2/5 converted (1/1 chunks)", out) + + class PromptOverwriteTests(unittest.TestCase): def test_single_file_yes(self): with patch("builtins.input", return_value="y"): diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py index f9fa794..2bd262a 100644 --- a/tests/test_make_audiocpp_server_json.py +++ b/tests/test_make_audiocpp_server_json.py @@ -304,20 +304,6 @@ class PromptHelperTests(unittest.TestCase): def tearDown(self): self._tmp.cleanup() - def test_ask_wav_dir_reprompts_until_valid(self): - with patch("builtins.input", - side_effect=[str(self.folder / "nope"), - str(self.folder)]): - self.assertEqual(make_server.ask_wav_dir(), self.folder) - - def test_ask_wav_dir_empty_skips(self): - with patch("builtins.input", return_value=""): - self.assertIsNone(make_server.ask_wav_dir()) - - def test_ask_wav_dir_eof_returns_none(self): - with patch("builtins.input", side_effect=EOFError): - self.assertIsNone(make_server.ask_wav_dir()) - def test_ask_port_reprompts_until_valid(self): with patch("builtins.input", side_effect=["abc", "8081"]): self.assertEqual(make_server.ask_port(8080), 8081) @@ -341,6 +327,46 @@ class PromptHelperTests(unittest.TestCase): "one") +class ResolveWavDirArgTests(unittest.TestCase): + """Path normalization for the required WAV_DIR argument.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_resolves_to_absolute(self): + self.assertEqual(make_server.resolve_wav_dir_arg(str(self.folder)), + self.folder.resolve()) + + def test_strips_surrounding_quotes(self): + quoted = f'"{self.folder}"' + self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.folder.resolve()) + + def test_strips_single_quotes(self): + quoted = f"'{self.folder}'" + self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.folder.resolve()) + + def test_strips_whitespace(self): + self.assertEqual(make_server.resolve_wav_dir_arg(f" {self.folder} "), + self.folder.resolve()) + + def test_expands_tilde(self): + with patch.object(make_server.os.path, "expanduser", + return_value=str(self.folder)) as mock_expand: + result = make_server.resolve_wav_dir_arg("~/voices") + mock_expand.assert_called_once_with("~/voices") + self.assertEqual(result, self.folder.resolve()) + + def test_trailing_slash_preserved_as_dir(self): + self.assertEqual(make_server.resolve_wav_dir_arg(f"{self.folder}/"), + self.folder.resolve()) + + class MainTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() @@ -370,14 +396,21 @@ class MainTests(unittest.TestCase): return make_server.main() def _defaults(self, models="", host="", port="", backend="", - lazy="", custom_path="", clone_path="", wav_dir="", + lazy="", custom_path="", clone_path="", confirm="y", prefix=()): - # First input selects the model family (default: Qwen3-TTS). + # First input selects the model family (default: Qwen3-TTS). The + # wav directory is always a positional argument, never prompted. return list(prefix) + ["", models, host, port, backend, lazy, - custom_path, clone_path, wav_dir, confirm] + custom_path, clone_path, confirm] + + def test_required_wav_dir_missing_prints_usage(self): + with self.assertRaises(SystemExit) as ctx: + self._run(["--output", str(self.output)], inputs=[]) + self.assertEqual(ctx.exception.code, 2) + self.assertFalse(self.output.exists()) def test_default_run_hosts_both_models(self): - exit_code = self._run(["--output", str(self.output)], + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=self._defaults()) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -395,7 +428,7 @@ class MainTests(unittest.TestCase): self.assertNotIn("voice_presets", data["models"][1]) def test_eof_uses_all_defaults(self): - exit_code = self._run(["--output", str(self.output)]) + exit_code = self._run([str(self.folder), "--output", str(self.output)]) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(data["host"], "127.0.0.1") @@ -427,7 +460,7 @@ class MainTests(unittest.TestCase): def test_custom_only_single_model(self): inputs = ["", "", "", "", "", "", "y"] exit_code = self._run( - ["--output", str(self.output), "--models", "custom"], + [str(self.folder), "--output", str(self.output), "--models", "custom"], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -437,8 +470,8 @@ class MainTests(unittest.TestCase): def test_duplicate_ids_prompt_for_distinct_clone_id(self): with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): - inputs = ["", "1", "qwen-clone-2", "", "", "", "", "", "", "", "y"] - exit_code = self._run(["--output", str(self.output)], + inputs = ["", "1", "qwen-clone-2", "", "", "", "", "", "", "y"] + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -449,7 +482,7 @@ class MainTests(unittest.TestCase): with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): with self.assertRaises(SystemExit) as ctx: - self._run(["--output", str(self.output)]) + self._run([str(self.folder), "--output", str(self.output)]) self.assertNotEqual(ctx.exception.code, 0) self.assertFalse(self.output.exists()) @@ -457,7 +490,7 @@ class MainTests(unittest.TestCase): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): inputs = ["", "", "", "y", "", "", "", "", "", "y"] - exit_code = self._run(["--output", str(self.output), + exit_code = self._run([str(self.folder), "--output", str(self.output), "--port", "8080"], inputs=inputs) self.assertEqual(exit_code, 0) @@ -470,7 +503,7 @@ class MainTests(unittest.TestCase): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): inputs = ["", "", "", "n", "", "", "", "", "", "y"] - exit_code = self._run(["--output", str(self.output), + exit_code = self._run([str(self.folder), "--output", str(self.output), "--port", "8080"], inputs=inputs) self.assertEqual(exit_code, 0) @@ -481,7 +514,7 @@ class MainTests(unittest.TestCase): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:8080"): inputs = self._defaults() - exit_code = self._run(["--output", str(self.output)], + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=inputs) self.assertEqual(exit_code, 0) self.assertEqual(self.fake_config.read_text(encoding="utf-8"), @@ -489,8 +522,8 @@ class MainTests(unittest.TestCase): def test_invalid_menu_choice_reprompts(self): # Family menu default, then an invalid models-menu choice retried. - inputs = ["", "9", "", "", "", "", "", "", "", "", "y"] - exit_code = self._run(["--output", str(self.output)], + inputs = ["", "9", "", "", "", "", "", "", "", "y"] + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -498,14 +531,14 @@ class MainTests(unittest.TestCase): def test_confirm_declined_writes_nothing(self): inputs = self._defaults(confirm="n") - exit_code = self._run(["--output", str(self.output)], + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=inputs) self.assertEqual(exit_code, 1) self.assertFalse(self.output.exists()) def test_existing_output_declined_keeps_file(self): self.output.write_text('{"old": true}', encoding="utf-8") - exit_code = self._run(["--output", str(self.output)], + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=["n"]) self.assertEqual(exit_code, 1) self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), @@ -514,7 +547,7 @@ class MainTests(unittest.TestCase): def test_existing_output_accepted_overwrites(self): self.output.write_text('{"old": true}', encoding="utf-8") inputs = ["y"] + self._defaults() - exit_code = self._run(["--output", str(self.output)], + exit_code = self._run([str(self.folder), "--output", str(self.output)], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -523,7 +556,8 @@ class MainTests(unittest.TestCase): def test_force_overwrites_without_prompt(self): self.output.write_text('{"old": true}', encoding="utf-8") inputs = self._defaults() - exit_code = self._run(["--output", str(self.output), "--force"], + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--force"], inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) @@ -531,13 +565,13 @@ class MainTests(unittest.TestCase): def test_flags_skip_prompts(self): # Family still asked (no --family flag); port 9000 differs from the - # config port so its sync prompt fires; custom/clone paths and the - # wav dir use their defaults. + # config port so its sync prompt fires; custom/clone paths use + # their defaults. exit_code = self._run( - ["--output", str(self.output), "--models", "both", + [str(self.folder), "--output", str(self.output), "--models", "both", "--host", "0.0.0.0", "--port", "9000", "--backend", "cpu", "--lazy-load"], - inputs=["", "y", "", "", "", "y"]) + inputs=["", "y", "", "", "y"]) self.assertEqual(exit_code, 0) self.assertIn('"http://127.0.0.1:9000"', self.fake_config.read_text(encoding="utf-8")) @@ -548,11 +582,15 @@ class MainTests(unittest.TestCase): self.assertTrue(data["lazy_load"]) def test_missing_positional_wav_dir_errors(self): - with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder / "nope"), - "--output", str(self.output)], + missing = self.folder / "nope" + with self.assertRaises(SystemExit) as ctx, \ + patch("sys.stderr") as mock_stderr: + self._run([str(missing), "--output", str(self.output)], inputs=self._defaults()) self.assertEqual(ctx.exception.code, 2) + shown = "".join(call[0][0] for call in mock_stderr.write.call_args_list) + self.assertIn(f"WAV directory not found: {missing.resolve()}", shown) + self.assertIn("directory containing the .wav", shown) class NonQwenFamilyMainTests(unittest.TestCase): @@ -613,10 +651,10 @@ class NonQwenFamilyMainTests(unittest.TestCase): self.fake_config.read_text(encoding="utf-8")) def test_model_id_sync_declined_keeps_config(self): - # sync declined, host, port, backend, lazy, wav dir skipped, confirm - inputs = ["n", "", "", "", "", "", "y"] + # sync declined, host, port, backend, lazy, confirm + inputs = ["n", "", "", "", "", "y"] exit_code = self._run( - ["--output", str(self.output), "--family", "voxcpm2", + [str(self.folder), "--output", str(self.output), "--family", "voxcpm2", "--model-id", "voxcpm2", "--model-path", "models/VoxCPM2-GGUF"], inputs=inputs) self.assertEqual(exit_code, 0) @@ -628,13 +666,14 @@ class NonQwenFamilyMainTests(unittest.TestCase): def test_no_voice_presets_warns(self): buf = io.StringIO() - # sync accepted, host, port, backend, lazy, wav dir skipped, confirm + # sync accepted, host, port, backend, lazy, confirm with patch.object(sys, "argv", ["make_audiocpp_server_json.py", + str(self.folder), "--output", str(self.output), "--family", "index_tts2", "--model-id", "indextts2", "--model-path", "models/IndexTTS2-GGUF"]), \ - patch("builtins.input", side_effect=["y", "", "", "", "", "", "y"]), \ + patch("builtins.input", side_effect=["y", "", "", "", "", "y"]), \ patch.object(make_server, "transcribe_reference_audio"), \ patch.object(make_server, "whisper_backend_available", return_value="faster_whisper"), \ @@ -649,7 +688,7 @@ class NonQwenFamilyMainTests(unittest.TestCase): def test_models_flag_rejected_for_non_qwen_family(self): with self.assertRaises(SystemExit) as ctx: - self._run(["--output", str(self.output), + self._run([str(self.folder), "--output", str(self.output), "--family", "higgs_audio_tts", "--models", "both"]) self.assertEqual(ctx.exception.code, 2) diff --git a/tests/test_tts.py b/tests/test_tts.py index e2fe921..0b6da02 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -3,8 +3,10 @@ import io import json import tempfile +import time import unittest import wave +from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch @@ -1016,6 +1018,69 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): self.assertEqual(remaining, ["chunk_0001.wav"]) +class AudioCppHeartbeatTests(unittest.TestCase): + """The heartbeat label drops 'Chunk' when the server does its own + long-form chunking (chunk_text=False, the default).""" + + 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() + + @staticmethod + def _client(chunk_text): + client = AudioCppTTSClient.__new__(AudioCppTTSClient) + client.api_url = "http://127.0.0.1:8080" + client.model_id = config.AUDIOCPP_MODEL_ID + client.preset_mode = False + client.voice = "Vivian" + client.language = "English" + client._seed = -1 + client.chunk_text = chunk_text + client.family = "qwen3_tts" + client.profile = tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE + return client + + @staticmethod + def _wav_bytes(): + buffer = io.BytesIO() + with wave.open(buffer, "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(tts.SAMPLE_RATE) + wav_file.writeframes(b"\x01\x00" * 10) + return buffer.getvalue() + + def _run(self, chunk_text): + client = self._client(chunk_text) + + def slow_request(*_args, **_kwargs): + time.sleep(0.12) + return self._wav_bytes() + + buf = io.StringIO() + with patch.object(config, "HEARTBEAT_INTERVAL_SECONDS", 0.03), \ + patch.object(client, "_request_wav_with_retry", + side_effect=slow_request), \ + redirect_stdout(buf): + result = client.generate_chunk("Hello.", 1) + self.assertTrue(result) + return buf.getvalue() + + def test_server_side_chunking_heartbeat_has_no_chunk_word(self): + out = self._run(chunk_text=False) + self.assertIn("Request still generating", out) + self.assertNotIn("Chunk", out) + + def test_client_side_chunking_heartbeat_keeps_chunk_word(self): + out = self._run(chunk_text=True) + self.assertIn("Chunk 1 still generating", out) + + class AudioCppTTSClientTruncationTests(unittest.TestCase): """Audio far shorter than its text implies fails the request.""" diff --git a/tools/make_audiocpp_server_json.py b/tools/make_audiocpp_server_json.py index 1273e43..d24f895 100755 --- a/tools/make_audiocpp_server_json.py +++ b/tools/make_audiocpp_server_json.py @@ -14,28 +14,30 @@ v3 TTS 4B, VoxCPM2-2B, and IndexTTS-2 / 2.5 (see the "Option 4" section of the README). The converter works with other audio.cpp TTS families too; host them by writing server.json by hand. -Reference .wav files for voice cloning (a directory argument or an -interactive prompt) are transcribed with a local Whisper backend -(faster_whisper or whisper) and added as voice_presets on the cloning -model entry. +Reference .wav files for voice cloning (the required WAV_DIR argument) +are transcribed with a local Whisper backend (faster_whisper or whisper) +and added as voice_presets on the cloning model entry. Every value can also be supplied as a command-line flag; anything missing is asked interactively with the default shown in brackets. Pressing Enter -accepts the default, so running the tool with no arguments and pressing -Enter through produces a server.json hosting both Qwen3-TTS models on -127.0.0.1:8080 with the cuda backend. +accepts the default. Usage: - python tools/make_audiocpp_server_json.py [WAV_DIR] [--output PATH] + python tools/make_audiocpp_server_json.py WAV_DIR [--output PATH] [--family {qwen3_tts,higgs_audio_tts,voxcpm2,index_tts2,index_tts2_5}] [--model-id ID] [--model-path PATH] [--host HOST] [--port PORT] [--models {both,custom,clone}] [--backend {cuda,vulkan,hip,cpu}] [--lazy-load] [--whisper-model NAME] [--force] + +WAV_DIR is required: a directory of .wav reference files used as voice +cloning presets. It is checked up front and reported with its resolved +absolute path if it does not exist. """ import argparse import json +import os import re import sys import urllib.parse @@ -108,6 +110,19 @@ FAMILY_KEYS = tuple(entry["key"] for entry in FAMILY_ENTRIES) FAMILY_BY_KEY = {entry["key"]: entry for entry in FAMILY_ENTRIES} +def resolve_wav_dir_arg(value: str) -> Path: + """Normalize a user-supplied wav directory argument. + + Strips surrounding quotes (a common copy-paste artifact), expands a + leading ``~``, and resolves the result to an absolute path so relative + paths are always validated against the current working directory. + """ + cleaned = value.strip() + if len(cleaned) >= 2 and cleaned[0] == cleaned[-1] and cleaned[0] in "\"'": + cleaned = cleaned[1:-1] + return Path(os.path.expanduser(cleaned)).resolve() + + def find_wav_files(input_dir: Path) -> list: """Return the .wav files in INPUT_DIR, sorted alphabetically by name.""" return sorted( @@ -241,23 +256,6 @@ def ask_distinct_clone_id(primary_id: str) -> str: f"equal to '{primary_id}'.") -def ask_wav_dir() -> Optional[Path]: - """Prompt for a directory of .wav clone references; Enter skips.""" - while True: - try: - answer = input("Directory with .wav files to clone " - "(Enter to skip): ").strip() - except EOFError: - return None - if not answer: - return None - path = Path(answer) - if path.is_dir(): - return path - print(f"[WARNING] {answer} is not a directory; try again " - "(or press Enter to skip).") - - def config_port() -> int: """Return the port of AUDIOCPP_API_URL in converter/config.py.""" try: @@ -452,27 +450,20 @@ def _ask_host_port_backend_lazy(args: argparse.Namespace def _collect_voice_presets(args: argparse.Namespace, include_clone: bool) -> Dict[str, dict]: - """Resolve the clone-reference wav directory and transcribe it. + """Transcribe the wav directory into the voice_presets mapping. - Returns the voice_presets mapping (empty when no wavs were given or - found). Cloning entries only: a run without any cloning model ignores - the wav directory entirely. + Returns the voice_presets mapping (empty when no wavs were found). + Cloning entries only: a run without any cloning model ignores the wav + directory entirely. """ - wav_dir: Optional[Path] = None - if args.input_dir is not None: - if include_clone: - wav_dir = args.input_dir - else: - print(f"[WARNING] Ignoring {args.input_dir}: no cloning model " - "selected, so voice presets are not used") - elif include_clone: - wav_dir = ask_wav_dir() - if wav_dir is None: + if not include_clone: + print(f"[WARNING] Ignoring {args.input_dir}: no cloning model " + "selected, so voice presets are not used") return {} - wav_files = find_wav_files(wav_dir) + wav_files = find_wav_files(args.input_dir) if not wav_files: - print(f"[WARNING] No .wav files found in {wav_dir}; writing the " + print(f"[WARNING] No .wav files found in {args.input_dir}; writing the " "config without voice presets") return {} if whisper_backend_available() is None: @@ -512,9 +503,9 @@ def main() -> int: parser = argparse.ArgumentParser( description="Generate a server.json for the audio.cpp audiocpp_server " "hosting a TTS model used by this converter.") - parser.add_argument("input_dir", type=Path, nargs="?", default=None, - help="Optional directory with .wav reference files " - "to add as voice cloning presets") + parser.add_argument("input_dir", type=resolve_wav_dir_arg, metavar="WAV_DIR", + help="Directory with .wav reference files to add as " + "voice cloning presets (required)") parser.add_argument("--output", type=Path, default=Path("server.json"), help="Output path for server.json (default: " "server.json in the current directory)") @@ -551,8 +542,13 @@ def main() -> int: help="Overwrite the output file without prompting") args = parser.parse_args() - if args.input_dir is not None and not args.input_dir.is_dir(): - parser.error(f"WAV directory not found: {args.input_dir}") + if not args.input_dir.is_dir(): + parser.error( + f"WAV directory not found: {args.input_dir}\n" + f" (resolved from the current working directory: " + f"{Path.cwd()})\n" + " WAV_DIR must be a directory containing the .wav " + "reference files to use as voice cloning presets") if args.output.exists() and not args.force \ and not prompt_overwrite(args.output): |