feat: support for faster-qwen3-tts backend server

3 files changed, 391 insertions, 2 deletions

diff --git a/tests/test_converter.py b/tests/test_converter.py
index ae307a2..48dc162 100644
--- a/tests/test_converter.py
+++ b/tests/test_converter.py
@@ -114,6 +114,8 @@ class NarratorTagTests(unittest.TestCase):
         converter = AudiobookConverter.__new__(AudiobookConverter)
         converter.voice_mode = voice_mode
         converter.voice_clone_ref_audio = ref_audio
+        converter.faster = False
+        converter.faster_voice = None
         return converter
 
     def test_custom_voice_uses_speaker_display_name(self):
@@ -189,6 +191,8 @@ class RunOverwritePromptTests(unittest.TestCase):
         self.converter = AudiobookConverter.__new__(AudiobookConverter)
         self.converter.voice_mode = config.VOICE_MODE_CUSTOM
         self.converter.voice_clone_ref_audio = None
+        self.converter.faster = False
+        self.converter.faster_voice = None
         self.converter.speed = 1.0
         self.converter.single_file = False
         self.converter.output_format = "mp3"
diff --git a/tests/test_make_voices.py b/tests/test_make_voices.py
new file mode 100644
index 0000000..cf76a57
--- /dev/null
+++ b/tests/test_make_voices.py
@@ -0,0 +1,169 @@
+"""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 b605daa..dfeda6f 100644
--- a/tests/test_tts.py
+++ b/tests/test_tts.py
@@ -1,12 +1,15 @@
-"""Tests for the Qwen TTS client wrapper (language handling and payloads)."""
+"""Tests for the TTS client wrappers (language handling and payloads)."""
 
+import json
 import tempfile
 import unittest
+import wave
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 from converter import config
-from converter.tts import QwenTTSClient, normalize_language
+from converter.converter import AudiobookConverter
+from converter.tts import FasterTTSClient, QwenTTSClient, normalize_language
 
 
 class NormalizeLanguageTests(unittest.TestCase):
@@ -151,5 +154,218 @@ class PayloadLanguageTests(unittest.TestCase):
         self.assertNotIn("max_chunk_chars", kwargs)
 
 
+class FasterTTSClientHealthTests(unittest.TestCase):
+    """Connection behavior of the faster-qwen3-tts client."""
+
+    def _health_response(self, model_loaded=True):
+        response = MagicMock()
+        response.__enter__.return_value = response
+        response.read.return_value = json.dumps(
+            {"status": "ok", "model_loaded": model_loaded}).encode("utf-8")
+        return response
+
+    def test_unreachable_server_raises_with_readme_pointer(self):
+        import urllib.error
+        with patch("converter.tts.urllib.request.urlopen",
+                   side_effect=urllib.error.URLError("Connection refused")):
+            with self.assertRaises(RuntimeError) as ctx:
+                FasterTTSClient()
+        message = str(ctx.exception)
+        self.assertIn("not reachable", message)
+        self.assertIn("README", message)
+
+    def test_model_not_loaded_raises(self):
+        with patch("converter.tts.urllib.request.urlopen",
+                   return_value=self._health_response(model_loaded=False)):
+            with self.assertRaises(RuntimeError) as ctx:
+                FasterTTSClient()
+        self.assertIn("not loaded", str(ctx.exception))
+
+    def test_healthy_server_defaults_from_config(self):
+        with patch("converter.tts.urllib.request.urlopen",
+                   return_value=self._health_response()):
+            client = FasterTTSClient()
+        self.assertEqual(client.voice, config.FASTER_TTS_VOICE)
+        self.assertEqual(client.api_url, config.FASTER_TTS_API_URL.rstrip("/"))
+
+    def test_explicit_voice_and_url_override_config(self):
+        with patch("converter.tts.urllib.request.urlopen",
+                   return_value=self._health_response()):
+            client = FasterTTSClient(voice="narrator", api_url="http://10.0.0.5:9000/")
+        self.assertEqual(client.voice, "narrator")
+        self.assertEqual(client.api_url, "http://10.0.0.5:9000")
+
+
+class FasterTTSClientGenerateTests(unittest.TestCase):
+    """Chunk generation: sub-chunking, WAV output, retries, bookkeeping."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self._chunks = patch.object(config, "CHUNKS_FOLDER", Path(self._tmp.name))
+        self._chunks.start()
+        self._sleep = patch("converter.tts.time.sleep")
+        self._sleep.start()
+
+    def tearDown(self):
+        self._sleep.stop()
+        self._chunks.stop()
+        self._tmp.cleanup()
+
+    def _make_client(self):
+        client = FasterTTSClient.__new__(FasterTTSClient)
+        client.voice = "default"
+        client.api_url = "http://127.0.0.1:8000"
+        return client
+
+    def _read_wav(self, path):
+        with wave.open(str(path), "rb") as wav_file:
+            return (wav_file.getnchannels(), wav_file.getsampwidth(),
+                    wav_file.getframerate(), wav_file.readframes(wav_file.getnframes()))
+
+    def test_generate_chunk_writes_valid_wav(self):
+        client = self._make_client()
+        pcm = b"\x01\x00" * 100
+        with patch.object(client, "_request_pcm", return_value=pcm):
+            result = client.generate_chunk("Hello world.", 1)
+        self.assertIsNotNone(result)
+        path = Path(result)
+        self.assertEqual(path.name, "chunk_0001.wav")
+        channels, sampwidth, framerate, frames = self._read_wav(path)
+        self.assertEqual(channels, 1)
+        self.assertEqual(sampwidth, 2)
+        self.assertEqual(framerate, config.FASTER_TTS_SAMPLE_RATE)
+        self.assertEqual(frames, pcm)
+
+    def test_long_text_is_subchunked_and_concatenated_in_order(self):
+        client = self._make_client()
+        sentences = [" ".join(f"word{i}" for i in range(6)) + "." for _ in range(3)]
+        text = " ".join(sentences)
+        pcm_parts = [b"\x01\x00" * 10, b"\x02\x00" * 20, b"\x03\x00" * 30]
+        with patch.object(config, "FASTER_SUBCHUNK_WORDS", 10), \
+                patch.object(client, "_request_pcm", side_effect=pcm_parts) as mock_pcm:
+            result = client.generate_chunk(text, 1)
+        self.assertEqual(mock_pcm.call_count, 3)
+        _, _, _, frames = self._read_wav(Path(result))
+        self.assertEqual(frames, b"".join(pcm_parts))
+
+    def test_stale_chunk_files_are_removed(self):
+        stale = Path(self._tmp.name) / "chunk_0001.mp3"
+        stale.write_bytes(b"old")
+        client = self._make_client()
+        with patch.object(client, "_request_pcm", return_value=b"\x01\x00"):
+            client.generate_chunk("Hello.", 1)
+        remaining = sorted(path.name for path in Path(self._tmp.name).glob("chunk_0001.*"))
+        self.assertEqual(remaining, ["chunk_0001.wav"])
+
+    def test_transient_failure_is_retried(self):
+        client = self._make_client()
+        pcm = b"\x01\x00" * 10
+        with patch.object(client, "_request_pcm",
+                          side_effect=[RuntimeError("boom"), pcm]) as mock_pcm:
+            result = client.generate_chunk("Hello.", 1)
+        self.assertIsNotNone(result)
+        self.assertEqual(mock_pcm.call_count, 2)
+
+    def test_empty_pcm_response_is_treated_as_failure(self):
+        client = self._make_client()
+        pcm = b"\x01\x00" * 10
+
+        def _response(body):
+            response = MagicMock()
+            response.__enter__.return_value = response
+            response.read.return_value = body
+            return response
+
+        with patch("converter.tts.urllib.request.urlopen",
+                   side_effect=[_response(b""), _response(pcm)]) as mock_urlopen:
+            result = client.generate_chunk("Hello.", 1)
+        self.assertIsNotNone(result)
+        self.assertEqual(mock_urlopen.call_count, 2)
+        _, _, _, frames = self._read_wav(Path(result))
+        self.assertEqual(frames, pcm)
+
+    def test_exhausted_subchunk_retries_fail_the_chunk(self):
+        client = self._make_client()
+        with patch.object(client, "_request_pcm",
+                          side_effect=RuntimeError("down")) as mock_pcm:
+            result = client.generate_chunk("Hello.", 1)
+        self.assertIsNone(result)
+        self.assertEqual(mock_pcm.call_count, config.FASTER_SUBCHUNK_RETRIES)
+
+    def test_empty_text_fails_the_chunk(self):
+        client = self._make_client()
+        with patch.object(client, "_request_pcm") as mock_pcm:
+            result = client.generate_chunk("   ", 1)
+        self.assertIsNone(result)
+        mock_pcm.assert_not_called()
+
+    def test_request_payload_includes_voice_text_and_format(self):
+        client = self._make_client()
+        response = MagicMock()
+        response.__enter__.return_value = response
+        response.read.return_value = b"\x01\x00" * 10
+        with patch("converter.tts.urllib.request.urlopen",
+                   return_value=response) as mock_urlopen:
+            pcm = client._request_pcm("Hello world.")
+        self.assertEqual(pcm, b"\x01\x00" * 10)
+        request = mock_urlopen.call_args[0][0]
+        self.assertEqual(request.full_url, "http://127.0.0.1:8000/v1/audio/speech")
+        payload = json.loads(request.data.decode("utf-8"))
+        self.assertEqual(payload["input"], "Hello world.")
+        self.assertEqual(payload["voice"], "default")
+        self.assertEqual(payload["response_format"], "pcm")
+
+
+class FasterModeWiringTests(unittest.TestCase):
+    """AudiobookConverter wiring for the --faster backend."""
+
+    def test_faster_mode_uses_faster_client_without_reference(self):
+        with patch("converter.converter.FasterTTSClient") as mock_faster, \
+                patch("converter.converter.QwenTTSClient") as mock_qwen:
+            AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE,
+                               faster=True, faster_voice="narrator")
+        mock_faster.assert_called_once_with(voice="narrator")
+        mock_qwen.assert_not_called()
+
+    def test_non_faster_clone_mode_still_requires_reference(self):
+        with patch("converter.converter.QwenTTSClient"):
+            with self.assertRaises(ValueError):
+                AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE)
+
+    def test_faster_mode_still_validates_other_settings(self):
+        with patch("converter.converter.FasterTTSClient"):
+            with self.assertRaises(ValueError):
+                AudiobookConverter(faster=True, speed=0)
+            with self.assertRaises(ValueError):
+                AudiobookConverter(faster=True, language="klingon")
+
+    def _faster_converter(self, faster_voice=None):
+        with patch("converter.converter.FasterTTSClient"):
+            return AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE,
+                                      faster=True, faster_voice=faster_voice)
+
+    def test_narrator_tag_uses_faster_voice_name(self):
+        converter = self._faster_converter(faster_voice="male_richard_poe")
+        self.assertEqual(converter._narrator_tag(), "male_richard_poe")
+
+    def test_narrator_tag_falls_back_to_config_voice(self):
+        converter = self._faster_converter()
+        self.assertEqual(converter._narrator_tag(), config.FASTER_TTS_VOICE)
+
+    def test_banner_and_narrator_work_without_reference_audio(self):
+        converter = self._faster_converter(faster_voice="male_richard_poe")
+        converter._print_banner()  # must not raise (regression: Path(None))
+        self.assertIsNone(converter.voice_clone_ref_audio)
+
+    def test_non_faster_narrator_tag_unchanged(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            ref = Path(tmp) / "ref.wav"
+            ref.write_bytes(b"x")
+            with patch("converter.converter.QwenTTSClient"):
+                converter = AudiobookConverter(voice_mode=config.VOICE_MODE_CLONE,
+                                               voice_clone_ref_audio=str(ref))
+        self.assertEqual(converter._narrator_tag(), "ref")
+
+
 if __name__ == "__main__":
     unittest.main()