From acbd9ff2c91182d96c57ffb57bee6e9b3fcbcbd4 Mon Sep 17 00:00:00 2001 From: historia Date: Wed, 26 Aug 2026 01:43:41 -0400 Subject: refactor: split tts.py into per-backend packages --- app/backends/__init__.py | 2 +- app/backends/audiocpp.py | 2 +- app/backends/faster.py | 2 +- app/backends/probe.py | 2 +- app/backends/qwen.py | 2 +- app/converter/audio.py | 47 +- app/converter/clients/__init__.py | 68 ++ app/converter/clients/audiocpp.py | 716 +++++++++++++++++ app/converter/clients/base.py | 155 ++++ app/converter/clients/faster.py | 123 +++ app/converter/clients/languages.py | 74 ++ app/converter/clients/qwen.py | 322 ++++++++ app/converter/clients/speakers.py | 57 ++ app/converter/clients/transcribe.py | 54 ++ app/converter/converter.py | 21 +- app/converter/tts.py | 1469 ---------------------------------- app/tests/test_audio.py | 19 +- app/tests/test_backends.py | 2 +- app/tests/test_converter.py | 53 +- app/tests/test_converter_progress.py | 31 +- app/tests/test_tts.py | 355 ++++---- app/ui/hub.py | 2 +- 22 files changed, 1858 insertions(+), 1720 deletions(-) create mode 100644 app/converter/clients/__init__.py create mode 100644 app/converter/clients/audiocpp.py create mode 100644 app/converter/clients/base.py create mode 100644 app/converter/clients/faster.py create mode 100644 app/converter/clients/languages.py create mode 100644 app/converter/clients/qwen.py create mode 100644 app/converter/clients/speakers.py create mode 100644 app/converter/clients/transcribe.py delete mode 100644 app/converter/tts.py (limited to 'app') diff --git a/app/backends/__init__.py b/app/backends/__init__.py index 63f0709..f6a1d9b 100644 --- a/app/backends/__init__.py +++ b/app/backends/__init__.py @@ -10,7 +10,7 @@ drives the hub's "Configure backends" menu. The registry is built lazily on the first call to ``get``/``detect_all``/ ``detect`` (not at package import time), because the backend modules pull -in ``converter.tts`` and its third-party dependencies, which are only +in ``converter.clients`` and its third-party dependencies, which are only available inside the managed venv that ``audiobook.py`` bootstraps before importing them. ``backends.envs`` is imported during that bootstrap, so importing this package must stay cheap and dependency-free. diff --git a/app/backends/audiocpp.py b/app/backends/audiocpp.py index 4bcf8fa..3cca94c 100755 --- a/app/backends/audiocpp.py +++ b/app/backends/audiocpp.py @@ -82,7 +82,7 @@ from backends.common import ( wav_dir_preview as _wav_dir_preview, ) from converter import config -from converter.tts import transcribe_reference_audio, whisper_backend_available +from converter.clients import transcribe_reference_audio, whisper_backend_available from ui import taskview, tui DEFAULT_HOST = "127.0.0.1" diff --git a/app/backends/faster.py b/app/backends/faster.py index 50cf61f..cec59a6 100755 --- a/app/backends/faster.py +++ b/app/backends/faster.py @@ -47,7 +47,7 @@ from backends.common import ( normalize_dir_arg, ) from converter import config -from converter.tts import ( +from converter.clients import ( normalize_language, transcribe_reference_audio, whisper_backend_available, diff --git a/app/backends/probe.py b/app/backends/probe.py index ada143a..e996725 100644 --- a/app/backends/probe.py +++ b/app/backends/probe.py @@ -35,7 +35,7 @@ IDENTITY_QWEN_CUSTOM = "qwen-custom" IDENTITY_QWEN_CLONE = "qwen-clone" # Endpoint names the converter resolves for each qwen demo server (see -# converter.tts QwenTTSClient). Mirror them here so identification matches +# converter.clients QwenTTSClient). Mirror them here so identification matches # exactly what the converter would call. _QWEN_CUSTOM_ENDPOINTS = ( "/run_instruct", "/run_custom_voice", "/generate_custom_voice") diff --git a/app/backends/qwen.py b/app/backends/qwen.py index 33e7114..501d929 100644 --- a/app/backends/qwen.py +++ b/app/backends/qwen.py @@ -30,7 +30,7 @@ from backends import ( servers, ) from converter import config -from converter.tts import QWEN3_TTS_SPEAKERS +from converter.clients import QWEN3_TTS_SPEAKERS from ui import taskview, tui QWEN_PIP_PKG = "qwen-tts" diff --git a/app/converter/audio.py b/app/converter/audio.py index eb970ff..90ce5ca 100644 --- a/app/converter/audio.py +++ b/app/converter/audio.py @@ -1,4 +1,9 @@ -"""Audio assembly: combining chunks, speed adjustment, cleanup.""" +"""Audio assembly: combining chunks, speed adjustment, cleanup. + +Every function that touches the run's scratch audio takes its folder as an +explicit CHUNKS_DIR argument — the converter owns the folder constants and +threads them through, so there is no module-global path to mutate. +""" import logging import re @@ -13,8 +18,6 @@ from . import config logger = logging.getLogger(__name__) -CHUNKS_FOLDER = Path(__file__).resolve().parent.parent.parent / "app" / "chunks" - # Tolerance for "is this speed 1.0?" comparisons (banner display, atempo # filter elision); shared by every speed check. SPEED_EPSILON = 1e-6 @@ -362,6 +365,7 @@ def _collect_chunk_files(total_chunks: int, def combine_chunks(total_chunks: int, output_path: Path, chunk_results: Dict[int, Optional[Path]], + *, chunks_dir: Path, speed: float = 1.0, output_format: str = config.AUDIO_FORMAT, intermediate: bool = False, meta: Optional[TrackMeta] = None, @@ -369,14 +373,15 @@ def combine_chunks(total_chunks: int, output_path: Path, """Combine audio chunks into the final audiobook using ffmpeg's concat demuxer. ``chunk_results`` maps chunk numbers to the audio file each chunk produced - (None for failed chunks); failed and missing chunks are skipped. When - ``speed`` differs from 1.0, an additional speed-adjusted copy is written - next to the normal-speed file. ``meta``/``cover`` embed tags and cover - art into the output (skipped for intermediate chapter scratch audio). - Chunks are streamed by ffmpeg, so the whole book is never held in - memory. Set ``intermediate`` for scratch chapter audio on the way to a - larger output (e.g. a chaptered m4b) so save messages don't present it - as the final audiobook. + (None for failed chunks); failed and missing chunks are skipped. The + concat scratch list is written to ``chunks_dir``. When ``speed`` differs + from 1.0, an additional speed-adjusted copy is written next to the + normal-speed file. ``meta``/``cover`` embed tags and cover art into the + output (skipped for intermediate chapter scratch audio). Chunks are + streamed by ffmpeg, so the whole book is never held in memory. Set + ``intermediate`` for scratch chapter audio on the way to a larger output + (e.g. a chaptered m4b) so save messages don't present it as the final + audiobook. """ if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None: logger.error("ffmpeg and ffprobe are required to combine audio chunks (install ffmpeg)") @@ -391,7 +396,7 @@ def combine_chunks(total_chunks: int, output_path: Path, if missing_chunks: logger.warning("Missing chunks: %s", missing_chunks) - concat_list = CHUNKS_FOLDER / "_concat_list.txt" + concat_list = Path(chunks_dir) / "_concat_list.txt" try: with open(concat_list, "w", encoding="utf-8") as list_file: for chunk_file in chunk_files: @@ -463,12 +468,12 @@ def combine_chunks(total_chunks: int, output_path: Path, pass -def cleanup_chunks() -> None: - """Remove temporary chunk and chapter files from the scratch folder.""" +def cleanup_chunks(chunks_dir: Path) -> None: + """Remove temporary chunk and chapter files from the CHUNKS_DIR scratch folder.""" try: chunk_count = 0 for pattern in ("chunk_*", "chapter_*"): - for chunk_file in CHUNKS_FOLDER.glob(pattern): + for chunk_file in Path(chunks_dir).glob(pattern): try: if chunk_file.is_file(): chunk_file.unlink() @@ -532,13 +537,15 @@ def build_ffmetadata(chapters: List[tuple], path: Path) -> None: def combine_chapters_to_m4b(chapter_files: List[Path], titles: List[str], - output_path: Path, speed: float = 1.0, + output_path: Path, *, chunks_dir: Path, + speed: float = 1.0, meta: Optional[TrackMeta] = None, cover: Optional[Path] = None) -> bool: """Concatenate per-chapter audio into a single m4b with embedded chapter markers. Chapter start/end times are derived from each chapter file's duration and - written as ffmpeg chapter metadata. ``meta``/``cover`` embed tags and + written as ffmpeg chapter metadata; the concat list and metadata scratch + files are written to ``chunks_dir``. ``meta``/``cover`` embed tags and cover art. When ``speed`` differs from 1.0, a speed-adjusted copy (with rescaled chapter markers) is written alongside the normal-speed file. """ @@ -550,9 +557,9 @@ def combine_chapters_to_m4b(chapter_files: List[Path], titles: List[str], logger.error("No chapter files provided") return False - concat_list = CHUNKS_FOLDER / "_concat_list.txt" - metadata_file = CHUNKS_FOLDER / "_chapters.txt" - speed_metadata_file = CHUNKS_FOLDER / "_chapters_speed.txt" + concat_list = Path(chunks_dir) / "_concat_list.txt" + metadata_file = Path(chunks_dir) / "_chapters.txt" + speed_metadata_file = Path(chunks_dir) / "_chapters_speed.txt" try: chapters = [] start_ms = 0 diff --git a/app/converter/clients/__init__.py b/app/converter/clients/__init__.py new file mode 100644 index 0000000..e216011 --- /dev/null +++ b/app/converter/clients/__init__.py @@ -0,0 +1,68 @@ +"""TTS client implementations — one module per backend server. + +Public API: the three client classes (QwenTTSClient, FasterTTSClient, +AudioCppTTSClient), the backend/voice-mode vocabulary, and the shared +helpers (normalize_language, speaker tables, whisper transcription) that +the UIs and setup wizards build on. +""" + +# The TTS backends a conversion can use, in Convert-form order. Each has a +# client module here; the backends package mirrors these keys for its +# install/setup wizards. +BACKEND_QWEN = "qwen" +BACKEND_FASTER = "faster" +BACKEND_AUDIOCPP = "audiocpp" +BACKENDS = (BACKEND_AUDIOCPP, BACKEND_QWEN, BACKEND_FASTER) + +from .base import BaseTTSClient, ConversionCancelled, VOICE_MODE_CLONE, \ + VOICE_MODE_CUSTOM, VOICE_MODES, resolve_request_seed +from .languages import LANGUAGE_ISO_CODES, TTS_LANGUAGES, \ + TTS_LANGUAGE_ALIASES, normalize_language +from .speakers import QWEN3_TTS_SPEAKERS, SPEAKER_DISPLAY_NAMES, \ + is_builtin_speaker, speaker_display_name, speaker_display_name_for +from .transcribe import transcribe_reference_audio, whisper_backend_available +from .qwen import CUSTOM_VOICE_MODEL_ID, MODEL_SIZE, QwenTTSClient +from .faster import SAMPLE_RATE, FasterTTSClient +from .audiocpp import ( + AUDIOCPP_DEFAULT_FAMILY_PROFILE, + AUDIOCPP_FAMILY_PROFILES, + AUDIOCPP_FAMILY_QWEN3_TTS, + AUDIOCPP_LANG_DISPLAY, + AUDIOCPP_LANG_ISO, + AUDIOCPP_LANG_OMIT, + AUDIOCPP_SYNTHESIS_TASKS, + AUDIOCPP_TASK_TTS, + AUDIOCPP_TASK_VDES, + AUDIOCPP_VOICE_CLONE, + AUDIOCPP_VOICE_DESIGN, + AUDIOCPP_VOICE_SPEAKER, + AudioCppFamilyProfile, + AudioCppTTSClient, + audiocpp_entry_voice_capability, +) + +__all__ = [ + # vocabulary + "BACKEND_QWEN", "BACKEND_FASTER", "BACKEND_AUDIOCPP", "BACKENDS", + "VOICE_MODE_CUSTOM", "VOICE_MODE_CLONE", "VOICE_MODES", + # clients + "BaseTTSClient", "ConversionCancelled", "resolve_request_seed", + "QwenTTSClient", "FasterTTSClient", "AudioCppTTSClient", + # model facts + "MODEL_SIZE", "CUSTOM_VOICE_MODEL_ID", "SAMPLE_RATE", + # languages + "TTS_LANGUAGES", "TTS_LANGUAGE_ALIASES", "LANGUAGE_ISO_CODES", + "normalize_language", + # speakers + "QWEN3_TTS_SPEAKERS", "SPEAKER_DISPLAY_NAMES", + "speaker_display_name", "speaker_display_name_for", "is_builtin_speaker", + # transcription + "transcribe_reference_audio", "whisper_backend_available", + # audio.cpp family profiles + "AUDIOCPP_LANG_DISPLAY", "AUDIOCPP_LANG_ISO", "AUDIOCPP_LANG_OMIT", + "AUDIOCPP_FAMILY_QWEN3_TTS", "AUDIOCPP_TASK_TTS", "AUDIOCPP_TASK_VDES", + "AUDIOCPP_SYNTHESIS_TASKS", "AUDIOCPP_VOICE_SPEAKER", + "AUDIOCPP_VOICE_CLONE", "AUDIOCPP_VOICE_DESIGN", + "AudioCppFamilyProfile", "AUDIOCPP_DEFAULT_FAMILY_PROFILE", + "AUDIOCPP_FAMILY_PROFILES", "audiocpp_entry_voice_capability", +] diff --git a/app/converter/clients/audiocpp.py b/app/converter/clients/audiocpp.py new file mode 100644 index 0000000..4a161cb --- /dev/null +++ b/app/converter/clients/audiocpp.py @@ -0,0 +1,716 @@ +"""Client for the audio.cpp audiocpp_server (native ggml TTS families).""" + +import json +import logging +import shutil +import tempfile +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path +from typing import Any, Dict, List, Optional + +from .. import config +from ..audio import concat_audio_files +from ..chunking import split_into_chunks +from .base import BaseTTSClient, ConversionCancelled, resolve_request_seed +from .languages import LANGUAGE_ISO_CODES, normalize_language +from .speakers import (is_builtin_speaker, speaker_display_name, + speaker_display_name_for) + +logger = logging.getLogger(__name__) + +# How the "language" request field is expressed by a family. +AUDIOCPP_LANG_DISPLAY = "display" # Qwen display names, e.g. "English" +AUDIOCPP_LANG_ISO = "iso" # ISO 639-1 codes, e.g. "en" +AUDIOCPP_LANG_OMIT = "omit" # no language field; the model detects it + +# The Qwen3-TTS family. Unlike every other family (one model type each), +# qwen3_tts hosts several model *types* under one family id, distinguished +# only by the server entry's id/task: the CustomVoice model (built-in +# speakers, e.g. Vivian/Ryan), the Base model (voice cloning via a +# server-side preset), and the VoiceDesign model (task "vdes"). The +# per-entry voice capability below (audiocpp_entry_voice_capability) +# resolves which is which, driving both the Convert form (which voice +# list to show) and the converter's mode selection. +AUDIOCPP_FAMILY_QWEN3_TTS = "qwen3_tts" + +# Server model entry tasks this client can synthesize audiobooks with, +# taken from GET /v1/models (the "task" field of each entry; a missing task +# is treated as "tts" — a harmless generic default). "vdes" entries are +# voice design models: the voice is described with --instructions instead +# of coming from a speaker or a reference clip. Entries with any other task +# (asr, vc, diar, ...) are rejected at connect time with a hint to pick a +# synthesis entry. +AUDIOCPP_TASK_TTS = "tts" +AUDIOCPP_TASK_VDES = "vdes" +AUDIOCPP_SYNTHESIS_TASKS = (AUDIOCPP_TASK_TTS, "clon", AUDIOCPP_TASK_VDES) + +# The voice capability of a server model entry — how its voice is supplied. +# Resolved per entry from (family, task, id) by +# audiocpp_entry_voice_capability; drives both the Convert form (which +# voice list to show) and the converter (speaker vs preset vs design mode). +# Most families are clone-only; only the Qwen3-TTS CustomVoice entry has +# built-in speakers, and only VoiceDesign entries take a description. +AUDIOCPP_VOICE_SPEAKER = "speaker" # built-in speaker name (Qwen CustomVoice) +AUDIOCPP_VOICE_CLONE = "clone" # server-side preset / voice_dir (Base, others) +AUDIOCPP_VOICE_DESIGN = "design" # voice described by --instructions (vdes) + + +class AudioCppFamilyProfile: + """Request conventions of one audio.cpp model family. + + Language style and whether the family reads a style/instruction prompt; + these are family-level (every entry of a family shares them). Whether a + *specific entry* has built-in speakers is an entry-level concern, decided + by audiocpp_entry_voice_capability, not this profile. + """ + + def __init__(self, language_style: str = AUDIOCPP_LANG_OMIT, + sends_instructions: bool = False): + self.language_style = language_style + self.sends_instructions = sends_instructions + + +# Generic profile for families not listed in AUDIOCPP_FAMILY_PROFILES: +# clone-only, no style instructions, and no language field (the model +# detects the language itself). Describes higgs_audio_tts, voxcpm2, +# fish_audio, dots_tts, dramabox, omnivoice, outetts, glm_tts, miotts, +# moss_tts_*, pocket_tts, vibevoice, ... as well as families added to +# audio.cpp after this table was written. +AUDIOCPP_DEFAULT_FAMILY_PROFILE = AudioCppFamilyProfile() + +AUDIOCPP_FAMILY_PROFILES = { + AUDIOCPP_FAMILY_QWEN3_TTS: AudioCppFamilyProfile( + language_style=AUDIOCPP_LANG_DISPLAY, + sends_instructions=True, + ), + # Families whose language option takes a code (e.g. "en") instead of + # a Qwen display name; otherwise clone-only like the default profile. + "chatterbox": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), + "confucius4_tts": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), + "index_tts2": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), + "magpie_tts": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), + "supertonic": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), +} + + +def audiocpp_entry_voice_capability(family: str, task: str, + model_id: str) -> str: + """How a server model entry's voice is supplied — speaker/clone/design. + + Resolved from the entry's family, task and id — the same {id, family, + task} triple GET /v1/models reports, so it works for local server.json + entries and remote live-queried entries alike. Qwen3-TTS is the one + family hosting several model *types* under one family id: the + CustomVoice model (id contains "customvoice") has built-in speakers, the + Base model and any other entry are clone-only, and VoiceDesign entries + (task "vdes") take a description. Every other family is clone-only. + """ + if task == AUDIOCPP_TASK_VDES: + return AUDIOCPP_VOICE_DESIGN + if family == AUDIOCPP_FAMILY_QWEN3_TTS \ + and "customvoice" in (model_id or "").lower(): + return AUDIOCPP_VOICE_SPEAKER + return AUDIOCPP_VOICE_CLONE + + +class AudioCppTTSClient(BaseTTSClient): + """Generates audio chunks through an audio.cpp audiocpp_server. + + Talks to the OpenAI-style HTTP API of audiocpp_server, which hosts TTS + model families through a native ggml runtime (GGUF weights, no Python + serving stack). The server API is family-agnostic; the family and task + of the configured model entry are read from GET /v1/models at startup + and adapt the request payload (language field style, style instructions) + through AUDIOCPP_FAMILY_PROFILES. The entry's voice capability + (audiocpp_entry_voice_capability: speaker / clone / design) decides how + its voice is supplied; all three are resolved server-side from the + request's "voice"/"instructions" fields: + + - Speaker mode (--voice with a built-in speaker name, or no flag on a + CustomVoice entry): Qwen3-TTS CustomVoice only. A built-in speaker + name (e.g. "Vivian") is passed through, plus the INSTRUCT style + prompt. The selected entry must be the CustomVoice model (capability + == speaker); a speaker name on a non-speaker entry is treated as a + server-side preset instead. + - Preset mode (--voice NAME): a voice configured on the server + (``voice_presets`` or ``voice_dir`` in its config, e.g. a cloning + reference). The name is validated against GET /v1/audio/voices at + startup because an unresolvable name would silently fall back to + plain TTS on a clone-based model instead of failing. When + AUDIOCPP_CLONE_MODEL_ID names a second server entry of the same + family (typically the Qwen Base model), preset requests are routed + to it. Selecting a non-speaker --voice on a CustomVoice primary with + a clone id configured is the documented way to switch a speaker setup + to cloning; without a clone id the voice is validated against the + server's voice library. + - Voice design (task "vdes" entries, e.g. Qwen3-TTS VoiceDesign): the + voice is described in natural language through ``instructions``, + which is required and sent with every request (no ``voice`` field). + A constant per-run seed keeps the designed voice consistent across + chunk boundaries. + + The entry's capability decides how an explicit --voice is read: on a + speaker-capable entry a name that matches a built-in speaker selects + speaker mode, and every other name is a server-side preset. With no + --voice the entry's capability picks the mode: design entries require + --instructions; speaker entries use the built-in CustomVoice speaker + in config.SPEAKER; clone entries (the Base model, and every other + family) fail fast with a hint to pass --voice, instead of silently + synthesizing with a random default voice. + + ``instructions`` also works on non-design entries, where it acts as a + generic style/delivery instruction (voice control): families that read + it (OmniVoice, Qwen3-TTS CustomVoice, ...) shape the voice or delivery + accordingly, and others ignore it. On instruction-conditioned families + without built-in speakers it may replace --voice entirely (the + instruction defines the voice). Extra request options (``--option + KEY=VALUE``, e.g. emotion, voice_id, speed) are forwarded verbatim in + the request's "options" object, which is the server's generic + pass-through for per-model controls. + + Chunking: text is split client-side into sub-requests of at most + config.CHUNK_SIZE words each; each sub-request returns a complete + WAV file and the parts are concatenated with the same lossless path + used for the Qwen client. + """ + + def __init__(self, chunks_dir: Path, + voice: Optional[str] = None, language: Optional[str] = None, + api_url: Optional[str] = None, + model_id: Optional[str] = None, + instructions: Optional[str] = None, + request_options: Optional[Dict[str, str]] = None, + quiet: bool = False): + super().__init__(chunks_dir, quiet=quiet) + self.api_url = (api_url or config.AUDIOCPP_API_URL).rstrip("/") + # Per-run model selection: the --model CLI flag overrides config; an + # empty value is resolved at connect time when the server hosts exactly + # one entry, so multi-model servers don't require editing config.py. + self.model_id = (model_id if model_id is not None + else config.AUDIOCPP_MODEL_ID) or "" + self._model_id_explicit = bool(self.model_id) + # Validate before connecting so bad values fail fast without a server. + self.language = normalize_language( + language if language is not None else config.LANGUAGE) + # One seed value per run, reused for every request (see + # resolve_request_seed). Unlike the Qwen demo, audio.cpp has no + # negative "randomize" seed, so a negative value means "send no seed + # at all" (see _request_wav) and the server randomizes. + self._seed = resolve_request_seed() + # Voice selection (the --voice name). preset_mode / speaker_mode are + # resolved in _connect: a --voice that names a built-in CustomVoice + # speaker on a speaker-capable entry selects speaker mode; every + # other name (and any name on a clone-capable entry) is a server-side + # preset. preset_mode gates _select_model's reroute to + # AUDIOCPP_CLONE_MODEL_ID and the INSTRUCT style-prompt logic. The + # request's "voice" field (self.voice) is filled in _connect per the + # mode. + self.preset_mode = False + self.speaker_mode = False + self.voice = voice or None + # Style/voice-design instruction sent with every request (the CLI + # --instructions flag overrides AUDIOCPP_INSTRUCTIONS in config.py). + # For task "vdes" entries it describes the voice to design; for other + # families it is a generic style instruction when the model reads one. + self.instructions = (instructions if instructions is not None + else config.AUDIOCPP_INSTRUCTIONS or "").strip() + # Free-form per-request options (--option KEY=VALUE) forwarded in the + # request's "options" object; models ignore keys they don't know. + self.request_options: Dict[str, str] = dict(request_options or {}) + # Set during _connect: design_mode for "vdes" entries, instruction_voice + # when a family without built-in speakers gets its voice from the + # instruction alone (no voice field). self.voice is also finalized + # there (the speaker/preset name, or config.SPEAKER for the default). + self.design_mode = False + self.instruction_voice = False + # Family and task of the selected model entry and the family's request + # profile; all are resolved from GET /v1/models during _connect. + self.family = "" + self.task = AUDIOCPP_TASK_TTS + self.profile = AUDIOCPP_DEFAULT_FAMILY_PROFILE + self._connect() + + # ------------------------------------------------------------------ + # Connection + # ------------------------------------------------------------------ + + def _connected(self, mode: str) -> None: + """Report the resolved connection (MODE: speaker/voice/... label).""" + self._report(f"[OK] Connected to audio.cpp server at {self.api_url} " + f"(model '{self.model_id}', family '{self.family}', " + f"{mode})") + + def _connect(self) -> None: + """Health-check the server and resolve the model, family, task, and voice. + + The entry's voice capability (audiocpp_entry_voice_capability, from + family/task/id) plus the caller's --voice/--instructions pick the + mode. An explicit --voice on a speaker-capable (CustomVoice) entry + that names a built-in speaker selects speaker mode; every other + --voice is a server-side preset, validated against the server's + voice library (and rerouted to AUDIOCPP_CLONE_MODEL_ID when set). + With no --voice, design entries require --instructions, speaker- + capable entries use the built-in config.SPEAKER, and clone entries + fail fast with a hint instead of silently synthesizing with a + random default voice. + """ + self._check_health() + models = self._list_models() + self._auto_pick_model_id(models) + if self.voice is not None: + # Explicit --voice: decide between speaker mode and a server-side + # preset. A name matching a built-in CustomVoice speaker on a + # speaker-capable primary selects speaker mode; every other name + # (and any name when the primary entry is absent) is a preset, + # validated against the server's voice library and rerouted to + # AUDIOCPP_CLONE_MODEL_ID when configured. + primary = next((m for m in models if m["id"] == self.model_id), + None) + if primary is not None: + self._resolve_family(models) + self._resolve_task(models) + capability = audiocpp_entry_voice_capability( + self.family, self.task, self.model_id) + if capability == AUDIOCPP_VOICE_SPEAKER \ + and is_builtin_speaker(self.voice): + self._require_synthesis_task(models) + self.voice = speaker_display_name_for(self.voice) + self.speaker_mode = True + self._connected(f"speaker '{self.voice}'") + if not self.speaker_mode: + # Server-side preset (--voice): validate it and route to + # the clone model entry when AUDIOCPP_CLONE_MODEL_ID is set. + self.preset_mode = True + self._select_model(models) + self._require_model_id(models) + self._resolve_family(models) + self._resolve_task(models) + self._require_synthesis_task(models) + if self.design_mode: + raise RuntimeError( + f"--voice cannot be used with the voice design model " + f"'{self.model_id}': the voice is described by the " + "--instructions text instead (see README).") + self._check_voice() + self._connected(f"voice '{self.voice}'") + else: + # No flag: the entry's capability picks the default mode. + self._require_model_id(models) + self._resolve_family(models) + self._resolve_task(models) + self._require_synthesis_task(models) + capability = audiocpp_entry_voice_capability( + self.family, self.task, self.model_id) + if self.design_mode: + if not self.instructions: + raise RuntimeError( + f"The audio.cpp model '{self.model_id}' (family " + f"'{self.family}') is a voice design model: pass a " + "description of the voice to synthesize with, e.g. " + '--instructions "A warm adult female narrator with a ' + 'British accent" (see README).') + self._connected("voice design") + self._report(f"[INFO] Designing the voice from: {self.instructions}") + elif capability == AUDIOCPP_VOICE_SPEAKER: + # No flag on a CustomVoice entry: the built-in config.SPEAKER. + self.voice = speaker_display_name() + self.speaker_mode = True + self._connected(f"speaker '{self.voice}'") + elif self.instructions: + # Families without built-in speakers can still get their voice + # from the instruction alone (e.g. OmniVoice voice design). + self.instruction_voice = True + self._connected("instruction voice") + self._report(f"[INFO] Designing the voice from: {self.instructions}") + else: + raise RuntimeError( + f"The audio.cpp model '{self.model_id}' (family " + f"'{self.family}') has no built-in speakers, so its voice " + "must come from the server: rerun with --voice NAME " + "matching a voice_preset or voice_dir entry in the server " + "config, or describe a voice with --instructions for " + "families that support it, or select the CustomVoice entry " + "for built-in speakers (see README).") + if self.instructions and not self.design_mode and not self.instruction_voice: + self._report(f"[INFO] Sending instruction with every request: {self.instructions}") + self._report("[INFO] Its effect (style, emotion, delivery) depends on the " + "model family; models without instruction support ignore it.") + if config.AUDIOCPP_UNLOAD_MODELS: + self._unload_server_models() + + def _require_synthesis_task(self, models: List[Dict[str, str]]) -> None: + """Reject model entries whose task is not a TTS synthesis task.""" + if self.task in AUDIOCPP_SYNTHESIS_TASKS: + return + available = ", ".join(model["id"] for model in models) or "none" + raise RuntimeError( + f"The audio.cpp model '{self.model_id}' has task " + f"'{self.task}'; audiobook.py can only synthesize with TTS " + f"model entries (tasks {', '.join(AUDIOCPP_SYNTHESIS_TASKS)}). " + f"Pick a synthesis entry with --model (available: {available})." + ) + + def _unload_server_models(self) -> None: + """Ask the server to unload every loaded model before generating. + + Lazy-loaded entries stay resident until the server exits (unless its + max_loaded_models setting bounds residency), so switching between + configured models across runs can exhaust device memory. Unloading + first frees those leftovers; this run's model reloads transparently + on its first request. Failures only warn: an older server without + the endpoint, or a busy one, must not block a working setup. + Controlled by config.AUDIOCPP_UNLOAD_MODELS (the TUI Settings + "Unload models" option). + """ + request = urllib.request.Request( + f"{self.api_url}/v1/tasks/unload_all_models", data=b"", + method="POST", headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(request, timeout=10) as response: + payload = json.loads(response.read().decode("utf-8")) + except Exception as exc: + self._report(f"[WARNING] Could not unload previously loaded models at " + f"{self.api_url}: {exc}") + return + unloaded = [entry for entry in (payload.get("unloaded") or []) + if isinstance(entry, str)] + if unloaded: + self._report(f"[OK] Unloaded {len(unloaded)} model(s) from server memory: " + f"{', '.join(unloaded)}") + else: + logger.debug("No loaded audio.cpp models to unload at %s", self.api_url) + + def _get_json(self, path: str, timeout: int = 10) -> Dict[str, Any]: + """GET a JSON document from the server.""" + url = f"{self.api_url}{path}" + try: + with urllib.request.urlopen(url, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + detail = "" + try: + detail = exc.read().decode("utf-8", errors="replace")[:200] + except Exception: + pass + raise RuntimeError( + f"audio.cpp server returned HTTP {exc.code} for {path}: {detail}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"audio.cpp request failed for {path}: {exc.reason}") from exc + + def _check_health(self) -> None: + """Verify the server is reachable and reports healthy.""" + try: + payload = self._get_json("/health") + except Exception as exc: + raise RuntimeError( + f"audio.cpp server not reachable at {self.api_url}: {exc}. " + "Start audiocpp_server first (see the 'audio.cpp backend' " + "section of the README)." + ) from exc + if payload.get("status") != "ok": + raise RuntimeError( + f"The audio.cpp server at {self.api_url} reports status " + f"{payload.get('status')!r} instead of 'ok'") + + def _list_models(self) -> List[Dict[str, str]]: + """Fetch the (id, family, task) triples reported by the server.""" + try: + payload = self._get_json("/v1/models") + except Exception as exc: + raise RuntimeError( + f"The audio.cpp server at {self.api_url} did not answer " + f"/v1/models: {exc}") from exc + entries = payload.get("data") or [] + models: List[Dict[str, str]] = [] + for entry in entries: + if isinstance(entry, dict) and entry.get("id"): + models.append({ + "id": entry["id"], + "family": entry.get("family") or "", + "task": entry.get("task") or "", + }) + return models + + def _auto_pick_model_id(self, models: List[Dict[str, str]]) -> None: + """Resolve an empty model id when the server hosts exactly one entry. + + Multi-model servers generated with several lazily-loaded entries can + be used without editing app/converter/config.py: leave AUDIOCPP_MODEL_ID + (and ``--model``) unset, and the single hosted entry is chosen + automatically. With more than one entry an explicit choice is required + (via ``--model`` or AUDIOCPP_MODEL_ID), since guessing would risk + synthesizing a whole book with the wrong family. + """ + if self.model_id: + return + if len(models) == 1: + self.model_id = models[0]["id"] + logger.info( + "AUDIOCPP_MODEL_ID is unset; using the only server entry '%s'", + self.model_id) + else: + logger.debug( + "AUDIOCPP_MODEL_ID is unset and the server hosts %d entries; " + "an explicit --model or config id is required", + len(models)) + + def _require_model_id(self, models: List[Dict[str, str]]) -> None: + """Verify the model id chosen for this run exists on the server. + + Speaker mode needs AUDIOCPP_MODEL_ID (the CustomVoice entry). + Preset mode validates whichever id _select_model resolved, so a + server hosting only a cloning model works for --voice. The default + error distinguishes the two so the fix is obvious. + """ + model_ids = [model["id"] for model in models] + if self.model_id and self.model_id in model_ids: + return + configured = ", ".join(model_ids) or "none" + if not self.model_id: + raise RuntimeError( + f"The audio.cpp server at {self.api_url} hosts {len(model_ids)} " + f"model entries ({configured}); audiobook.py needs to know which " + "one to use. Pass --model when converting, or set " + "AUDIOCPP_MODEL_ID in app/converter/config.py to one of them " + "(see README)." + ) + if self.preset_mode: + raise RuntimeError( + f"The audio.cpp server at {self.api_url} has no model id " + f"'{self.model_id}' or clone model id " + f"'{config.AUDIOCPP_CLONE_MODEL_ID}' (configured: {configured}). " + "Add a TTS model entry for the family you want to the server " + "config and match AUDIOCPP_MODEL_ID / AUDIOCPP_CLONE_MODEL_ID " + "in app/converter/config.py to its id, or select it per run with " + "--model (see README)." + ) + raise RuntimeError( + f"The audio.cpp server at {self.api_url} has no model id " + f"'{self.model_id}' (configured: {configured}). Select the " + "Qwen3-TTS CustomVoice entry for built-in speakers, or rerun " + "with --voice NAME matching a voice_preset or voice_dir entry " + "on any TTS model (see README)." + ) + + def _select_model(self, models: List[Dict[str, str]]) -> None: + """Pick the model for preset (cloning) requests. + + Defaults to the primary model id. When AUDIOCPP_CLONE_MODEL_ID is + configured and present on the server, preset requests are routed + to it instead, so one server can host the CustomVoice model for + speaker mode and the Base model for cloning (Qwen3-TTS setups). + A clone id that names a model of a different family is ignored + with a warning, since preset requests must synthesize with the + family the run is configured for. + """ + clone_model_id = config.AUDIOCPP_CLONE_MODEL_ID + if not clone_model_id or clone_model_id == self.model_id: + return + families = {model["id"]: model["family"] for model in models} + if clone_model_id not in families: + # A qwen3_tts primary without its clone entry silently degrades + # (presets are ignored on the CustomVoice model), so that case + # keeps the warning; single-model servers of other families are + # the normal configuration and only get a debug note. + primary_family = families.get(self.model_id) or "" + if primary_family == AUDIOCPP_FAMILY_QWEN3_TTS: + logger.warning( + "AUDIOCPP_CLONE_MODEL_ID %r is not configured on the audio.cpp " + "server; preset requests use '%s' instead", + clone_model_id, self.model_id) + else: + logger.debug( + "AUDIOCPP_CLONE_MODEL_ID %r is not configured on the audio.cpp " + "server; preset requests use '%s' instead", + clone_model_id, self.model_id) + return + primary_family = families.get(self.model_id) + clone_family = families[clone_model_id] + if primary_family and clone_family and primary_family != clone_family: + logger.warning( + "AUDIOCPP_CLONE_MODEL_ID %r hosts family %r, but " + "AUDIOCPP_MODEL_ID %r hosts %r; preset requests stay on " + "'%s'. Point both ids at the same model entry in " + "app/converter/config.py (single-model servers use the same id " + "for both)", + clone_model_id, clone_family, self.model_id, primary_family, + self.model_id) + return + self.model_id = clone_model_id + + def _resolve_family(self, models: List[Dict[str, str]]) -> None: + """Resolve the selected model's family and its request profile. + + The family comes from GET /v1/models; a missing family is an unknown + family that falls through to the generic (clone-only) profile rather + than guessing a specific one — audiocpp_server always reports family + for entries its server.json describes. + """ + entry = next( + (model for model in models if model["id"] == self.model_id), None) + family = (entry["family"] if entry is not None else "") or "" + self.family = family + self.profile = AUDIOCPP_FAMILY_PROFILES.get( + family, AUDIOCPP_DEFAULT_FAMILY_PROFILE) + if not family: + logger.debug("Model '%s' reported no family; using the generic " + "profile", self.model_id) + elif family not in AUDIOCPP_FAMILY_PROFILES: + logger.info( + "audio.cpp family '%s' has no dedicated profile; using the " + "generic profile (voice cloning via --voice, model-detected " + "language)", family) + + def _resolve_task(self, models: List[Dict[str, str]]) -> None: + """Resolve the selected model's task (tts, clon, vdes, ...) and set + design mode for voice design entries. + + The task comes from GET /v1/models and is fixed per server entry by + its server.json config (a VoiceDesign model must be hosted with + "task": "vdes"). Servers that predate the task field hosted plain + TTS models, so a missing task is treated as tts. + """ + entry = next( + (model for model in models if model["id"] == self.model_id), None) + task = (entry["task"] if entry is not None else "") or "" + if not task: + task = AUDIOCPP_TASK_TTS + logger.debug("Model '%s' reported no task; assuming tts", + self.model_id) + self.task = task + self.design_mode = task == AUDIOCPP_TASK_VDES + + def _check_voice(self) -> None: + """Verify the requested voice is available on the server. + + A voice name that matches no server preset or voice-library wav + would be passed through to the model as a cached voice id; on the + Base (cloning) model that is silently ignored and plain TTS audio + comes back, so preset names are validated up front. When the + voices endpoint cannot be queried, validation is skipped with a + warning rather than blocking the run. + """ + query = urllib.parse.urlencode({"model": self.model_id}) + try: + payload = self._get_json(f"/v1/audio/voices?{query}") + except Exception as exc: + logger.warning("Could not list server voices; skipping voice " + "validation: %s", exc) + return + voices = payload.get("voices") or [] + if self.voice not in voices: + available = ", ".join(str(v) for v in voices) or "none" + raise RuntimeError( + f"Voice '{self.voice}' is not available on the audio.cpp server " + f"(available: {available}). Configure it as a voice_preset or " + "voice_dir entry in the server config, or pass a listed name " + "with --voice (see README)." + ) + + # ------------------------------------------------------------------ + # HTTP requests + # ------------------------------------------------------------------ + + def _request_wav(self, text: str) -> bytes: + """POST one sub-chunk and return the raw WAV bytes.""" + url = f"{self.api_url}/v1/audio/speech" + payload: Dict[str, Any] = { + "model": self.model_id, + "input": text, + } + # Design models take no voice field (the voice comes from the + # instruction); instruction-voice runs on families without built-in + # speakers omit it too, since no speaker or preset was requested. + if not self.design_mode and not self.instruction_voice: + payload["voice"] = self.voice + if self.profile.language_style == AUDIOCPP_LANG_DISPLAY: + payload["language"] = self.language + elif self.profile.language_style == AUDIOCPP_LANG_ISO: + iso_code = LANGUAGE_ISO_CODES.get(self.language) + if iso_code: + payload["language"] = iso_code + else: + # "Auto": no code to send, so let the server pick its default. + logger.debug("%s: no language code for %r; omitted from request", + self.family, self.language) + if self._seed >= 0: + # audio.cpp has no negative "randomize" seed; a negative seed + # means "let the server randomize", so the field is omitted. + payload["seed"] = self._seed + if self.instructions: + # Explicit voice-design or style instruction (required for task + # "vdes" entries; a Ctrl/style control on families that read it). + payload["instructions"] = self.instructions + elif not self.preset_mode and config.INSTRUCT \ + and self.profile.sends_instructions: + # Style instruction for the Qwen3-TTS CustomVoice speakers; + # ignored by the Base (cloning) model and other families. + payload["instructions"] = config.INSTRUCT + if self.request_options: + # Generic per-model controls (--option KEY=VALUE): forwarded + # verbatim; the model ignores keys it does not know. + payload["options"] = dict(self.request_options) + request = urllib.request.Request( + url, data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, method="POST") + timeout = config.API_TIMEOUT + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + wav = response.read() + except urllib.error.HTTPError as exc: + detail = "" + try: + detail = exc.read().decode("utf-8", errors="replace")[:200] + except Exception: + pass + raise RuntimeError(f"audio.cpp server returned HTTP {exc.code}: {detail}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"audio.cpp request failed: {exc.reason}") from exc + if len(wav) < 12 or wav[:4] != b"RIFF" or wav[8:12] != b"WAVE": + raise RuntimeError("audio.cpp server returned audio that is not a WAV file") + return wav + + # ------------------------------------------------------------------ + # Chunk generation + # ------------------------------------------------------------------ + + def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: + """Generate one audio chunk; returns its path in the chunks folder. + + The text is split into sub-requests of at most ``config.CHUNK_SIZE`` + words each; 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=config.CHUNK_SIZE) + if not sub_texts: + raise RuntimeError("No text to synthesize") + + output_path: Optional[Path] = None + with tempfile.TemporaryDirectory(prefix="tts_parts_") as parts_dir, \ + self._chunk_heartbeat(chunk_num): + part_paths = [] + for sub_num, sub_text in enumerate(sub_texts, 1): + wav = self._request_wav(sub_text) + destination = Path(parts_dir) / f"part_{sub_num:02d}.wav" + destination.write_bytes(wav) + part_paths.append(destination) + if len(part_paths) == 1: + output_path = self._chunk_path(chunk_num, ".wav") + shutil.copy2(part_paths[0], output_path) + else: + output_path = self._chunk_path(chunk_num, ".wav") + concat_audio_files(part_paths, output_path) + + logger.debug("Chunk %d generated successfully (%d sub-request(s))", + chunk_num, len(sub_texts)) + return str(output_path) + + except ConversionCancelled: + raise + except Exception as exc: + logger.error("audio.cpp chunk processing failed for chunk %d: %s", + chunk_num, exc) + return None diff --git a/app/converter/clients/base.py b/app/converter/clients/base.py new file mode 100644 index 0000000..a0f28cf --- /dev/null +++ b/app/converter/clients/base.py @@ -0,0 +1,155 @@ +"""Shared TTS client plumbing: cancellation, retries, chunk bookkeeping.""" + +import contextlib +import logging +import random +import threading +import time +from pathlib import Path +from typing import Optional + +from .. import config + +logger = logging.getLogger(__name__) + + +class ConversionCancelled(Exception): + """Raised when the run's cancel event is set (between requests).""" + + +# How a run supplies its voice: a built-in CustomVoice speaker, or by +# cloning a reference audio clip (the faster and audiocpp backends always +# clone server-side; only the Qwen client branches on this at request time). +VOICE_MODE_CUSTOM = "custom_voice" +VOICE_MODE_CLONE = "voice_clone" +VOICE_MODES = (VOICE_MODE_CUSTOM, VOICE_MODE_CLONE) + + +def resolve_request_seed() -> int: + """Resolve the seed sent with every request. + + Returns config.SEED as-is, or (with CONSTANT_SEED and SEED < 0) one + random value drawn per run, meant to be reused for every request so + the voice stays consistent across chunk boundaries. Without + CONSTANT_SEED, -1 is returned so the server re-samples the voice on + every generation. + """ + seed = config.SEED + if config.CONSTANT_SEED and seed < 0: + seed = random.randrange(2 ** 31) + return seed + + +class BaseTTSClient: + """Shared chunk retry logic, heartbeat, and chunk file bookkeeping. + + CHUNKS_DIR is the scratch folder the generated chunk files are written + to — provided by the converter that owns the run's folders, never a + module global, so concurrent runs (and tests) cannot step on each other. + """ + + # Class-level defaults so a partially-constructed instance behaves like + # a plain console run (tests build clients via __new__). + cancel = None + quiet = False + + def __init__(self, chunks_dir: Path, quiet: bool = False): + self.chunks_dir = Path(chunks_dir) + # Quiet silences console prints (the run view owns the screen). + self.quiet = bool(quiet) + # Set by the converter when the run is cancellable (the TUI run + # view): a threading.Event that, once set, aborts the run between + # requests (and interrupts retry back-off sleeps). + self.cancel = None + + def _report(self, message: str) -> None: + """Print a console line unless quiet (the run view owns the screen).""" + if not self.quiet: + print(message) + + def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: + """Generate one audio chunk; returns its path in the chunks folder.""" + raise NotImplementedError + + def _cancel_requested(self) -> bool: + """True when the run's cancel event has been set (if any).""" + return isinstance(self.cancel, threading.Event) \ + and self.cancel.is_set() + + def _check_cancelled(self) -> None: + """Raise ConversionCancelled when the cancel event is set.""" + if self._cancel_requested(): + raise ConversionCancelled("Cancelled by user") + + def _sleep(self, seconds: float) -> None: + """Sleep SECONDS, cut short (raising) when the cancel event sets.""" + if isinstance(self.cancel, threading.Event): + if self.cancel.wait(seconds): + raise ConversionCancelled("Cancelled by user") + else: + time.sleep(seconds) + + def _chunk_path(self, chunk_num: int, suffix: str) -> Path: + """Resolve the target path for a chunk, removing stale files first. + + Any stale chunk file for this index is removed so a retry or extension + change can never leave two files matching chunk_NNNN.*. + """ + for stale in self.chunks_dir.glob(f"chunk_{chunk_num:04d}.*"): + try: + stale.unlink() + except OSError as exc: + logger.debug("Could not remove stale chunk file %s: %s", stale, exc) + return self.chunks_dir / f"chunk_{chunk_num:04d}{suffix}" + + def process_chunk_with_retry(self, chunk_num: int, text: str) -> Optional[Path]: + """Process a chunk with retry logic. + + Returns the generated chunk file's path, or None when all attempts + failed. Raises ConversionCancelled when the run was cancelled. + """ + for attempt in range(config.MAX_RETRIES): + self._check_cancelled() + try: + result = self.generate_chunk(text, chunk_num) + if result and Path(result).exists(): + return Path(result) + logger.warning("Chunk %d attempt %d failed", chunk_num, attempt + 1) + except ConversionCancelled: + raise + except Exception as exc: + logger.warning("Chunk %d attempt %d error: %s", chunk_num, attempt + 1, exc) + + if attempt < config.MAX_RETRIES - 1: + sleep_time = 5 + (2 ** attempt) + logger.info("Waiting %ds before retry...", sleep_time) + self._sleep(sleep_time) + + logger.error("Chunk %d failed after %d attempts", chunk_num, config.MAX_RETRIES) + return None + + @contextlib.contextmanager + def _chunk_heartbeat(self, chunk_num: int): + """Log a periodic "still working" record while a request generates.""" + stop = threading.Event() + subject = f"Chunk {chunk_num}" + + def _beat(): + start = time.time() + while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS): + elapsed = time.time() - start + if self.quiet: + logger.info("%s still generating — %dm %ds elapsed", + subject, int(elapsed // 60), int(elapsed % 60)) + else: + print(f"[...] {subject} still generating — " + f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed", + flush=True) + + thread = threading.Thread(target=_beat, daemon=True) + thread.start() + try: + yield + finally: + stop.set() + thread.join() diff --git a/app/converter/clients/faster.py b/app/converter/clients/faster.py new file mode 100644 index 0000000..df98546 --- /dev/null +++ b/app/converter/clients/faster.py @@ -0,0 +1,123 @@ +"""Client for the faster-qwen3-tts OpenAI-compatible server.""" + +import json +import logging +import urllib.error +import urllib.request +import wave +from pathlib import Path +from typing import List, Optional + +from .. import config +from ..chunking import split_into_chunks +from .base import BaseTTSClient, ConversionCancelled + +logger = logging.getLogger(__name__) + +# The 12Hz codec the faster server synthesizes with outputs 24 kHz audio. +SAMPLE_RATE = 24000 + + +class FasterTTSClient(BaseTTSClient): + """Generates audio chunks through a faster-qwen3-tts server. + + Talks to the OpenAI-compatible server shipped in the faster-qwen3-tts + repository (examples/openai_server.py). The reference voice (ref audio, + ref text) and language are configured on the server itself via + --ref-audio/--ref-text or a --voices JSON file; this client only sends + text. Unlike the Qwen demo, the server performs one generation per + request, so long chunks are sub-chunked client-side. + """ + + def __init__(self, chunks_dir: Path, + voice: Optional[str] = None, api_url: Optional[str] = None, + quiet: bool = False): + super().__init__(chunks_dir, quiet=quiet) + self.voice = voice or config.FASTER_VOICE + self.api_url = (api_url or config.FASTER_API_URL).rstrip("/") + self._check_health() + + def _check_health(self) -> None: + """Verify the server is reachable and its model is loaded.""" + url = f"{self.api_url}/health" + try: + with urllib.request.urlopen(url, timeout=10) as response: + payload = json.loads(response.read().decode("utf-8")) + except Exception as exc: + raise RuntimeError( + f"Faster TTS server not reachable at {url}: {exc}. " + "Start the faster-qwen3-tts OpenAI-compatible server first " + "(see the 'Faster backend' section of the README)." + ) from exc + if not payload.get("model_loaded"): + raise RuntimeError( + "The faster TTS server is running but its model is not loaded yet; " + "wait for model download and startup to finish, then retry." + ) + self._report(f"[OK] Connected to faster TTS API at {self.api_url} (voice '{self.voice}')") + self._report(f"[INFO] The server silently falls back to its first configured voice if " + f"'{self.voice}' is not defined in its voice config (see README).") + + # ------------------------------------------------------------------ + # HTTP requests + # ------------------------------------------------------------------ + + def _request_pcm(self, text: str) -> bytes: + """POST one sub-chunk and return raw 16-bit mono PCM bytes.""" + url = f"{self.api_url}/v1/audio/speech" + payload = json.dumps({ + "model": "tts-1", + "input": text, + "voice": self.voice, + "response_format": "pcm", + }).encode("utf-8") + request = urllib.request.Request( + url, data=payload, headers={"Content-Type": "application/json"}, method="POST") + try: + with urllib.request.urlopen(request, timeout=config.API_TIMEOUT) as response: + pcm = response.read() + except urllib.error.HTTPError as exc: + detail = "" + try: + detail = exc.read().decode("utf-8", errors="replace")[:200] + except Exception: + pass + raise RuntimeError(f"Faster TTS server returned HTTP {exc.code}: {detail}") from exc + except urllib.error.URLError as exc: + raise RuntimeError(f"Faster TTS request failed: {exc.reason}") from exc + if not pcm: + raise RuntimeError("Faster TTS server returned empty audio") + return pcm + + # ------------------------------------------------------------------ + # Chunk generation + # ------------------------------------------------------------------ + + def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: + """Generate one audio chunk; returns its path in the chunks folder.""" + try: + sub_chunks = split_into_chunks(text, max_words=config.CHUNK_SIZE) + if not sub_chunks: + raise RuntimeError("No text to synthesize") + + pcm_parts: List[bytes] = [] + with self._chunk_heartbeat(chunk_num): + for sub_num, sub_text in enumerate(sub_chunks, 1): + pcm = self._request_pcm(sub_text) + pcm_parts.append(pcm) + + output_path = self._chunk_path(chunk_num, ".wav") + with wave.open(str(output_path), "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(SAMPLE_RATE) + wav_file.writeframes(b"".join(pcm_parts)) + + logger.debug("Chunk %d generated (%d sub-chunks)", chunk_num, len(sub_chunks)) + return str(output_path) + + except ConversionCancelled: + raise + except Exception as exc: + logger.error("Faster chunk processing failed for chunk %d: %s", chunk_num, exc) + return None diff --git a/app/converter/clients/languages.py b/app/converter/clients/languages.py new file mode 100644 index 0000000..079dead --- /dev/null +++ b/app/converter/clients/languages.py @@ -0,0 +1,74 @@ +"""Language tables shared by the TTS clients and their UIs.""" + +from typing import Optional + +# Languages the Qwen3-TTS demo accepts as display names (its API silently +# falls back to "Auto" for anything else, so unknown names are rejected +# before a run starts instead of mispronouncing a whole book). +TTS_LANGUAGES = ( + "Auto", "Chinese", "English", "German", "Italian", "Portuguese", + "Spanish", "Japanese", "Korean", "French", "Russian", +) + +# Short aliases accepted on the command line (ISO 639-1 codes and common +# shorthands), mapped to the display names above. +TTS_LANGUAGE_ALIASES = { + "zh": "Chinese", + "en": "English", + "de": "German", + "it": "Italian", + "pt": "Portuguese", + "es": "Spanish", + "ja": "Japanese", + "ko": "Korean", + "fr": "French", + "ru": "Russian", + "zh-cn": "Chinese", + "zh-tw": "Chinese", + "pt-br": "Portuguese", + "en-us": "English", + "en-gb": "English", +} + +# Qwen display names -> ISO 639-1 codes, for audio.cpp families whose +# language request option takes a code instead of a display name. "Auto" +# has no code and maps to None so the field is omitted and the server +# applies its own default. +LANGUAGE_ISO_CODES = { + "Chinese": "zh", + "English": "en", + "German": "de", + "Italian": "it", + "Portuguese": "pt", + "Spanish": "es", + "Japanese": "ja", + "Korean": "ko", + "French": "fr", + "Russian": "ru", +} + + +def normalize_language(value: Optional[str]) -> str: + """Normalize a user-provided language name to a Qwen3-TTS display name. + + Accepts the display names in TTS_LANGUAGES case-insensitively as + well as the short aliases in TTS_LANGUAGE_ALIASES (ISO 639-1 codes + and common shorthands). Raises ValueError for anything else, since the + Qwen3-TTS demo silently falls back to "Auto" for unrecognized languages. + """ + if value is None: + raise ValueError("Language must not be None") + candidate = value.strip() + if not candidate: + raise ValueError("Language must not be empty") + for name in TTS_LANGUAGES: + if candidate.lower() == name.lower(): + return name + alias = TTS_LANGUAGE_ALIASES.get(candidate.lower()) + if alias: + return alias + raise ValueError( + f"Unknown language: {value!r}. Expected one of " + f"{', '.join(TTS_LANGUAGES)} (or an alias: " + f"{', '.join(sorted(TTS_LANGUAGE_ALIASES))})." + ) diff --git a/app/converter/clients/qwen.py b/app/converter/clients/qwen.py new file mode 100644 index 0000000..354ee04 --- /dev/null +++ b/app/converter/clients/qwen.py @@ -0,0 +1,322 @@ +"""Client for the qwen-tts Gradio demo servers (CustomVoice + Base).""" + +import io +import logging +import shutil +import sys +import tempfile +from pathlib import Path +from typing import Any, Dict, Optional, Tuple + +from .. import config +from ..audio import concat_audio_files +from ..chunking import split_into_chunks +from .base import (BaseTTSClient, ConversionCancelled, resolve_request_seed, + VOICE_MODE_CLONE, VOICE_MODE_CUSTOM, VOICE_MODES) +from .languages import normalize_language +from .speakers import speaker_display_name + +logger = logging.getLogger(__name__) + +# Fixed model facts: both demos run the 1.7B model (the CustomVoice demo +# takes its full HuggingFace id), and the 12Hz codec outputs 24 kHz audio. +MODEL_SIZE = "1.7B" +CUSTOM_VOICE_MODEL_ID = "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice" + + +class QwenTTSClient(BaseTTSClient): + """Generates audio chunks through a Qwen3-TTS demo server.""" + + def __init__(self, chunks_dir: Path, + voice_mode: str = "custom_voice", voice_clone_ref_audio: Optional[str] = None, + voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False, + language: Optional[str] = None, api_url: Optional[str] = None, + quiet: bool = False): + super().__init__(chunks_dir, quiet=quiet) + if voice_mode not in VOICE_MODES: + raise ValueError( + f"Unknown voice mode: {voice_mode!r} (expected one of {VOICE_MODES})" + ) + self.voice_mode = voice_mode + self.voice_clone_ref_audio = voice_clone_ref_audio + self.voice_clone_ref_text = (voice_clone_ref_text or "").strip() + self.skip_transcription = skip_transcription + # api_url overrides the configured endpoint for the active voice mode + # (used by the hub's "[remote]" backend entries and --api-url). + self.api_url = (api_url or "").strip() or None + # Seed sent with every request: config.SEED as-is, or (with + # CONSTANT_SEED and SEED < 0) one random value drawn per run and + # reused for every request so the voice stays consistent across + # chunk boundaries. Without CONSTANT_SEED, -1 is forwarded so the + # server re-samples the voice on every generation. + self._seed = resolve_request_seed() + if language is None: + language = config.LANGUAGE + # Validate before connecting so bad values fail fast without a server. + self.language = normalize_language(language) + self.client = None + self.api_info: Dict[str, Any] = {} + self.clone_client = None + self.clone_api_info: Dict[str, Any] = {} + self._ref_audio_filedata: Optional[Dict[str, Any]] = None + self._connect() + + # ------------------------------------------------------------------ + # Connection + # ------------------------------------------------------------------ + + def _connect(self) -> None: + api_url = self.api_url or ( + config.CLONE_API_URL if self.voice_mode == VOICE_MODE_CLONE + else config.QWEN_API_URL) + try: + if self.voice_mode == VOICE_MODE_CLONE: + # Voice clone uses the Base-model demo, which is a separate server + # from the CustomVoice demo (that one only exposes /run_instruct). + self._init_client(api_url, clone=True) + self._report(f"[OK] Connected to Voice Clone API at {api_url}") + self._resolve_reference_text() + else: + self._init_client(api_url, clone=False) + self._report("[OK] Connected to Qwen API") + except Exception as exc: + raise RuntimeError( + f"Qwen API initialization failed at {api_url}: {exc}. " + "Make sure the Qwen demo server is running and reachable, and that your " + "installed Qwen3-TTS version matches this converter's API expectations " + "(voice clone requires the Base-model demo: Qwen/Qwen3-TTS-12Hz-1.7B-Base)." + ) from exc + + def _resolve_reference_text(self) -> None: + """Resolve the reference transcript: explicit text, then local + transcription, then x-vector-only mode.""" + if not self.voice_clone_ref_text and self.voice_clone_ref_audio: + if self.skip_transcription: + self._report("[INFO] Skipping reference audio transcription (--no-transcription).") + else: + self._report("[INFO] Transcribing reference audio for voice cloning...") + self.voice_clone_ref_text = self.transcribe_audio(self.voice_clone_ref_audio) or "" + if not self.voice_clone_ref_text: + self._report("[WARNING] No reference text available; using " + "x-vector-only clone mode (lower quality).") + self._report(' Pass --transcription "..." for higher-quality in-context cloning.') + else: + self._report(f"[OK] Reference text:\n{self.voice_clone_ref_text}") + + def _init_client(self, url: str, clone: bool = False) -> None: + """Initialize a Gradio client and store its API metadata. + + gradio_client prints its usage info directly to stdout while the + client is created and its API metadata loaded, so stdout is swapped + for a buffer for the whole process; the captured text is re-emitted + at DEBUG level for troubleshooting. + """ + from gradio_client import Client + + logger.info("Connecting to Qwen API at %s...", url) + old_stdout = sys.stdout + captured = io.StringIO() + sys.stdout = captured + try: + try: + client = Client(url, httpx_kwargs={"timeout": config.API_TIMEOUT}) + except TypeError: + # Older gradio_client versions don't support httpx_kwargs. + client = Client(url) + if clone: + self.clone_client = client + self.clone_api_info = self._load_api_info(client) + else: + self.client = client + self.api_info = self._load_api_info(client) + finally: + sys.stdout = old_stdout + usage_info = captured.getvalue().strip() + if usage_info: + logger.debug("Gradio client output for %s:\n%s", url, usage_info) + logger.info("Connected to Qwen API") + + @staticmethod + def _load_api_info(client) -> Dict[str, Any]: + """Load available API metadata from the Gradio app.""" + try: + return client.view_api(return_format="dict") + except Exception as exc: + logger.warning("Unable to read API metadata: %s", exc) + return {} + + def _resolve_api_name(self, *candidates: str, api_info: Optional[Dict[str, Any]] = None) -> str: + """Return the first available api_name from candidate list.""" + info = api_info if api_info is not None else self.api_info + named_endpoints = info.get("named_endpoints", {}) + for candidate in candidates: + if candidate in named_endpoints: + return candidate + return candidates[0] + + def _endpoint_accepts_param(self, api_name: str, param_name: str, + api_info: Optional[Dict[str, Any]] = None) -> bool: + """Check whether endpoint input schema includes the given parameter.""" + info = api_info if api_info is not None else self.api_info + endpoint = info.get("named_endpoints", {}).get(api_name, {}) + parameters = endpoint.get("parameters", []) + return any(parameter.get("parameter_name") == param_name for parameter in parameters) + + # ------------------------------------------------------------------ + # Reference audio transcription (voice clone) + # ------------------------------------------------------------------ + + def transcribe_audio(self, audio_path: str) -> Optional[str]: + """Transcribe reference audio locally using an optional Whisper backend.""" + from .transcribe import transcribe_reference_audio + return transcribe_reference_audio(audio_path) + + # ------------------------------------------------------------------ + # Chunk generation + # ------------------------------------------------------------------ + + def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: + """Generate one audio chunk; returns its path in the chunks folder. + + The text is split into sub-requests of at most + ``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=config.CHUNK_SIZE) + if not sub_texts: + raise RuntimeError("No text to synthesize") + + output_path: Optional[Path] = None + with tempfile.TemporaryDirectory(prefix="tts_parts_") as parts_dir, \ + self._chunk_heartbeat(chunk_num): + part_paths = [ + self._generate_sub_request(sub_text, parts_dir, sub_num, + len(sub_texts), chunk_num) + for sub_num, sub_text in enumerate(sub_texts, 1) + ] + if len(part_paths) == 1: + suffix = part_paths[0].suffix or ".wav" + output_path = self._chunk_path(chunk_num, suffix) + shutil.copy2(part_paths[0], output_path) + else: + output_path = self._chunk_path(chunk_num, ".wav") + concat_audio_files(part_paths, output_path) + + logger.debug("Chunk %d generated successfully (%d sub-request(s))", + chunk_num, len(sub_texts)) + return str(output_path) + + except ConversionCancelled: + raise + except Exception as exc: + logger.error("Qwen chunk processing failed for chunk %d: %s", chunk_num, exc) + return None + + def _generate_sub_request(self, text: str, parts_dir: str, sub_num: int, + sub_total: int, chunk_num: int) -> Path: + """Run one API generation for ``text``; returns the downloaded audio.""" + if sub_total > 1: + logger.info("Chunk %d: oversized input split into %d requests " + "(sub-request %d/%d)", chunk_num, sub_total, sub_num, sub_total) + if self.voice_mode == VOICE_MODE_CUSTOM: + result = self._generate_custom_voice(text) + elif self.voice_mode == VOICE_MODE_CLONE: + result = self._generate_voice_clone(text) + else: + raise ValueError(f"Unknown voice mode: {self.voice_mode}") + + if not isinstance(result, (tuple, list)) or not result: + raise RuntimeError("Qwen API returned an invalid result") + + audio_path = result[0] # First element is the audio file path + if not isinstance(audio_path, (str, Path)) or not audio_path: + raise RuntimeError("Qwen API did not return an audio file path") + + source = Path(audio_path) + if not source.exists(): + raise RuntimeError(f"Generated audio file not found: {audio_path}") + + destination = Path(parts_dir) / f"part_{sub_num:02d}{source.suffix or '.wav'}" + shutil.copy2(source, destination) + + return destination + + # ------------------------------------------------------------------ + # API payloads + # ------------------------------------------------------------------ + + def _generate_custom_voice(self, text: str) -> Tuple: + """Generate audio using CustomVoice mode.""" + custom_api = self._resolve_api_name("/run_instruct", "/run_custom_voice", "/generate_custom_voice") + if custom_api == "/run_instruct": + payload = dict( + text=text, + lang_disp=self.language, + spk_disp=speaker_display_name(), + instruct=config.INSTRUCT, + ) + else: + payload = dict( + text=text, + language=self.language, + speaker=config.SPEAKER, + instruct=config.INSTRUCT, + ) + if self._endpoint_accepts_param(custom_api, "model_id_cv"): + payload["model_id_cv"] = CUSTOM_VOICE_MODEL_ID + elif self._endpoint_accepts_param(custom_api, "model_size"): + payload["model_size"] = MODEL_SIZE + + if self._endpoint_accepts_param(custom_api, "seed"): + payload["seed"] = self._seed + + return self.client.predict(**payload, api_name=custom_api) + + def _ref_audio_payload(self) -> Dict[str, Any]: + """Gradio file payload for the reference audio (built once, reused).""" + if self._ref_audio_filedata is None: + from gradio_client import handle_file + self._ref_audio_filedata = handle_file(self.voice_clone_ref_audio) + return self._ref_audio_filedata + + def _generate_voice_clone(self, text: str) -> Tuple: + """Generate audio using Voice Clone mode.""" + if not Path(self.voice_clone_ref_audio).exists(): + raise FileNotFoundError(f"Reference audio not found: {self.voice_clone_ref_audio}") + + if self.clone_client is None: + raise RuntimeError("Voice Clone client is not initialized. Is the Base-model demo running?") + + clone_api = self._resolve_api_name("/run_voice_clone", "/generate_voice_clone", + api_info=self.clone_api_info) + use_xvector = config.XVECTOR_ONLY or not self.voice_clone_ref_text + + if clone_api == "/run_voice_clone": + payload = dict( + ref_aud=self._ref_audio_payload(), + ref_txt=self.voice_clone_ref_text, + use_xvec=use_xvector, + text=text, + lang_disp=self.language, + ) + else: + payload = dict( + ref_audio=self._ref_audio_payload(), + ref_text=self.voice_clone_ref_text, + target_text=text, + language=self.language, + use_xvector_only=use_xvector, + ) + optional_params = { + "model_size": MODEL_SIZE, + "seed": self._seed, + } + for name, value in optional_params.items(): + if self._endpoint_accepts_param(clone_api, name, api_info=self.clone_api_info): + payload[name] = value + + return self.clone_client.predict(**payload, api_name=clone_api) diff --git a/app/converter/clients/speakers.py b/app/converter/clients/speakers.py new file mode 100644 index 0000000..eecd52a --- /dev/null +++ b/app/converter/clients/speakers.py @@ -0,0 +1,57 @@ +"""Qwen3-TTS built-in speaker names and their wire (display) forms.""" + +from typing import Optional + +from .. import config + +# Built-in CustomVoice speaker names for the Qwen3-TTS family. Shared by the +# qwen-tts demo backend (config.SPEAKER, the qwen setup/form) and the +# audio.cpp audiocpp backend's CustomVoice entry (the Convert form's Speaker +# picker). Entries are the canonical/config form; speaker_display_name() +# maps them to the wire (display) form via SPEAKER_DISPLAY_NAMES below. +QWEN3_TTS_SPEAKERS = ("Vivian", "Serena", "Uncle_Fu", "Dylan", "Eric", + "Ryan", "Aiden", "Ono_Anna", "Sohee") + +# Canonical speaker names -> display names used by the qwen-tts demo. +SPEAKER_DISPLAY_NAMES = { + "ryan": "Ryan", + "serena": "Serena", + "vivian": "Vivian", + "uncle_fu": "Uncle Fu", + "aiden": "Aiden", + "ono_anna": "Ono Anna", + "sohee": "Sohee", + "eric": "Eric", + "dylan": "Dylan", +} + + +def speaker_display_name_for(name: str) -> str: + """Return the wire (display) form of a Qwen3-TTS CustomVoice speaker NAME. + + Accepts either the canonical/config form (e.g. "uncle_fu", "Uncle_Fu") + or the display form ("Uncle Fu"), case-insensitively; unknown names pass + through unchanged. Used by AudioCppTTSClient to normalize the --voice / + Speaker-picker value into what audiocpp_server expects in the request's + voice field. + """ + return SPEAKER_DISPLAY_NAMES.get((name or "").lower(), name) + + +def is_builtin_speaker(name: Optional[str]) -> bool: + """True when NAME is one of the Qwen3-TTS CustomVoice built-in speakers. + + Matches case-insensitively across the canonical ("Uncle_Fu"), display + ("Uncle Fu") and shorthand ("uncle_fu") forms, so the --voice flag and + the Convert form's Speaker picker resolve to the same set. + """ + if not name: + return False + norm = name.lower().replace("_", " ").replace("-", " ") + return any(norm == speaker.lower().replace("_", " ") + for speaker in QWEN3_TTS_SPEAKERS) + + +def speaker_display_name() -> str: + """Return the display name for the configured custom speaker.""" + return speaker_display_name_for(config.SPEAKER) diff --git a/app/converter/clients/transcribe.py b/app/converter/clients/transcribe.py new file mode 100644 index 0000000..d2db9f1 --- /dev/null +++ b/app/converter/clients/transcribe.py @@ -0,0 +1,54 @@ +"""Optional local Whisper transcription of reference audio.""" + +import logging +from typing import Optional + +logger = logging.getLogger(__name__) + + +def transcribe_reference_audio(audio_path: str, model_name: str = "base") -> Optional[str]: + """Transcribe reference audio locally using an optional Whisper backend. + + The current qwen-tts demo does not expose a transcription endpoint, so + transcription is done client-side when a Whisper package is available. + Returns None if no backend is installed. + """ + for backend in ("faster_whisper", "whisper"): + try: + if backend == "faster_whisper": + from faster_whisper import WhisperModel + model = WhisperModel(model_name, device="cpu", compute_type="int8") + segments, _ = model.transcribe(audio_path) + text = " ".join(seg.text.strip() for seg in segments).strip() + else: + import whisper + model = whisper.load_model(model_name) + result = model.transcribe(audio_path) + text = (result.get("text") or "").strip() + if text: + logger.info("Transcription complete via %s: %s", backend, text) + return text + except ImportError: + continue + except Exception as exc: + logger.warning("%s transcription failed: %s", backend, exc) + logger.warning("No Whisper backend available; transcription skipped.") + return None + + +def whisper_backend_available() -> Optional[str]: + """Return the name of an importable Whisper backend, or None. + + Checks faster_whisper first (preferred), then the openai-whisper + package, without importing the heavy model code: a bare import probe + is enough to tell whether the package is installed in the current + environment. Used by the make_audiocpp_server_json tool to warn when + neither is present (e.g. the wrong conda environment is active). + """ + for backend in ("faster_whisper", "whisper"): + try: + __import__(backend) + except ImportError: + continue + return backend + return None diff --git a/app/converter/converter.py b/app/converter/converter.py index 1abc85c..48d4987 100644 --- a/app/converter/converter.py +++ b/app/converter/converter.py @@ -15,7 +15,7 @@ from typing import Callable, Dict, List, Optional, Tuple from . import audio, chunking, config, cover, extractors from .audio import TrackMeta -from .tts import ( +from .clients import ( BACKENDS, BACKEND_AUDIOCPP, BACKEND_FASTER, @@ -228,7 +228,8 @@ class AudiobookConverter: if backend == BACKEND_FASTER: # The faster backend always voice-clones using a reference voice # configured on the server, so no local reference audio is needed. - self.tts = FasterTTSClient(voice=voice, api_url=api_url, + self.tts = FasterTTSClient(chunks_dir=CHUNKS_FOLDER, + voice=voice, api_url=api_url, quiet=quiet) elif backend == BACKEND_AUDIOCPP: # --voice picks the voice: a built-in speaker name on the @@ -237,13 +238,15 @@ class AudiobookConverter: # multi-model servers; instructions describe or style the # voice, request_options pass per-model controls through to # the server. - self.tts = AudioCppTTSClient(voice=voice, language=self.language, + self.tts = AudioCppTTSClient(chunks_dir=CHUNKS_FOLDER, + voice=voice, language=self.language, model_id=model_id, instructions=instructions, request_options=self.request_options, api_url=api_url, quiet=quiet) else: self.tts = QwenTTSClient( + chunks_dir=CHUNKS_FOLDER, voice_mode=voice_mode, voice_clone_ref_audio=voice_clone_ref_audio, voice_clone_ref_text=voice_clone_ref_text, @@ -398,7 +401,7 @@ class AudiobookConverter: self.current_outputs = [] # Start from a clean scratch folder so a previous crash can never # affect this run - audio.cleanup_chunks() + audio.cleanup_chunks(CHUNKS_FOLDER) logger.info("Extracting text...") book = extractors.extract_book(file_path) @@ -476,7 +479,7 @@ class AudiobookConverter: return False finally: # Always cleanup, even on failure or interrupt - audio.cleanup_chunks() + audio.cleanup_chunks(CHUNKS_FOLDER) def _convert_m4b_with_chapters(self, sections, stem: str, start_time: float, meta: Optional[TrackMeta] = None, @@ -519,7 +522,9 @@ class AudiobookConverter: return False output_path = AUDIOBOOKS_FOLDER / f"{stem}.{self.output_format}" - if not audio.combine_chapters_to_m4b(chapter_files, titles, output_path, speed=self.speed, + if not audio.combine_chapters_to_m4b(chapter_files, titles, output_path, + chunks_dir=CHUNKS_FOLDER, + speed=self.speed, meta=meta, cover=cover): return False duration = time.time() - start_time @@ -653,7 +658,9 @@ class AudiobookConverter: successful_chunks, total_chunks) return False - success = audio.combine_chunks(total_chunks, output_path, chunk_results=results, + success = audio.combine_chunks(total_chunks, output_path, + chunk_results=results, + chunks_dir=CHUNKS_FOLDER, speed=speed, output_format=output_format, intermediate=chapter is not None, meta=meta, cover=cover) diff --git a/app/converter/tts.py b/app/converter/tts.py deleted file mode 100644 index 8130a44..0000000 --- a/app/converter/tts.py +++ /dev/null @@ -1,1469 +0,0 @@ -"""Client wrappers for the TTS backends. - -QwenTTSClient talks to the Qwen3-TTS demo server (custom voice / voice clone). -FasterTTSClient talks to the OpenAI-compatible server from the -faster-qwen3-tts repository (voice cloning only; the reference voice is -configured server-side — see the "Faster backend" section of the README). -AudioCppTTSClient talks to the audiocpp_server from the audio.cpp -repository, which can host any TTS model family audio.cpp supports -(Qwen3-TTS, Higgs Audio, VoxCPM2, IndexTTS2, ...) through one OpenAI-style -API; the family is detected from the server at startup (see the -"audio.cpp backend" sections of the README). -""" - -import contextlib -import io -import json -import logging -import random -import shutil -import sys -import tempfile -import threading -import time -import urllib.error -import urllib.parse -import urllib.request -import wave -from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple - -from . import config -from .audio import concat_audio_files -from .chunking import split_into_chunks - -logger = logging.getLogger(__name__) - - -class ConversionCancelled(Exception): - """Raised inside a conversion whose cancel event was set. - - The TUI run view sets a ``threading.Event`` on the TTS client (and the - converter checks it between chunks/chapters/books); the retry loops - raise this so the cancellation propagates out of a sleeping or retrying - request promptly instead of finishing the retry ladder. - """ - - -# Voice modes (re-exported for the CLI and the converter orchestrator). -VOICE_MODE_CUSTOM = "custom_voice" -VOICE_MODE_CLONE = "voice_clone" -VOICE_MODES = (VOICE_MODE_CUSTOM, VOICE_MODE_CLONE) - -# TTS backends (re-exported for the CLI and the converter orchestrator). -BACKEND_QWEN = "qwen" -BACKEND_FASTER = "faster" -BACKEND_AUDIOCPP = "audiocpp" -BACKENDS = (BACKEND_AUDIOCPP, BACKEND_QWEN, BACKEND_FASTER) - -# Languages understood by the Qwen3-TTS API. Display names must match the -# demo dropdown exactly (the demo silently falls back to "Auto" for -# unrecognized values, so languages are validated client-side first). -TTS_LANGUAGES = ( - "Auto", - "Chinese", - "English", - "German", - "Italian", - "Portuguese", - "Spanish", - "Japanese", - "Korean", - "French", - "Russian", -) - -# Short aliases accepted on the command line (ISO 639-1 codes and common -# shorthands), mapped to the display names above. -TTS_LANGUAGE_ALIASES = { - "zh": "Chinese", - "en": "English", - "de": "German", - "it": "Italian", - "pt": "Portuguese", - "es": "Spanish", - "ja": "Japanese", - "ko": "Korean", - "fr": "French", - "ru": "Russian", - "zh-cn": "Chinese", - "zh-tw": "Chinese", - "pt-br": "Portuguese", - "en-us": "English", - "en-gb": "English", -} - -# Qwen display names -> ISO 639-1 codes, for audio.cpp families whose -# language request option takes a code instead of a display name. "Auto" -# has no code and maps to None so the field is omitted and the server -# applies its own default. -LANGUAGE_ISO_CODES = { - "Chinese": "zh", - "English": "en", - "German": "de", - "Italian": "it", - "Portuguese": "pt", - "Spanish": "es", - "Japanese": "ja", - "Korean": "ko", - "French": "fr", - "Russian": "ru", -} - -# --- audio.cpp model families --------------------------------------------- -# -# audiocpp_server exposes the same OpenAI-style API for every TTS family it -# hosts; families only differ in a few request conventions, captured here as -# profiles. Families that are not listed use the default profile below. - -# How the "language" request field is expressed by a family. -AUDIOCPP_LANG_DISPLAY = "display" # Qwen display names, e.g. "English" -AUDIOCPP_LANG_ISO = "iso" # ISO 639-1 codes, e.g. "en" -AUDIOCPP_LANG_OMIT = "omit" # no language field; the model detects it - -# The Qwen3-TTS family. Unlike every other family (one model type each), -# qwen3_tts hosts several model *types* under one family id, distinguished -# only by the server entry's id/task: the CustomVoice model (built-in -# speakers, e.g. Vivian/Ryan), the Base model (voice cloning via a -# server-side preset), and the VoiceDesign model (task "vdes"). The -# per-entry voice capability below (audiocpp_entry_voice_capability) -# resolves which is which, driving both the Convert form (which voice -# list to show) and the converter's mode selection. -AUDIOCPP_FAMILY_QWEN3_TTS = "qwen3_tts" - -# Server model entry tasks this client can synthesize audiobooks with, -# taken from GET /v1/models (the "task" field of each entry; a missing task -# is treated as "tts" — a harmless generic default). "vdes" entries are -# voice design models: the voice is described with --instructions instead -# of coming from a speaker or a reference clip. Entries with any other task -# (asr, vc, diar, ...) are rejected at connect time with a hint to pick a -# synthesis entry. -AUDIOCPP_TASK_TTS = "tts" -AUDIOCPP_TASK_VDES = "vdes" -AUDIOCPP_SYNTHESIS_TASKS = (AUDIOCPP_TASK_TTS, "clon", AUDIOCPP_TASK_VDES) - -# The voice capability of a server model entry — how its voice is supplied. -# Resolved per entry from (family, task, id) by -# audiocpp_entry_voice_capability; drives both the Convert form (which -# voice list to show) and the converter (speaker vs preset vs design mode). -# Most families are clone-only; only the Qwen3-TTS CustomVoice entry has -# built-in speakers, and only VoiceDesign entries take a description. -AUDIOCPP_VOICE_SPEAKER = "speaker" # built-in speaker name (Qwen CustomVoice) -AUDIOCPP_VOICE_CLONE = "clone" # server-side preset / voice_dir (Base, others) -AUDIOCPP_VOICE_DESIGN = "design" # voice described by --instructions (vdes) - - -class AudioCppFamilyProfile: - """Request conventions of one audio.cpp model family. - - Language style and whether the family reads a style/instruction prompt; - these are family-level (every entry of a family shares them). Whether a - *specific entry* has built-in speakers is an entry-level concern, decided - by audiocpp_entry_voice_capability, not this profile. - """ - - def __init__(self, language_style: str = AUDIOCPP_LANG_OMIT, - sends_instructions: bool = False): - self.language_style = language_style - self.sends_instructions = sends_instructions - - -# Generic profile for families not listed in AUDIOCPP_FAMILY_PROFILES: -# clone-only, no style instructions, and no language field (the model -# detects the language itself). Describes higgs_audio_tts, voxcpm2, -# fish_audio, dots_tts, dramabox, omnivoice, outetts, glm_tts, miotts, -# moss_tts_*, pocket_tts, vibevoice, ... as well as families added to -# audio.cpp after this table was written. -AUDIOCPP_DEFAULT_FAMILY_PROFILE = AudioCppFamilyProfile() - -AUDIOCPP_FAMILY_PROFILES = { - AUDIOCPP_FAMILY_QWEN3_TTS: AudioCppFamilyProfile( - language_style=AUDIOCPP_LANG_DISPLAY, - sends_instructions=True, - ), - # Families whose language option takes a code (e.g. "en") instead of - # a Qwen display name; otherwise clone-only like the default profile. - "chatterbox": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), - "confucius4_tts": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), - "index_tts2": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), - "magpie_tts": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), - "supertonic": AudioCppFamilyProfile(language_style=AUDIOCPP_LANG_ISO), -} - - -def audiocpp_entry_voice_capability(family: str, task: str, - model_id: str) -> str: - """How a server model entry's voice is supplied — speaker/clone/design. - - Resolved from the entry's family, task and id — the same {id, family, - task} triple GET /v1/models reports, so it works for local server.json - entries and remote live-queried entries alike. Qwen3-TTS is the one - family hosting several model *types* under one family id: the - CustomVoice model (id contains "customvoice") has built-in speakers, the - Base model and any other entry are clone-only, and VoiceDesign entries - (task "vdes") take a description. Every other family is clone-only. - """ - if task == AUDIOCPP_TASK_VDES: - return AUDIOCPP_VOICE_DESIGN - if family == AUDIOCPP_FAMILY_QWEN3_TTS \ - and "customvoice" in (model_id or "").lower(): - return AUDIOCPP_VOICE_SPEAKER - return AUDIOCPP_VOICE_CLONE - -# Built-in CustomVoice speaker names for the Qwen3-TTS family. Shared by the -# qwen-tts demo backend (config.SPEAKER, the qwen setup/form) and the -# audio.cpp audiocpp backend's CustomVoice entry (the Convert form's Speaker -# picker). Entries are the canonical/config form; speaker_display_name() -# maps them to the wire (display) form via SPEAKER_DISPLAY_NAMES below. -QWEN3_TTS_SPEAKERS = ("Vivian", "Serena", "Uncle_Fu", "Dylan", "Eric", - "Ryan", "Aiden", "Ono_Anna", "Sohee") - -# Canonical speaker names -> display names used by the qwen-tts demo. -SPEAKER_DISPLAY_NAMES = { - "ryan": "Ryan", - "serena": "Serena", - "vivian": "Vivian", - "uncle_fu": "Uncle Fu", - "aiden": "Aiden", - "ono_anna": "Ono Anna", - "sohee": "Sohee", - "eric": "Eric", - "dylan": "Dylan", -} - -# Fixed model facts: both demos run the 1.7B model (the CustomVoice demo -# takes its full HuggingFace id), and the 12Hz codec outputs 24 kHz audio. -MODEL_SIZE = "1.7B" -CUSTOM_VOICE_MODEL_ID = "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice" -SAMPLE_RATE = 24000 - -CHUNKS_FOLDER = Path(__file__).resolve().parent.parent.parent / "app" / "chunks" - - -def _resolve_request_seed() -> int: - """Resolve the seed sent with every request. - - Returns config.SEED as-is, or (with CONSTANT_SEED and SEED < 0) one - random value drawn per run, meant to be reused for every request so - the voice stays consistent across chunk boundaries. Without - CONSTANT_SEED, -1 is returned so the server re-samples the voice on - every generation. - """ - seed = config.SEED - if config.CONSTANT_SEED and seed < 0: - seed = random.randrange(2 ** 31) - return seed - - -def speaker_display_name_for(name: str) -> str: - """Return the wire (display) form of a Qwen3-TTS CustomVoice speaker NAME. - - Accepts either the canonical/config form (e.g. "uncle_fu", "Uncle_Fu") - or the display form ("Uncle Fu"), case-insensitively; unknown names pass - through unchanged. Used by AudioCppTTSClient to normalize the --voice / - Speaker-picker value into what audiocpp_server expects in the request's - voice field. - """ - return SPEAKER_DISPLAY_NAMES.get((name or "").lower(), name) - - -def is_builtin_speaker(name: Optional[str]) -> bool: - """True when NAME is one of the Qwen3-TTS CustomVoice built-in speakers. - - Matches case-insensitively across the canonical ("Uncle_Fu"), display - ("Uncle Fu") and shorthand ("uncle_fu") forms, so the --voice flag and - the Convert form's Speaker picker resolve to the same set. - """ - if not name: - return False - norm = name.lower().replace("_", " ").replace("-", " ") - return any(norm == speaker.lower().replace("_", " ") - for speaker in QWEN3_TTS_SPEAKERS) - - -def speaker_display_name() -> str: - """Return the display name for the configured custom speaker.""" - return speaker_display_name_for(config.SPEAKER) - - -def normalize_language(value: Optional[str]) -> str: - """Normalize a user-provided language name to a Qwen3-TTS display name. - - Accepts the display names in TTS_LANGUAGES case-insensitively as - well as the short aliases in TTS_LANGUAGE_ALIASES (ISO 639-1 codes - and common shorthands). Raises ValueError for anything else, since the - Qwen3-TTS demo silently falls back to "Auto" for unrecognized languages. - """ - if value is None: - raise ValueError("Language must not be None") - candidate = value.strip() - if not candidate: - raise ValueError("Language must not be empty") - for name in TTS_LANGUAGES: - if candidate.lower() == name.lower(): - return name - alias = TTS_LANGUAGE_ALIASES.get(candidate.lower()) - if alias: - return alias - raise ValueError( - f"Unknown language: {value!r}. Expected one of " - f"{', '.join(TTS_LANGUAGES)} (or an alias: " - f"{', '.join(sorted(TTS_LANGUAGE_ALIASES))})." - ) - - -def transcribe_reference_audio(audio_path: str, model_name: str = "base") -> Optional[str]: - """Transcribe reference audio locally using an optional Whisper backend. - - The current qwen-tts demo does not expose a transcription endpoint, so - transcription is done client-side when a Whisper package is available. - Returns None if no backend is installed. - """ - for backend in ("faster_whisper", "whisper"): - try: - if backend == "faster_whisper": - from faster_whisper import WhisperModel - model = WhisperModel(model_name, device="cpu", compute_type="int8") - segments, _ = model.transcribe(audio_path) - text = " ".join(seg.text.strip() for seg in segments).strip() - else: - import whisper - model = whisper.load_model(model_name) - result = model.transcribe(audio_path) - text = (result.get("text") or "").strip() - if text: - logger.info("Transcription complete via %s: %s", backend, text) - return text - except ImportError: - continue - except Exception as exc: - logger.warning("%s transcription failed: %s", backend, exc) - logger.warning("No Whisper backend available; transcription skipped.") - return None - - -def whisper_backend_available() -> Optional[str]: - """Return the name of an importable Whisper backend, or None. - - Checks faster_whisper first (preferred), then the openai-whisper - package, without importing the heavy model code: a bare import probe - is enough to tell whether the package is installed in the current - environment. Used by the make_audiocpp_server_json tool to warn when - neither is present (e.g. the wrong conda environment is active). - """ - for backend in ("faster_whisper", "whisper"): - try: - __import__(backend) - except ImportError: - continue - return backend - return None - - -class _BaseTTSClient: - """Shared chunk retry logic, heartbeat, and chunk file bookkeeping.""" - - # Set by the converter when the run is cancellable (the TUI run view): - # a threading.Event that, once set, aborts the run between requests - # (and interrupts retry back-off sleeps). ``quiet`` silences console - # prints (the run view owns the screen). - cancel = None - quiet = False - - def _report(self, message: str) -> None: - """Print a console line unless quiet (the run view owns the screen).""" - if not self.quiet: - print(message) - - def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: - """Generate one audio chunk; returns its path in the chunks folder.""" - raise NotImplementedError - - def _cancel_requested(self) -> bool: - """True when the run's cancel event has been set (if any).""" - return isinstance(self.cancel, threading.Event) \ - and self.cancel.is_set() - - def _check_cancelled(self) -> None: - """Raise ConversionCancelled when the cancel event is set.""" - if self._cancel_requested(): - raise ConversionCancelled("Cancelled by user") - - def _sleep(self, seconds: float) -> None: - """Sleep SECONDS, cut short (raising) when the cancel event sets.""" - if isinstance(self.cancel, threading.Event): - if self.cancel.wait(seconds): - raise ConversionCancelled("Cancelled by user") - else: - time.sleep(seconds) - - def _chunk_path(self, chunk_num: int, suffix: str) -> Path: - """Resolve the target path for a chunk, removing stale files first. - - Any stale chunk file for this index is removed so a retry or extension - change can never leave two files matching chunk_NNNN.*. - """ - for stale in CHUNKS_FOLDER.glob(f"chunk_{chunk_num:04d}.*"): - try: - stale.unlink() - except OSError as exc: - logger.debug("Could not remove stale chunk file %s: %s", stale, exc) - return CHUNKS_FOLDER / f"chunk_{chunk_num:04d}{suffix}" - - def process_chunk_with_retry(self, chunk_num: int, text: str) -> Optional[Path]: - """Process a chunk with retry logic. - - Returns the generated chunk file's path, or None when all attempts - failed. Raises ConversionCancelled when the run was cancelled. - """ - for attempt in range(config.MAX_RETRIES): - self._check_cancelled() - try: - result = self.generate_chunk(text, chunk_num) - if result and Path(result).exists(): - return Path(result) - logger.warning("Chunk %d attempt %d failed", chunk_num, attempt + 1) - except ConversionCancelled: - raise - except Exception as exc: - logger.warning("Chunk %d attempt %d error: %s", chunk_num, attempt + 1, exc) - - if attempt < config.MAX_RETRIES - 1: - sleep_time = 5 + (2 ** attempt) - logger.info("Waiting %ds before retry...", sleep_time) - self._sleep(sleep_time) - - logger.error("Chunk %d failed after %d attempts", chunk_num, config.MAX_RETRIES) - return None - - @contextlib.contextmanager - def _chunk_heartbeat(self, chunk_num: int): - """Log a periodic "still working" record while a request generates.""" - stop = threading.Event() - subject = f"Chunk {chunk_num}" - - def _beat(): - start = time.time() - while not stop.wait(config.HEARTBEAT_INTERVAL_SECONDS): - elapsed = time.time() - start - if self.quiet: - logger.info("%s still generating — %dm %ds elapsed", - subject, int(elapsed // 60), int(elapsed % 60)) - else: - print(f"[...] {subject} still generating — " - f"{int(elapsed // 60)}m {int(elapsed % 60)}s elapsed", - flush=True) - - thread = threading.Thread(target=_beat, daemon=True) - thread.start() - try: - yield - finally: - stop.set() - thread.join() - - -class QwenTTSClient(_BaseTTSClient): - """Generates audio chunks through a Qwen3-TTS demo server.""" - - def __init__(self, voice_mode: str = "custom_voice", voice_clone_ref_audio: Optional[str] = None, - voice_clone_ref_text: Optional[str] = None, skip_transcription: bool = False, - language: Optional[str] = None, api_url: Optional[str] = None, - quiet: bool = False): - # Quiet before connecting so connect-time status lines never reach - # a screen the TUI run view owns. - self.quiet = bool(quiet) - if voice_mode not in VOICE_MODES: - raise ValueError( - f"Unknown voice mode: {voice_mode!r} (expected one of {VOICE_MODES})" - ) - self.voice_mode = voice_mode - self.voice_clone_ref_audio = voice_clone_ref_audio - self.voice_clone_ref_text = (voice_clone_ref_text or "").strip() - self.skip_transcription = skip_transcription - # api_url overrides the configured endpoint for the active voice mode - # (used by the hub's "[remote]" backend entries and --api-url). - self.api_url = (api_url or "").strip() or None - # Seed sent with every request: config.SEED as-is, or (with - # CONSTANT_SEED and SEED < 0) one random value drawn per run and - # reused for every request so the voice stays consistent across - # chunk boundaries. Without CONSTANT_SEED, -1 is forwarded so the - # server re-samples the voice on every generation. - self._seed = _resolve_request_seed() - if language is None: - language = config.LANGUAGE - # Validate before connecting so bad values fail fast without a server. - self.language = normalize_language(language) - self.client = None - self.api_info: Dict[str, Any] = {} - self.clone_client = None - self.clone_api_info: Dict[str, Any] = {} - self._ref_audio_filedata: Optional[Dict[str, Any]] = None - self._connect() - - # ------------------------------------------------------------------ - # Connection - # ------------------------------------------------------------------ - - def _connect(self) -> None: - api_url = self.api_url or ( - config.CLONE_API_URL if self.voice_mode == VOICE_MODE_CLONE - else config.QWEN_API_URL) - try: - if self.voice_mode == VOICE_MODE_CLONE: - # Voice clone uses the Base-model demo, which is a separate server - # from the CustomVoice demo (that one only exposes /run_instruct). - self._init_client(api_url, clone=True) - self._report(f"[OK] Connected to Voice Clone API at {api_url}") - self._resolve_reference_text() - else: - self._init_client(api_url, clone=False) - self._report("[OK] Connected to Qwen API") - except Exception as exc: - raise RuntimeError( - f"Qwen API initialization failed at {api_url}: {exc}. " - "Make sure the Qwen demo server is running and reachable, and that your " - "installed Qwen3-TTS version matches this converter's API expectations " - "(voice clone requires the Base-model demo: Qwen/Qwen3-TTS-12Hz-1.7B-Base)." - ) from exc - - def _resolve_reference_text(self) -> None: - """Resolve the reference transcript: explicit text, then local - transcription, then x-vector-only mode.""" - if not self.voice_clone_ref_text and self.voice_clone_ref_audio: - if self.skip_transcription: - self._report("[INFO] Skipping reference audio transcription (--no-transcription).") - else: - self._report("[INFO] Transcribing reference audio for voice cloning...") - self.voice_clone_ref_text = self.transcribe_audio(self.voice_clone_ref_audio) or "" - if not self.voice_clone_ref_text: - self._report("[WARNING] No reference text available; using " - "x-vector-only clone mode (lower quality).") - self._report(' Pass --transcription "..." for higher-quality in-context cloning.') - else: - self._report(f"[OK] Reference text:\n{self.voice_clone_ref_text}") - - def _init_client(self, url: str, clone: bool = False) -> None: - """Initialize a Gradio client and store its API metadata. - - gradio_client prints its usage info directly to stdout while the - client is created and its API metadata loaded, so stdout is swapped - for a buffer for the whole process; the captured text is re-emitted - at DEBUG level for troubleshooting. - """ - from gradio_client import Client - - logger.info("Connecting to Qwen API at %s...", url) - old_stdout = sys.stdout - captured = io.StringIO() - sys.stdout = captured - try: - try: - client = Client(url, httpx_kwargs={"timeout": config.API_TIMEOUT}) - except TypeError: - # Older gradio_client versions don't support httpx_kwargs. - client = Client(url) - if clone: - self.clone_client = client - self.clone_api_info = self._load_api_info(client) - else: - self.client = client - self.api_info = self._load_api_info(client) - finally: - sys.stdout = old_stdout - usage_info = captured.getvalue().strip() - if usage_info: - logger.debug("Gradio client output for %s:\n%s", url, usage_info) - logger.info("Connected to Qwen API") - - @staticmethod - def _load_api_info(client) -> Dict[str, Any]: - """Load available API metadata from the Gradio app.""" - try: - return client.view_api(return_format="dict") - except Exception as exc: - logger.warning("Unable to read API metadata: %s", exc) - return {} - - def _resolve_api_name(self, *candidates: str, api_info: Optional[Dict[str, Any]] = None) -> str: - """Return the first available api_name from candidate list.""" - info = api_info if api_info is not None else self.api_info - named_endpoints = info.get("named_endpoints", {}) - for candidate in candidates: - if candidate in named_endpoints: - return candidate - return candidates[0] - - def _endpoint_accepts_param(self, api_name: str, param_name: str, - api_info: Optional[Dict[str, Any]] = None) -> bool: - """Check whether endpoint input schema includes the given parameter.""" - info = api_info if api_info is not None else self.api_info - endpoint = info.get("named_endpoints", {}).get(api_name, {}) - parameters = endpoint.get("parameters", []) - return any(parameter.get("parameter_name") == param_name for parameter in parameters) - - # ------------------------------------------------------------------ - # Reference audio transcription (voice clone) - # ------------------------------------------------------------------ - - def transcribe_audio(self, audio_path: str) -> Optional[str]: - """Transcribe reference audio locally using an optional Whisper backend.""" - return transcribe_reference_audio(audio_path) - - # ------------------------------------------------------------------ - # Chunk generation - # ------------------------------------------------------------------ - - def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: - """Generate one audio chunk; returns its path in the chunks folder. - - The text is split into sub-requests of at most - ``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=config.CHUNK_SIZE) - if not sub_texts: - raise RuntimeError("No text to synthesize") - - output_path: Optional[Path] = None - with tempfile.TemporaryDirectory(prefix="tts_parts_") as parts_dir, \ - self._chunk_heartbeat(chunk_num): - part_paths = [ - self._generate_sub_request(sub_text, parts_dir, sub_num, - len(sub_texts), chunk_num) - for sub_num, sub_text in enumerate(sub_texts, 1) - ] - if len(part_paths) == 1: - suffix = part_paths[0].suffix or ".wav" - output_path = self._chunk_path(chunk_num, suffix) - shutil.copy2(part_paths[0], output_path) - else: - output_path = self._chunk_path(chunk_num, ".wav") - concat_audio_files(part_paths, output_path) - - logger.debug("Chunk %d generated successfully (%d sub-request(s))", - chunk_num, len(sub_texts)) - return str(output_path) - - except ConversionCancelled: - raise - except Exception as exc: - logger.error("Qwen chunk processing failed for chunk %d: %s", chunk_num, exc) - return None - - def _generate_sub_request(self, text: str, parts_dir: str, sub_num: int, - sub_total: int, chunk_num: int) -> Path: - """Run one API generation for ``text``; returns the downloaded audio.""" - if sub_total > 1: - logger.info("Chunk %d: oversized input split into %d requests " - "(sub-request %d/%d)", chunk_num, sub_total, sub_num, sub_total) - if self.voice_mode == VOICE_MODE_CUSTOM: - result = self._generate_custom_voice(text) - elif self.voice_mode == VOICE_MODE_CLONE: - result = self._generate_voice_clone(text) - else: - raise ValueError(f"Unknown voice mode: {self.voice_mode}") - - if not isinstance(result, (tuple, list)) or not result: - raise RuntimeError("Qwen API returned an invalid result") - - audio_path = result[0] # First element is the audio file path - if not isinstance(audio_path, (str, Path)) or not audio_path: - raise RuntimeError("Qwen API did not return an audio file path") - - source = Path(audio_path) - if not source.exists(): - raise RuntimeError(f"Generated audio file not found: {audio_path}") - - destination = Path(parts_dir) / f"part_{sub_num:02d}{source.suffix or '.wav'}" - shutil.copy2(source, destination) - - return destination - - # ------------------------------------------------------------------ - # API payloads - # ------------------------------------------------------------------ - - def _generate_custom_voice(self, text: str) -> Tuple: - """Generate audio using CustomVoice mode.""" - custom_api = self._resolve_api_name("/run_instruct", "/run_custom_voice", "/generate_custom_voice") - if custom_api == "/run_instruct": - payload = dict( - text=text, - lang_disp=self.language, - spk_disp=speaker_display_name(), - instruct=config.INSTRUCT, - ) - else: - payload = dict( - text=text, - language=self.language, - speaker=config.SPEAKER, - instruct=config.INSTRUCT, - ) - if self._endpoint_accepts_param(custom_api, "model_id_cv"): - payload["model_id_cv"] = CUSTOM_VOICE_MODEL_ID - elif self._endpoint_accepts_param(custom_api, "model_size"): - payload["model_size"] = MODEL_SIZE - - if self._endpoint_accepts_param(custom_api, "seed"): - payload["seed"] = self._seed - - return self.client.predict(**payload, api_name=custom_api) - - def _ref_audio_payload(self) -> Dict[str, Any]: - """Gradio file payload for the reference audio (built once, reused).""" - if self._ref_audio_filedata is None: - from gradio_client import handle_file - self._ref_audio_filedata = handle_file(self.voice_clone_ref_audio) - return self._ref_audio_filedata - - def _generate_voice_clone(self, text: str) -> Tuple: - """Generate audio using Voice Clone mode.""" - if not Path(self.voice_clone_ref_audio).exists(): - raise FileNotFoundError(f"Reference audio not found: {self.voice_clone_ref_audio}") - - if self.clone_client is None: - raise RuntimeError("Voice Clone client is not initialized. Is the Base-model demo running?") - - clone_api = self._resolve_api_name("/run_voice_clone", "/generate_voice_clone", - api_info=self.clone_api_info) - use_xvector = config.XVECTOR_ONLY or not self.voice_clone_ref_text - - if clone_api == "/run_voice_clone": - payload = dict( - ref_aud=self._ref_audio_payload(), - ref_txt=self.voice_clone_ref_text, - use_xvec=use_xvector, - text=text, - lang_disp=self.language, - ) - else: - payload = dict( - ref_audio=self._ref_audio_payload(), - ref_text=self.voice_clone_ref_text, - target_text=text, - language=self.language, - use_xvector_only=use_xvector, - ) - optional_params = { - "model_size": MODEL_SIZE, - "seed": self._seed, - } - for name, value in optional_params.items(): - if self._endpoint_accepts_param(clone_api, name, api_info=self.clone_api_info): - payload[name] = value - - return self.clone_client.predict(**payload, api_name=clone_api) - - -class FasterTTSClient(_BaseTTSClient): - """Generates audio chunks through a faster-qwen3-tts server. - - Talks to the OpenAI-compatible server shipped in the faster-qwen3-tts - repository (examples/openai_server.py). The reference voice (ref audio, - ref text) and language are configured on the server itself via - --ref-audio/--ref-text or a --voices JSON file; this client only sends - text. Unlike the Qwen demo, the server performs one generation per - request, so long chunks are sub-chunked client-side. - """ - - def __init__(self, voice: Optional[str] = None, api_url: Optional[str] = None, - quiet: bool = False): - # Quiet before connecting so connect-time status lines never reach - # a screen the TUI run view owns. - self.quiet = bool(quiet) - self.voice = voice or config.FASTER_VOICE - self.api_url = (api_url or config.FASTER_API_URL).rstrip("/") - self._check_health() - - def _check_health(self) -> None: - """Verify the server is reachable and its model is loaded.""" - url = f"{self.api_url}/health" - try: - with urllib.request.urlopen(url, timeout=10) as response: - payload = json.loads(response.read().decode("utf-8")) - except Exception as exc: - raise RuntimeError( - f"Faster TTS server not reachable at {url}: {exc}. " - "Start the faster-qwen3-tts OpenAI-compatible server first " - "(see the 'Faster backend' section of the README)." - ) from exc - if not payload.get("model_loaded"): - raise RuntimeError( - "The faster TTS server is running but its model is not loaded yet; " - "wait for model download and startup to finish, then retry." - ) - self._report(f"[OK] Connected to faster TTS API at {self.api_url} (voice '{self.voice}')") - self._report(f"[INFO] The server silently falls back to its first configured voice if " - f"'{self.voice}' is not defined in its voice config (see README).") - - # ------------------------------------------------------------------ - # HTTP requests - # ------------------------------------------------------------------ - - def _request_pcm(self, text: str) -> bytes: - """POST one sub-chunk and return raw 16-bit mono PCM bytes.""" - url = f"{self.api_url}/v1/audio/speech" - payload = json.dumps({ - "model": "tts-1", - "input": text, - "voice": self.voice, - "response_format": "pcm", - }).encode("utf-8") - request = urllib.request.Request( - url, data=payload, headers={"Content-Type": "application/json"}, method="POST") - try: - with urllib.request.urlopen(request, timeout=config.API_TIMEOUT) as response: - pcm = response.read() - except urllib.error.HTTPError as exc: - detail = "" - try: - detail = exc.read().decode("utf-8", errors="replace")[:200] - except Exception: - pass - raise RuntimeError(f"Faster TTS server returned HTTP {exc.code}: {detail}") from exc - except urllib.error.URLError as exc: - raise RuntimeError(f"Faster TTS request failed: {exc.reason}") from exc - if not pcm: - raise RuntimeError("Faster TTS server returned empty audio") - return pcm - - # ------------------------------------------------------------------ - # Chunk generation - # ------------------------------------------------------------------ - - def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: - """Generate one audio chunk; returns its path in the chunks folder.""" - try: - sub_chunks = split_into_chunks(text, max_words=config.CHUNK_SIZE) - if not sub_chunks: - raise RuntimeError("No text to synthesize") - - pcm_parts: List[bytes] = [] - with self._chunk_heartbeat(chunk_num): - for sub_num, sub_text in enumerate(sub_chunks, 1): - pcm = self._request_pcm(sub_text) - pcm_parts.append(pcm) - - output_path = self._chunk_path(chunk_num, ".wav") - with wave.open(str(output_path), "wb") as wav_file: - wav_file.setnchannels(1) - wav_file.setsampwidth(2) - wav_file.setframerate(SAMPLE_RATE) - wav_file.writeframes(b"".join(pcm_parts)) - - logger.debug("Chunk %d generated (%d sub-chunks)", chunk_num, len(sub_chunks)) - return str(output_path) - - except ConversionCancelled: - raise - except Exception as exc: - logger.error("Faster chunk processing failed for chunk %d: %s", chunk_num, exc) - return None - - -class AudioCppTTSClient(_BaseTTSClient): - """Generates audio chunks through an audio.cpp audiocpp_server. - - Talks to the OpenAI-style HTTP API of audiocpp_server, which hosts TTS - model families through a native ggml runtime (GGUF weights, no Python - serving stack). The server API is family-agnostic; the family and task - of the configured model entry are read from GET /v1/models at startup - and adapt the request payload (language field style, style instructions) - through AUDIOCPP_FAMILY_PROFILES. The entry's voice capability - (audiocpp_entry_voice_capability: speaker / clone / design) decides how - its voice is supplied; all three are resolved server-side from the - request's "voice"/"instructions" fields: - - - Speaker mode (--voice with a built-in speaker name, or no flag on a - CustomVoice entry): Qwen3-TTS CustomVoice only. A built-in speaker - name (e.g. "Vivian") is passed through, plus the INSTRUCT style - prompt. The selected entry must be the CustomVoice model (capability - == speaker); a speaker name on a non-speaker entry is treated as a - server-side preset instead. - - Preset mode (--voice NAME): a voice configured on the server - (``voice_presets`` or ``voice_dir`` in its config, e.g. a cloning - reference). The name is validated against GET /v1/audio/voices at - startup because an unresolvable name would silently fall back to - plain TTS on a clone-based model instead of failing. When - AUDIOCPP_CLONE_MODEL_ID names a second server entry of the same - family (typically the Qwen Base model), preset requests are routed - to it. Selecting a non-speaker --voice on a CustomVoice primary with - a clone id configured is the documented way to switch a speaker setup - to cloning; without a clone id the voice is validated against the - server's voice library. - - Voice design (task "vdes" entries, e.g. Qwen3-TTS VoiceDesign): the - voice is described in natural language through ``instructions``, - which is required and sent with every request (no ``voice`` field). - A constant per-run seed keeps the designed voice consistent across - chunk boundaries. - - The entry's capability decides how an explicit --voice is read: on a - speaker-capable entry a name that matches a built-in speaker selects - speaker mode, and every other name is a server-side preset. With no - --voice the entry's capability picks the mode: design entries require - --instructions; speaker entries use the built-in CustomVoice speaker - in config.SPEAKER; clone entries (the Base model, and every other - family) fail fast with a hint to pass --voice, instead of silently - synthesizing with a random default voice. - - ``instructions`` also works on non-design entries, where it acts as a - generic style/delivery instruction (voice control): families that read - it (OmniVoice, Qwen3-TTS CustomVoice, ...) shape the voice or delivery - accordingly, and others ignore it. On instruction-conditioned families - without built-in speakers it may replace --voice entirely (the - instruction defines the voice). Extra request options (``--option - KEY=VALUE``, e.g. emotion, voice_id, speed) are forwarded verbatim in - the request's "options" object, which is the server's generic - pass-through for per-model controls. - - Chunking: text is split client-side into sub-requests of at most - config.CHUNK_SIZE words each; each sub-request returns a complete - WAV file and the parts are concatenated with the same lossless path - used for the Qwen client. - """ - - def __init__(self, voice: Optional[str] = None, language: Optional[str] = None, - api_url: Optional[str] = None, - model_id: Optional[str] = None, - instructions: Optional[str] = None, - request_options: Optional[Dict[str, str]] = None, - quiet: bool = False): - # Quiet before connecting so connect-time status lines never reach - # a screen the TUI run view owns. - self.quiet = bool(quiet) - self.api_url = (api_url or config.AUDIOCPP_API_URL).rstrip("/") - # Per-run model selection: the --model CLI flag overrides config; an - # empty value is resolved at connect time when the server hosts exactly - # one entry, so multi-model servers don't require editing config.py. - self.model_id = (model_id if model_id is not None - else config.AUDIOCPP_MODEL_ID) or "" - self._model_id_explicit = bool(self.model_id) - # Validate before connecting so bad values fail fast without a server. - self.language = normalize_language( - language if language is not None else config.LANGUAGE) - # One seed value per run, reused for every request (see - # _resolve_request_seed). Unlike the Qwen demo, audio.cpp has no - # negative "randomize" seed, so a negative value means "send no seed - # at all" (see _request_wav) and the server randomizes. - self._seed = _resolve_request_seed() - # Voice selection (the --voice name). preset_mode / speaker_mode are - # resolved in _connect: a --voice that names a built-in CustomVoice - # speaker on a speaker-capable entry selects speaker mode; every - # other name (and any name on a clone-capable entry) is a server-side - # preset. preset_mode gates _select_model's reroute to - # AUDIOCPP_CLONE_MODEL_ID and the INSTRUCT style-prompt logic. The - # request's "voice" field (self.voice) is filled in _connect per the - # mode. - self.preset_mode = False - self.speaker_mode = False - self.voice = voice or None - # Style/voice-design instruction sent with every request (the CLI - # --instructions flag overrides AUDIOCPP_INSTRUCTIONS in config.py). - # For task "vdes" entries it describes the voice to design; for other - # families it is a generic style instruction when the model reads one. - self.instructions = (instructions if instructions is not None - else config.AUDIOCPP_INSTRUCTIONS or "").strip() - # Free-form per-request options (--option KEY=VALUE) forwarded in the - # request's "options" object; models ignore keys they don't know. - self.request_options: Dict[str, str] = dict(request_options or {}) - # Set during _connect: design_mode for "vdes" entries, instruction_voice - # when a family without built-in speakers gets its voice from the - # instruction alone (no voice field). self.voice is also finalized - # there (the speaker/preset name, or config.SPEAKER for the default). - self.design_mode = False - self.instruction_voice = False - # Family and task of the selected model entry and the family's request - # profile; all are resolved from GET /v1/models during _connect. - self.family = "" - self.task = AUDIOCPP_TASK_TTS - self.profile = AUDIOCPP_DEFAULT_FAMILY_PROFILE - self._connect() - - # ------------------------------------------------------------------ - # Connection - # ------------------------------------------------------------------ - - def _connected(self, mode: str) -> None: - """Report the resolved connection (MODE: speaker/voice/... label).""" - self._report(f"[OK] Connected to audio.cpp server at {self.api_url} " - f"(model '{self.model_id}', family '{self.family}', " - f"{mode})") - - def _connect(self) -> None: - """Health-check the server and resolve the model, family, task, and voice. - - The entry's voice capability (audiocpp_entry_voice_capability, from - family/task/id) plus the caller's --voice/--instructions pick the - mode. An explicit --voice on a speaker-capable (CustomVoice) entry - that names a built-in speaker selects speaker mode; every other - --voice is a server-side preset, validated against the server's - voice library (and rerouted to AUDIOCPP_CLONE_MODEL_ID when set). - With no --voice, design entries require --instructions, speaker- - capable entries use the built-in config.SPEAKER, and clone entries - fail fast with a hint instead of silently synthesizing with a - random default voice. - """ - self._check_health() - models = self._list_models() - self._auto_pick_model_id(models) - if self.voice is not None: - # Explicit --voice: decide between speaker mode and a server-side - # preset. A name matching a built-in CustomVoice speaker on a - # speaker-capable primary selects speaker mode; every other name - # (and any name when the primary entry is absent) is a preset, - # validated against the server's voice library and rerouted to - # AUDIOCPP_CLONE_MODEL_ID when configured. - primary = next((m for m in models if m["id"] == self.model_id), - None) - if primary is not None: - self._resolve_family(models) - self._resolve_task(models) - capability = audiocpp_entry_voice_capability( - self.family, self.task, self.model_id) - if capability == AUDIOCPP_VOICE_SPEAKER \ - and is_builtin_speaker(self.voice): - self._require_synthesis_task(models) - self.voice = speaker_display_name_for(self.voice) - self.speaker_mode = True - self._connected(f"speaker '{self.voice}'") - if not self.speaker_mode: - # Server-side preset (--voice): validate it and route to - # the clone model entry when AUDIOCPP_CLONE_MODEL_ID is set. - self.preset_mode = True - self._select_model(models) - self._require_model_id(models) - self._resolve_family(models) - self._resolve_task(models) - self._require_synthesis_task(models) - if self.design_mode: - raise RuntimeError( - f"--voice cannot be used with the voice design model " - f"'{self.model_id}': the voice is described by the " - "--instructions text instead (see README).") - self._check_voice() - self._connected(f"voice '{self.voice}'") - else: - # No flag: the entry's capability picks the default mode. - self._require_model_id(models) - self._resolve_family(models) - self._resolve_task(models) - self._require_synthesis_task(models) - capability = audiocpp_entry_voice_capability( - self.family, self.task, self.model_id) - if self.design_mode: - if not self.instructions: - raise RuntimeError( - f"The audio.cpp model '{self.model_id}' (family " - f"'{self.family}') is a voice design model: pass a " - "description of the voice to synthesize with, e.g. " - '--instructions "A warm adult female narrator with a ' - 'British accent" (see README).') - self._connected("voice design") - self._report(f"[INFO] Designing the voice from: {self.instructions}") - elif capability == AUDIOCPP_VOICE_SPEAKER: - # No flag on a CustomVoice entry: the built-in config.SPEAKER. - self.voice = speaker_display_name() - self.speaker_mode = True - self._connected(f"speaker '{self.voice}'") - elif self.instructions: - # Families without built-in speakers can still get their voice - # from the instruction alone (e.g. OmniVoice voice design). - self.instruction_voice = True - self._connected("instruction voice") - self._report(f"[INFO] Designing the voice from: {self.instructions}") - else: - raise RuntimeError( - f"The audio.cpp model '{self.model_id}' (family " - f"'{self.family}') has no built-in speakers, so its voice " - "must come from the server: rerun with --voice NAME " - "matching a voice_preset or voice_dir entry in the server " - "config, or describe a voice with --instructions for " - "families that support it, or select the CustomVoice entry " - "for built-in speakers (see README).") - if self.instructions and not self.design_mode and not self.instruction_voice: - self._report(f"[INFO] Sending instruction with every request: {self.instructions}") - self._report("[INFO] Its effect (style, emotion, delivery) depends on the " - "model family; models without instruction support ignore it.") - if config.AUDIOCPP_UNLOAD_MODELS: - self._unload_server_models() - - def _require_synthesis_task(self, models: List[Dict[str, str]]) -> None: - """Reject model entries whose task is not a TTS synthesis task.""" - if self.task in AUDIOCPP_SYNTHESIS_TASKS: - return - available = ", ".join(model["id"] for model in models) or "none" - raise RuntimeError( - f"The audio.cpp model '{self.model_id}' has task " - f"'{self.task}'; audiobook.py can only synthesize with TTS " - f"model entries (tasks {', '.join(AUDIOCPP_SYNTHESIS_TASKS)}). " - f"Pick a synthesis entry with --model (available: {available})." - ) - - def _unload_server_models(self) -> None: - """Ask the server to unload every loaded model before generating. - - Lazy-loaded entries stay resident until the server exits (unless its - max_loaded_models setting bounds residency), so switching between - configured models across runs can exhaust device memory. Unloading - first frees those leftovers; this run's model reloads transparently - on its first request. Failures only warn: an older server without - the endpoint, or a busy one, must not block a working setup. - Controlled by config.AUDIOCPP_UNLOAD_MODELS (the TUI Settings - "Unload models" option). - """ - request = urllib.request.Request( - f"{self.api_url}/v1/tasks/unload_all_models", data=b"", - method="POST", headers={"Content-Type": "application/json"}) - try: - with urllib.request.urlopen(request, timeout=10) as response: - payload = json.loads(response.read().decode("utf-8")) - except Exception as exc: - self._report(f"[WARNING] Could not unload previously loaded models at " - f"{self.api_url}: {exc}") - return - unloaded = [entry for entry in (payload.get("unloaded") or []) - if isinstance(entry, str)] - if unloaded: - self._report(f"[OK] Unloaded {len(unloaded)} model(s) from server memory: " - f"{', '.join(unloaded)}") - else: - logger.debug("No loaded audio.cpp models to unload at %s", self.api_url) - - def _get_json(self, path: str, timeout: int = 10) -> Dict[str, Any]: - """GET a JSON document from the server.""" - url = f"{self.api_url}{path}" - try: - with urllib.request.urlopen(url, timeout=timeout) as response: - return json.loads(response.read().decode("utf-8")) - except urllib.error.HTTPError as exc: - detail = "" - try: - detail = exc.read().decode("utf-8", errors="replace")[:200] - except Exception: - pass - raise RuntimeError( - f"audio.cpp server returned HTTP {exc.code} for {path}: {detail}") from exc - except urllib.error.URLError as exc: - raise RuntimeError(f"audio.cpp request failed for {path}: {exc.reason}") from exc - - def _check_health(self) -> None: - """Verify the server is reachable and reports healthy.""" - try: - payload = self._get_json("/health") - except Exception as exc: - raise RuntimeError( - f"audio.cpp server not reachable at {self.api_url}: {exc}. " - "Start audiocpp_server first (see the 'audio.cpp backend' " - "section of the README)." - ) from exc - if payload.get("status") != "ok": - raise RuntimeError( - f"The audio.cpp server at {self.api_url} reports status " - f"{payload.get('status')!r} instead of 'ok'") - - def _list_models(self) -> List[Dict[str, str]]: - """Fetch the (id, family, task) triples reported by the server.""" - try: - payload = self._get_json("/v1/models") - except Exception as exc: - raise RuntimeError( - f"The audio.cpp server at {self.api_url} did not answer " - f"/v1/models: {exc}") from exc - entries = payload.get("data") or [] - models: List[Dict[str, str]] = [] - for entry in entries: - if isinstance(entry, dict) and entry.get("id"): - models.append({ - "id": entry["id"], - "family": entry.get("family") or "", - "task": entry.get("task") or "", - }) - return models - - def _auto_pick_model_id(self, models: List[Dict[str, str]]) -> None: - """Resolve an empty model id when the server hosts exactly one entry. - - Multi-model servers generated with several lazily-loaded entries can - be used without editing app/converter/config.py: leave AUDIOCPP_MODEL_ID - (and ``--model``) unset, and the single hosted entry is chosen - automatically. With more than one entry an explicit choice is required - (via ``--model`` or AUDIOCPP_MODEL_ID), since guessing would risk - synthesizing a whole book with the wrong family. - """ - if self.model_id: - return - if len(models) == 1: - self.model_id = models[0]["id"] - logger.info( - "AUDIOCPP_MODEL_ID is unset; using the only server entry '%s'", - self.model_id) - else: - logger.debug( - "AUDIOCPP_MODEL_ID is unset and the server hosts %d entries; " - "an explicit --model or config id is required", - len(models)) - - def _require_model_id(self, models: List[Dict[str, str]]) -> None: - """Verify the model id chosen for this run exists on the server. - - Speaker mode needs AUDIOCPP_MODEL_ID (the CustomVoice entry). - Preset mode validates whichever id _select_model resolved, so a - server hosting only a cloning model works for --voice. The default - error distinguishes the two so the fix is obvious. - """ - model_ids = [model["id"] for model in models] - if self.model_id and self.model_id in model_ids: - return - configured = ", ".join(model_ids) or "none" - if not self.model_id: - raise RuntimeError( - f"The audio.cpp server at {self.api_url} hosts {len(model_ids)} " - f"model entries ({configured}); audiobook.py needs to know which " - "one to use. Pass --model when converting, or set " - "AUDIOCPP_MODEL_ID in app/converter/config.py to one of them " - "(see README)." - ) - if self.preset_mode: - raise RuntimeError( - f"The audio.cpp server at {self.api_url} has no model id " - f"'{self.model_id}' or clone model id " - f"'{config.AUDIOCPP_CLONE_MODEL_ID}' (configured: {configured}). " - "Add a TTS model entry for the family you want to the server " - "config and match AUDIOCPP_MODEL_ID / AUDIOCPP_CLONE_MODEL_ID " - "in app/converter/config.py to its id, or select it per run with " - "--model (see README)." - ) - raise RuntimeError( - f"The audio.cpp server at {self.api_url} has no model id " - f"'{self.model_id}' (configured: {configured}). Select the " - "Qwen3-TTS CustomVoice entry for built-in speakers, or rerun " - "with --voice NAME matching a voice_preset or voice_dir entry " - "on any TTS model (see README)." - ) - - def _select_model(self, models: List[Dict[str, str]]) -> None: - """Pick the model for preset (cloning) requests. - - Defaults to the primary model id. When AUDIOCPP_CLONE_MODEL_ID is - configured and present on the server, preset requests are routed - to it instead, so one server can host the CustomVoice model for - speaker mode and the Base model for cloning (Qwen3-TTS setups). - A clone id that names a model of a different family is ignored - with a warning, since preset requests must synthesize with the - family the run is configured for. - """ - clone_model_id = config.AUDIOCPP_CLONE_MODEL_ID - if not clone_model_id or clone_model_id == self.model_id: - return - families = {model["id"]: model["family"] for model in models} - if clone_model_id not in families: - # A qwen3_tts primary without its clone entry silently degrades - # (presets are ignored on the CustomVoice model), so that case - # keeps the warning; single-model servers of other families are - # the normal configuration and only get a debug note. - primary_family = families.get(self.model_id) or "" - if primary_family == AUDIOCPP_FAMILY_QWEN3_TTS: - logger.warning( - "AUDIOCPP_CLONE_MODEL_ID %r is not configured on the audio.cpp " - "server; preset requests use '%s' instead", - clone_model_id, self.model_id) - else: - logger.debug( - "AUDIOCPP_CLONE_MODEL_ID %r is not configured on the audio.cpp " - "server; preset requests use '%s' instead", - clone_model_id, self.model_id) - return - primary_family = families.get(self.model_id) - clone_family = families[clone_model_id] - if primary_family and clone_family and primary_family != clone_family: - logger.warning( - "AUDIOCPP_CLONE_MODEL_ID %r hosts family %r, but " - "AUDIOCPP_MODEL_ID %r hosts %r; preset requests stay on " - "'%s'. Point both ids at the same model entry in " - "app/converter/config.py (single-model servers use the same id " - "for both)", - clone_model_id, clone_family, self.model_id, primary_family, - self.model_id) - return - self.model_id = clone_model_id - - def _resolve_family(self, models: List[Dict[str, str]]) -> None: - """Resolve the selected model's family and its request profile. - - The family comes from GET /v1/models; a missing family is an unknown - family that falls through to the generic (clone-only) profile rather - than guessing a specific one — audiocpp_server always reports family - for entries its server.json describes. - """ - entry = next( - (model for model in models if model["id"] == self.model_id), None) - family = (entry["family"] if entry is not None else "") or "" - self.family = family - self.profile = AUDIOCPP_FAMILY_PROFILES.get( - family, AUDIOCPP_DEFAULT_FAMILY_PROFILE) - if not family: - logger.debug("Model '%s' reported no family; using the generic " - "profile", self.model_id) - elif family not in AUDIOCPP_FAMILY_PROFILES: - logger.info( - "audio.cpp family '%s' has no dedicated profile; using the " - "generic profile (voice cloning via --voice, model-detected " - "language)", family) - - def _resolve_task(self, models: List[Dict[str, str]]) -> None: - """Resolve the selected model's task (tts, clon, vdes, ...) and set - design mode for voice design entries. - - The task comes from GET /v1/models and is fixed per server entry by - its server.json config (a VoiceDesign model must be hosted with - "task": "vdes"). Servers that predate the task field hosted plain - TTS models, so a missing task is treated as tts. - """ - entry = next( - (model for model in models if model["id"] == self.model_id), None) - task = (entry["task"] if entry is not None else "") or "" - if not task: - task = AUDIOCPP_TASK_TTS - logger.debug("Model '%s' reported no task; assuming tts", - self.model_id) - self.task = task - self.design_mode = task == AUDIOCPP_TASK_VDES - - def _check_voice(self) -> None: - """Verify the requested voice is available on the server. - - A voice name that matches no server preset or voice-library wav - would be passed through to the model as a cached voice id; on the - Base (cloning) model that is silently ignored and plain TTS audio - comes back, so preset names are validated up front. When the - voices endpoint cannot be queried, validation is skipped with a - warning rather than blocking the run. - """ - query = urllib.parse.urlencode({"model": self.model_id}) - try: - payload = self._get_json(f"/v1/audio/voices?{query}") - except Exception as exc: - logger.warning("Could not list server voices; skipping voice " - "validation: %s", exc) - return - voices = payload.get("voices") or [] - if self.voice not in voices: - available = ", ".join(str(v) for v in voices) or "none" - raise RuntimeError( - f"Voice '{self.voice}' is not available on the audio.cpp server " - f"(available: {available}). Configure it as a voice_preset or " - "voice_dir entry in the server config, or pass a listed name " - "with --voice (see README)." - ) - - # ------------------------------------------------------------------ - # HTTP requests - # ------------------------------------------------------------------ - - def _request_wav(self, text: str) -> bytes: - """POST one sub-chunk and return the raw WAV bytes.""" - url = f"{self.api_url}/v1/audio/speech" - payload: Dict[str, Any] = { - "model": self.model_id, - "input": text, - } - # Design models take no voice field (the voice comes from the - # instruction); instruction-voice runs on families without built-in - # speakers omit it too, since no speaker or preset was requested. - if not self.design_mode and not self.instruction_voice: - payload["voice"] = self.voice - if self.profile.language_style == AUDIOCPP_LANG_DISPLAY: - payload["language"] = self.language - elif self.profile.language_style == AUDIOCPP_LANG_ISO: - iso_code = LANGUAGE_ISO_CODES.get(self.language) - if iso_code: - payload["language"] = iso_code - else: - # "Auto": no code to send, so let the server pick its default. - logger.debug("%s: no language code for %r; omitted from request", - self.family, self.language) - if self._seed >= 0: - # audio.cpp has no negative "randomize" seed; a negative seed - # means "let the server randomize", so the field is omitted. - payload["seed"] = self._seed - if self.instructions: - # Explicit voice-design or style instruction (required for task - # "vdes" entries; a Ctrl/style control on families that read it). - payload["instructions"] = self.instructions - elif not self.preset_mode and config.INSTRUCT \ - and self.profile.sends_instructions: - # Style instruction for the Qwen3-TTS CustomVoice speakers; - # ignored by the Base (cloning) model and other families. - payload["instructions"] = config.INSTRUCT - if self.request_options: - # Generic per-model controls (--option KEY=VALUE): forwarded - # verbatim; the model ignores keys it does not know. - payload["options"] = dict(self.request_options) - request = urllib.request.Request( - url, data=json.dumps(payload).encode("utf-8"), - headers={"Content-Type": "application/json"}, method="POST") - timeout = config.API_TIMEOUT - try: - with urllib.request.urlopen(request, timeout=timeout) as response: - wav = response.read() - except urllib.error.HTTPError as exc: - detail = "" - try: - detail = exc.read().decode("utf-8", errors="replace")[:200] - except Exception: - pass - raise RuntimeError(f"audio.cpp server returned HTTP {exc.code}: {detail}") from exc - except urllib.error.URLError as exc: - raise RuntimeError(f"audio.cpp request failed: {exc.reason}") from exc - if len(wav) < 12 or wav[:4] != b"RIFF" or wav[8:12] != b"WAVE": - raise RuntimeError("audio.cpp server returned audio that is not a WAV file") - return wav - - # ------------------------------------------------------------------ - # Chunk generation - # ------------------------------------------------------------------ - - def generate_chunk(self, text: str, chunk_num: int) -> Optional[str]: - """Generate one audio chunk; returns its path in the chunks folder. - - The text is split into sub-requests of at most ``config.CHUNK_SIZE`` - words each; 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=config.CHUNK_SIZE) - if not sub_texts: - raise RuntimeError("No text to synthesize") - - output_path: Optional[Path] = None - with tempfile.TemporaryDirectory(prefix="tts_parts_") as parts_dir, \ - self._chunk_heartbeat(chunk_num): - part_paths = [] - for sub_num, sub_text in enumerate(sub_texts, 1): - wav = self._request_wav(sub_text) - destination = Path(parts_dir) / f"part_{sub_num:02d}.wav" - destination.write_bytes(wav) - part_paths.append(destination) - if len(part_paths) == 1: - output_path = self._chunk_path(chunk_num, ".wav") - shutil.copy2(part_paths[0], output_path) - else: - output_path = self._chunk_path(chunk_num, ".wav") - concat_audio_files(part_paths, output_path) - - logger.debug("Chunk %d generated successfully (%d sub-request(s))", - chunk_num, len(sub_texts)) - return str(output_path) - - except ConversionCancelled: - raise - except Exception as exc: - logger.error("audio.cpp chunk processing failed for chunk %d: %s", - chunk_num, exc) - return None diff --git a/app/tests/test_audio.py b/app/tests/test_audio.py index ef5e92a..6c67294 100644 --- a/app/tests/test_audio.py +++ b/app/tests/test_audio.py @@ -60,12 +60,7 @@ class CleanupChunksTests(unittest.TestCase): (chunks_dir / "chunk_0002.wav").write_bytes(b"stale") (chunks_dir / "keep.txt").write_bytes(b"keep") - original = audio.CHUNKS_FOLDER - audio.CHUNKS_FOLDER = chunks_dir - try: - cleanup_chunks() - finally: - audio.CHUNKS_FOLDER = original + cleanup_chunks(chunks_dir) self.assertFalse((chunks_dir / "chunk_0001.wav").exists()) self.assertFalse((chunks_dir / "chunk_0002.wav").exists()) @@ -77,12 +72,7 @@ class CleanupChunksTests(unittest.TestCase): (chunks_dir / "chapter_0001.m4b").write_bytes(b"stale") (chunks_dir / "chunk_0001.wav").write_bytes(b"stale") - original = audio.CHUNKS_FOLDER - audio.CHUNKS_FOLDER = chunks_dir - try: - cleanup_chunks() - finally: - audio.CHUNKS_FOLDER = original + cleanup_chunks(chunks_dir) self.assertFalse((chunks_dir / "chapter_0001.m4b").exists()) self.assertFalse((chunks_dir / "chunk_0001.wav").exists()) @@ -477,9 +467,7 @@ class CombineChunksPrintTests(unittest.TestCase): 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) + self.addCleanup(self._tmp.cleanup) def _combine(self, total_chunks, chunk_results, intermediate=False): buf = io.StringIO() @@ -495,6 +483,7 @@ class CombineChunksPrintTests(unittest.TestCase): redirect_stdout(buf): ok = audio.combine_chunks( total_chunks, Path("out.m4b"), chunk_results, + chunks_dir=Path(self._tmp.name), output_format="m4b", intermediate=intermediate) self.assertTrue(ok) return buf.getvalue() diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index 1f146e5..47dc5d4 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -22,7 +22,7 @@ class FormatLaunchHintTests(unittest.TestCase): class RegistryTests(unittest.TestCase): def setUp(self): # The registry is built lazily on first access (the backend modules - # pull in converter.tts and its deps, which are only available inside + # pull in converter.clients and its deps, which are only available inside # the managed venv). Trigger the build so these tests don't depend on # another test class having called detect_all() first. get("audiocpp") diff --git a/app/tests/test_converter.py b/app/tests/test_converter.py index 7aa9c69..53e2897 100644 --- a/app/tests/test_converter.py +++ b/app/tests/test_converter.py @@ -9,7 +9,14 @@ from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch -from converter import config, tts +from converter import config +from converter.clients import ( + BACKEND_AUDIOCPP, + BACKEND_FASTER, + BACKEND_QWEN, + VOICE_MODE_CLONE, + VOICE_MODE_CUSTOM, +) from converter import converter as converter_mod from converter.converter import ( AudiobookConverter, @@ -56,17 +63,17 @@ class ConfigurationValidationTests(unittest.TestCase): def test_language_defaults_to_config(self): with patch("converter.converter.QwenTTSClient") as mock_tts: - AudiobookConverter(backend=tts.BACKEND_QWEN) + AudiobookConverter(backend=BACKEND_QWEN) self.assertEqual(mock_tts.call_args.kwargs["language"], config.LANGUAGE) def test_output_format_defaults_to_config(self): with patch("converter.converter.QwenTTSClient"): - converter = AudiobookConverter(backend=tts.BACKEND_QWEN) + converter = AudiobookConverter(backend=BACKEND_QWEN) self.assertEqual(converter.output_format, config.AUDIO_FORMAT) def test_language_normalized_before_tts_client(self): with patch("converter.converter.QwenTTSClient") as mock_tts: - converter = AudiobookConverter(language="ja", backend=tts.BACKEND_QWEN) + converter = AudiobookConverter(language="ja", backend=BACKEND_QWEN) self.assertEqual(converter.language, "Japanese") self.assertEqual(mock_tts.call_args.kwargs["language"], "Japanese") @@ -129,40 +136,40 @@ class NarratorTagTests(unittest.TestCase): converter = AudiobookConverter.__new__(AudiobookConverter) converter.voice_mode = voice_mode converter.voice_clone_ref_audio = ref_audio - converter.backend = tts.BACKEND_QWEN + converter.backend = BACKEND_QWEN converter.voice = None converter.instructions = instructions return converter def test_custom_voice_uses_speaker_display_name(self): - self.assertEqual(self._converter(tts.VOICE_MODE_CUSTOM)._narrator_tag(), + self.assertEqual(self._converter(VOICE_MODE_CUSTOM)._narrator_tag(), "Vivian") def test_multi_word_display_name_gets_underscores(self): with patch.object(config, "SPEAKER", "uncle_fu"): - self.assertEqual(self._converter(tts.VOICE_MODE_CUSTOM)._narrator_tag(), + self.assertEqual(self._converter(VOICE_MODE_CUSTOM)._narrator_tag(), "Uncle_Fu") def test_clone_uses_reference_audio_stem(self): - self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/ref.wav")._narrator_tag(), + self.assertEqual(self._converter(VOICE_MODE_CLONE, "/x/ref.wav")._narrator_tag(), "ref") def test_clone_stem_spaces_become_underscores(self): - self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/my voice.wav")._narrator_tag(), + self.assertEqual(self._converter(VOICE_MODE_CLONE, "/x/my voice.wav")._narrator_tag(), "my_voice") def test_invalid_characters_sanitized(self): - self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/bad:name?.wav")._narrator_tag(), + self.assertEqual(self._converter(VOICE_MODE_CLONE, "/x/bad:name?.wav")._narrator_tag(), "bad_name") def test_empty_after_sanitize_falls_back(self): - self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/???.wav")._narrator_tag(), + self.assertEqual(self._converter(VOICE_MODE_CLONE, "/x/???.wav")._narrator_tag(), "narrator") def _audiocpp_converter(self, voice=None, instructions=None): - converter = self._converter(tts.VOICE_MODE_CUSTOM, + converter = self._converter(VOICE_MODE_CUSTOM, instructions=instructions) - converter.backend = tts.BACKEND_AUDIOCPP + converter.backend = BACKEND_AUDIOCPP converter.voice = voice return converter @@ -202,7 +209,7 @@ class NarratorTagTests(unittest.TestCase): with patch("builtins.input", side_effect=AssertionError("should not prompt")): _, planned = AudiobookConverter.preflight_overwrites( - tts.BACKEND_AUDIOCPP, None, tts.VOICE_MODE_CUSTOM, + BACKEND_AUDIOCPP, None, VOICE_MODE_CUSTOM, None, "mp3", instructions="A warm narrator") self.assertEqual(planned, [(converter_mod.BOOKS_FOLDER / "book.txt", "book_designed")]) @@ -324,8 +331,8 @@ class DebugDumpTests(unittest.TestCase): def test_debug_flag_wiring(self): with patch("converter.converter.QwenTTSClient"): - self.assertFalse(AudiobookConverter(backend=tts.BACKEND_QWEN).debug) - self.assertTrue(AudiobookConverter(debug=True, backend=tts.BACKEND_QWEN).debug) + self.assertFalse(AudiobookConverter(backend=BACKEND_QWEN).debug) + self.assertTrue(AudiobookConverter(debug=True, backend=BACKEND_QWEN).debug) class SetupLoggingTests(unittest.TestCase): @@ -415,7 +422,7 @@ class ChunkProgressOutputTests(unittest.TestCase): def _converter(self): converter = AudiobookConverter.__new__(AudiobookConverter) - converter.backend = tts.BACKEND_AUDIOCPP + converter.backend = BACKEND_AUDIOCPP converter.speed = 1.0 converter.output_format = "mp3" converter.tts = MagicMock() @@ -514,14 +521,14 @@ class PreflightOverwritesTests(unittest.TestCase): (converter_mod.BOOKS_FOLDER / "book.txt").unlink() with patch("builtins.input", side_effect=AssertionError("should not prompt")): book_files, planned = AudiobookConverter.preflight_overwrites( - tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + BACKEND_QWEN, None, VOICE_MODE_CUSTOM, None, "mp3") self.assertEqual(book_files, []) self.assertEqual(planned, []) def test_new_book_planned_without_prompt(self): with patch("builtins.input", side_effect=AssertionError("should not prompt")): book_files, planned = AudiobookConverter.preflight_overwrites( - tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + BACKEND_QWEN, None, VOICE_MODE_CUSTOM, None, "mp3") self.assertEqual(len(book_files), 1) self.assertEqual(planned, [(book_files[0], "book_Vivian")]) @@ -529,14 +536,14 @@ class PreflightOverwritesTests(unittest.TestCase): (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") with patch("builtins.input", return_value=""): book_files, planned = AudiobookConverter.preflight_overwrites( - tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + BACKEND_QWEN, None, VOICE_MODE_CUSTOM, None, "mp3") self.assertEqual(planned, [(book_files[0], "book_Vivian")]) def test_existing_output_declined_is_skipped(self): (converter_mod.AUDIOBOOKS_FOLDER / "book_Vivian.mp3").write_bytes(b"existing") with patch("builtins.input", return_value="n"): book_files, planned = AudiobookConverter.preflight_overwrites( - tts.BACKEND_QWEN, None, tts.VOICE_MODE_CUSTOM, None, "mp3") + BACKEND_QWEN, None, VOICE_MODE_CUSTOM, None, "mp3") self.assertEqual(len(book_files), 1) self.assertEqual(planned, []) @@ -552,9 +559,9 @@ class RunOverwritePromptTests(unittest.TestCase): converter_mod.AUDIOBOOKS_FOLDER = Path(self._output_tmp.name) (converter_mod.BOOKS_FOLDER / "book.txt").write_text("hello world", encoding="utf-8") self.converter = AudiobookConverter.__new__(AudiobookConverter) - self.converter.voice_mode = tts.VOICE_MODE_CUSTOM + self.converter.voice_mode = VOICE_MODE_CUSTOM self.converter.voice_clone_ref_audio = None - self.converter.backend = tts.BACKEND_QWEN + self.converter.backend = BACKEND_QWEN self.converter.voice = None self.converter.instructions = None self.converter.speed = 1.0 diff --git a/app/tests/test_converter_progress.py b/app/tests/test_converter_progress.py index 0f708a1..00bcc46 100644 --- a/app/tests/test_converter_progress.py +++ b/app/tests/test_converter_progress.py @@ -14,7 +14,14 @@ from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch -from converter import config, tts +from converter import config +from converter.clients import ( + BACKEND_AUDIOCPP, + BACKEND_FASTER, + BACKEND_QWEN, + VOICE_MODE_CLONE, + VOICE_MODE_CUSTOM, +) from converter import converter as converter_mod from converter.converter import ( AudiobookConverter, @@ -26,24 +33,24 @@ from converter.converter import ( class VoiceModeForTests(unittest.TestCase): def test_faster_always_clones(self): - self.assertEqual(voice_mode_for(tts.BACKEND_FASTER), - tts.VOICE_MODE_CLONE) + self.assertEqual(voice_mode_for(BACKEND_FASTER), + VOICE_MODE_CLONE) def test_audiocpp_voice_clones(self): - self.assertEqual(voice_mode_for(tts.BACKEND_AUDIOCPP, voice="narrator"), - tts.VOICE_MODE_CLONE) + self.assertEqual(voice_mode_for(BACKEND_AUDIOCPP, voice="narrator"), + VOICE_MODE_CLONE) def test_audiocpp_no_voice_is_custom(self): - self.assertEqual(voice_mode_for(tts.BACKEND_AUDIOCPP), - tts.VOICE_MODE_CUSTOM) + self.assertEqual(voice_mode_for(BACKEND_AUDIOCPP), + VOICE_MODE_CUSTOM) def test_qwen_clone_wav_clones(self): - self.assertEqual(voice_mode_for(tts.BACKEND_QWEN, clone="x.wav"), - tts.VOICE_MODE_CLONE) + self.assertEqual(voice_mode_for(BACKEND_QWEN, clone="x.wav"), + VOICE_MODE_CLONE) def test_qwen_no_clone_is_custom(self): - self.assertEqual(voice_mode_for(tts.BACKEND_QWEN), - tts.VOICE_MODE_CUSTOM) + self.assertEqual(voice_mode_for(BACKEND_QWEN), + VOICE_MODE_CUSTOM) class PromptOverwriteConfirmTests(unittest.TestCase): @@ -96,7 +103,7 @@ class _ConvertFixture: with patch.object(converter_mod, "QwenTTSClient", return_value=MagicMock()): converter = AudiobookConverter( - voice_mode=tts.VOICE_MODE_CUSTOM, backend=tts.BACKEND_QWEN, + voice_mode=VOICE_MODE_CUSTOM, backend=BACKEND_QWEN, output_format="mp3", language="English", progress=progress, cancel=cancel) converter.tts.process_chunk_with_retry.return_value = "chunk_0001.wav" diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py index 02b7dc4..2b2ac1c 100644 --- a/app/tests/test_tts.py +++ b/app/tests/test_tts.py @@ -4,20 +4,43 @@ import io import json import tempfile import time +import urllib.error import unittest import wave from contextlib import redirect_stdout from pathlib import Path from unittest.mock import MagicMock, patch -from converter import config, tts -from converter.converter import AudiobookConverter -from converter.tts import ( +from converter import config +from converter import converter as converter_mod +from converter.clients import ( + AUDIOCPP_DEFAULT_FAMILY_PROFILE, + AUDIOCPP_FAMILY_PROFILES, + AUDIOCPP_LANG_OMIT, + AUDIOCPP_TASK_TTS, + AUDIOCPP_TASK_VDES, + AUDIOCPP_VOICE_CLONE, + AUDIOCPP_VOICE_DESIGN, + AUDIOCPP_VOICE_SPEAKER, + BACKEND_AUDIOCPP, + BACKEND_FASTER, + BACKEND_QWEN, + LANGUAGE_ISO_CODES, + MODEL_SIZE, + SAMPLE_RATE, + TTS_LANGUAGES, + VOICE_MODE_CLONE, + VOICE_MODE_CUSTOM, AudioCppTTSClient, FasterTTSClient, QwenTTSClient, + audiocpp_entry_voice_capability, normalize_language, ) +from converter.converter import AudiobookConverter + +# Chunks folder handed to clients whose tests never write chunk files. +_DUMMY_CHUNKS = Path(tempfile.gettempdir()) / "audiobook_tts_test_chunks" class NormalizeLanguageTests(unittest.TestCase): @@ -43,7 +66,7 @@ class NormalizeLanguageTests(unittest.TestCase): self.assertEqual(normalize_language("it"), "Italian") def test_all_supported_languages_round_trip(self): - for name in tts.TTS_LANGUAGES: + for name in TTS_LANGUAGES: self.assertEqual(normalize_language(name.lower()), name) def test_unknown_language_rejected_with_guidance(self): @@ -65,34 +88,34 @@ class QwenTTSClientLanguageTests(unittest.TestCase): def _make_client(self, **kwargs): with patch.object(QwenTTSClient, "_connect"): - return QwenTTSClient(**kwargs) + return QwenTTSClient(_DUMMY_CHUNKS, **kwargs) def test_default_follows_config_for_each_mode(self): - custom = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM) + custom = self._make_client(voice_mode=VOICE_MODE_CUSTOM) self.assertEqual(custom.language, config.LANGUAGE) - clone = self._make_client(voice_mode=tts.VOICE_MODE_CLONE, + clone = self._make_client(voice_mode=VOICE_MODE_CLONE, voice_clone_ref_audio="ref.wav") self.assertEqual(clone.language, config.LANGUAGE) def test_explicit_language_normalized(self): - client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM, language="ja") + client = self._make_client(voice_mode=VOICE_MODE_CUSTOM, language="ja") self.assertEqual(client.language, "Japanese") def test_invalid_language_fails_before_connect(self): with patch.object(QwenTTSClient, "_connect") as mock_connect: with self.assertRaises(ValueError): - QwenTTSClient(language="klingon") + QwenTTSClient(_DUMMY_CHUNKS, language="klingon") mock_connect.assert_not_called() def test_api_url_override_stored(self): - client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM, + client = self._make_client(voice_mode=VOICE_MODE_CUSTOM, api_url="http://10.0.0.5:7860") self.assertEqual(client.api_url, "http://10.0.0.5:7860") def test_api_url_override_used_by_connect(self): with patch.object(QwenTTSClient, "_init_client") as mk_init: client = QwenTTSClient.__new__(QwenTTSClient) - client.voice_mode = tts.VOICE_MODE_CUSTOM + client.voice_mode = VOICE_MODE_CUSTOM client.api_url = "http://10.0.0.5:7860" client._connect() mk_init.assert_called_once_with("http://10.0.0.5:7860", clone=False) @@ -105,24 +128,24 @@ class SeedResolutionTests(unittest.TestCase): def _make_client(self, **kwargs): with patch.object(QwenTTSClient, "_connect"): - return QwenTTSClient(**kwargs) + return QwenTTSClient(_DUMMY_CHUNKS, **kwargs) def test_constant_seed_draws_one_nonnegative_seed(self): with patch.object(config, "CONSTANT_SEED", True), \ patch.object(config, "SEED", -1): - client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM) + client = self._make_client(voice_mode=VOICE_MODE_CUSTOM) self.assertGreaterEqual(client._seed, 0) def test_explicit_seed_wins_over_constant_seed(self): with patch.object(config, "CONSTANT_SEED", True), \ patch.object(config, "SEED", 42): - client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM) + client = self._make_client(voice_mode=VOICE_MODE_CUSTOM) self.assertEqual(client._seed, 42) def test_without_constant_seed_minus_one_is_forwarded(self): with patch.object(config, "CONSTANT_SEED", False), \ patch.object(config, "SEED", -1): - client = self._make_client(voice_mode=tts.VOICE_MODE_CUSTOM) + client = self._make_client(voice_mode=VOICE_MODE_CUSTOM) self.assertEqual(client._seed, -1) def test_resolved_seed_is_reused_across_requests(self): @@ -134,7 +157,7 @@ class SeedResolutionTests(unittest.TestCase): } } client = QwenTTSClient.__new__(QwenTTSClient) - client.voice_mode = tts.VOICE_MODE_CUSTOM + client.voice_mode = VOICE_MODE_CUSTOM client.language = "English" client._seed = 1234 client.api_info = api_info @@ -159,7 +182,7 @@ class PayloadLanguageTests(unittest.TestCase): def _custom_client(self, language, endpoint, api_info=None): client = QwenTTSClient.__new__(QwenTTSClient) - client.voice_mode = tts.VOICE_MODE_CUSTOM + client.voice_mode = VOICE_MODE_CUSTOM client.language = language client._seed = config.SEED client.api_info = api_info if api_info is not None else { @@ -170,7 +193,7 @@ class PayloadLanguageTests(unittest.TestCase): def _clone_client(self, language, endpoint, api_info=None, ref_text="hello"): client = QwenTTSClient.__new__(QwenTTSClient) - client.voice_mode = tts.VOICE_MODE_CLONE + client.voice_mode = VOICE_MODE_CLONE client.language = language client._seed = config.SEED client.voice_clone_ref_audio = str(self.ref_audio) @@ -220,7 +243,7 @@ class PayloadLanguageTests(unittest.TestCase): client = self._clone_client("English", "/generate_voice_clone", api_info=api_info) client._generate_voice_clone("text") kwargs = client.clone_client.predict.call_args.kwargs - self.assertEqual(kwargs["model_size"], tts.MODEL_SIZE) + self.assertEqual(kwargs["model_size"], MODEL_SIZE) self.assertEqual(kwargs["seed"], config.SEED) @@ -236,32 +259,33 @@ class FasterTTSClientHealthTests(unittest.TestCase): def test_unreachable_server_raises_with_readme_pointer(self): import urllib.error - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", side_effect=urllib.error.URLError("Connection refused")): with self.assertRaises(RuntimeError) as ctx: - FasterTTSClient() + FasterTTSClient(_DUMMY_CHUNKS) 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", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._health_response(model_loaded=False)): with self.assertRaises(RuntimeError) as ctx: - FasterTTSClient() + FasterTTSClient(_DUMMY_CHUNKS) self.assertIn("not loaded", str(ctx.exception)) def test_healthy_server_defaults_from_config(self): - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._health_response()): - client = FasterTTSClient() + client = FasterTTSClient(_DUMMY_CHUNKS) self.assertEqual(client.voice, config.FASTER_VOICE) self.assertEqual(client.api_url, config.FASTER_API_URL.rstrip("/")) def test_explicit_voice_and_url_override_config(self): - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._health_response()): - client = FasterTTSClient(voice="narrator", api_url="http://10.0.0.5:9000/") + client = FasterTTSClient(_DUMMY_CHUNKS, + 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") @@ -271,18 +295,16 @@ class FasterTTSClientGenerateTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() - self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) - self._chunks.start() - self._sleep = patch("converter.tts.time.sleep") + self._sleep = patch("converter.clients.base.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.chunks_dir = Path(self._tmp.name) client.voice = "default" client.api_url = "http://127.0.0.1:8000" return client @@ -303,7 +325,7 @@ class FasterTTSClientGenerateTests(unittest.TestCase): channels, sampwidth, framerate, frames = self._read_wav(path) self.assertEqual(channels, 1) self.assertEqual(sampwidth, 2) - self.assertEqual(framerate, tts.SAMPLE_RATE) + self.assertEqual(framerate, SAMPLE_RATE) self.assertEqual(frames, pcm) def test_long_text_is_subchunked_and_concatenated_in_order(self): @@ -360,7 +382,7 @@ class FasterTTSClientGenerateTests(unittest.TestCase): response.read.return_value = body return response - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", side_effect=[_response(b"")]) as mock_urlopen: result = client.generate_chunk("Hello.", 1) self.assertIsNone(result) @@ -386,7 +408,7 @@ class FasterTTSClientGenerateTests(unittest.TestCase): response = MagicMock() response.__enter__.return_value = response response.read.return_value = b"\x01\x00" * 10 - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=response) as mock_urlopen: pcm = client._request_pcm("Hello world.") self.assertEqual(pcm, b"\x01\x00" * 10) @@ -401,7 +423,7 @@ class FasterTTSClientGenerateTests(unittest.TestCase): client = self._make_client() text = " ".join(f"word{i}" for i in range(12)) # 12 words -> expected 4.8s, half is 2.4s -> 2.5s of audio passes. - pcm = b"\x01\x00" * int(2.5 * tts.SAMPLE_RATE) + pcm = b"\x01\x00" * int(2.5 * SAMPLE_RATE) with patch.object(client, "_request_pcm", return_value=pcm): result = client.generate_chunk(text, 1) self.assertIsNotNone(result) @@ -412,16 +434,14 @@ class QwenTTSClientGenerateTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() - self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) - self._chunks.start() def tearDown(self): - self._chunks.stop() self._tmp.cleanup() def _make_client(self): client = QwenTTSClient.__new__(QwenTTSClient) - client.voice_mode = tts.VOICE_MODE_CUSTOM + client.chunks_dir = Path(self._tmp.name) + client.voice_mode = VOICE_MODE_CUSTOM return client @staticmethod @@ -429,7 +449,7 @@ class QwenTTSClientGenerateTests(unittest.TestCase): with wave.open(str(path), "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) - wav_file.setframerate(tts.SAMPLE_RATE) + wav_file.setframerate(SAMPLE_RATE) wav_file.writeframes(frames) return path @@ -513,17 +533,17 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): return _dispatch def _client(self, voice=None, language=None, model_id=None, **kwargs): - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", side_effect=self._get_responses(**kwargs)): - return AudioCppTTSClient(voice=voice, language=language, - model_id=model_id) + return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice, + language=language, model_id=model_id) def test_unreachable_server_raises_with_readme_pointer(self): import urllib.error - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", side_effect=urllib.error.URLError("Connection refused")): with self.assertRaises(RuntimeError) as ctx: - AudioCppTTSClient() + AudioCppTTSClient(_DUMMY_CHUNKS) message = str(ctx.exception) self.assertIn("not reachable", message) self.assertIn("README", message) @@ -625,9 +645,9 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): self.assertEqual(client.voice, "narrator") def test_invalid_language_fails_before_connect(self): - with patch("converter.tts.urllib.request.urlopen") as mock_urlopen: + with patch("converter.clients.faster.urllib.request.urlopen") as mock_urlopen: with self.assertRaises(ValueError): - AudioCppTTSClient(language="klingon") + AudioCppTTSClient(_DUMMY_CHUNKS, language="klingon") mock_urlopen.assert_not_called() def test_explicit_language_normalized(self): @@ -651,7 +671,7 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): def test_preset_mode_falls_back_when_clone_model_not_on_server(self): 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: + self.assertLogs("converter.clients.audiocpp", level="WARNING") as logs: client = self._client( voice="narrator", models={"data": [{"id": "qwen3-tts", "family": "qwen3_tts"}, @@ -730,7 +750,7 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): # requirement error lists both configured ids instead. with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen3-tts"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen3-tts-clone"), \ - self.assertNoLogs("converter.tts", level="WARNING"): + self.assertNoLogs("converter.clients.audiocpp", level="WARNING"): with self.assertRaises(RuntimeError) as ctx: self._client(voice="narrator", models={"data": [{"id": "pocket-tts"}]}) @@ -774,16 +794,17 @@ class AudioCppTaskDetectionTests(unittest.TestCase): return self._json_response({"voices": ["narrator"]}) raise AssertionError(f"unexpected URL: {url}") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", side_effect=_dispatch): - return AudioCppTTSClient(voice=voice, instructions=instructions, + return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice, + instructions=instructions, request_options=request_options) def test_missing_task_falls_back_to_tts(self): # Servers that predate the task field hosted plain TTS models. client = self._client(models={"data": [ {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]}) - self.assertEqual(client.task, tts.AUDIOCPP_TASK_TTS) + self.assertEqual(client.task, AUDIOCPP_TASK_TTS) self.assertFalse(client.design_mode) def test_task_detected_from_models_endpoint(self): @@ -791,7 +812,7 @@ class AudioCppTaskDetectionTests(unittest.TestCase): {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", "task": "vdes"}]}, instructions="A warm adult narrator") - self.assertEqual(client.task, tts.AUDIOCPP_TASK_VDES) + self.assertEqual(client.task, AUDIOCPP_TASK_VDES) self.assertTrue(client.design_mode) def test_clon_task_entry_connects_in_preset_mode(self): @@ -912,15 +933,15 @@ class AudioCppFamilyDetectionTests(unittest.TestCase): return self._json_response({"voices": [voice] if voice else []}) raise AssertionError(f"unexpected URL: {url}") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", side_effect=_dispatch): - return AudioCppTTSClient(voice=voice) + return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice) def test_family_detected_from_models_endpoint(self): client = self._client(models={"data": [ {"id": config.AUDIOCPP_MODEL_ID, "family": "higgs_audio_tts"}]}) self.assertEqual(client.family, "higgs_audio_tts") - self.assertIs(client.profile, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) + self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE) def test_missing_family_uses_generic_profile(self): # A missing family is unknown (not guessed as qwen3_tts): it falls @@ -928,14 +949,14 @@ class AudioCppFamilyDetectionTests(unittest.TestCase): client = self._client(models={"data": [ {"id": config.AUDIOCPP_MODEL_ID}]}) self.assertEqual(client.family, "") - self.assertIs(client.profile, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) + self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE) def test_unknown_family_uses_generic_profile(self): client = self._client(models={"data": [ {"id": config.AUDIOCPP_MODEL_ID, "family": "future_tts"}]}) self.assertEqual(client.family, "future_tts") - self.assertIs(client.profile, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) - self.assertEqual(client.profile.language_style, tts.AUDIOCPP_LANG_OMIT) + self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE) + self.assertEqual(client.profile.language_style, AUDIOCPP_LANG_OMIT) def test_speaker_mode_rejected_for_clone_only_family(self): client = None @@ -977,7 +998,7 @@ class AudioCppFamilyDetectionTests(unittest.TestCase): def test_clone_model_id_of_different_family_is_ignored(self): with patch.object(config, "AUDIOCPP_MODEL_ID", "higgs"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen-clone"), \ - self.assertLogs("converter.tts", level="WARNING") as logs: + self.assertLogs("converter.clients.audiocpp", level="WARNING") as logs: client = self._client(models={"data": [ {"id": "higgs", "family": "higgs_audio_tts"}, {"id": "qwen-clone", "family": "qwen3_tts"}]}) @@ -989,7 +1010,7 @@ class AudioCppFamilyDetectionTests(unittest.TestCase): def test_clone_model_id_missing_on_non_qwen_server_is_debug_only(self): with patch.object(config, "AUDIOCPP_MODEL_ID", "higgs"), \ patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen-clone"), \ - self.assertNoLogs("converter.tts", level="WARNING"): + self.assertNoLogs("converter.clients.audiocpp", level="WARNING"): client = self._client(models={"data": [ {"id": "higgs", "family": "higgs_audio_tts"}]}) self.assertEqual(client.model_id, "higgs") @@ -997,7 +1018,7 @@ class AudioCppFamilyDetectionTests(unittest.TestCase): def test_clone_model_id_missing_on_qwen_server_still_warns(self): 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: + self.assertLogs("converter.clients.audiocpp", level="WARNING") as logs: client = self._client(models={"data": [ {"id": "qwen3-tts", "family": "qwen3_tts"}, {"id": "pocket-tts", "family": "pocket_tts"}]}) @@ -1005,54 +1026,54 @@ class AudioCppFamilyDetectionTests(unittest.TestCase): self.assertTrue(any("qwen3-tts-clone" in line for line in logs.output)) def test_iso_language_code_helper(self): - self.assertEqual(tts.LANGUAGE_ISO_CODES["English"], "en") - self.assertIsNone(tts.LANGUAGE_ISO_CODES.get("Auto")) + self.assertEqual(LANGUAGE_ISO_CODES["English"], "en") + self.assertIsNone(LANGUAGE_ISO_CODES.get("Auto")) class AudiocppEntryVoiceCapabilityTests(unittest.TestCase): """The per-entry voice capability resolver (speaker/clone/design).""" def _cap(self, family="", task="tts", model_id=""): - return tts.audiocpp_entry_voice_capability(family, task, model_id) + return audiocpp_entry_voice_capability(family, task, model_id) def test_vdes_task_is_design(self): self.assertEqual(self._cap("qwen3_tts", "vdes", "Qwen3-TTS-VoiceDesign-GGUF"), - tts.AUDIOCPP_VOICE_DESIGN) + AUDIOCPP_VOICE_DESIGN) def test_qwen_customvoice_entry_is_speaker(self): self.assertEqual(self._cap("qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"), - tts.AUDIOCPP_VOICE_SPEAKER) + AUDIOCPP_VOICE_SPEAKER) def test_qwen_base_entry_is_clone(self): self.assertEqual(self._cap("qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-Base-GGUF"), - tts.AUDIOCPP_VOICE_CLONE) + AUDIOCPP_VOICE_CLONE) def test_qwen_unidentified_entry_is_clone(self): self.assertEqual(self._cap("qwen3_tts", "tts", "qwen"), - tts.AUDIOCPP_VOICE_CLONE) + AUDIOCPP_VOICE_CLONE) def test_other_families_are_clone(self): self.assertEqual(self._cap("higgs_audio_tts", "tts", "higgs"), - tts.AUDIOCPP_VOICE_CLONE) + AUDIOCPP_VOICE_CLONE) def test_missing_family_is_clone(self): self.assertEqual(self._cap("", "tts", "legacy"), - tts.AUDIOCPP_VOICE_CLONE) + AUDIOCPP_VOICE_CLONE) def test_customvoice_match_is_case_insensitive(self): self.assertEqual(self._cap("qwen3_tts", "tts", "Qwen3-TTS-12Hz-1.7B-CUSTOMVOICE-GGUF"), - tts.AUDIOCPP_VOICE_SPEAKER) + AUDIOCPP_VOICE_SPEAKER) def test_customvoice_id_in_other_family_is_not_speaker(self): # The "customvoice" substring only marks a speaker for the qwen3_tts # family; another family with a lookalike id stays clone-only. self.assertEqual(self._cap("future_tts", "tts", "Qwen3-TTS-CustomVoice"), - tts.AUDIOCPP_VOICE_CLONE) + AUDIOCPP_VOICE_CLONE) class AudioCppTTSClientRequestTests(unittest.TestCase): @@ -1060,21 +1081,18 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() - self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) - self._chunks.start() - self._sleep = patch("converter.tts.time.sleep") + self._sleep = patch("converter.clients.base.time.sleep") self._sleep.start() def tearDown(self): self._sleep.stop() - self._chunks.stop() self._tmp.cleanup() - @staticmethod - def _make_client(preset_mode=False, voice="Vivian", language="English", seed=-1, + def _make_client(self, preset_mode=False, voice="Vivian", language="English", seed=-1, family="qwen3_tts", task="tts", instructions=None, request_options=None): client = AudioCppTTSClient.__new__(AudioCppTTSClient) + client.chunks_dir = Path(self._tmp.name) client.api_url = "http://127.0.0.1:8080" client.model_id = config.AUDIOCPP_MODEL_ID client.preset_mode = preset_mode @@ -1083,23 +1101,23 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): client._seed = seed client.family = family client.task = task - client.profile = tts.AUDIOCPP_FAMILY_PROFILES.get( - family, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) + client.profile = AUDIOCPP_FAMILY_PROFILES.get( + family, AUDIOCPP_DEFAULT_FAMILY_PROFILE) client.instructions = instructions or "" client.request_options = dict(request_options or {}) - client.design_mode = task == tts.AUDIOCPP_TASK_VDES + client.design_mode = task == AUDIOCPP_TASK_VDES # Mirrors the connect-time rule: an instruction-defined voice on a # clone-capable entry with no --voice (design mode takes precedence). - capability = tts.audiocpp_entry_voice_capability( + capability = audiocpp_entry_voice_capability( family, task, client.model_id) client.instruction_voice = ( not preset_mode and not client.design_mode - and capability == tts.AUDIOCPP_VOICE_CLONE + and capability == AUDIOCPP_VOICE_CLONE and bool(client.instructions)) return client @staticmethod - def _wav_bytes(frames=b"\x01\x00" * 10, rate=tts.SAMPLE_RATE): + def _wav_bytes(frames=b"\x01\x00" * 10, rate=SAMPLE_RATE): buffer = io.BytesIO() with wave.open(buffer, "wb") as wav_file: wav_file.setnchannels(1) @@ -1117,7 +1135,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_payload_includes_model_input_voice_language_and_seed(self): client = self._make_client(preset_mode=True, voice="narrator", language="Japanese", seed=1234) - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello world.") request = mock_urlopen.call_args[0][0] @@ -1133,7 +1151,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_negative_seed_omitted_from_payload(self): client = self._make_client(preset_mode=True, voice="narrator", seed=-1) - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello world.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1142,7 +1160,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_request_timeout_is_the_configured_api_timeout(self): client = self._make_client() long_text = " ".join(f"word{i}" for i in range(1500)) - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav(long_text) timeout = mock_urlopen.call_args[1]["timeout"] @@ -1150,7 +1168,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_speaker_mode_sends_instruct(self): client = self._make_client(preset_mode=False) - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1160,7 +1178,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): # --instructions overrides the INSTRUCT default in speaker mode. client = self._make_client(preset_mode=False, instructions="Read whisper quiet.") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1171,7 +1189,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): # instruction reach the model. client = self._make_client(preset_mode=True, voice="narrator", instructions="Calm and steady.") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1181,7 +1199,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_design_mode_payload_omits_voice_and_sends_instructions(self): client = self._make_client(task="vdes", instructions="A warm adult narrator") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1193,7 +1211,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): # takes Qwen display names like the other variants. client = self._make_client(task="vdes", language="Japanese", instructions="A warm adult narrator") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1204,7 +1222,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): # no speaker name is invented, the instruction carries the voice. client = self._make_client(family="omnivoice", instructions="female, young adult") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1216,7 +1234,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): client = self._make_client(preset_mode=True, voice="narrator", request_options={"emotion": "neutral", "speed": "1.1"}) - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1225,7 +1243,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_empty_request_options_omit_options_field(self): client = self._make_client(preset_mode=True, voice="narrator") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1235,7 +1253,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): # Clone-only families (higgs_audio_tts, voxcpm2, ...) detect the # language themselves and take no style instruction. client = self._make_client(preset_mode=False, family="higgs_audio_tts") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1245,7 +1263,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_iso_family_sends_language_code(self): client = self._make_client(preset_mode=True, voice="narrator", language="Japanese", family="index_tts2") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1254,7 +1272,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_iso_family_auto_omits_language(self): client = self._make_client(preset_mode=True, voice="narrator", language="Auto", family="index_tts2") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1263,7 +1281,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_qwen_language_display_name_still_sent(self): client = self._make_client(preset_mode=True, voice="narrator", language="Japanese", family="qwen3_tts") - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(self._wav_bytes())) as mock_urlopen: client._request_wav("Hello.") payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) @@ -1272,7 +1290,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): def test_non_wav_response_rejected(self): client = self._make_client() for body in (b"", b"RIFFxxxx", b"MP3DATA-MP3DATA", b"RIFF\x00\x00\x00\x00mpeg"): - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.faster.urllib.request.urlopen", return_value=self._post_response(body)): with self.assertRaises(RuntimeError): client._request_wav("Hello.") @@ -1283,7 +1301,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): error = urllib.error.HTTPError( "http://127.0.0.1:8080/v1/audio/speech", 500, "Server Error", {}, io.BytesIO(b'{"error":"bad voice"}')) - with patch("converter.tts.urllib.request.urlopen", side_effect=error): + with patch("converter.clients.faster.urllib.request.urlopen", side_effect=error): with self.assertRaises(RuntimeError) as ctx: client._request_wav("Hello.") self.assertIn("500", str(ctx.exception)) @@ -1327,7 +1345,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): with wave.open(str(path), "rb") as wav_file: self.assertEqual(wav_file.getnchannels(), 1) self.assertEqual(wav_file.getsampwidth(), 2) - self.assertEqual(wav_file.getframerate(), tts.SAMPLE_RATE) + self.assertEqual(wav_file.getframerate(), SAMPLE_RATE) self.assertEqual(wav_file.readframes(wav_file.getnframes()), frames) def test_long_text_is_subchunked_and_concatenated_in_order(self): @@ -1360,16 +1378,13 @@ class AudioCppHeartbeatTests(unittest.TestCase): 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(): + def _client(self): client = AudioCppTTSClient.__new__(AudioCppTTSClient) + client.chunks_dir = Path(self._tmp.name) client.api_url = "http://127.0.0.1:8080" client.model_id = config.AUDIOCPP_MODEL_ID client.preset_mode = False @@ -1377,7 +1392,7 @@ class AudioCppHeartbeatTests(unittest.TestCase): client.language = "English" client._seed = -1 client.family = "qwen3_tts" - client.profile = tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE + client.profile = AUDIOCPP_DEFAULT_FAMILY_PROFILE return client @staticmethod @@ -1386,7 +1401,7 @@ class AudioCppHeartbeatTests(unittest.TestCase): with wave.open(buffer, "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) - wav_file.setframerate(tts.SAMPLE_RATE) + wav_file.setframerate(SAMPLE_RATE) wav_file.writeframes(b"\x01\x00" * 10) return buffer.getvalue() @@ -1416,15 +1431,13 @@ class AudioCppTTSClientTruncationTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() - self._chunks = patch.object(tts, "CHUNKS_FOLDER", Path(self._tmp.name)) - self._chunks.start() def tearDown(self): - self._chunks.stop() self._tmp.cleanup() def _make_client(self): client = AudioCppTTSClient.__new__(AudioCppTTSClient) + client.chunks_dir = Path(self._tmp.name) client.api_url = "http://127.0.0.1:8080" client.model_id = config.AUDIOCPP_MODEL_ID client.preset_mode = True @@ -1432,7 +1445,7 @@ class AudioCppTTSClientTruncationTests(unittest.TestCase): client.language = "English" client._seed = -1 client.family = "qwen3_tts" - client.profile = tts.AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] + client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] return client @staticmethod @@ -1441,7 +1454,7 @@ class AudioCppTTSClientTruncationTests(unittest.TestCase): with wave.open(buffer, "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) - wav_file.setframerate(tts.SAMPLE_RATE) + wav_file.setframerate(SAMPLE_RATE) wav_file.writeframes(frames) return buffer.getvalue() @@ -1449,7 +1462,7 @@ class AudioCppTTSClientTruncationTests(unittest.TestCase): client = self._make_client() text = " ".join(f"word{i}" for i in range(12)) # 12 words -> expected 4.8s, half is 2.4s -> 2.5s of audio passes. - wav = self._wav_bytes(b"\x01\x00" * int(2.5 * tts.SAMPLE_RATE)) + wav = self._wav_bytes(b"\x01\x00" * int(2.5 * SAMPLE_RATE)) with patch.object(client, "_request_wav", return_value=wav): result = client.generate_chunk(text, 1) self.assertIsNotNone(result) @@ -1473,7 +1486,7 @@ class AudioCppUnloadModelsTests(unittest.TestCase): def test_posts_to_unload_all_models(self): client = self._client() - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.audiocpp.urllib.request.urlopen", return_value=self._response(b'{"unloaded": ["qwen"]}')) as mock_urlopen: client._unload_server_models() request = mock_urlopen.call_args[0][0] @@ -1485,7 +1498,7 @@ class AudioCppUnloadModelsTests(unittest.TestCase): def test_reports_unloaded_ids(self): client = self._client() buf = io.StringIO() - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.audiocpp.urllib.request.urlopen", return_value=self._response(b'{"unloaded": ["a", "b"]}')), \ redirect_stdout(buf): client._unload_server_models() @@ -1495,7 +1508,7 @@ class AudioCppUnloadModelsTests(unittest.TestCase): def test_no_loaded_models_is_silent(self): client = self._client() buf = io.StringIO() - with patch("converter.tts.urllib.request.urlopen", + with patch("converter.clients.audiocpp.urllib.request.urlopen", return_value=self._response(b'{"unloaded": []}')), \ redirect_stdout(buf): client._unload_server_models() @@ -1504,8 +1517,8 @@ class AudioCppUnloadModelsTests(unittest.TestCase): def test_http_error_warns_and_continues(self): client = self._client() buf = io.StringIO() - with patch("converter.tts.urllib.request.urlopen", - side_effect=tts.urllib.error.HTTPError( + with patch("converter.clients.audiocpp.urllib.request.urlopen", + side_effect=urllib.error.HTTPError( "http://127.0.0.1:8080/v1/tasks/unload_all_models", 404, "Not Found", None, io.BytesIO())), \ redirect_stdout(buf): @@ -1517,8 +1530,8 @@ class AudioCppUnloadModelsTests(unittest.TestCase): def test_connection_error_warns_and_continues(self): client = self._client() buf = io.StringIO() - with patch("converter.tts.urllib.request.urlopen", - side_effect=tts.urllib.error.URLError("refused")), \ + with patch("converter.clients.audiocpp.urllib.request.urlopen", + side_effect=urllib.error.URLError("refused")), \ redirect_stdout(buf): client._unload_server_models() self.assertIn("[WARNING]", buf.getvalue()) @@ -1532,8 +1545,8 @@ class AudioCppUnloadModelsTests(unittest.TestCase): client.language = "English" client._seed = -1 client.family = "qwen3_tts" - client.task = tts.AUDIOCPP_TASK_TTS - client.profile = tts.AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] + client.task = AUDIOCPP_TASK_TTS + client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] client.design_mode = False client.instruction_voice = False client.speaker_mode = False @@ -1562,8 +1575,8 @@ class AudioCppUnloadModelsTests(unittest.TestCase): client.language = "English" client._seed = -1 client.family = "qwen3_tts" - client.task = tts.AUDIOCPP_TASK_TTS - client.profile = tts.AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] + client.task = AUDIOCPP_TASK_TTS + client.profile = AUDIOCPP_FAMILY_PROFILES["qwen3_tts"] client.design_mode = False client.instruction_voice = False client.speaker_mode = False @@ -1579,7 +1592,7 @@ class AudioCppUnloadModelsTests(unittest.TestCase): patch.object(client, "_resolve_family"), \ patch.object(client, "_resolve_task"), \ patch.object(client, "_check_voice"), \ - patch.object(tts.config, "AUDIOCPP_UNLOAD_MODELS", False), \ + patch.object(config, "AUDIOCPP_UNLOAD_MODELS", False), \ patch.object(client, "_unload_server_models") as mock_unload: client._connect() mock_unload.assert_not_called() @@ -1592,9 +1605,10 @@ class BackendWiringTests(unittest.TestCase): with patch("converter.converter.FasterTTSClient") as mock_faster, \ patch("converter.converter.QwenTTSClient") as mock_qwen, \ patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_FASTER, voice="narrator") - mock_faster.assert_called_once_with(voice="narrator", api_url=None, + AudiobookConverter(voice_mode=VOICE_MODE_CLONE, + backend=BACKEND_FASTER, voice="narrator") + mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, + voice="narrator", api_url=None, quiet=False) mock_qwen.assert_not_called() mock_audiocpp.assert_not_called() @@ -1603,10 +1617,11 @@ class BackendWiringTests(unittest.TestCase): with patch("converter.converter.FasterTTSClient") as mock_faster, \ patch("converter.converter.QwenTTSClient") as mock_qwen, \ patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_AUDIOCPP, voice="narrator", + AudiobookConverter(voice_mode=VOICE_MODE_CLONE, + backend=BACKEND_AUDIOCPP, voice="narrator", language="ja") - mock_audiocpp.assert_called_once_with(voice="narrator", language="Japanese", + mock_audiocpp.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, + voice="narrator", language="Japanese", model_id=None, instructions=None, request_options={}, @@ -1616,9 +1631,10 @@ class BackendWiringTests(unittest.TestCase): def test_audiocpp_backend_without_voice_uses_audiocpp_client(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, - backend=tts.BACKEND_AUDIOCPP) - mock_audiocpp.assert_called_once_with(voice=None, language=config.LANGUAGE, + AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, + backend=BACKEND_AUDIOCPP) + mock_audiocpp.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, + voice=None, language=config.LANGUAGE, model_id=None, instructions=None, request_options={}, @@ -1626,22 +1642,24 @@ class BackendWiringTests(unittest.TestCase): def test_audiocpp_backend_model_id_is_wired_through(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_AUDIOCPP, voice="narrator", + AudiobookConverter(voice_mode=VOICE_MODE_CLONE, + backend=BACKEND_AUDIOCPP, voice="narrator", model_id="higgs") mock_audiocpp.assert_called_once_with( + chunks_dir=converter_mod.CHUNKS_FOLDER, voice="narrator", language=config.LANGUAGE, model_id="higgs", instructions=None, request_options={}, api_url=None, quiet=False) def test_audiocpp_backend_instructions_and_options_are_wired_through(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, - backend=tts.BACKEND_AUDIOCPP, + AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, + backend=BACKEND_AUDIOCPP, instructions="A warm adult narrator", request_options={"emotion": "neutral", "speed": "1.1"}) mock_audiocpp.assert_called_once_with( + chunks_dir=converter_mod.CHUNKS_FOLDER, voice=None, language=config.LANGUAGE, model_id=None, instructions="A warm adult narrator", @@ -1652,8 +1670,8 @@ class BackendWiringTests(unittest.TestCase): with patch("converter.converter.FasterTTSClient") as mock_faster, \ patch("converter.converter.QwenTTSClient") as mock_qwen, \ patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, - backend=tts.BACKEND_QWEN) + AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, + backend=BACKEND_QWEN) mock_qwen.assert_called_once() mock_faster.assert_not_called() mock_audiocpp.assert_not_called() @@ -1661,32 +1679,35 @@ class BackendWiringTests(unittest.TestCase): def test_qwen_clone_mode_still_requires_reference(self): with patch("converter.converter.QwenTTSClient"): with self.assertRaises(ValueError): - AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_QWEN) + AudiobookConverter(voice_mode=VOICE_MODE_CLONE, + backend=BACKEND_QWEN) def test_api_url_override_reaches_each_client(self): # A remote conversion threads api_url through to the selected client. with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_AUDIOCPP, voice="narrator", + AudiobookConverter(voice_mode=VOICE_MODE_CLONE, + backend=BACKEND_AUDIOCPP, voice="narrator", api_url="http://10.0.0.5:8080") mock_audiocpp.assert_called_once_with( + chunks_dir=converter_mod.CHUNKS_FOLDER, voice="narrator", language=config.LANGUAGE, model_id=None, instructions=None, request_options={}, api_url="http://10.0.0.5:8080", quiet=False) with patch("converter.converter.FasterTTSClient") as mock_faster: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_FASTER, voice="narrator", + AudiobookConverter(voice_mode=VOICE_MODE_CLONE, + backend=BACKEND_FASTER, voice="narrator", api_url="http://10.0.0.5:8000") - mock_faster.assert_called_once_with(voice="narrator", + mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, + voice="narrator", api_url="http://10.0.0.5:8000", quiet=False) with patch("converter.converter.QwenTTSClient") as mock_qwen: - AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, - backend=tts.BACKEND_QWEN, + AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, + backend=BACKEND_QWEN, api_url="http://10.0.0.5:7860") mock_qwen.assert_called_once_with( - voice_mode=tts.VOICE_MODE_CUSTOM, voice_clone_ref_audio=None, + chunks_dir=converter_mod.CHUNKS_FOLDER, + voice_mode=VOICE_MODE_CUSTOM, voice_clone_ref_audio=None, voice_clone_ref_text=None, skip_transcription=False, language=config.LANGUAGE, api_url="http://10.0.0.5:7860", quiet=False) @@ -1695,8 +1716,8 @@ class BackendWiringTests(unittest.TestCase): # Cloning is server-side for the audiocpp backend, so the # clone-mode voice can be selected without local reference audio. with patch("converter.converter.AudioCppTTSClient"): - converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_AUDIOCPP, + converter = AudiobookConverter(voice_mode=VOICE_MODE_CLONE, + backend=BACKEND_AUDIOCPP, voice="narrator") self.assertIsNone(converter.voice_clone_ref_audio) @@ -1710,8 +1731,8 @@ class BackendWiringTests(unittest.TestCase): def test_chapter_chunks_qwen_always_splits(self): with patch("converter.converter.QwenTTSClient"): - converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, - backend=tts.BACKEND_QWEN) + converter = AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, + backend=BACKEND_QWEN) text = " ".join(f"word{i}" for i in range(50)) with patch.object(config, "CHUNK_SIZE", 10): chunks = converter._chapter_chunks(text) @@ -1720,27 +1741,27 @@ class BackendWiringTests(unittest.TestCase): def test_faster_backend_still_validates_other_settings(self): with patch("converter.converter.FasterTTSClient"): with self.assertRaises(ValueError): - AudiobookConverter(backend=tts.BACKEND_FASTER, speed=0) + AudiobookConverter(backend=BACKEND_FASTER, speed=0) with self.assertRaises(ValueError): - AudiobookConverter(backend=tts.BACKEND_FASTER, language="klingon") + AudiobookConverter(backend=BACKEND_FASTER, language="klingon") def test_audiocpp_backend_still_validates_other_settings(self): with patch("converter.converter.AudioCppTTSClient"): with self.assertRaises(ValueError): - AudiobookConverter(backend=tts.BACKEND_AUDIOCPP, speed=0) + AudiobookConverter(backend=BACKEND_AUDIOCPP, speed=0) with self.assertRaises(ValueError): - AudiobookConverter(backend=tts.BACKEND_AUDIOCPP, language="klingon") + AudiobookConverter(backend=BACKEND_AUDIOCPP, language="klingon") def _faster_converter(self, voice=None): with patch("converter.converter.FasterTTSClient"): - return AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, - backend=tts.BACKEND_FASTER, voice=voice) + return AudiobookConverter(voice_mode=VOICE_MODE_CLONE, + backend=BACKEND_FASTER, voice=voice) def _audiocpp_converter(self, voice=None): with patch("converter.converter.AudioCppTTSClient"): return AudiobookConverter( - voice_mode=tts.VOICE_MODE_CLONE if voice else tts.VOICE_MODE_CUSTOM, - backend=tts.BACKEND_AUDIOCPP, voice=voice) + voice_mode=VOICE_MODE_CLONE if voice else VOICE_MODE_CUSTOM, + backend=BACKEND_AUDIOCPP, voice=voice) def test_narrator_tag_uses_faster_voice_name(self): converter = self._faster_converter(voice="male_richard_poe") @@ -1783,9 +1804,9 @@ class BackendWiringTests(unittest.TestCase): ref = Path(tmp) / "ref.wav" ref.write_bytes(b"x") with patch("converter.converter.QwenTTSClient"): - converter = AudiobookConverter(voice_mode=tts.VOICE_MODE_CLONE, + converter = AudiobookConverter(voice_mode=VOICE_MODE_CLONE, voice_clone_ref_audio=str(ref), - backend=tts.BACKEND_QWEN) + backend=BACKEND_QWEN) self.assertEqual(converter._narrator_tag(), "ref") diff --git a/app/ui/hub.py b/app/ui/hub.py index 1adb54e..5f29ed1 100644 --- a/app/ui/hub.py +++ b/app/ui/hub.py @@ -50,7 +50,7 @@ from converter.converter import ( LOGS_FOLDER, voice_mode_for, ) -from converter.tts import ( +from converter.clients import ( AUDIOCPP_VOICE_CLONE, AUDIOCPP_VOICE_DESIGN, AUDIOCPP_VOICE_SPEAKER, -- cgit v1.2.3