From 1f2142e7f610871a6bbe6498d0709d310fcbebb1 Mon Sep 17 00:00:00 2001 From: historia Date: Thu, 20 Aug 2026 17:05:21 -0400 Subject: feat: tool scripts for server.json and voices.json --- converter/chunking.py | 30 +-- converter/config.py | 9 +- converter/tts.py | 12 +- tests/test_chunking.py | 48 ++-- tests/test_make_audiocpp_server_json.py | 457 ++++++++++++++++++++++++++++++++ tests/test_make_faster_voices_json.py | 169 ++++++++++++ tests/test_make_voices.py | 169 ------------ tests/test_tts.py | 27 +- tools/make_audiocpp_server_json.py | 443 +++++++++++++++++++++++++++++++ tools/make_faster_voices_json.py | 121 +++++++++ tools/make_voices.py | 120 --------- 11 files changed, 1250 insertions(+), 355 deletions(-) create mode 100644 tests/test_make_audiocpp_server_json.py create mode 100644 tests/test_make_faster_voices_json.py delete mode 100644 tests/test_make_voices.py create mode 100755 tools/make_audiocpp_server_json.py create mode 100755 tools/make_faster_voices_json.py delete mode 100755 tools/make_voices.py diff --git a/converter/chunking.py b/converter/chunking.py index 1ce69a3..800b76a 100644 --- a/converter/chunking.py +++ b/converter/chunking.py @@ -2,28 +2,21 @@ import logging import re -from typing import List +from typing import List, Optional from . import config logger = logging.getLogger(__name__) -# Hard ceiling on words per request, regardless of the configured chunk -# size. 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 - -def split_into_chunks(text: str, max_words: int = config.CHUNK_SIZE) -> List[str]: +def split_into_chunks(text: str, max_words: Optional[int] = None) -> List[str]: """Split text into chunks of at most ``max_words`` words. - ``max_words`` is clamped to ``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. + ``max_words`` defaults to ``config.CHUNK_SIZE`` (read at call time). + There is no ceiling beyond that setting, but note that the TTS + servers silently truncate audio when a single generation runs too + long without reporting an error, so very large values are at your + own risk (see CHUNK_SIZE in converter/config.py). Splits on sentence boundaries. Sentences longer than the limit are split further at clause punctuation (which is kept attached for TTS @@ -33,13 +26,8 @@ def split_into_chunks(text: str, max_words: int = config.CHUNK_SIZE) -> List[str at word boundaries as a last resort: individual tokens stay intact, but whitespace between them is normalized. """ - if max_words > 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/chunking.py)", - max_words, MAX_REQUEST_WORDS, MAX_REQUEST_WORDS) - max_words = MAX_REQUEST_WORDS + if max_words is None: + max_words = config.CHUNK_SIZE if max_words < 1: max_words = 1 diff --git a/converter/config.py b/converter/config.py index 4cb5ab8..ed49f12 100644 --- a/converter/config.py +++ b/converter/config.py @@ -8,8 +8,13 @@ MAX_RETRIES = 3 # Attempts per chunk request HEARTBEAT_INTERVAL_SECONDS = 30 # Print "still working" in console logs every N seconds # Words per TTS generation request. -# Note that qwen-tts-demo does no chunking at all, but faster-qwen-tts and +# Note that qwen-tts-demo does no chunking at all, but faster-qwen3-tts and # audio.cpp may do chunking as well, so you may be needlessly double-chunking. +# This is the only size limit: there is no hard ceiling. However, servers +# silently truncate audio when a single generation runs too long (roughly +# ~2.5 min on the faster backend's static KV cache, ~11 min on the Gradio +# demo) without reporting an error, so raising this is at your own risk. +# The client-side truncation check still catches and retries gross cases. CHUNK_SIZE = 250 # Default TTS backend. @@ -54,5 +59,5 @@ FASTER_VOICE = "default" AUDIOCPP_API_URL = "http://127.0.0.1:8080" # audio.cpp audiocpp_server # Model ids in the audio.cpp server.json config. -AUDIOCPP_MODEL_ID = "qwen"https://github.com/0xShug0/audio.cpp +AUDIOCPP_MODEL_ID = "qwen" AUDIOCPP_CLONE_MODEL_ID = "qwen-clone" diff --git a/converter/tts.py b/converter/tts.py index 741d9d3..a1c0b09 100644 --- a/converter/tts.py +++ b/converter/tts.py @@ -28,7 +28,7 @@ from typing import Any, Dict, List, Optional, Tuple from . import config from .audio import concat_audio_files, probe_duration_ms -from .chunking import MAX_REQUEST_WORDS, split_into_chunks +from .chunking import split_into_chunks logger = logging.getLogger(__name__) @@ -440,14 +440,14 @@ class QwenTTSClient(_BaseTTSClient): """Generate one audio chunk; returns its path in the chunks folder. The text is split into sub-requests of at most - ``MAX_REQUEST_WORDS`` words each (the book-level chunker + ``config.CHUNK_SIZE`` 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: - sub_texts = split_into_chunks(text, max_words=MAX_REQUEST_WORDS) + sub_texts = split_into_chunks(text, max_words=config.CHUNK_SIZE) if not sub_texts: raise RuntimeError("No text to synthesize") @@ -673,7 +673,7 @@ 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=MAX_REQUEST_WORDS) + sub_chunks = split_into_chunks(text, max_words=config.CHUNK_SIZE) if not sub_chunks: raise RuntimeError("No text to synthesize") @@ -914,13 +914,13 @@ class AudioCppTTSClient(_BaseTTSClient): """Generate one audio chunk; returns its path in the chunks folder. The text is split into sub-requests of at most - ``MAX_REQUEST_WORDS`` words each (defense in depth against + ``config.CHUNK_SIZE`` words each (defense in depth against pathological input, matching the Gradio client), each sub-request returns a complete WAV file, and the parts are concatenated into one chunk file. """ try: - sub_texts = split_into_chunks(text, max_words=MAX_REQUEST_WORDS) + sub_texts = split_into_chunks(text, max_words=config.CHUNK_SIZE) if not sub_texts: raise RuntimeError("No text to synthesize") diff --git a/tests/test_chunking.py b/tests/test_chunking.py index 7f37f80..3cd926b 100644 --- a/tests/test_chunking.py +++ b/tests/test_chunking.py @@ -4,42 +4,41 @@ import unittest from unittest.mock import patch from converter import config -from converter.chunking import MAX_REQUEST_WORDS, split_into_chunks +from converter.chunking import split_into_chunks class ChunkSizeDefaultTests(unittest.TestCase): - """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.""" + """Guard the request-size setting: each API call is one model + generation, and the servers silently truncate audio when a single + generation runs too long (~2.5 min faster backend, ~11 min Gradio + demo), so the default chunk size must stay well inside that budget. + There is no hard ceiling beyond CHUNK_SIZE; users raising it accept + the truncation risk themselves.""" - def test_default_chunk_size_within_request_ceiling(self): - self.assertLessEqual(config.CHUNK_SIZE, MAX_REQUEST_WORDS) + def test_default_chunk_size_within_single_generation_budget(self): + self.assertLessEqual(config.CHUNK_SIZE, 300) - def test_request_ceiling_within_single_generation_budget(self): - self.assertLessEqual(MAX_REQUEST_WORDS, 300) - - def test_sizes_are_positive(self): + def test_default_chunk_size_is_positive(self): self.assertGreaterEqual(config.CHUNK_SIZE, 1) - self.assertGreaterEqual(MAX_REQUEST_WORDS, 1) -class RequestCeilingClampTests(unittest.TestCase): - def test_oversized_chunk_size_is_clamped_with_warning(self): +class RequestSizeTests(unittest.TestCase): + def test_oversized_chunk_size_is_honored(self): + # No clamping: whatever size is configured (or requested) is used. text = " ".join(f"word{i}" for i in range(30)) + "." - with patch("converter.chunking.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)) + chunks = split_into_chunks(text, max_words=5000) + self.assertEqual(len(chunks), 1) + self.assertEqual(len(chunks[0].split()), 30) - def test_default_ceiling_clamps_realistic_configuration(self): + def test_default_uses_runtime_config_chunk_size(self): + # The default resolves config.CHUNK_SIZE at call time, so + # patching the config changes the default split size. sentences = " ".join( f"S{i} " + " ".join(["word"] * 8) + "." for i in range(60)) - chunks = split_into_chunks(sentences, max_words=5000) + with patch.object(config, "CHUNK_SIZE", 120): + chunks = split_into_chunks(sentences) self.assertGreater(len(chunks), 1) - self.assertTrue(all(len(chunk.split()) <= MAX_REQUEST_WORDS - for chunk in chunks)) + self.assertTrue(all(len(chunk.split()) <= 120 for chunk in chunks)) class SplitIntoChunksTests(unittest.TestCase): @@ -94,8 +93,7 @@ class SplitIntoChunksTests(unittest.TestCase): 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. + # boundaries so no single request exceeds the configured size. sentence = " ".join(["word"] * 30) + "." chunks = split_into_chunks(sentence, max_words=10) self.assertGreater(len(chunks), 1) diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py new file mode 100644 index 0000000..63260ca --- /dev/null +++ b/tests/test_make_audiocpp_server_json.py @@ -0,0 +1,457 @@ +"""Tests for the audio.cpp server.json generator tool.""" + +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from converter import config +from tools import make_audiocpp_server_json as make_server + +FAKE_CONFIG = ( + 'LANGUAGE = "English"\n' + "\n" + 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' + "\n" + "CHUNK_SIZE = 250\n" +) + + +class FindWavFilesTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _touch(self, name): + path = self.folder / name + path.write_bytes(b"x") + return path + + def test_finds_only_wavs_case_insensitive(self): + self._touch("b.wav") + self._touch("a.WAV") + self._touch("notes.txt") + (self.folder / "sub").mkdir() + (self.folder / "sub" / "c.wav").write_bytes(b"x") + names = [path.name for path in make_server.find_wav_files(self.folder)] + self.assertEqual(names, ["a.WAV", "b.wav"]) + + def test_sorted_alphabetically_case_insensitive(self): + for name in ("Zed.wav", "alpha.wav", "Beta.wav"): + self._touch(name) + names = [path.name for path in make_server.find_wav_files(self.folder)] + self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) + + def test_empty_directory_returns_empty_list(self): + self.assertEqual(make_server.find_wav_files(self.folder), []) + + +class ConfigPortTests(unittest.TestCase): + def test_port_parsed_from_config_url(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:8080"): + self.assertEqual(make_server.config_port(), 8080) + + def test_missing_port_falls_back(self): + with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1"): + self.assertEqual(make_server.config_port(), + make_server.FALLBACK_PORT) + + def test_invalid_url_falls_back(self): + with patch.object(config, "AUDIOCPP_API_URL", "not a url"): + self.assertEqual(make_server.config_port(), + make_server.FALLBACK_PORT) + + def test_url_with_port_replaces_port(self): + self.assertEqual( + make_server._url_with_port("http://127.0.0.1:8080", 9000), + "http://127.0.0.1:9000") + + def test_url_without_port_adds_port(self): + self.assertEqual( + make_server._url_with_port("http://localhost", 8080), + "http://localhost:8080") + + +class UpdateConfigPortTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.config_path = Path(self._tmp.name) / "config.py" + self.config_path.write_text(FAKE_CONFIG, encoding="utf-8") + + def tearDown(self): + self._tmp.cleanup() + + def test_rewrites_port_preserving_comment(self): + changed = make_server.update_config_api_url_port( + 8080, config_path=self.config_path) + self.assertTrue(changed) + text = self.config_path.read_text(encoding="utf-8") + self.assertIn( + 'AUDIOCPP_API_URL = "http://127.0.0.1:8080" # audio.cpp audiocpp_server', + text) + self.assertIn('LANGUAGE = "English"', text) + self.assertIn("CHUNK_SIZE = 250", text) + + def test_returns_false_when_no_url_line(self): + path = Path(self._tmp.name) / "other.py" + path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") + self.assertFalse(make_server.update_config_api_url_port( + 8080, config_path=path)) + + def test_returns_false_when_port_unchanged(self): + self.assertFalse(make_server.update_config_api_url_port( + 9999, config_path=self.config_path)) + self.assertEqual(self.config_path.read_text(encoding="utf-8"), + FAKE_CONFIG) + + def test_returns_false_when_file_missing(self): + self.assertFalse(make_server.update_config_api_url_port( + 8080, config_path=Path(self._tmp.name) / "nope.py")) + + +class BuildVoicePresetsTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self.narrator = self.folder / "narrator.wav" + self.narrator.write_bytes(b"x") + self.other = self.folder / "other.wav" + self.other.write_bytes(b"x") + + def tearDown(self): + self._tmp.cleanup() + + def test_presets_named_after_basenames_with_absolute_paths(self): + transcripts = {str(self.narrator): "First transcript.", + str(self.other): "Second transcript."} + with patch.object(make_server, "transcribe_reference_audio", + side_effect=lambda path, model_name="base": + transcripts[path]): + presets = make_server.build_voice_presets( + [self.narrator, self.other], "base") + self.assertEqual(list(presets), ["narrator", "other"]) + self.assertEqual(presets["narrator"]["reference_text"], + "First transcript.") + self.assertEqual(Path(presets["narrator"]["voice_ref"]), + self.narrator.resolve()) + + def test_failed_transcription_keeps_entry_with_empty_text(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value=None): + presets = make_server.build_voice_presets([self.narrator], "base") + self.assertEqual(presets["narrator"]["reference_text"], "") + + def test_whisper_model_name_is_passed_through(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value="text") as mock_transcribe: + make_server.build_voice_presets([self.narrator], "large-v3") + self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], + "large-v3") + + +class BuildServerConfigTests(unittest.TestCase): + def test_both_models_with_presets(self): + presets = {"narrator": {"voice_ref": "/x.wav", + "reference_text": "hi"}} + server_config = make_server.build_server_config( + host="127.0.0.1", port=8080, backend="cuda", lazy_load=False, + include_custom=True, include_clone=True, + custom_voice_id="qwen", clone_model_id="qwen-clone", + custom_voice_path="models/custom", base_path="models/base", + voice_presets=presets) + self.assertEqual(server_config["host"], "127.0.0.1") + self.assertEqual(server_config["port"], 8080) + self.assertEqual(server_config["backend"], "cuda") + self.assertFalse(server_config["lazy_load"]) + self.assertEqual([model["id"] for model in server_config["models"]], + ["qwen", "qwen-clone"]) + custom_entry, clone_entry = server_config["models"] + self.assertNotIn("voice_presets", custom_entry) + self.assertEqual(custom_entry["family"], "qwen3_tts") + self.assertEqual(custom_entry["path"], "models/custom") + self.assertEqual(clone_entry["path"], "models/base") + self.assertEqual(clone_entry["voice_presets"], presets) + + def test_custom_only_has_single_entry(self): + server_config = make_server.build_server_config( + host="0.0.0.0", port=9000, backend="cpu", lazy_load=True, + include_custom=True, include_clone=False, + custom_voice_id="qwen", clone_model_id="qwen-clone", + custom_voice_path="models/custom", base_path=None, + voice_presets={}) + self.assertEqual(len(server_config["models"]), 1) + self.assertEqual(server_config["models"][0]["id"], "qwen") + self.assertNotIn("voice_presets", server_config["models"][0]) + + def test_clone_only_without_presets_omits_key(self): + server_config = make_server.build_server_config( + host="127.0.0.1", port=8080, backend="vulkan", lazy_load=False, + include_custom=False, include_clone=True, + custom_voice_id="qwen", clone_model_id="qwen-clone", + custom_voice_path=None, base_path="models/base", + voice_presets={}) + self.assertEqual(len(server_config["models"]), 1) + self.assertEqual(server_config["models"][0]["id"], "qwen-clone") + self.assertNotIn("voice_presets", server_config["models"][0]) + + +class PromptHelperTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + 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) + + def test_ask_port_eof_returns_default(self): + with patch("builtins.input", side_effect=EOFError): + self.assertEqual(make_server.ask_port(8080), 8080) + + def test_ask_menu_reprompts_until_valid(self): + options = [("One", "one"), ("Two", "two")] + with patch("builtins.input", side_effect=["9", "2"]): + self.assertEqual( + make_server.ask_menu("Pick:", options, default_index=1), + "two") + + def test_ask_menu_eof_returns_default(self): + options = [("One", "one"), ("Two", "two")] + with patch("builtins.input", side_effect=EOFError): + self.assertEqual( + make_server.ask_menu("Pick:", options, default_index=1), + "one") + + +class MainTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self.output = self.folder / "server.json" + # Isolate the config.py rewrite target so no test can ever + # modify the repository's real converter/config.py. + self.fake_config = self.folder / "config.py" + self.fake_config.write_text(FAKE_CONFIG, encoding="utf-8") + patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) + patcher.start() + self.addCleanup(patcher.stop) + + def tearDown(self): + self._tmp.cleanup() + + def _run(self, argv, inputs=None, transcribe=None): + argv = ["make_audiocpp_server_json.py"] + argv + input_effect = inputs if inputs is not None else EOFError + transcribe_effect = transcribe if transcribe is not None else MagicMock() + with patch.object(sys, "argv", argv), \ + patch("builtins.input", side_effect=input_effect), \ + patch.object(make_server, "transcribe_reference_audio", + side_effect=transcribe_effect): + return make_server.main() + + def _defaults(self, models="", host="", port="", backend="", + lazy="", custom_path="", clone_path="", wav_dir="", + confirm="y", prefix=()): + return list(prefix) + [models, host, port, backend, lazy, + custom_path, clone_path, wav_dir, confirm] + + def test_default_run_hosts_both_models(self): + exit_code = self._run(["--output", str(self.output)], + inputs=self._defaults()) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["host"], "127.0.0.1") + self.assertEqual(data["port"], make_server.config_port()) + self.assertEqual(data["backend"], "cuda") + self.assertFalse(data["lazy_load"]) + self.assertEqual( + [model["id"] for model in data["models"]], + [config.AUDIOCPP_MODEL_ID, config.AUDIOCPP_CLONE_MODEL_ID]) + self.assertEqual( + [model["path"] for model in data["models"]], + [make_server.DEFAULT_CUSTOM_VOICE_PATH, + make_server.DEFAULT_BASE_PATH]) + self.assertNotIn("voice_presets", data["models"][1]) + + def test_eof_uses_all_defaults(self): + exit_code = self._run(["--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") + self.assertEqual(data["port"], make_server.config_port()) + self.assertEqual(data["backend"], "cuda") + self.assertFalse(data["lazy_load"]) + self.assertEqual(len(data["models"]), 2) + + def test_clone_only_with_positional_wav_dir(self): + (self.folder / "narrator.wav").write_bytes(b"x") + (self.folder / "alpha.wav").write_bytes(b"x") + inputs = ["3", "", "", "", "", "", "y"] + exit_code = self._run( + [str(self.folder), "--output", str(self.output)], + inputs=inputs, + transcribe=lambda path, model_name="base": + f"transcript of {Path(path).name}") + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 1) + clone_entry = data["models"][0] + self.assertEqual(clone_entry["id"], config.AUDIOCPP_CLONE_MODEL_ID) + self.assertEqual(sorted(clone_entry["voice_presets"]), + ["alpha", "narrator"]) + self.assertEqual(clone_entry["voice_presets"]["narrator"], + {"voice_ref": str((self.folder / "narrator.wav").resolve()), + "reference_text": "transcript of narrator.wav"}) + + def test_custom_only_single_model(self): + inputs = ["", "", "", "", "", "y"] + exit_code = self._run( + ["--output", str(self.output), "--models", "custom"], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([model["id"] for model in data["models"]], + [config.AUDIOCPP_MODEL_ID]) + + 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=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([model["id"] for model in data["models"]], + ["qwen", "qwen-clone-2"]) + + def test_duplicate_ids_eof_exits(self): + 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.assertNotEqual(ctx.exception.code, 0) + self.assertFalse(self.output.exists()) + + def test_port_sync_accepted_updates_config(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:9999"): + inputs = ["", "", "y", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output), + "--port", "8080"], + inputs=inputs) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:8080"', + self.fake_config.read_text(encoding="utf-8")) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["port"], 8080) + + def test_port_sync_declined_keeps_config(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:9999"): + inputs = ["", "", "n", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output), + "--port", "8080"], + inputs=inputs) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:9999"', + self.fake_config.read_text(encoding="utf-8")) + + def test_matching_port_does_not_prompt_for_sync(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:8080"): + inputs = self._defaults() + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + self.assertEqual(self.fake_config.read_text(encoding="utf-8"), + FAKE_CONFIG) + + def test_invalid_menu_choice_reprompts(self): + inputs = ["9", "", "", "", "", "", "", "", "", "y"] + exit_code = self._run(["--output", str(self.output)], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 2) + + def test_confirm_declined_writes_nothing(self): + inputs = self._defaults(confirm="n") + exit_code = self._run(["--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)], + inputs=["n"]) + self.assertEqual(exit_code, 1) + self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), + {"old": True}) + + 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)], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 2) + + 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"], + inputs=inputs) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 2) + + def test_flags_skip_prompts(self): + exit_code = self._run( + ["--output", str(self.output), "--models", "both", + "--host", "0.0.0.0", "--port", "9000", "--backend", "cpu", + "--lazy-load"], + inputs=["y", "", "", "", "y"]) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:9000"', + self.fake_config.read_text(encoding="utf-8")) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["host"], "0.0.0.0") + self.assertEqual(data["port"], 9000) + self.assertEqual(data["backend"], "cpu") + 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)], + inputs=self._defaults()) + self.assertEqual(ctx.exception.code, 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_make_faster_voices_json.py b/tests/test_make_faster_voices_json.py new file mode 100644 index 0000000..f26e645 --- /dev/null +++ b/tests/test_make_faster_voices_json.py @@ -0,0 +1,169 @@ +"""Tests for the faster-qwen3-tts voices.json generator tool.""" + +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from tools import make_faster_voices_json as make_voices + + +class FindWavFilesTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _touch(self, name): + path = self.folder / name + path.write_bytes(b"x") + return path + + def test_finds_only_wavs_case_insensitive(self): + self._touch("b.wav") + self._touch("a.WAV") + self._touch("notes.txt") + (self.folder / "sub").mkdir() + (self.folder / "sub" / "c.wav").write_bytes(b"x") + names = [path.name for path in make_voices.find_wav_files(self.folder)] + self.assertEqual(names, ["a.WAV", "b.wav"]) + + def test_sorted_alphabetically_case_insensitive(self): + for name in ("Zed.wav", "alpha.wav", "Beta.wav"): + self._touch(name) + names = [path.name for path in make_voices.find_wav_files(self.folder)] + self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) + + def test_empty_directory_returns_empty_list(self): + self.assertEqual(make_voices.find_wav_files(self.folder), []) + + +class BuildVoicesTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self.narrator = self.folder / "narrator.wav" + self.narrator.write_bytes(b"x") + self.other = self.folder / "other.wav" + self.other.write_bytes(b"x") + + def tearDown(self): + self._tmp.cleanup() + + def test_voices_named_after_basenames_with_absolute_paths(self): + transcripts = {str(self.narrator): "First transcript.", + str(self.other): "Second transcript."} + with patch.object(make_voices, "transcribe_reference_audio", + side_effect=lambda path, model_name="base": transcripts[path]): + voices = make_voices.build_voices([self.narrator, self.other], + "English", "base") + self.assertEqual(list(voices), ["narrator", "other"]) + self.assertEqual(voices["narrator"]["ref_text"], "First transcript.") + self.assertEqual(voices["narrator"]["language"], "English") + self.assertTrue(Path(voices["narrator"]["ref_audio"]).is_absolute()) + self.assertEqual(Path(voices["narrator"]["ref_audio"]), self.narrator.resolve()) + + def test_failed_transcription_keeps_entry_with_empty_text(self): + with patch.object(make_voices, "transcribe_reference_audio", + return_value=None): + voices = make_voices.build_voices([self.narrator], "English", "base") + self.assertEqual(voices["narrator"]["ref_text"], "") + + def test_whisper_model_name_is_passed_through(self): + with patch.object(make_voices, "transcribe_reference_audio", + return_value="text") as mock_transcribe: + make_voices.build_voices([self.narrator], "English", "large-v3") + self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") + + +class MainTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + (self.folder / "narrator.wav").write_bytes(b"x") + (self.folder / "alpha.wav").write_bytes(b"x") + self.output = self.folder / "voices.json" + + def tearDown(self): + self._tmp.cleanup() + + def _run(self, argv): + with patch.object(sys, "argv", ["make_voices.py"] + argv): + return make_voices.main() + + def test_writes_json_with_alphabetical_voice_order(self): + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"): + exit_code = self._run([str(self.folder)]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(list(data), ["alpha", "narrator"]) + self.assertEqual(data["alpha"]["ref_text"], "hello") + self.assertEqual(data["alpha"]["language"], "English") + + def test_custom_output_path(self): + custom = Path(self._tmp.name) / "custom.json" + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"): + self._run([str(self.folder), "--output", str(custom)]) + self.assertTrue(custom.exists()) + self.assertFalse(self.output.exists()) + + def test_invalid_language_errors_before_work(self): + with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe: + with self.assertRaises(SystemExit) as ctx: + self._run([str(self.folder), "--language", "klingon"]) + self.assertEqual(ctx.exception.code, 2) + mock_transcribe.assert_not_called() + + def test_missing_input_dir_errors(self): + with self.assertRaises(SystemExit) as ctx: + self._run([str(self.folder / "nope")]) + self.assertEqual(ctx.exception.code, 2) + + def test_no_wav_files_errors(self): + empty = Path(tempfile.mkdtemp()) + try: + with self.assertRaises(SystemExit) as ctx: + self._run([str(empty)]) + self.assertEqual(ctx.exception.code, 2) + finally: + empty.rmdir() + + def test_existing_output_declined_keeps_file(self): + self.output.write_text('{"old": true}', encoding="utf-8") + with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe, \ + patch("builtins.input", return_value="n"): + exit_code = self._run([str(self.folder)]) + self.assertEqual(exit_code, 1) + mock_transcribe.assert_not_called() + self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), + {"old": True}) + + def test_existing_output_accepted_overwrites(self): + self.output.write_text('{"old": true}', encoding="utf-8") + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"), \ + patch("builtins.input", return_value="y"): + exit_code = self._run([str(self.folder)]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(list(data), ["alpha", "narrator"]) + + def test_force_overwrites_without_prompt(self): + self.output.write_text('{"old": true}', encoding="utf-8") + with patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"), \ + patch("builtins.input", side_effect=AssertionError("prompted")): + exit_code = self._run([str(self.folder), "--force"]) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(list(data), ["alpha", "narrator"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_make_voices.py b/tests/test_make_voices.py deleted file mode 100644 index cf76a57..0000000 --- a/tests/test_make_voices.py +++ /dev/null @@ -1,169 +0,0 @@ -"""Tests for the voices.json generator tool.""" - -import json -import sys -import tempfile -import unittest -from pathlib import Path -from unittest.mock import patch - -from tools import make_voices - - -class FindWavFilesTests(unittest.TestCase): - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - - def tearDown(self): - self._tmp.cleanup() - - def _touch(self, name): - path = self.folder / name - path.write_bytes(b"x") - return path - - def test_finds_only_wavs_case_insensitive(self): - self._touch("b.wav") - self._touch("a.WAV") - self._touch("notes.txt") - (self.folder / "sub").mkdir() - (self.folder / "sub" / "c.wav").write_bytes(b"x") - names = [path.name for path in make_voices.find_wav_files(self.folder)] - self.assertEqual(names, ["a.WAV", "b.wav"]) - - def test_sorted_alphabetically_case_insensitive(self): - for name in ("Zed.wav", "alpha.wav", "Beta.wav"): - self._touch(name) - names = [path.name for path in make_voices.find_wav_files(self.folder)] - self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) - - def test_empty_directory_returns_empty_list(self): - self.assertEqual(make_voices.find_wav_files(self.folder), []) - - -class BuildVoicesTests(unittest.TestCase): - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - self.narrator = self.folder / "narrator.wav" - self.narrator.write_bytes(b"x") - self.other = self.folder / "other.wav" - self.other.write_bytes(b"x") - - def tearDown(self): - self._tmp.cleanup() - - def test_voices_named_after_basenames_with_absolute_paths(self): - transcripts = {str(self.narrator): "First transcript.", - str(self.other): "Second transcript."} - with patch.object(make_voices, "transcribe_reference_audio", - side_effect=lambda path, model_name="base": transcripts[path]): - voices = make_voices.build_voices([self.narrator, self.other], - "English", "base") - self.assertEqual(list(voices), ["narrator", "other"]) - self.assertEqual(voices["narrator"]["ref_text"], "First transcript.") - self.assertEqual(voices["narrator"]["language"], "English") - self.assertTrue(Path(voices["narrator"]["ref_audio"]).is_absolute()) - self.assertEqual(Path(voices["narrator"]["ref_audio"]), self.narrator.resolve()) - - def test_failed_transcription_keeps_entry_with_empty_text(self): - with patch.object(make_voices, "transcribe_reference_audio", - return_value=None): - voices = make_voices.build_voices([self.narrator], "English", "base") - self.assertEqual(voices["narrator"]["ref_text"], "") - - def test_whisper_model_name_is_passed_through(self): - with patch.object(make_voices, "transcribe_reference_audio", - return_value="text") as mock_transcribe: - make_voices.build_voices([self.narrator], "English", "large-v3") - self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") - - -class MainTests(unittest.TestCase): - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - (self.folder / "narrator.wav").write_bytes(b"x") - (self.folder / "alpha.wav").write_bytes(b"x") - self.output = self.folder / "voices.json" - - def tearDown(self): - self._tmp.cleanup() - - def _run(self, argv): - with patch.object(sys, "argv", ["make_voices.py"] + argv): - return make_voices.main() - - def test_writes_json_with_alphabetical_voice_order(self): - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"): - exit_code = self._run([str(self.folder)]) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(list(data), ["alpha", "narrator"]) - self.assertEqual(data["alpha"]["ref_text"], "hello") - self.assertEqual(data["alpha"]["language"], "English") - - def test_custom_output_path(self): - custom = Path(self._tmp.name) / "custom.json" - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"): - self._run([str(self.folder), "--output", str(custom)]) - self.assertTrue(custom.exists()) - self.assertFalse(self.output.exists()) - - def test_invalid_language_errors_before_work(self): - with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe: - with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder), "--language", "klingon"]) - self.assertEqual(ctx.exception.code, 2) - mock_transcribe.assert_not_called() - - def test_missing_input_dir_errors(self): - with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder / "nope")]) - self.assertEqual(ctx.exception.code, 2) - - def test_no_wav_files_errors(self): - empty = Path(tempfile.mkdtemp()) - try: - with self.assertRaises(SystemExit) as ctx: - self._run([str(empty)]) - self.assertEqual(ctx.exception.code, 2) - finally: - empty.rmdir() - - def test_existing_output_declined_keeps_file(self): - self.output.write_text('{"old": true}', encoding="utf-8") - with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe, \ - patch("builtins.input", return_value="n"): - exit_code = self._run([str(self.folder)]) - self.assertEqual(exit_code, 1) - mock_transcribe.assert_not_called() - self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), - {"old": True}) - - def test_existing_output_accepted_overwrites(self): - self.output.write_text('{"old": true}', encoding="utf-8") - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"), \ - patch("builtins.input", return_value="y"): - exit_code = self._run([str(self.folder)]) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(list(data), ["alpha", "narrator"]) - - def test_force_overwrites_without_prompt(self): - self.output.write_text('{"old": true}', encoding="utf-8") - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"), \ - patch("builtins.input", side_effect=AssertionError("prompted")): - exit_code = self._run([str(self.folder), "--force"]) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(list(data), ["alpha", "narrator"]) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_tts.py b/tests/test_tts.py index 813332d..f2dda0e 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -296,23 +296,23 @@ 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(tts, "MAX_REQUEST_WORDS", 10), \ + with patch.object(config, "CHUNK_SIZE", 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): + def test_subchunk_size_follows_config_chunk_size(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", 4), \ patch.object(client, "_request_pcm", return_value=pcm) as mock_pcm: result = client.generate_chunk(text, 1) - # CHUNK_SIZE 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) + # The sub-chunk split follows config.CHUNK_SIZE, so the whole + # (8-word) text needs two 4-word requests here. + self.assertEqual(mock_pcm.call_count, 2) self.assertIsNotNone(result) def test_stale_chunk_files_are_removed(self): @@ -489,7 +489,7 @@ class QwenTTSClientGenerateTests(unittest.TestCase): 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(tts, "MAX_REQUEST_WORDS", 5), \ + with patch.object(config, "CHUNK_SIZE", 5), \ patch.object(client, "_generate_custom_voice", side_effect=[(str(first),), (str(second),), (str(first),)]) as mock_generate: @@ -627,26 +627,29 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): self.assertGreaterEqual(client._seed, 0) def test_preset_mode_routes_to_clone_model_when_configured(self): - with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): client = self._client( voice="narrator", models={"data": [{"id": "qwen3-tts"}, {"id": "qwen3-tts-clone"}]}) self.assertEqual(client.model_id, "qwen3-tts-clone") def test_preset_mode_falls_back_when_clone_model_not_on_server(self): - with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"), \ + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"), \ self.assertLogs("converter.tts", level="WARNING") as logs: client = self._client( voice="narrator", models={"data": [{"id": "qwen3-tts"}, {"id": "pocket-tts"}]}) - self.assertEqual(client.model_id, config.AUDIOCPP_MODEL_ID) + self.assertEqual(client.model_id, "qwen3-tts") self.assertTrue(any("qwen3-tts-clone" in line for line in logs.output)) def test_clone_model_id_ignored_for_speaker_mode(self): - with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): + with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ + patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"): client = self._client( models={"data": [{"id": "qwen3-tts"}, {"id": "qwen3-tts-clone"}]}) - self.assertEqual(client.model_id, config.AUDIOCPP_MODEL_ID) + self.assertEqual(client.model_id, "qwen3-tts") def test_clone_model_id_equal_to_primary_is_noop(self): with patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", @@ -787,7 +790,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): parts = [self._wav_bytes(b"\x01\x00" * 10), self._wav_bytes(b"\x02\x00" * 20), self._wav_bytes(b"\x03\x00" * 30)] - with patch.object(tts, "MAX_REQUEST_WORDS", 10), \ + with patch.object(config, "CHUNK_SIZE", 10), \ patch.object(client, "_request_wav", side_effect=parts) as mock_request: result = client.generate_chunk(text, 1) self.assertEqual(mock_request.call_count, 3) diff --git a/tools/make_audiocpp_server_json.py b/tools/make_audiocpp_server_json.py new file mode 100755 index 0000000..40354eb --- /dev/null +++ b/tools/make_audiocpp_server_json.py @@ -0,0 +1,443 @@ +#!/usr/bin/env python3 +"""Interactively generate a server.json for the audio.cpp audiocpp_server. + +Asks which Qwen3-TTS models to host, pulls the model ids expected by this +converter (AUDIOCPP_MODEL_ID / AUDIOCPP_CLONE_MODEL_ID) from +converter/config.py, and writes a server.json that can be passed to +audiocpp_server: + + audiocpp_server --config server.json + +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 Base-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 models on +127.0.0.1:8080 with the cuda backend. + +Usage: + python tools/make_audiocpp_server_json.py [WAV_DIR] [--output PATH] + [--host HOST] [--port PORT] [--models {both,custom,clone}] + [--backend {cuda,vulkan,hip,cpu}] [--lazy-load] + [--whisper-model NAME] [--force] +""" + +import argparse +import json +import re +import sys +import urllib.parse +from pathlib import Path +from typing import Dict, Optional + +# Allow running from any working directory. +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from converter import config +from converter.tts import transcribe_reference_audio + +DEFAULT_HOST = "127.0.0.1" +FALLBACK_PORT = 8080 +DEFAULT_CUSTOM_VOICE_PATH = "models/Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF" +DEFAULT_BASE_PATH = "models/Qwen3-TTS-12Hz-1.7B-Base-GGUF" +CONFIG_PATH = Path(__file__).resolve().parent.parent / "converter" / "config.py" + +MODEL_SELECTIONS = ("both", "custom", "clone") +BACKENDS = ("cuda", "vulkan", "hip", "cpu") + + +def find_wav_files(input_dir: Path) -> list: + """Return the .wav files in INPUT_DIR, sorted alphabetically by name.""" + return sorted( + (path for path in input_dir.iterdir() + if path.is_file() and path.suffix.lower() == ".wav"), + key=lambda path: path.name.lower(), + ) + + +def prompt_overwrite(output_path: Path) -> bool: + """Ask whether to overwrite an existing output file.""" + while True: + try: + answer = input(f"{output_path} already exists. Overwrite? (y/n): ").strip().lower() + except EOFError: + print("\n[WARNING] No interactive input available; keeping existing file") + return False + if answer in ("y", "yes"): + return True + if answer in ("n", "no"): + return False + print("Please answer 'y' or 'n'.") + + +def ask(prompt: str, default: Optional[str] = None) -> Optional[str]: + """Prompt for a free-text value with a default; EOF returns the default.""" + suffix = f" [{default}]" if default is not None else "" + try: + answer = input(f"{prompt}{suffix}: ").strip() + except EOFError: + return default + return answer or default + + +def ask_bool(prompt: str, default: bool = False) -> bool: + """Prompt for a yes/no answer; Enter or EOF accepts the default.""" + suffix = " [Y/n]" if default else " [y/N]" + while True: + try: + answer = input(f"{prompt}{suffix}: ").strip().lower() + except EOFError: + return default + if not answer: + return default + if answer in ("y", "yes"): + return True + if answer in ("n", "no"): + return False + print("Please answer 'y' or 'n'.") + + +def ask_port(default: int) -> int: + """Prompt for a port number; Enter or EOF accepts the default.""" + while True: + try: + answer = input(f"Port [{default}]: ").strip() + except EOFError: + return default + if not answer: + return default + try: + value = int(answer) + except ValueError: + value = None + if value is not None and 1 <= value <= 65535: + return value + print("Please enter a port number between 1 and 65535.") + + +def ask_menu(title: str, options: list, default_index: int = 1) -> str: + """Show a numbered menu and return the chosen option's value.""" + print(title) + for number, (label, _) in enumerate(options, 1): + print(f" {number}) {label}") + while True: + try: + answer = input(f"Choice [{default_index}]: ").strip() + except EOFError: + return options[default_index - 1][1] + if not answer: + return options[default_index - 1][1] + if answer.isdigit() and 1 <= int(answer) <= len(options): + return options[int(answer) - 1][1] + print(f"Please enter a number between 1 and {len(options)}.") + + +def ask_models() -> str: + return ask_menu( + "Which models should the server host?", + [ + ("Both (recommended) - built-in speakers + voice cloning", "both"), + ("CustomVoice only - built-in speakers", "custom"), + ("Base only - voice cloning (converting then requires --voice)", "clone"), + ]) + + +def ask_backend() -> str: + return ask_menu( + "Which inference backend was audiocpp_server built for?", + [ + ("cuda - NVIDIA GPUs (fastest)", "cuda"), + ("vulkan - cross-vendor GPU", "vulkan"), + ("hip - AMD GPUs", "hip"), + ("cpu - no GPU required", "cpu"), + ]) + + +def ask_distinct_clone_id(primary_id: str) -> str: + """Prompt until a non-empty id different from PRIMARY_ID is entered.""" + prompt = (f"Enter a new id for the cloning (Base) model " + f"(must differ from '{primary_id}'): ") + while True: + try: + answer = input(prompt).strip() + except EOFError: + print() + sys.exit("[FATAL] No interactive input available to resolve the " + "duplicate model id; give the two models distinct " + "AUDIOCPP_MODEL_ID / AUDIOCPP_CLONE_MODEL_ID values in " + "converter/config.py first") + if answer and answer != primary_id: + return answer + print(f"[WARNING] The id must be unique; it cannot be empty or " + 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: + return urllib.parse.urlsplit(config.AUDIOCPP_API_URL).port or FALLBACK_PORT + except ValueError: + return FALLBACK_PORT + + +def _url_with_port(url: str, port: int) -> str: + parts = urllib.parse.urlsplit(url) + host = parts.hostname or "127.0.0.1" + return urllib.parse.urlunsplit( + (parts.scheme or "http", f"{host}:{port}", parts.path, "", "")) + + +def update_config_api_url_port(port: int, config_path: Optional[Path] = None) -> bool: + """Rewrite the port inside AUDIOCPP_API_URL in converter/config.py. + + Only the quoted URL literal is replaced; surrounding lines and the + trailing comment are preserved. Returns True when the file was changed. + """ + path = Path(config_path) if config_path is not None else CONFIG_PATH + try: + text = path.read_text(encoding="utf-8") + except OSError: + return False + match = re.search(r'(?m)^(\s*AUDIOCPP_API_URL\s*=\s*")([^"]*)(")', text) + if not match: + return False + new_url = _url_with_port(match.group(2), port) + if new_url == match.group(2): + return False + text = text[:match.start(2)] + new_url + text[match.end(2):] + try: + path.write_text(text, encoding="utf-8") + except OSError: + return False + return True + + +def build_voice_presets(wav_files: list, whisper_model: str) -> Dict[str, dict]: + """Transcribe each wav file and build the voice_presets mapping.""" + presets: Dict[str, dict] = {} + for wav_file in wav_files: + name = wav_file.stem + print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...") + text = transcribe_reference_audio(str(wav_file), model_name=whisper_model) + if text: + print(f"[OK] {name}: {text}") + else: + print(f"[WARNING] No transcript for '{name}'; cloning works best " + "with an accurate transcript — consider editing server.json " + "by hand before starting the server") + presets[name] = { + "voice_ref": str(wav_file.resolve()), + "reference_text": text or "", + } + return presets + + +def build_server_config(host: str, port: int, backend: str, lazy_load: bool, + include_custom: bool, include_clone: bool, + custom_voice_id: str, clone_model_id: str, + custom_voice_path: str, base_path: str, + voice_presets: Dict[str, dict]) -> dict: + """Assemble the server.json document.""" + models = [] + if include_custom: + models.append({ + "id": custom_voice_id, + "family": "qwen3_tts", + "path": custom_voice_path, + "task": "tts", + "mode": "offline", + }) + if include_clone: + clone_entry = { + "id": clone_model_id, + "family": "qwen3_tts", + "path": base_path, + "task": "tts", + "mode": "offline", + } + if voice_presets: + clone_entry["voice_presets"] = voice_presets + models.append(clone_entry) + return { + "host": host, + "port": port, + "backend": backend, + "lazy_load": lazy_load, + "models": models, + } + + +def _print_next_steps(output_path: Path, include_custom: bool, + include_clone: bool, voice_presets: Dict[str, dict]) -> None: + print("\nNext steps:") + print(" 1. Start the server (build path varies by platform, e.g.") + print(" ./build/linux-cuda-release/bin/):") + print(f" audiocpp_server --config {output_path}") + print(" 2. Convert a book from this repository:") + if include_custom: + print(" python audiobook.py --backend audiocpp" + " # built-in speaker") + if include_clone: + names = ", ".join(voice_presets) or "none configured yet" + print(" python audiobook.py --backend audiocpp --voice NAME" + f" # cloned voice ({names})") + if include_clone and not include_custom: + print("[INFO] Only the Base model is hosted: --voice is required, " + "since speaker mode needs the CustomVoice model.") + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Generate a server.json for the audio.cpp audiocpp_server " + "hosting the Qwen3-TTS models 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("--output", type=Path, default=Path("server.json"), + help="Output path for server.json (default: " + "server.json in the current directory)") + parser.add_argument("--host", type=str, default=None, + help="Bind host for the server (default: 127.0.0.1)") + parser.add_argument("--port", type=int, default=None, + help="Port for the server (default: the port in " + "AUDIOCPP_API_URL from converter/config.py)") + parser.add_argument("--models", choices=MODEL_SELECTIONS, default=None, + help="Which models to host: both (default), custom " + "(CustomVoice speakers only), or clone " + "(Base voice cloning only)") + parser.add_argument("--backend", choices=BACKENDS, default=None, + help="Inference backend audiocpp_server was built " + "for (default: cuda)") + parser.add_argument("--lazy-load", action="store_true", + help="Load models on first use instead of at startup " + "(default: load at startup)") + parser.add_argument("--whisper-model", type=str, default="base", + help="Whisper model size for transcription " + "(default: base)") + parser.add_argument("--force", action="store_true", + 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 args.output.exists() and not args.force \ + and not prompt_overwrite(args.output): + print("[INFO] Aborted; existing server.json kept") + return 1 + + print("[INFO] Model ids from converter/config.py:") + print(f" built-in speakers (CustomVoice): '{config.AUDIOCPP_MODEL_ID}'") + print(f" voice cloning (Base): '{config.AUDIOCPP_CLONE_MODEL_ID}'") + + selection = args.models if args.models is not None else ask_models() + include_custom = selection in ("both", "custom") + include_clone = selection in ("both", "clone") + + custom_voice_id = config.AUDIOCPP_MODEL_ID + clone_model_id = config.AUDIOCPP_CLONE_MODEL_ID + if include_custom and include_clone and custom_voice_id == clone_model_id: + print(f"[WARNING] AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID are " + f"both '{custom_voice_id}' in converter/config.py, but server " + "model ids must be unique.") + clone_model_id = ask_distinct_clone_id(custom_voice_id) + + host = args.host if args.host else ask("Bind host", DEFAULT_HOST) + port = args.port if args.port is not None else ask_port(config_port()) + if port != config_port(): + if ask_bool(f"Update AUDIOCPP_API_URL in converter/config.py to port " + f"{port} so audiobook.py talks to this server", True): + if update_config_api_url_port(port): + print(f"[OK] Updated AUDIOCPP_API_URL in {CONFIG_PATH}") + else: + print(f"[WARNING] Could not update {CONFIG_PATH}; edit " + "AUDIOCPP_API_URL by hand so audiobook.py uses the " + "new port") + else: + print("[WARNING] Left AUDIOCPP_API_URL unchanged; audiobook.py " + f"will still use port {config_port()}") + + backend = args.backend if args.backend else ask_backend() + lazy_load = args.lazy_load or ask_bool( + "Load models lazily (on first use instead of at startup)", False) + + custom_voice_path = base_path = None + if include_custom: + custom_voice_path = ask("Path to the Qwen3-TTS CustomVoice GGUF package", + DEFAULT_CUSTOM_VOICE_PATH) + if include_clone: + base_path = ask("Path to the Qwen3-TTS Base GGUF package", + DEFAULT_BASE_PATH) + + 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 (Base) " + "model selected, so voice presets are not used") + elif include_clone: + wav_dir = ask_wav_dir() + + voice_presets: Dict[str, dict] = {} + if wav_dir is not None: + wav_files = find_wav_files(wav_dir) + if wav_files: + voice_presets = build_voice_presets(wav_files, args.whisper_model) + else: + print(f"[WARNING] No .wav files found in {wav_dir}; writing the " + "config without voice presets") + + server_config = build_server_config( + host=host, + port=port, + backend=backend, + lazy_load=lazy_load, + include_custom=include_custom, + include_clone=include_clone, + custom_voice_id=custom_voice_id, + clone_model_id=clone_model_id, + custom_voice_path=custom_voice_path, + base_path=base_path, + voice_presets=voice_presets, + ) + + print("\nGenerated server.json:") + print(json.dumps(server_config, indent=2, ensure_ascii=False)) + if not ask_bool(f"\nWrite this to {args.output}", True): + print("[INFO] Aborted; nothing written") + return 1 + + with args.output.open("w", encoding="utf-8") as handle: + json.dump(server_config, handle, indent=2, ensure_ascii=False) + handle.write("\n") + + print(f"\n[OK] Wrote {args.output} with {len(server_config['models'])} " + f"model(s) and {len(voice_presets)} voice preset(s)") + _print_next_steps(args.output, include_custom, include_clone, voice_presets) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/make_faster_voices_json.py b/tools/make_faster_voices_json.py new file mode 100755 index 0000000..2e00f72 --- /dev/null +++ b/tools/make_faster_voices_json.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +"""Generate a voices.json file for the faster-qwen3-tts server. + +Scans a directory for .wav files, transcribes each with a local Whisper +backend (faster_whisper or whisper), and writes a voices.json + +Usage: + python tools/make_faster_voices_json.py INPUT_DIR [--output PATH] + [--language LANG] + [--whisper-model NAME] [--force] + +The output can be passed to the faster server: + python examples/openai_server.py --voices voices.json --port 8000 +""" + +import argparse +import json +import sys +from pathlib import Path + +# Allow running from any working directory. +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from converter.tts import normalize_language, transcribe_reference_audio + + +def find_wav_files(input_dir: Path) -> list: + """Return the .wav files in INPUT_DIR, sorted alphabetically by name.""" + return sorted( + (path for path in input_dir.iterdir() + if path.is_file() and path.suffix.lower() == ".wav"), + key=lambda path: path.name.lower(), + ) + + +def prompt_overwrite(output_path: Path) -> bool: + """Ask whether to overwrite an existing output file.""" + while True: + try: + answer = input(f"{output_path} already exists. Overwrite? (y/n): ").strip().lower() + except EOFError: + print("\n[WARNING] No interactive input available; keeping existing file") + return False + if answer in ("y", "yes"): + return True + if answer in ("n", "no"): + return False + print("Please answer 'y' or 'n'.") + + +def build_voices(wav_files: list, language: str, whisper_model: str) -> dict: + """Transcribe each wav file and build the voices mapping.""" + voices = {} + for wav_file in wav_files: + name = wav_file.stem + print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...") + text = transcribe_reference_audio(str(wav_file), model_name=whisper_model) + if text: + print(f"[OK] {name}: {text}") + else: + print(f"[WARNING] No transcript for '{name}'; the faster backend " + "strongly recommends an accurate transcript — consider editing " + "voices.json by hand before starting the server") + voices[name] = { + "ref_audio": str(wav_file.resolve()), + "ref_text": text or "", + "language": language, + } + return voices + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Generate a voices.json for the faster-qwen3-tts server " + "from a directory of .wav reference files.") + parser.add_argument("input_dir", type=Path, + help="Directory containing .wav reference audio files") + parser.add_argument("--output", type=Path, default=None, + help="Output path for voices.json " + "(default: INPUT_DIR/voices.json)") + parser.add_argument("--language", type=str, default="English", + help="Language for all voices, as passed to the TTS model " + "(default: English; names and short codes accepted)") + parser.add_argument("--whisper-model", type=str, default="base", + help="Whisper model size for transcription " + "(default: base)") + parser.add_argument("--force", action="store_true", + help="Overwrite the output file without prompting") + args = parser.parse_args() + + try: + language = normalize_language(args.language) + except ValueError as exc: + parser.error(str(exc)) + + if not args.input_dir.is_dir(): + parser.error(f"Input directory not found: {args.input_dir}") + + wav_files = find_wav_files(args.input_dir) + if not wav_files: + parser.error(f"No .wav files found in {args.input_dir}") + + output_path = args.output if args.output is not None \ + else args.input_dir / "voices.json" + if output_path.exists() and not args.force and not prompt_overwrite(output_path): + print("[INFO] Aborted; existing voices.json kept") + return 1 + + voices = build_voices(wav_files, language, args.whisper_model) + + with output_path.open("w", encoding="utf-8") as handle: + json.dump(voices, handle, indent=4, ensure_ascii=False) + handle.write("\n") + + print(f"[OK] Wrote {output_path} with {len(voices)} voice(s): " + f"{', '.join(voices)}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/make_voices.py b/tools/make_voices.py deleted file mode 100755 index 913c454..0000000 --- a/tools/make_voices.py +++ /dev/null @@ -1,120 +0,0 @@ -#!/usr/bin/env python3 -"""Generate a voices.json file for the faster-qwen3-tts server. - -Scans a directory for .wav files, transcribes each with a local Whisper -backend (faster_whisper or whisper), and writes a voices.json - -Usage: - python tools/make_voices.py INPUT_DIR [--output PATH] [--language LANG] - [--whisper-model NAME] [--force] - -The output can be passed to the faster server: - python examples/openai_server.py --voices voices.json --port 8000 -""" - -import argparse -import json -import sys -from pathlib import Path - -# Allow running from any working directory. -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) - -from converter.tts import normalize_language, transcribe_reference_audio - - -def find_wav_files(input_dir: Path) -> list: - """Return the .wav files in INPUT_DIR, sorted alphabetically by name.""" - return sorted( - (path for path in input_dir.iterdir() - if path.is_file() and path.suffix.lower() == ".wav"), - key=lambda path: path.name.lower(), - ) - - -def prompt_overwrite(output_path: Path) -> bool: - """Ask whether to overwrite an existing output file.""" - while True: - try: - answer = input(f"{output_path} already exists. Overwrite? (y/n): ").strip().lower() - except EOFError: - print("\n[WARNING] No interactive input available; keeping existing file") - return False - if answer in ("y", "yes"): - return True - if answer in ("n", "no"): - return False - print("Please answer 'y' or 'n'.") - - -def build_voices(wav_files: list, language: str, whisper_model: str) -> dict: - """Transcribe each wav file and build the voices mapping.""" - voices = {} - for wav_file in wav_files: - name = wav_file.stem - print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...") - text = transcribe_reference_audio(str(wav_file), model_name=whisper_model) - if text: - print(f"[OK] {name}: {text}") - else: - print(f"[WARNING] No transcript for '{name}'; the faster backend " - "strongly recommends an accurate transcript — consider editing " - "voices.json by hand before starting the server") - voices[name] = { - "ref_audio": str(wav_file.resolve()), - "ref_text": text or "", - "language": language, - } - return voices - - -def main() -> int: - parser = argparse.ArgumentParser( - description="Generate a voices.json for the faster-qwen3-tts server " - "from a directory of .wav reference files.") - parser.add_argument("input_dir", type=Path, - help="Directory containing .wav reference audio files") - parser.add_argument("--output", type=Path, default=None, - help="Output path for voices.json " - "(default: INPUT_DIR/voices.json)") - parser.add_argument("--language", type=str, default="English", - help="Language for all voices, as passed to the TTS model " - "(default: English; names and short codes accepted)") - parser.add_argument("--whisper-model", type=str, default="base", - help="Whisper model size for transcription " - "(default: base)") - parser.add_argument("--force", action="store_true", - help="Overwrite the output file without prompting") - args = parser.parse_args() - - try: - language = normalize_language(args.language) - except ValueError as exc: - parser.error(str(exc)) - - if not args.input_dir.is_dir(): - parser.error(f"Input directory not found: {args.input_dir}") - - wav_files = find_wav_files(args.input_dir) - if not wav_files: - parser.error(f"No .wav files found in {args.input_dir}") - - output_path = args.output if args.output is not None \ - else args.input_dir / "voices.json" - if output_path.exists() and not args.force and not prompt_overwrite(output_path): - print("[INFO] Aborted; existing voices.json kept") - return 1 - - voices = build_voices(wav_files, language, args.whisper_model) - - with output_path.open("w", encoding="utf-8") as handle: - json.dump(voices, handle, indent=4, ensure_ascii=False) - handle.write("\n") - - print(f"[OK] Wrote {output_path} with {len(voices)} voice(s): " - f"{', '.join(voices)}") - return 0 - - -if __name__ == "__main__": - sys.exit(main()) -- cgit v1.2.3