diff options
| -rw-r--r-- | README.md | 14 | ||||
| -rwxr-xr-x | audiobook.py | 51 | ||||
| -rw-r--r-- | converter/config.py | 7 | ||||
| -rw-r--r-- | converter/converter.py | 54 | ||||
| -rw-r--r-- | converter/tts.py | 150 | ||||
| -rw-r--r-- | requirements.txt | 1 | ||||
| -rw-r--r-- | tests/test_converter.py | 45 | ||||
| -rw-r--r-- | tests/test_make_audiocpp_server_json.py | 585 | ||||
| -rw-r--r-- | tests/test_tts.py | 254 | ||||
| -rwxr-xr-x | tools/make_audiocpp_server_json.py | 1097 | ||||
| -rw-r--r-- | tools/tui.py | 511 |
11 files changed, 2307 insertions, 462 deletions
@@ -11,7 +11,7 @@ The converter sends text extracted from your books to a locally running TTS serv - Supports [audio.cpp](https://github.com/0xShug0/audio.cpp), [qwen-tts](https://pypi.org/project/qwen-tts/), and [faster-qwen3-tts](https://github.com/andimarafioti/faster-qwen3-tts) backend servers (audio.cpp can host any of its TTS model families, not just Qwen3-TTS) - Output a single file or one per chapter - Automatic metadata (title/artist/album tags, chapter track numbers) and a generated cover -- Clone voices from .wav reference files or use the built-in speaker in the CustomVoice model. +- Supports text-to-speech, voice cloning, voice design, and per-model controls like emotion/speed ## Prerequisites @@ -50,7 +50,9 @@ You need to install one of the following backends (see below for installation/us | `--single-file` | Merge all chapters into a single file. `m4b` is always one file. | | `--language <lang>` | Output language for the synthesized speech. Can add an accent even if the text is English. | | `--chunk` | Force client-side chunking into `CHUNK_SIZE`-word requests. Only matters for `--backend audiocpp`, which otherwise sends each chapter as one request and lets the server chunk long text itself (may double-chunk); the `qwen` and `faster` backends always chunk. | -| `--model <id>` | `--backend audiocpp` only: Choose the model from `server.json` | +| `--model <id>` | `audiocpp`: Choose the model from `server.json` | +| `--instructions "..."` | `audiocpp`: voice design or style instruction. Required for voice design models (`vdes`) | +| `--option KEY=VALUE` | `audiocpp`: Some models support custom options (e.g. `emotion=netural`) that can be passed with this flag | | `--debug` | Dump each chunk's raw audio and sent text to `debug/` and log every request. | | `--voice <name>` | `audiocpp`, `faster`: Server-side voice to request | | `--clone <path>` | `qwen`: Reference audio (`wav`) for voice cloning. | @@ -89,7 +91,7 @@ You can run `python tools/model_manager_v2.py list` to see all available models. Create a `server.json` config file. One server can host multiple models and multiple cloned voices. The `id:` fields are the model names you will set for `tts-audiobook-generator` with `--model`. -A helper tool is available in this repo `tools/make_audiocpp_server_json.py path/to/clone/wavs` that will interactively make this file for you, including automatically transcribing `.wav` voices to clone with `whisper`. Just pass it a path of `.wav` files. Make sure you're in a Python environment that has `whisper` (i.e. `conda activate audiobook` before running) +A helper tool is available in this repo, `tools/make_audiocpp_server_json.py`, that will interactively make this file for you, including automatically transcribing `.wav` voices to clone with `whisper`. It runs as a minimal full-screen TUI: browse to your `audio.cpp` checkout, pick model families and packages from an expandable checkbox tree, and accept the defaults on the remaining screens (host, port, backend, lazy loading). Pass `--wavs path/to/clone/wavs` to skip the voice-directory browser (it is prompted for otherwise), and `--notui` to use classic line prompts instead (selected automatically when `curses` is unavailable, e.g. on Windows without `pip install windows-curses`, or when not running in a terminal). Make sure you're in a Python environment that has `whisper` (i.e. `conda activate audiobook` before running). The Qwen3-TTS model tree also offers hosting the VoiceDesign package as a `vdes` entry (see [Voice design](#voice-design) below). ```json { @@ -124,6 +126,8 @@ A helper tool is available in this repo `tools/make_audiocpp_server_json.py path } ``` +### Run audio.cpp and the audiobook script + Run the server with this config file. The `audiocpp_server` path will be slightly different depending on your platform and build options: ```bash @@ -141,6 +145,10 @@ python audiobook.py --backend audiocpp --model qwen # Qwen3-TTS voice cloning python audiobook.py --backend audiocpp --model qwen-clone --voice narrator + +# Qwen-TTS voice design +python audiobook.py --backend audiocpp --model qwen-design \ + --instructions "A warm adult female narrator with a British accent" ``` ## Other TTS Backends diff --git a/audiobook.py b/audiobook.py index d3a5426..6ad056c 100755 --- a/audiobook.py +++ b/audiobook.py @@ -48,6 +48,10 @@ Examples: # Use the audio.cpp audiocpp_server with a server-side voice preset python audiobook.py --backend audiocpp --voice narrator + # Use the audio.cpp audiocpp_server with a voice design model (task 'vdes') + python audiobook.py --backend audiocpp --model qwen-design \\ + --instructions "A warm adult female narrator with a British accent" + # Use the Qwen demo server with a custom voice python audiobook.py --backend qwen @@ -176,6 +180,35 @@ Examples: "auto-select when the server hosts exactly one entry.") ) + parser.add_argument( + "--instructions", + type=str, + default=None, + metavar="TEXT", + help=("Voice design or style instruction sent with every request " + "(--backend audiocpp only). Required for voice design models " + "(server entries with task 'vdes', e.g. Qwen3-TTS " + "VoiceDesign): describe the voice to synthesize with, e.g. " + "'A warm adult female narrator with a British accent'. On " + "other families it acts as a style/delivery instruction when " + "the model supports one and is ignored otherwise. Defaults to " + "AUDIOCPP_INSTRUCTIONS in converter/config.py (empty).") + ) + + parser.add_argument( + "--option", + action="append", + type=str, + default=None, + metavar="KEY=VALUE", + help=("Request option passed through to the audio.cpp model " + "(--backend audiocpp only); repeatable. Whatever the hosted " + "family supports (emotion, voice_id, speed, speaking_rate, " + "temperature, ...) — unsupported keys are ignored by the " + "model. See the audio.cpp docs for the model's valid option " + "keys, e.g. --option emotion=neutral --option speed=1.1.") + ) + args = parser.parse_args() if args.speed <= 0: @@ -242,6 +275,21 @@ Examples: parser.error("--model requires --backend audiocpp; it selects an " "audio.cpp server model entry id") + if args.instructions is not None and args.backend != BACKEND_AUDIOCPP: + parser.error("--instructions requires --backend audiocpp; it is " + "sent as the audio.cpp request's instructions field") + + request_options = {} + if args.option: + if args.backend != BACKEND_AUDIOCPP: + parser.error("--option requires --backend audiocpp; the options " + "are passed through to the audio.cpp model") + for item in args.option: + key, sep, value = item.partition("=") + if not sep or not key.strip(): + parser.error(f"--option expects KEY=VALUE (got {item!r})") + request_options[key.strip()] = value + setup_logging(debug=args.debug) setup_directories() @@ -261,6 +309,7 @@ Examples: voice_mode=voice_mode, voice_clone_ref_audio=args.clone, output_format=args.format, + instructions=args.instructions, ) if not book_files: @@ -287,6 +336,8 @@ Examples: debug=args.debug, chunk=args.chunk, model_id=args.model, + instructions=args.instructions, + request_options=request_options, ) converter._book_files = book_files converter._planned = planned diff --git a/converter/config.py b/converter/config.py index d15efa5..4686764 100644 --- a/converter/config.py +++ b/converter/config.py @@ -68,3 +68,10 @@ AUDIOCPP_API_URL = "http://127.0.0.1:8080" # audio.cpp audiocpp_server # run with the --model CLI flag. AUDIOCPP_MODEL_ID = "qwen" AUDIOCPP_CLONE_MODEL_ID = "qwen-clone" + +# Voice design / style instruction sent with every audio.cpp request when +# the --instructions CLI flag is not given. Required for server entries +# hosted with task "vdes" (voice design models such as Qwen3-TTS +# VoiceDesign); on other families it acts as a style/delivery instruction +# when the model supports one and is ignored otherwise. Empty by default. +AUDIOCPP_INSTRUCTIONS = "" diff --git a/converter/converter.py b/converter/converter.py index 3915fe4..b24ac09 100644 --- a/converter/converter.py +++ b/converter/converter.py @@ -138,7 +138,9 @@ class AudiobookConverter: speed: float = 1.0, single_file: bool = False, output_format: str = config.AUDIO_FORMAT, language: Optional[str] = None, backend: str = config.BACKEND, voice: Optional[str] = None, debug: bool = False, - chunk: bool = False, model_id: Optional[str] = None): + chunk: bool = False, model_id: Optional[str] = None, + instructions: Optional[str] = None, + request_options: Optional[Dict[str, str]] = None): if speed <= 0: raise ValueError(f"Speed must be a positive number, got {speed}") if output_format not in AUDIO_FORMATS: @@ -164,6 +166,11 @@ class AudiobookConverter: # defaults to one request per chapter; --chunk forces client-side # chunking on top (possible needless double-chunking). self.client_chunks = bool(chunk) or backend != BACKEND_AUDIOCPP + # Voice design / style instruction and free-form request options + # (audio.cpp only): forwarded to AudioCppTTSClient, which validates + # them against the server-hosted model at connect time. + self.instructions = instructions + self.request_options = dict(request_options or {}) self._validate_configuration() if backend == BACKEND_FASTER: # The faster backend always voice-clones using a reference voice @@ -172,10 +179,14 @@ class AudiobookConverter: elif backend == BACKEND_AUDIOCPP: # Speaker mode (no voice) uses a built-in CustomVoice speaker; # an explicit voice selects a server-side preset (cloning). - # model_id overrides AUDIOCPP_MODEL_ID for multi-model servers. + # model_id overrides AUDIOCPP_MODEL_ID for 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, chunk_text=self.client_chunks, - model_id=model_id) + model_id=model_id, + instructions=instructions, + request_options=self.request_options) else: self.tts = QwenTTSClient( voice_mode=voice_mode, @@ -214,18 +225,22 @@ class AudiobookConverter: def _narrator_tag(self) -> str: """Narrator name used in output file names (see compute_narrator_tag).""" return self.compute_narrator_tag( - self.backend, self.voice, self.voice_mode, self.voice_clone_ref_audio) + self.backend, self.voice, self.voice_mode, + self.voice_clone_ref_audio, self.instructions) @staticmethod def compute_narrator_tag(backend: str, voice: Optional[str], voice_mode: str, - voice_clone_ref_audio: Optional[str]) -> str: + voice_clone_ref_audio: Optional[str], + instructions: Optional[str] = None) -> str: """Narrator name used in output file names, without a server connection. Custom voice mode uses the built-in speaker's display name; voice clone mode uses the reference audio file's stem; the faster and audiocpp backends use the server-side voice name (falling back to - the built-in speaker for the audiocpp backend's speaker mode). + the built-in speaker for the audiocpp backend's speaker mode). An + instruction without a voice (voice design, or instruction-defined + voices on families without built-in speakers) uses "designed". Spaces become underscores (e.g. "Uncle Fu" -> "Uncle_Fu"). Pure (no I/O, no server) so the pre-flight overwrite check can @@ -235,7 +250,13 @@ class AudiobookConverter: if backend == BACKEND_FASTER: narrator = voice or config.FASTER_VOICE elif backend == BACKEND_AUDIOCPP: - narrator = voice or speaker_display_name() + if voice: + narrator = voice + elif instructions: + # The voice comes from the instruction, not a speaker name. + narrator = "designed" + else: + narrator = speaker_display_name() elif voice_mode == VOICE_MODE_CLONE: narrator = Path(voice_clone_ref_audio).stem else: @@ -590,9 +611,14 @@ class AudiobookConverter: if self.voice: print("Backend: audio.cpp (voice cloning, reference configured on server)") print(f"Voice: {self.voice}") + elif self.instructions: + print("Backend: audio.cpp (voice from --instructions description)") + print(f"Instruction: {self.instructions}") else: print("Backend: audio.cpp (custom voice, built-in speaker)") print(f"Speaker: {config.SPEAKER}") + if self.request_options: + print(f"Request options: {self.request_options}") if self.client_chunks: print("Chunking: client-side (--chunk; the server also chunks " "long text itself, so this may double-chunk)") @@ -629,8 +655,9 @@ class AudiobookConverter: def preflight_overwrites(backend: str, voice: Optional[str], voice_mode: str, voice_clone_ref_audio: Optional[str], - output_format: str) -> Tuple[List[Path], - List[Tuple[Path, str]]]: + output_format: str, + instructions: Optional[str] = None + ) -> Tuple[List[Path], List[Tuple[Path, str]]]: """Discover books and ask every overwrite question up front. Pure of the TTS server: it scans the books folder, computes the @@ -639,8 +666,8 @@ class AudiobookConverter: existing output files. Returns ``(book_files, planned)`` where ``planned`` is the subset the user agreed to (re)convert. - Asking before connecting means a user who declines a prompt (or - has nothing to convert) never waits on a slow server handshake. + Asking before connecting means a user who declines a prompt (or has + nothing to convert) never waits on a slow server handshake. """ book_files = sorted( f for f in BOOKS_FOLDER.iterdir() @@ -658,7 +685,7 @@ class AudiobookConverter: # starts, so the rest of the run is unattended. planned: List[Tuple[Path, str]] = [] narrator_tag = AudiobookConverter.compute_narrator_tag( - backend, voice, voice_mode, voice_clone_ref_audio) + backend, voice, voice_mode, voice_clone_ref_audio, instructions) for book_file in book_files: output_name = book_file.stem if stem_counts[book_file.stem] > 1: @@ -689,7 +716,8 @@ class AudiobookConverter: else: book_files, planned = AudiobookConverter.preflight_overwrites( self.backend, self.voice, self.voice_mode, - self.voice_clone_ref_audio, self.output_format) + self.voice_clone_ref_audio, self.output_format, + self.instructions) if not book_files: print(f"[INFO] No supported files found in {BOOKS_FOLDER}") diff --git a/converter/tts.py b/converter/tts.py index 9b54cf4..f5d54e2 100644 --- a/converter/tts.py +++ b/converter/tts.py @@ -115,6 +115,17 @@ AUDIOCPP_LANG_OMIT = "omit" # no language field; the model detects it # voice comes from a server-side preset requested with --voice. 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; servers that +# predate the field reported TTS models only, so a missing task is treated +# as "tts"). "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) + class AudioCppFamilyProfile: """Request conventions of one audio.cpp model family.""" @@ -748,17 +759,17 @@ class AudioCppTTSClient(_BaseTTSClient): 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 of the - configured model entry is read from GET /v1/models at startup and - adapts the request payload (language field style, style instructions) - through AUDIOCPP_FAMILY_PROFILES. Two voice modes, both resolved - server-side from the request's "voice" field: + 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. Three voice modes, all resolved + server-side from the request's "voice"/"instructions" fields: - Speaker mode (no ``voice``): Qwen3-TTS only. A built-in CustomVoice speaker name (e.g. "Vivian") is passed through, plus the INSTRUCT style prompt. The server must be configured with the CustomVoice model for this. Families without built-in speakers reject this mode - with a hint to pass --voice. + with a hint to pass --voice (or --instructions, see below). - 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 @@ -769,6 +780,21 @@ class AudioCppTTSClient(_BaseTTSClient): to it. Only the entry actually used needs to exist on the server: a clone-only (Base) server works for --voice runs, while speaker mode on such a server fails with a hint to pass --voice. + - 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. + + ``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: the server does its own long-form text chunking for every family (its ``text_chunk_size`` option, with a per-family default), so @@ -784,7 +810,9 @@ class AudioCppTTSClient(_BaseTTSClient): def __init__(self, voice: Optional[str] = None, language: Optional[str] = None, api_url: Optional[str] = None, chunk_text: bool = False, - model_id: Optional[str] = None): + model_id: Optional[str] = None, + instructions: Optional[str] = None, + request_options: Optional[Dict[str, str]] = None): 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 @@ -802,13 +830,28 @@ class AudioCppTTSClient(_BaseTTSClient): self._seed = _resolve_request_seed() self.preset_mode = bool(voice) self.voice = voice or speaker_display_name() + # 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 {}) + # Both set during _connect once the entry's task is known: 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.design_mode = False + self.instruction_voice = False # When False (default), each chapter is sent as one request and the # server does its own long-form chunking (text_chunk_size); when True, # text is split client-side into CHUNK_SIZE-word sub-requests first. self.chunk_text = bool(chunk_text) - # Family of the selected model entry and its request profile; both - # are resolved from GET /v1/models during _connect. + # 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() @@ -817,12 +860,14 @@ class AudioCppTTSClient(_BaseTTSClient): # ------------------------------------------------------------------ def _connect(self) -> None: - """Health-check the server and resolve the model, family, and voice. + """Health-check the server and resolve the model, family, task, and voice. Speaker mode is only offered to families with built-in speakers (Qwen3-TTS); every other family must select a server-side voice - with --voice, so it fails fast with a hint instead of silently - synthesizing with a random default voice. + with --voice or describe one with --instructions, so it fails fast + with a hint instead of silently synthesizing with a random default + voice. Voice design entries (task "vdes") require --instructions + and reject --voice. """ self._check_health() models = self._list_models() @@ -831,7 +876,33 @@ class AudioCppTTSClient(_BaseTTSClient): self._select_model(models) self._require_model_id(models) self._resolve_family(models) - if self.preset_mode: + self._resolve_task(models) + if self.task not in AUDIOCPP_SYNTHESIS_TASKS: + 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})." + ) + if self.design_mode: + if self.preset_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).") + 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).') + print(f"[OK] Connected to audio.cpp server at {self.api_url} " + f"(model '{self.model_id}', family '{self.family}', " + "voice design)") + print(f"[INFO] Designing the voice from: {self.instructions}") + elif self.preset_mode: self._check_voice() print(f"[OK] Connected to audio.cpp server at {self.api_url} " f"(model '{self.model_id}', family '{self.family}', " @@ -843,13 +914,26 @@ class AudioCppTTSClient(_BaseTTSClient): print("[INFO] Speaker mode expects the server to be configured with the " "CustomVoice model; with the Base model the speaker name is ignored " "and a random default voice is used (see README).") + 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 + print(f"[OK] Connected to audio.cpp server at {self.api_url} " + f"(model '{self.model_id}', family '{self.family}', " + "instruction voice)") + print(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 (see README).") + "config, or describe a voice with --instructions for " + "families that support it (see README).") + if self.instructions and not self.design_mode and not self.instruction_voice: + print(f"[INFO] Sending instruction with every request: {self.instructions}") + print("[INFO] Its effect (style, emotion, delivery) depends on the " + "model family; models without instruction support ignore it.") def _get_json(self, path: str, timeout: int = 10) -> Dict[str, Any]: """GET a JSON document from the server.""" @@ -884,7 +968,7 @@ class AudioCppTTSClient(_BaseTTSClient): f"{payload.get('status')!r} instead of 'ok'") def _list_models(self) -> List[Dict[str, str]]: - """Fetch the (id, family) pairs reported by the server.""" + """Fetch the (id, family, task) triples reported by the server.""" try: payload = self._get_json("/v1/models") except Exception as exc: @@ -898,6 +982,7 @@ class AudioCppTTSClient(_BaseTTSClient): models.append({ "id": entry["id"], "family": entry.get("family") or "", + "task": entry.get("task") or "", }) return models @@ -1034,6 +1119,25 @@ class AudioCppTTSClient(_BaseTTSClient): "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. @@ -1076,8 +1180,12 @@ class AudioCppTTSClient(_BaseTTSClient): payload: Dict[str, Any] = { "model": self.model_id, "input": text, - "voice": self.voice, } + # 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: @@ -1092,11 +1200,19 @@ class AudioCppTTSClient(_BaseTTSClient): # 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 not self.preset_mode and config.INSTRUCT \ + 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") diff --git a/requirements.txt b/requirements.txt index 038c7fb..1747be4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,6 +6,7 @@ ebooklib>=0.18 # Optional dependencies beautifulsoup4>=4.11.0 # better HTML cleaning for EPUB faster-whisper>=1.0.0 # reference-audio transcription for voice cloning +# windows-curses>=2.3 # Windows only: enables the tools/make_audiocpp_server_json.py TUI # Audio processing # Note: ffmpeg is required to concatenate and encode the final audiobook. diff --git a/tests/test_converter.py b/tests/test_converter.py index f09e151..2fe0f5d 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -125,12 +125,13 @@ class FindExistingOutputsTests(unittest.TestCase): class NarratorTagTests(unittest.TestCase): - def _converter(self, voice_mode, ref_audio=None): + def _converter(self, voice_mode, ref_audio=None, instructions=None): converter = AudiobookConverter.__new__(AudiobookConverter) converter.voice_mode = voice_mode converter.voice_clone_ref_audio = ref_audio converter.backend = tts.BACKEND_QWEN converter.voice = None + converter.instructions = instructions return converter def test_custom_voice_uses_speaker_display_name(self): @@ -158,6 +159,47 @@ class NarratorTagTests(unittest.TestCase): self.assertEqual(self._converter(tts.VOICE_MODE_CLONE, "/x/???.wav")._narrator_tag(), "narrator") + def _audiocpp_converter(self, voice=None, instructions=None): + converter = self._converter(tts.VOICE_MODE_CUSTOM, + instructions=instructions) + converter.backend = tts.BACKEND_AUDIOCPP + converter.voice = voice + return converter + + def test_audiocpp_design_run_uses_designed_tag(self): + # An instruction without a voice (voice design, or instruction- + # defined voices) must not be named after the built-in speaker. + converter = self._audiocpp_converter(instructions="A warm narrator") + self.assertEqual(converter._narrator_tag(), "designed") + + def test_audiocpp_instruction_with_voice_keeps_voice_tag(self): + converter = self._audiocpp_converter( + voice="narrator", instructions="Calm delivery") + self.assertEqual(converter._narrator_tag(), "narrator") + + def test_audiocpp_speaker_mode_keeps_speaker_tag(self): + converter = self._audiocpp_converter() + self.assertEqual(converter._narrator_tag(), "Vivian") + + def test_preflight_design_run_uses_designed_tag(self): + with tempfile.TemporaryDirectory() as books_tmp, \ + tempfile.TemporaryDirectory() as output_tmp: + original = (converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER) + converter_mod.BOOKS_FOLDER = Path(books_tmp) + converter_mod.AUDIOBOOKS_FOLDER = Path(output_tmp) + try: + (converter_mod.BOOKS_FOLDER / "book.txt").write_text( + "hello world", encoding="utf-8") + with patch("builtins.input", + side_effect=AssertionError("should not prompt")): + _, planned = AudiobookConverter.preflight_overwrites( + tts.BACKEND_AUDIOCPP, None, tts.VOICE_MODE_CUSTOM, + None, "mp3", instructions="A warm narrator") + self.assertEqual(planned, [(converter_mod.BOOKS_FOLDER / "book.txt", + "book_designed")]) + finally: + converter_mod.BOOKS_FOLDER, converter_mod.AUDIOBOOKS_FOLDER = original + class ChapterDebugDirTests(unittest.TestCase): """Per-chapter debug subfolder naming (chunk numbering restarts per chapter).""" @@ -538,6 +580,7 @@ class RunOverwritePromptTests(unittest.TestCase): self.converter.voice_clone_ref_audio = None self.converter.backend = tts.BACKEND_QWEN self.converter.voice = None + self.converter.instructions = None self.converter.speed = 1.0 self.converter.single_file = False self.converter.output_format = "mp3" diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py index d2d93bb..ca17e93 100644 --- a/tests/test_make_audiocpp_server_json.py +++ b/tests/test_make_audiocpp_server_json.py @@ -1,5 +1,6 @@ """Tests for the audio.cpp server.json generator tool.""" +import argparse import io import json import sys @@ -58,11 +59,17 @@ def _make_checkout(tmp: Path) -> Path: _write_spec(checkout, "qwen3_tts", display_name="Qwen3-TTS", tasks=("tts", "clone", "design"), languages=("zh", "en", "ja"), - packages=[{ - "id": "qwen3_tts_1_7b_base_q8_0", "default": True, - "format": "gguf", - "target_directory": "Qwen3-TTS-12Hz-1.7B-Base-GGUF", - }]) + packages=[ + {"id": "qwen3_tts_1_7b_base_q8_0", "default": True, + "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-1.7B-Base-GGUF"}, + {"id": "qwen3_tts_1_7b_customvoice_q8_0", + "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF"}, + {"id": "qwen3_tts_1_7b_voicedesign_q8_0", + "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF"}, + ]) _write_spec(checkout, "higgs_audio_tts", display_name="Higgs Audio v3 TTS 4B", languages=("auto",), packages=[{ @@ -286,6 +293,23 @@ class ResolveWavDirArgTests(unittest.TestCase): self.folder.resolve()) +class NormalizeDirArgTests(unittest.TestCase): + """Path normalization for the audio.cpp checkout argument.""" + + def test_expands_tilde_and_resolves(self): + with patch.object(make_server.os.path, "expanduser", + return_value="/home/u/audio.cpp") as mock_expand: + result = make_server.normalize_dir_arg("~/audio.cpp") + mock_expand.assert_called_once_with("~/audio.cpp") + self.assertEqual(result, Path("/home/u/audio.cpp").resolve()) + + def test_strips_quotes_and_whitespace(self): + with patch.object(make_server.os.path, "expanduser", + side_effect=lambda s: s): + result = make_server.normalize_dir_arg(' "/tmp/foo" ') + self.assertEqual(result, Path("/tmp/foo").resolve()) + + class DefaultModelIdTests(unittest.TestCase): def test_preferred_ids_for_tested_families(self): self.assertEqual(make_server.default_model_id("qwen3_tts"), "qwen") @@ -444,6 +468,12 @@ class BuildServerConfigTests(unittest.TestCase): self.assertEqual(entry["task"], "tts") self.assertEqual(entry["mode"], "offline") + def test_model_entry_design_task(self): + entry = make_server.build_model_entry( + "qwen3_tts", "qwen-design", "p", task="vdes") + self.assertEqual(entry["task"], "vdes") + self.assertEqual(entry["mode"], "offline") + class TranscribeWavDirTests(unittest.TestCase): def setUp(self): @@ -522,6 +552,67 @@ class PromptHelperTests(unittest.TestCase): "one") +class DesignPackageTests(unittest.TestCase): + """Voice-design package detection.""" + + def test_detects_voicedesign_in_id(self): + self.assertTrue(make_server.is_design_package( + {"id": "qwen3_tts_1_7b_voicedesign_q8_0"})) + + def test_detects_voicedesign_in_directory(self): + self.assertTrue(make_server.is_design_package( + {"target_directory": "Foo-VoiceDesign-GGUF"})) + + def test_detects_separated_voice_design(self): + self.assertTrue(make_server.is_design_package( + {"display_name": "Voice Design Q8_0"})) + + def test_ignores_other_packages(self): + self.assertFalse(make_server.is_design_package( + {"id": "higgs_audio_tts_4b_q8_0"})) + self.assertFalse(make_server.is_design_package({})) + + +class PackageDirOptionsTests(unittest.TestCase): + """Grouping a family's packages into distinct target directories.""" + + def test_groups_precisions_and_marks_recommended(self): + entry = { + "family": "qwen3_tts", + "packages": [ + {"id": "base_q8", "default": True, "format": "gguf", + "target_directory": "Base-GGUF"}, + {"id": "base_bf16", "format": "gguf", + "target_directory": "Base-GGUF"}, + {"id": "voicedesign_q8", "format": "gguf", + "target_directory": "VoiceDesign-GGUF"}, + ], + } + options = make_server.package_dir_options(entry) + self.assertEqual([o["target_directory"] for o in options], + ["Base-GGUF", "VoiceDesign-GGUF"]) + self.assertTrue(options[0]["recommended"]) + self.assertFalse(options[0]["design"]) + self.assertFalse(options[1]["recommended"]) + self.assertTrue(options[1]["design"]) + # The recommended precision inside the shared directory wins. + self.assertEqual(options[0]["install_id"], "base_q8") + + def test_recommended_comes_first_even_if_listed_later(self): + entry = { + "family": "demo_tts", + "packages": [ + {"id": "demo_other", "format": "gguf", + "target_directory": "Other-GGUF"}, + {"id": "demo_default", "default": True, "format": "gguf", + "target_directory": "Default-GGUF"}, + ], + } + options = make_server.package_dir_options(entry) + self.assertEqual([o["target_directory"] for o in options], + ["Default-GGUF", "Other-GGUF"]) + + class _MainTestBase(unittest.TestCase): """Shared fixtures for end-to-end main() tests.""" @@ -539,6 +630,12 @@ class _MainTestBase(unittest.TestCase): patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) patcher.start() self.addCleanup(patcher.stop) + # Force the line-prompt flow regardless of the test terminal, so + # the builtins.input patches below are what actually answer the + # questions (the TUI path is exercised separately). + patcher = patch.object(make_server, "_tui_enabled", return_value=False) + patcher.start() + self.addCleanup(patcher.stop) def tearDown(self): self._td.cleanup() @@ -555,30 +652,35 @@ class _MainTestBase(unittest.TestCase): return_value=whisper): return make_server.main() + # Default single-family run inputs (no flags, port matches config): + # family, host, port, backend, lazy, model-id-sync. + def _defaults(self, sync="y"): + return ["", "", "", "", "", sync] + class MainTests(_MainTestBase): - """The default Qwen3-TTS flow and shared server settings.""" + """The default single-family flow and shared server settings.""" def _args(self, *extra): - return [str(self.folder), "--output", str(self.output), + return ["--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout)] + list(extra) - # Default Qwen3-TTS "both" run inputs (no flags, port matches config): - # families, models, custom_path, base_path, host, port, backend, lazy, confirm - def _defaults(self, confirm="y"): - return ["", "", "", "", "", "", "", "", confirm] - - def test_required_wav_dir_missing_prints_usage(self): - with self.assertRaises(SystemExit) as ctx: - self._run(["--output", str(self.output), - "--audiocpp-dir", str(self.checkout)], inputs=[]) + def test_missing_wav_dir_prompted_errors(self): + # No --wavs and EOF at the prompt -> hard error. + buf = io.StringIO() + with patch.object(sys, "argv", + ["make_audiocpp_server_json.py", + "--output", str(self.output), + "--audiocpp-dir", str(self.checkout)]), \ + patch("builtins.input", side_effect=EOFError), \ + redirect_stdout(buf): + with self.assertRaises(SystemExit) as ctx: + make_server.main() self.assertEqual(ctx.exception.code, 2) - self.assertFalse(self.output.exists()) def test_missing_audiocpp_dir_errors(self): with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.root / "nope")], + self._run(self._args("--audiocpp-dir", str(self.root / "nope")), inputs=[]) self.assertEqual(ctx.exception.code, 2) @@ -587,29 +689,28 @@ class MainTests(_MainTestBase): buf = io.StringIO() with patch.object(sys, "argv", ["make_audiocpp_server_json.py", - str(self.folder), "--output", str(self.output)]), \ + "--wavs", str(self.folder), + "--output", str(self.output)]), \ patch("builtins.input", side_effect=EOFError), \ redirect_stdout(buf): with self.assertRaises(SystemExit) as ctx: make_server.main() self.assertEqual(ctx.exception.code, 2) - def test_default_run_hosts_both_models(self): + def test_default_run_hosts_recommended_entry(self): exit_code = self._run(self._args(), inputs=self._defaults()) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(data["host"], "127.0.0.1") self.assertEqual(data["port"], make_server.config_port()) self.assertEqual(data["backend"], "cuda") - # Single family (qwen3_tts) -> lazy defaults to False. + # Single family -> one entry, lazy defaults to False. self.assertFalse(data["lazy_load"]) - self.assertEqual( - [model["id"] for model in data["models"]], - [config.AUDIOCPP_MODEL_ID, config.AUDIOCPP_CLONE_MODEL_ID]) + self.assertEqual([model["id"] for model in data["models"]], ["qwen"]) self.assertEqual( [model["path"] for model in data["models"]], - [make_server.DEFAULT_CUSTOM_VOICE_PATH, - make_server.DEFAULT_BASE_PATH]) + ["models/Qwen3-TTS-12Hz-1.7B-Base-GGUF"]) + self.assertEqual(data["models"][0]["task"], "tts") # voice_dir only when wavs are present; this run has none. self.assertNotIn("voice_dir", data) @@ -617,71 +718,14 @@ class MainTests(_MainTestBase): exit_code = self._run(self._args()) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 2) - - def test_clone_only_run(self): - (self.folder / "narrator.wav").write_bytes(b"x") - (self.folder / "alpha.wav").write_bytes(b"x") - # families=default, models=3(clone), custom_path skipped, base_path, - # host, port, backend, lazy, confirm - inputs = ["", "3", "", "", "", "", "", "y"] - exit_code = self._run( - self._args(), - inputs=inputs, - transcribe=lambda path, model_name="base": - f"transcript of {Path(path).name}") - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(len(data["models"]), 1) - clone_entry = data["models"][0] - self.assertEqual(clone_entry["id"], config.AUDIOCPP_CLONE_MODEL_ID) - # Voice presets now live in a server-level voice_dir + prompt_text, - # not per-entry voice_presets. - self.assertNotIn("voice_presets", clone_entry) - self.assertIn("voice_dir", data) - self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - prompt = (self.folder / make_server.PROMPT_TEXT_FILENAME).read_text( - encoding="utf-8") - self.assertIn("narrator|transcript of narrator.wav", prompt) - self.assertIn("alpha|transcript of alpha.wav", prompt) - - def test_custom_only_single_model(self): - # families=default, models=2(custom), host, port, backend, lazy, confirm - inputs = ["", "2", "", "", "", "", "y"] - exit_code = self._run( - self._args("--models", "custom"), inputs=inputs) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual([model["id"] for model in data["models"]], - [config.AUDIOCPP_MODEL_ID]) - - def test_duplicate_ids_prompt_for_distinct_clone_id(self): - with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ - patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): - # families, models(default both), distinct_clone_id, custom_path, - # base_path, host, port, backend, lazy, confirm - inputs = ["", "", "qwen-clone-2", "", "", "", "", "", "", "y"] - exit_code = self._run(self._args(), inputs=inputs) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual([model["id"] for model in data["models"]], - ["qwen", "qwen-clone-2"]) - - def test_duplicate_ids_eof_exits(self): - with patch.object(config, "AUDIOCPP_MODEL_ID", "qwen"), \ - patch.object(config, "AUDIOCPP_CLONE_MODEL_ID", "qwen"): - with self.assertRaises(SystemExit) as ctx: - self._run(self._args()) - self.assertNotEqual(ctx.exception.code, 0) - self.assertFalse(self.output.exists()) def test_port_sync_accepted_updates_config(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): # --port 8080 differs from config port 9999 -> sync prompt fires. - # families, models, custom_path, base_path, host, port_sync(y), - # backend, lazy, confirm - inputs = ["", "", "", "", "", "y", "", "", "y"] + # family, host, port_sync(y), backend, lazy, sync(y) + inputs = ["", "", "y", "", "", "y"] exit_code = self._run( self._args("--port", "8080"), inputs=inputs) self.assertEqual(exit_code, 0) @@ -693,7 +737,7 @@ class MainTests(_MainTestBase): def test_port_sync_declined_keeps_config(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:9999"): - inputs = ["", "", "", "", "", "n", "", "", "y"] + inputs = ["", "", "n", "", "", "n"] exit_code = self._run( self._args("--port", "8080"), inputs=inputs) self.assertEqual(exit_code, 0) @@ -708,11 +752,12 @@ class MainTests(_MainTestBase): self.assertEqual(self.fake_config.read_text(encoding="utf-8"), FAKE_CONFIG) - def test_confirm_declined_writes_nothing(self): - inputs = self._defaults(confirm="n") - exit_code = self._run(self._args(), inputs=inputs) - self.assertEqual(exit_code, 1) - self.assertFalse(self.output.exists()) + def test_no_final_confirm_prompt_writes_file(self): + # There is no final confirmation prompt anymore; the config is always + # written once the (single) overwrite check has been passed. + exit_code = self._run(self._args(), inputs=EOFError) + self.assertEqual(exit_code, 0) + self.assertTrue(self.output.exists()) def test_existing_output_declined_keeps_file(self): self.output.write_text('{"old": true}', encoding="utf-8") @@ -727,7 +772,7 @@ class MainTests(_MainTestBase): exit_code = self._run(self._args(), inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 2) + self.assertEqual(len(data["models"]), 1) def test_force_overwrites_without_prompt(self): self.output.write_text('{"old": true}', encoding="utf-8") @@ -735,16 +780,16 @@ class MainTests(_MainTestBase): exit_code = self._run(self._args("--force"), inputs=inputs) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 2) + self.assertEqual(len(data["models"]), 1) def test_flags_skip_prompts(self): - # --families qwen3_tts --models both + server flags; port 9000 differs - # from config port 8080 -> the port sync prompt still fires. + # --families qwen3_tts + server flags; port 9000 differs from config + # port 8080 -> the port sync prompt still fires. exit_code = self._run( - self._args("--families", "qwen3_tts", "--models", "both", + self._args("--families", "qwen3_tts", "--host", "0.0.0.0", "--port", "9000", "--backend", "cpu", "--lazy-load"), - inputs=["y", "", "", "y"]) + inputs=["y", "y"]) self.assertEqual(exit_code, 0) self.assertIn('"http://127.0.0.1:9000"', self.fake_config.read_text(encoding="utf-8")) @@ -754,11 +799,11 @@ class MainTests(_MainTestBase): self.assertEqual(data["backend"], "cpu") self.assertTrue(data["lazy_load"]) - def test_missing_positional_wav_dir_errors(self): + def test_missing_wav_dir_flag_errors_with_message(self): missing = self.root / "nope" with self.assertRaises(SystemExit) as ctx, \ patch("sys.stderr") as mock_stderr: - self._run([str(missing), "--output", str(self.output), + self._run(["--wavs", str(missing), "--output", str(self.output), "--audiocpp-dir", str(self.checkout)], inputs=self._defaults()) self.assertEqual(ctx.exception.code, 2) @@ -766,12 +811,80 @@ class MainTests(_MainTestBase): self.assertIn(f"WAV directory not found: {missing.resolve()}", shown) self.assertIn("directory containing the .wav", shown) - def test_models_flag_rejected_without_qwen(self): - with self.assertRaises(SystemExit) as ctx: - self._run(self._args("--families", "higgs_audio_tts", - "--models", "both"), - inputs=[]) - self.assertEqual(ctx.exception.code, 2) + def _run_capturing(self, argv, inputs): + argv = ["make_audiocpp_server_json.py"] + argv + buf = io.StringIO() + with patch.object(sys, "argv", argv), \ + patch("builtins.input", side_effect=inputs), \ + patch.object(make_server, "transcribe_reference_audio"), \ + patch.object(make_server, "whisper_backend_available", + return_value="faster_whisper"), \ + redirect_stdout(buf): + code = make_server.main() + return code, buf.getvalue() + + def test_all_packages_design_hosts_vdes_entry(self): + # --all-packages: pick the VoiceDesign package (menu 3) and accept the + # "design" default so it is hosted with task "vdes". + self.fake_config.write_text(FAKE_CONFIG_WITH_MODEL_IDS, + encoding="utf-8") + # family, packages(3=VoiceDesign), task(design default Enter), host, + # port, backend, lazy, sync(y) + inputs = ["", "3", "", "", "", "", "", "y"] + code, out = self._run_capturing( + self._args("--all-packages"), inputs=inputs) + self.assertEqual(code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["models"], [{ + "id": "qwen-design", + "family": "qwen3_tts", + "path": "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF", + "task": "vdes", + "mode": "offline", + }]) + self.assertNotIn("voice_dir", data) + # Only the VoiceDesign package is installed (custom/base are not). + self.assertIn("install qwen3_tts_1_7b_voicedesign_q8_0", out) + self.assertNotIn("install qwen3_tts_1_7b_customvoice_q8_0", out) + self.assertNotIn("install qwen3_tts_1_7b_base_q8_0", out) + # Usage guidance points at the --instructions flow. + self.assertIn("--model qwen-design", out) + self.assertIn("--instructions", out) + # Single-entry server: the converter ids are synced to the entry. + text = self.fake_config.read_text(encoding="utf-8") + self.assertIn('AUDIOCPP_MODEL_ID = "qwen-design"', text) + self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "qwen-design"', text) + + def test_all_packages_non_design_package_gets_tts_no_prompt(self): + # CustomVoice (menu 2) is not a design package -> task "tts" with no + # task prompt. + inputs = ["", "2", "", "", "", "", "y"] + code, _ = self._run_capturing( + self._args("--all-packages"), inputs=inputs) + self.assertEqual(code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["models"], [{ + "id": "qwen", + "family": "qwen3_tts", + "path": "models/Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF", + "task": "tts", + "mode": "offline", + }]) + + def test_all_packages_both_tts_and_design(self): + # Pick Base (recommended) + VoiceDesign -> two entries; the design + # package prompts for its task. + # family, packages(1,3), task(design default Enter), host, port, + # backend, lazy + inputs = ["", "1,3", "", "", "", "", ""] + code, _ = self._run_capturing( + self._args("--all-packages"), inputs=inputs) + self.assertEqual(code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([model["id"] for model in data["models"]], + ["qwen", "qwen-design"]) + self.assertEqual([model["task"] for model in data["models"]], + ["tts", "vdes"]) class NonQwenFamilyMainTests(_MainTestBase): @@ -785,16 +898,15 @@ class NonQwenFamilyMainTests(_MainTestBase): encoding="utf-8") def _args(self, family, *extra): - return [str(self.folder), "--output", str(self.output), + return ["--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout), "--families", family] + list(extra) def test_higgs_family_run(self): (self.folder / "narrator.wav").write_bytes(b"x") - # Single non-qwen family -> path is asked; then host, port, backend, - # lazy, confirm, model-id sync(y). prompt_text is written (no overwrite - # prompt on a fresh directory). - inputs = ["", "", "", "", "", "y", "y"] + # Single family -> path comes from the catalog (no prompt); host, port, + # backend, lazy, model-id sync(y). + inputs = ["", "", "", "", "y"] exit_code = self._run( self._args("higgs_audio_tts"), inputs=inputs, transcribe=lambda path, model_name="base": "a transcript") @@ -813,15 +925,15 @@ class NonQwenFamilyMainTests(_MainTestBase): prompt = (self.folder / make_server.PROMPT_TEXT_FILENAME).read_text( encoding="utf-8") self.assertIn("narrator|a transcript", prompt) - # Single non-qwen entry -> both converter ids are synced to it. + # Single entry -> both converter ids are synced to it. text = self.fake_config.read_text(encoding="utf-8") self.assertIn('AUDIOCPP_MODEL_ID = "higgs"', text) self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "higgs"', text) def test_model_id_sync_declined_keeps_config(self): (self.folder / "narrator.wav").write_bytes(b"x") - # path, host, port, backend, lazy, confirm, sync(n) - inputs = ["", "", "", "", "", "y", "n"] + # host, port, backend, lazy, sync(n) + inputs = ["", "", "", "", "n"] exit_code = self._run( self._args("voxcpm2"), inputs=inputs, transcribe=lambda path, model_name="base": "t") @@ -834,11 +946,11 @@ class NonQwenFamilyMainTests(_MainTestBase): def test_no_wavs_warns_and_omits_voice_dir(self): buf = io.StringIO() - # path, host, port, backend, lazy, confirm, sync(y) - inputs = ["", "", "", "", "", "y", "y"] + # host, port, backend, lazy, sync(y) + inputs = ["", "", "", "", "y"] with patch.object(sys, "argv", ["make_audiocpp_server_json.py", - str(self.folder), "--output", str(self.output), + "--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout), "--families", "index_tts2"]), \ patch("builtins.input", side_effect=inputs), \ @@ -864,16 +976,14 @@ class MultiFamilyMainTests(_MainTestBase): """Hosting several families in one server.json.""" def _args(self, *extra): - return [str(self.folder), "--output", str(self.output), + return ["--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout)] + list(extra) def test_multiple_families_lazy_by_default_with_voice_dir(self): (self.folder / "narrator.wav").write_bytes(b"x") - # --families selects qwen3_tts + higgs_audio_tts. qwen is among them - # with others -> qwen sub-flow forced to "both" (no models prompt). - # custom_path, base_path, host, port, backend, lazy(default True->Enter), - # prompt_text overwrite(none yet->writes), confirm - inputs = ["", "", "", "", "", "", "", "y"] + # --families selects qwen3_tts + higgs_audio_tts; each hosts its + # recommended package. host, port, backend, lazy(default True->Enter). + inputs = ["", "", "", ""] exit_code = self._run( self._args("--families", "qwen3_tts,higgs_audio_tts"), inputs=inputs, @@ -881,18 +991,17 @@ class MultiFamilyMainTests(_MainTestBase): self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) ids = [model["id"] for model in data["models"]] - self.assertEqual(ids, ["qwen", "qwen-clone", "higgs"]) - # Two families -> lazy defaults to True. + self.assertEqual(ids, ["qwen", "higgs"]) + # Two entries -> lazy defaults to True. self.assertTrue(data["lazy_load"]) self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - # Multi-entry -> the tool prints a --model note instead of syncing. - higgs = data["models"][2] + higgs = data["models"][1] self.assertEqual(higgs["path"], "models/Higgs-Audio-v3-TTS-4B-GGUF") def test_two_non_qwen_families_use_catalog_paths(self): - # Multiple non-qwen families -> paths are NOT prompted (catalog defaults). - # qwen absent -> no models prompt; host, port, backend, lazy, confirm - inputs = ["", "", "", "", "y"] + # Multiple families -> paths come from the catalog (no prompts). + # host, port, backend, lazy + inputs = ["", "", "", ""] exit_code = self._run( self._args("--families", "higgs_audio_tts,voxcpm2"), inputs=inputs) @@ -902,17 +1011,17 @@ class MultiFamilyMainTests(_MainTestBase): self.assertEqual(by_id["higgs"]["path"], "models/Higgs-Audio-v3-TTS-4B-GGUF") self.assertEqual(by_id["voxcpm2"]["path"], "models/VoxCPM2-GGUF") - # No wavs and both clone-capable, but no wavs present -> no voice_dir. + # No wavs present -> no voice_dir. self.assertNotIn("voice_dir", data) def test_non_clone_family_selected_warns_about_wav_dir(self): buf = io.StringIO() # supertonic is TTS-only (no clone): wav dir is ignored. - # path, host, port, backend, lazy, confirm, sync(n) - inputs = ["", "", "", "", "y", "y", "n"] + # host, port, backend, lazy, sync(n) + inputs = ["", "", "", "", "n"] with patch.object(sys, "argv", ["make_audiocpp_server_json.py", - str(self.folder), "--output", str(self.output), + "--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout), "--families", "supertonic"]), \ patch("builtins.input", side_effect=inputs), \ @@ -929,11 +1038,43 @@ class MultiFamilyMainTests(_MainTestBase): self.assertEqual(data["models"][0]["family"], "supertonic") +class DefaultOutputTests(_MainTestBase): + """server.json defaults into the audio.cpp checkout unless declined.""" + + def test_default_output_written_into_checkout(self): + # No --output: server.json lands in the audio.cpp checkout. + argv = ["--wavs", str(self.folder), "--audiocpp-dir", str(self.checkout)] + exit_code = self._run(argv, inputs=self._defaults()) + self.assertEqual(exit_code, 0) + out = self.checkout / "server.json" + self.assertTrue(out.exists()) + data = json.loads(out.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 1) + + def test_declined_overwrite_falls_back_to_cwd(self): + # A pre-existing server.json in the checkout; declining the overwrite + # writes server.json into the current working directory instead. + checkout_out = self.checkout / "server.json" + checkout_out.write_text('{"old": true}', encoding="utf-8") + cwd = self.root / "run-cwd" + cwd.mkdir() + argv = ["--wavs", str(self.folder), "--audiocpp-dir", str(self.checkout)] + with patch.object(make_server.os, "getcwd", return_value=str(cwd)): + exit_code = self._run(argv, inputs=["n"] + self._defaults()) + self.assertEqual(exit_code, 0) + self.assertEqual(json.loads(checkout_out.read_text(encoding="utf-8")), + {"old": True}) + fallback = cwd / "server.json" + self.assertTrue(fallback.exists()) + data = json.loads(fallback.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 1) + + class TranscriptWarningTests(_MainTestBase): """Empty transcripts and a missing Whisper backend produce loud warnings.""" def _args(self, *extra): - return [str(self.folder), "--output", str(self.output), + return ["--wavs", str(self.folder), "--output", str(self.output), "--audiocpp-dir", str(self.checkout)] + list(extra) def _run_capturing(self, argv, inputs, transcribe, whisper): @@ -952,9 +1093,7 @@ class TranscriptWarningTests(_MainTestBase): def test_empty_transcript_prints_loud_end_warning(self): (self.folder / "narrator.wav").write_bytes(b"x") (self.folder / "alpha.wav").write_bytes(b"x") - # Qwen clone-only (menu 3); custom_path skipped, base_path, host, port, - # backend, lazy, prompt_text write, confirm - inputs = ["", "3", "", "", "", "", "", "", "y"] + inputs = self._defaults() code, out = self._run_capturing( self._args(), inputs=inputs, transcribe=lambda path, model_name="base": None, @@ -965,20 +1104,20 @@ class TranscriptWarningTests(_MainTestBase): self.assertIn("alpha", out) self.assertIn("prompt_text", out) - def test_missing_whisper_backend_prints_conda_warning(self): + def test_missing_whisper_backend_prints_install_warning(self): (self.folder / "narrator.wav").write_bytes(b"x") - inputs = ["", "3", "", "", "", "", "", "", "y"] + inputs = self._defaults() code, out = self._run_capturing( self._args(), inputs=inputs, transcribe=lambda path, model_name="base": "a transcript", whisper=None) self.assertEqual(code, 0) - self.assertIn("conda activate qwen3-tts", out) + self.assertIn("Install whisper", out) self.assertIn("faster_whisper", out) def test_all_transcripts_present_prints_no_end_warning(self): (self.folder / "narrator.wav").write_bytes(b"x") - inputs = ["", "3", "", "", "", "", "", "", "y"] + inputs = self._defaults() code, out = self._run_capturing( self._args(), inputs=inputs, transcribe=lambda path, model_name="base": "a real transcript", @@ -987,5 +1126,167 @@ class TranscriptWarningTests(_MainTestBase): self.assertNotIn("MANUAL TRANSCRIPTION REQUIRED", out) +class PromptTextReuseTests(_MainTestBase): + """Reusing an existing prompt_text and transcribing only new voices.""" + + def _args(self, *extra): + return ["--wavs", str(self.folder), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout), + "--families", "higgs_audio_tts"] + list(extra) + + def _run_capturing(self, argv, inputs, transcribe): + argv = ["make_audiocpp_server_json.py"] + argv + buf = io.StringIO() + with patch.object(sys, "argv", argv), \ + patch("builtins.input", side_effect=inputs), \ + patch.object(make_server, "transcribe_reference_audio", + side_effect=transcribe), \ + patch.object(make_server, "whisper_backend_available", + return_value="faster_whisper"), \ + redirect_stdout(buf): + code = make_server.main() + return code, buf.getvalue() + + def _transcribe(self, called, text): + def transcribe(path, model_name="base"): + called.append(path) + return text + return transcribe + + def test_all_present_decline_keeps_file_and_skips_transcribe(self): + (self.folder / "narrator.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("narrator|An existing transcript.\n", + encoding="utf-8") + called = [] + # host, port, backend, lazy, re-transcribe(n), sync(y) + inputs = ["", "", "", "", "n", "y"] + code, out = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(called, []) + self.assertEqual(prompt.read_text(encoding="utf-8"), + "narrator|An existing transcript.\n") + self.assertIn("Kept existing", out) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["voice_dir"], str(self.folder.resolve())) + + def test_all_present_accept_retranscribes_and_overwrites(self): + (self.folder / "narrator.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("narrator|Old.\n", encoding="utf-8") + called = [] + # host, port, backend, lazy, re-transcribe(y), sync(y) + inputs = ["", "", "", "", "y", "y"] + code, _ = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(called, [str(self.folder / "narrator.wav")]) + self.assertIn("narrator|Fresh.", prompt.read_text(encoding="utf-8")) + + def test_new_voice_merges_preserving_hand_edits(self): + (self.folder / "existing.wav").write_bytes(b"x") + (self.folder / "new.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("existing|Hand edited transcript.\n", + encoding="utf-8") + called = [] + # host, port, backend, lazy, only-new(Enter -> y), sync(y) + inputs = ["", "", "", "", "", "y"] + code, _ = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "New transcript.")) + self.assertEqual(code, 0) + self.assertEqual(called, [str(self.folder / "new.wav")]) + text = prompt.read_text(encoding="utf-8") + self.assertIn("existing|Hand edited transcript.", text) + self.assertIn("new|New transcript.", text) + + def test_new_voice_decline_retranscribes_all(self): + (self.folder / "existing.wav").write_bytes(b"x") + (self.folder / "new.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("existing|Old.\n", encoding="utf-8") + called = [] + # host, port, backend, lazy, only-new(n), sync(y) + inputs = ["", "", "", "", "n", "y"] + code, _ = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(sorted(called), sorted([ + str(self.folder / "existing.wav"), str(self.folder / "new.wav")])) + self.assertIn("existing|Fresh.", prompt.read_text(encoding="utf-8")) + + def test_force_retranscribes_without_prompt(self): + (self.folder / "narrator.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("narrator|Old.\n", encoding="utf-8") + called = [] + # host, port, backend, lazy, sync(y); no re-transcribe prompt with force. + inputs = ["", "", "", "", "y"] + code, _ = self._run_capturing( + self._args("--force"), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(called, [str(self.folder / "narrator.wav")]) + self.assertIn("narrator|Fresh.", prompt.read_text(encoding="utf-8")) + + def test_empty_transcript_counts_as_missing(self): + (self.folder / "narrator.wav").write_bytes(b"x") + prompt = self.folder / make_server.PROMPT_TEXT_FILENAME + prompt.write_text("narrator|\n", encoding="utf-8") + called = [] + # Empty transcript is treated as missing -> the "only new voices" + # prompt fires (Enter -> y). + # host, port, backend, lazy, only-new(Enter), sync(y) + inputs = ["", "", "", "", "", "y"] + code, _ = self._run_capturing( + self._args(), inputs=inputs, + transcribe=self._transcribe(called, "Fresh.")) + self.assertEqual(code, 0) + self.assertEqual(called, [str(self.folder / "narrator.wav")]) + self.assertIn("narrator|Fresh.", prompt.read_text(encoding="utf-8")) + + +class ModeSelectionTests(unittest.TestCase): + """Choosing between the TUI wizard and the line prompts.""" + + def _args(self, notui=False): + return argparse.Namespace(notui=notui) + + def test_notui_flag_forces_prompt_mode(self): + # Even with a tty and an importable curses, --notui disables the TUI. + with patch.object(make_server, "_curses_importable", return_value=True), \ + patch.object(make_server.sys.stdin, "isatty", return_value=True), \ + patch.object(make_server.sys.stdout, "isatty", return_value=True): + self.assertFalse(make_server._tui_enabled(self._args(notui=True))) + + def test_non_tty_forces_prompt_mode(self): + with patch.object(make_server, "_curses_importable", return_value=True), \ + patch.object(make_server.sys.stdin, "isatty", return_value=False), \ + patch.object(make_server.sys.stdout, "isatty", return_value=True): + self.assertFalse(make_server._tui_enabled(self._args())) + + def test_tty_with_curses_uses_tui(self): + with patch.object(make_server, "_curses_importable", return_value=True), \ + patch.object(make_server.sys.stdin, "isatty", return_value=True), \ + patch.object(make_server.sys.stdout, "isatty", return_value=True): + self.assertTrue(make_server._tui_enabled(self._args())) + + def test_missing_curses_forces_prompt_mode(self): + with patch.object(make_server, "_curses_importable", return_value=False), \ + patch.object(make_server.sys.stdin, "isatty", return_value=True), \ + patch.object(make_server.sys.stdout, "isatty", return_value=True): + self.assertFalse(make_server._tui_enabled(self._args())) + + def test_curses_is_importable_on_this_platform(self): + # The TUI widget module imports without curses at module load time, + # but the wizard still needs the real curses package to run. + self.assertTrue(make_server._curses_importable()) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_tts.py b/tests/test_tts.py index 89248f2..a2df07f 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -654,6 +654,144 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): self.assertIn("pocket-tts", message) +class AudioCppTaskDetectionTests(unittest.TestCase): + """Task auto-detection (tts/clon/vdes) and voice design validation.""" + + @staticmethod + def _json_response(payload): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps(payload).encode("utf-8") + return response + + def _client(self, voice=None, instructions=None, request_options=None, + models=None): + if models is None: + models = {"data": [{"id": config.AUDIOCPP_MODEL_ID, + "family": "qwen3_tts"}]} + + def _dispatch(request, **_kwargs): + url = request if isinstance(request, str) else request.full_url + if url.endswith("/health"): + return self._json_response({"status": "ok"}) + if url.endswith("/v1/models"): + return self._json_response(models) + if "/v1/audio/voices" in url: + return self._json_response({"voices": ["narrator"]}) + raise AssertionError(f"unexpected URL: {url}") + + with patch("converter.tts.urllib.request.urlopen", + side_effect=_dispatch): + return AudioCppTTSClient(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.assertFalse(client.design_mode) + + def test_task_detected_from_models_endpoint(self): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="A warm adult narrator") + self.assertEqual(client.task, tts.AUDIOCPP_TASK_VDES) + self.assertTrue(client.design_mode) + + def test_clon_task_entry_connects_in_preset_mode(self): + client = self._client(voice="narrator", models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "chatterbox", + "task": "clon"}]}) + self.assertEqual(client.task, "clon") + self.assertFalse(client.design_mode) + self.assertTrue(client.preset_mode) + + def test_unsupported_task_rejected_with_available_entries(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_asr", + "task": "asr"}, + {"id": "tts-1", "family": "qwen3_tts", "task": "tts"}]}, + instructions="unused") + message = str(ctx.exception) + self.assertIn("'asr'", message) + self.assertIn("--model", message) + self.assertIn("tts-1", message) + + def test_vdes_without_instructions_requires_description(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}) + message = str(ctx.exception) + self.assertIn("voice design", message) + self.assertIn("--instructions", message) + + def test_vdes_with_voice_rejected(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(voice="narrator", models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="A warm adult narrator") + self.assertIn("--voice", str(ctx.exception)) + self.assertIn("--instructions", str(ctx.exception)) + + def test_vdes_with_instructions_connects_in_design_mode(self): + buf = io.StringIO() + with redirect_stdout(buf): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="A warm adult narrator") + self.assertTrue(client.design_mode) + self.assertEqual(client.instructions, "A warm adult narrator") + out = buf.getvalue() + self.assertIn("voice design", out) + self.assertIn("A warm adult narrator", out) + + def test_instructions_without_voice_on_generic_family_connects(self): + # Families without built-in speakers can get their voice from the + # instruction alone (e.g. OmniVoice voice design). + buf = io.StringIO() + with redirect_stdout(buf): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "omnivoice", + "task": "tts"}]}, + instructions="female, young adult, moderate pitch") + self.assertFalse(client.design_mode) + self.assertTrue(client.instruction_voice) + self.assertIn("instruction voice", buf.getvalue()) + + def test_instructions_with_builtin_speaker_family_stays_speaker_mode(self): + buf = io.StringIO() + with redirect_stdout(buf): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "tts"}]}, + instructions="Very happy.") + self.assertFalse(client.design_mode) + self.assertFalse(client.instruction_voice) + self.assertIn("speaker 'Vivian'", buf.getvalue()) + + def test_config_instructions_used_when_flag_omitted(self): + with patch.object(config, "AUDIOCPP_INSTRUCTIONS", + "A calm elderly storyteller"): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}) + self.assertEqual(client.instructions, "A calm elderly storyteller") + + def test_explicit_instructions_override_config_default(self): + with patch.object(config, "AUDIOCPP_INSTRUCTIONS", "from config"): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="from flag") + self.assertEqual(client.instructions, "from flag") + + class AudioCppFamilyDetectionTests(unittest.TestCase): """Family auto-detection and per-family adaptations.""" @@ -768,7 +906,8 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): @staticmethod def _make_client(preset_mode=False, voice="Vivian", language="English", seed=-1, - chunk_text=True, family="qwen3_tts"): + chunk_text=True, family="qwen3_tts", task="tts", + instructions=None, request_options=None): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.api_url = "http://127.0.0.1:8080" client.model_id = config.AUDIOCPP_MODEL_ID @@ -778,8 +917,18 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): client._seed = seed client.chunk_text = chunk_text client.family = family + client.task = task client.profile = tts.AUDIOCPP_FAMILY_PROFILES.get( family, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) + client.instructions = instructions or "" + client.request_options = dict(request_options or {}) + client.design_mode = task == tts.AUDIOCPP_TASK_VDES + # Mirrors the connect-time rule: an instruction-defined voice on a + # family without built-in speakers (design mode takes precedence). + client.instruction_voice = ( + not preset_mode and not client.design_mode + and not client.profile.builtin_speakers + and bool(client.instructions)) return client @staticmethod @@ -862,6 +1011,81 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertEqual(payload["instructions"], config.INSTRUCT) + def test_explicit_instructions_replace_config_instruct(self): + # --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", + 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")) + self.assertEqual(payload["instructions"], "Read whisper quiet.") + + def test_preset_mode_sends_instructions_alongside_voice(self): + # Clone + style control: both the server-side voice and the + # instruction reach the model. + client = self._make_client(preset_mode=True, voice="narrator", + instructions="Calm and steady.") + with patch("converter.tts.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")) + self.assertEqual(payload["voice"], "narrator") + self.assertEqual(payload["instructions"], "Calm and steady.") + + 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", + 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")) + self.assertNotIn("voice", payload) + self.assertEqual(payload["instructions"], "A warm adult narrator") + + def test_design_mode_language_follows_family_profile(self): + # The VoiceDesign package is family qwen3_tts, whose language field + # 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", + 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")) + self.assertEqual(payload["language"], "Japanese") + + def test_instruction_voice_payload_omits_voice(self): + # Instruction-defined voice on a family without built-in speakers: + # 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", + 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")) + self.assertNotIn("voice", payload) + self.assertNotIn("language", payload) # generic profile: omitted + self.assertEqual(payload["instructions"], "female, young adult") + + def test_request_options_forwarded_in_payload(self): + client = self._make_client(preset_mode=True, voice="narrator", + request_options={"emotion": "neutral", + "speed": "1.1"}) + with patch("converter.tts.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")) + self.assertEqual(payload["options"], {"emotion": "neutral", + "speed": "1.1"}) + + 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", + 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")) + self.assertNotIn("options", payload) + def test_generic_family_omits_language_and_instructions(self): # Clone-only families (higgs_audio_tts, voxcpm2, ...) detect the # language themselves and take no style instruction. @@ -1112,7 +1336,9 @@ class BackendWiringTests(unittest.TestCase): backend=tts.BACKEND_AUDIOCPP, voice="narrator", language="ja") mock_audiocpp.assert_called_once_with(voice="narrator", language="Japanese", - chunk_text=False, model_id=None) + chunk_text=False, model_id=None, + instructions=None, + request_options={}) mock_faster.assert_not_called() mock_qwen.assert_not_called() @@ -1121,7 +1347,9 @@ class BackendWiringTests(unittest.TestCase): AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, backend=tts.BACKEND_AUDIOCPP) mock_audiocpp.assert_called_once_with(voice=None, language=config.LANGUAGE, - chunk_text=False, model_id=None) + chunk_text=False, model_id=None, + instructions=None, + request_options={}) def test_audiocpp_backend_chunk_flag_forces_client_chunking(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: @@ -1130,7 +1358,9 @@ class BackendWiringTests(unittest.TestCase): voice="narrator", chunk=True) mock_audiocpp.assert_called_once_with(voice="narrator", language=config.LANGUAGE, - chunk_text=True, model_id=None) + chunk_text=True, model_id=None, + instructions=None, + request_options={}) self.assertTrue(converter.client_chunks) def test_audiocpp_backend_model_id_is_wired_through(self): @@ -1140,7 +1370,21 @@ class BackendWiringTests(unittest.TestCase): model_id="higgs") mock_audiocpp.assert_called_once_with( voice="narrator", language=config.LANGUAGE, - chunk_text=False, model_id="higgs") + chunk_text=False, model_id="higgs", instructions=None, + request_options={}) + + 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, + instructions="A warm adult narrator", + request_options={"emotion": "neutral", + "speed": "1.1"}) + mock_audiocpp.assert_called_once_with( + voice=None, language=config.LANGUAGE, + chunk_text=False, model_id=None, + instructions="A warm adult narrator", + request_options={"emotion": "neutral", "speed": "1.1"}) def test_qwen_backend_uses_qwen_client(self): with patch("converter.converter.FasterTTSClient") as mock_faster, \ diff --git a/tools/make_audiocpp_server_json.py b/tools/make_audiocpp_server_json.py index bda50e3..fb16a43 100755 --- a/tools/make_audiocpp_server_json.py +++ b/tools/make_audiocpp_server_json.py @@ -2,36 +2,61 @@ """Interactively generate a server.json for the audio.cpp audiocpp_server. Reads the model catalog (``model_specs/*.json``) from a local audio.cpp -checkout and offers every TTS model family audio.cpp supports as a -multi-select checklist, so one server.json can host several lazily-loaded -model entries at once. The converter itself is family-agnostic (it detects -the family of the selected entry from ``GET /v1/models`` at startup), so any -TTS family listed in the catalog works without further changes. - -Cloning reference .wav files (the required WAV_DIR argument) are transcribed -with a local Whisper backend (faster_whisper or whisper) and published as a -server-level ``voice_dir`` plus a ``prompt_text`` mapping file written into -WAV_DIR, so every hosted clone-capable family can use them with ``--voice``. - -Every value can also be supplied as a command-line flag; anything missing is -asked interactively with the default shown in brackets. Pressing Enter accepts -the default (the Qwen3-TTS built-in-speakers + voice-cloning flow). +checkout and offers every TTS model family audio.cpp supports, so one +server.json can host several lazily-loaded model entries at once. The +converter itself is family-agnostic (it detects the family of the selected +entry from ``GET /v1/models`` at startup), so any TTS family listed in the +catalog works without further changes. + +By default the tool runs as a minimal full-screen TUI (curses): a file +browser for the audio.cpp checkout and the .wav directory, an expandable +checkbox tree of model families and their installable packages, and a +series of single-question screens for the server settings. Pass ``--notui`` +to use the classic numbered line prompts instead (also selected +automatically when stdin/stdout is not a terminal, or when curses is +unavailable such as on Windows without ``windows-curses``). Every value +can also be supplied as a command-line flag, which skips the corresponding +screen or prompt. + +Each family is hosted through its recommended package by default; the TUI +tree always lists every installable package (distinct ``target_directory`` +values) as checkboxes, while ``--all-packages`` in prompt mode offers a +per-family package checklist (and pre-expands every family in the TUI). +Packages whose name marks them as voice-design models are asked whether to +host them with task "vdes" (describe the voice with ``--instructions``) or +plain "tts". + +Cloning reference .wav files (``--wavs DIR``) are transcribed with a local +Whisper backend (faster_whisper or whisper) and published as a server-level +``voice_dir`` plus a ``prompt_text`` mapping file written into the wav +directory, so every hosted clone-capable family can use them with +``--voice``. If ``prompt_text`` already exists, only voices that are missing +(or have an empty transcript) are re-transcribed, and you are asked first +when everything is already transcribed or when a mix of existing and new +voices is detected. Transcription runs in the plain console after the TUI +has gathered every setting. Usage: - python tools/make_audiocpp_server_json.py WAV_DIR [--output PATH] - [--audiocpp-dir PATH] [--families FAM1,FAM2] - [--models {both,custom,clone}] [--host HOST] [--port PORT] + python tools/make_audiocpp_server_json.py [--wavs WAV_DIR] + [--output PATH] [--audiocpp-dir PATH] [--families FAM1,FAM2] + [--all-packages] [--host HOST] [--port PORT] [--backend {cuda,vulkan,hip,cpu}] [--lazy-load] - [--whisper-model NAME] [--force] + [--whisper-model NAME] [--force] [--notui] -WAV_DIR is required: a directory of .wav reference files used as voice -cloning presets. It is checked up front and reported with its resolved -absolute path if it does not exist. +--wavs is the directory of .wav reference files used as voice cloning +presets; when omitted it is asked for. It is checked up front and reported +with its resolved absolute path if it does not exist. + +server.json is written into the audio.cpp checkout by default (next to +model_specs/). If that file already exists you are prompted [Y/n] before +overwriting; answering "n" writes server.json in the current working +directory instead. --audiocpp-dir defaults to a detected audio.cpp checkout (the AUDIOCPP_DIR environment variable, or an ``audio.cpp`` directory next to or above the current working directory); if none is found it is asked interactively. The -checkout must contain a ``model_specs/`` directory. +checkout must contain a ``model_specs/`` directory. A leading ``~`` in a +path argument or prompt answer is expanded. """ import argparse @@ -41,7 +66,7 @@ import re import sys import urllib.parse from pathlib import Path -from typing import Dict, List, Optional, Tuple +from typing import Callable, Dict, List, Optional, Set, Tuple # Allow running from any working directory. sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) @@ -51,15 +76,16 @@ from converter.tts import transcribe_reference_audio, whisper_backend_available DEFAULT_HOST = "127.0.0.1" FALLBACK_PORT = 8080 -DEFAULT_CUSTOM_VOICE_PATH = "models/Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF" -DEFAULT_BASE_PATH = "models/Qwen3-TTS-12Hz-1.7B-Base-GGUF" CONFIG_PATH = Path(__file__).resolve().parent.parent / "converter" / "config.py" -MODEL_SELECTIONS = ("both", "custom", "clone") BACKENDS = ("cuda", "vulkan", "hip", "cpu") -FAMILY_QWEN3_TTS = "qwen3_tts" PROMPT_TEXT_FILENAME = "prompt_text" +TASK_TTS = "tts" +TASK_VDES = "vdes" + +# Package names that mark a voice-design model (hosted with task "vdes"). +DESIGN_PACKAGE_RE = re.compile(r"voice[\s_\-]?design", re.IGNORECASE) # Families explicitly tested with this converter, in display order. These are # listed first in the checklist and marked "[tested]"; every other TTS family @@ -81,8 +107,46 @@ PREFERRED_IDS = { } -def resolve_wav_dir_arg(value: str) -> Path: - """Normalize a user-supplied wav directory argument. +class _TuiError(Exception): + """A fatal error raised from inside the TUI wizard. + + The message is reported to stderr after the terminal is restored; the + process exits with code 2 (matching a parser error). + """ + + +def _curses_importable() -> bool: + """Return True when the curses module can be imported.""" + try: + import curses # noqa: F401 + return True + except ImportError: + return False + + +def _load_tui(): + """Import the TUI widget module (tools/tui.py).""" + try: + from tools import tui + except ImportError: # executed directly from the tools/ directory + import tui + return tui + + +def _tui_enabled(args: argparse.Namespace) -> bool: + """Decide whether to run the TUI or fall back to line prompts.""" + if args.notui: + return False + if not _curses_importable(): + return False + try: + return sys.stdin.isatty() and sys.stdout.isatty() + except (AttributeError, ValueError): + return False + + +def normalize_dir_arg(value: str) -> Path: + """Normalize a user-supplied path argument. Strips surrounding quotes (a common copy-paste artifact), expands a leading ``~``, and resolves the result to an absolute path so relative @@ -94,6 +158,11 @@ def resolve_wav_dir_arg(value: str) -> Path: return Path(os.path.expanduser(cleaned)).resolve() +def resolve_wav_dir_arg(value: str) -> Path: + """Normalize a user-supplied wav directory argument.""" + return normalize_dir_arg(value) + + def find_wav_files(input_dir: Path) -> list: """Return the .wav files in INPUT_DIR, sorted alphabetically by name.""" return sorted( @@ -103,21 +172,6 @@ def find_wav_files(input_dir: Path) -> list: ) -def prompt_overwrite(output_path: Path) -> bool: - """Ask whether to overwrite an existing output file.""" - while True: - try: - answer = input(f"{output_path} already exists. Overwrite? (y/n): ").strip().lower() - except EOFError: - print("\n[WARNING] No interactive input available; keeping existing file") - return False - if answer in ("y", "yes"): - return True - if answer in ("n", "no"): - return False - print("Please answer 'y' or 'n'.") - - def ask(prompt: str, default: Optional[str] = None) -> Optional[str]: """Prompt for a free-text value with a default; EOF returns the default.""" suffix = f" [{default}]" if default is not None else "" @@ -180,14 +234,37 @@ def ask_menu(title: str, options: list, default_index: int = 1) -> str: print(f"Please enter a number between 1 and {len(options)}.") -def ask_models() -> str: - return ask_menu( - "Which Qwen3-TTS models should the server host?", - [ - ("Both (recommended) - built-in speakers + voice cloning", "both"), - ("CustomVoice only - built-in speakers", "custom"), - ("Base only - voice cloning (converting then requires --voice)", "clone"), - ]) +def ask_checklist(title: str, options: list, default: Set[str]) -> Set[str]: + """Show a numbered multi-select checklist and return the chosen values. + + Input is comma/space-separated numbers; Enter or EOF selects every option + in DEFAULT. At least one option is required. + """ + print(title) + for number, (label, _) in enumerate(options, 1): + print(f" {number}) {label}") + default_numbers = [str(number) for number, (_, value) in enumerate(options, 1) + if value in default] + suffix = f" [{', '.join(default_numbers)}]" + while True: + try: + answer = input(f"Choice{suffix}: ").strip() + except EOFError: + return set(default) + if not answer: + return set(default) + parts = [p for p in re.split(r"[,\s]+", answer) if p] + indices: List[int] = [] + valid = True + for part in parts: + if part.isdigit() and 1 <= int(part) <= len(options): + indices.append(int(part)) + else: + valid = False + break + if valid and indices: + return {options[index - 1][1] for index in indices} + print(f"Please enter comma-separated numbers between 1 and {len(options)}.") def ask_backend() -> str: @@ -201,25 +278,6 @@ def ask_backend() -> str: ]) -def ask_distinct_clone_id(primary_id: str) -> str: - """Prompt until a non-empty id different from PRIMARY_ID is entered.""" - prompt = (f"Enter a new id for the cloning (Base) model " - f"(must differ from '{primary_id}'): ") - while True: - try: - answer = input(prompt).strip() - except EOFError: - print() - sys.exit("[FATAL] No interactive input available to resolve the " - "duplicate model id; give the two models distinct " - "AUDIOCPP_MODEL_ID / AUDIOCPP_CLONE_MODEL_ID values in " - "converter/config.py first") - if answer and answer != primary_id: - return answer - print(f"[WARNING] The id must be unique; it cannot be empty or " - f"equal to '{primary_id}'.") - - def config_port() -> int: """Return the port of AUDIOCPP_API_URL in converter/config.py.""" try: @@ -311,7 +369,7 @@ def detect_audiocpp_dir() -> Optional[Path]: candidates: List[Path] = [] env_dir = os.environ.get("AUDIOCPP_DIR") if env_dir: - candidates.append(Path(env_dir)) + candidates.append(Path(os.path.expanduser(env_dir))) cwd = Path.cwd() candidates.append(cwd / "audio.cpp") candidates.append(cwd.parent / "audio.cpp") @@ -326,14 +384,12 @@ def detect_audiocpp_dir() -> Optional[Path]: return None -def _default_package(spec: dict) -> Optional[dict]: - """Pick the default installable package from a model spec. +def _default_package(packages: List[dict]) -> Optional[dict]: + """Pick the default package from a list of packages. Prefers the package flagged ``default: true``, then the first GGUF - package, then the first package overall. Returns None if the spec - declares no packages. + package, then the first package overall. Returns None for an empty list. """ - packages = spec.get("packages") or [] if not packages: return None for package in packages: @@ -349,10 +405,10 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: """Read model_specs/*.json and return the TTS-capable families. Each returned entry has: family, display_name, description, languages, - clone_capable, install_id (default package id), default_path - (``models/<target_directory>``), tested, and preferred_id. Tested - families come first (in TESTED_FAMILIES order), the rest follow - alphabetically by display name. + clone_capable, packages (the full list from the spec), install_id + (recommended package id), default_path (``models/<target_directory>``), + tested, and preferred_id. Tested families come first (in TESTED_FAMILIES + order), the rest follow alphabetically by display name. """ specs_dir = audiocpp_dir / "model_specs" if not specs_dir.is_dir(): @@ -369,7 +425,8 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: if "tts" not in tasks and spec.get("category") != "tts": continue family = spec.get("family") or spec_path.stem - package = _default_package(spec) + packages = spec.get("packages") or [] + package = _default_package(packages) if package is None: # No installable package: skip (cannot be hosted from a path). continue @@ -382,7 +439,9 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: "display_name": display_name, "description": description, "languages": languages, + "tasks": list(tasks), "clone_capable": "clone" in tasks, + "packages": packages, "install_id": package.get("id") or family, "default_path": f"models/{target_directory}", "tested": family in TESTED_FAMILIES, @@ -399,26 +458,118 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: return entries +def is_design_package(package: dict) -> bool: + """Return True when a package's name marks it a voice-design model. + + audio.cpp voice-design packages (whose id, display name, or target + directory mentions "voice design") are the only packages that must be + hosted with task "vdes"; their role is not in the schema, only in those + strings, so it is detected from them. + """ + text = " ".join(str(package.get(key, "")) + for key in ("id", "display_name", "target_directory")) + return bool(DESIGN_PACKAGE_RE.search(text)) + + +def package_dir_options(entry: dict) -> List[dict]: + """Return one option per distinct target_directory of a family's packages. + + Each option is a dict with: target_directory, install_id (the recommended + package id inside that directory), design (voice-design package flag), and + recommended (whether it holds the family's default package). Precisions + that share a directory (q8_0/bf16/...) collapse to a single option. + """ + packages = entry.get("packages") or [] + default_pkg = _default_package(packages) + default_dir = (default_pkg or {}).get("target_directory") or entry["family"] + by_dir: Dict[str, List[dict]] = {} + order: List[str] = [] + for package in packages: + directory = package.get("target_directory") or entry["family"] + if directory not in by_dir: + by_dir[directory] = [] + order.append(directory) + by_dir[directory].append(package) + options: List[dict] = [] + for directory in order: + package = _default_package(by_dir[directory]) + options.append({ + "target_directory": directory, + "install_id": (package or {}).get("id") or directory, + "design": is_design_package(package or {}), + "recommended": directory == default_dir, + }) + # Put the recommended package first for a friendlier checklist. + options.sort(key=lambda opt: not opt["recommended"]) + return options + + +def ask_package_dirs(entry: dict) -> List[dict]: + """Choose which of a family's packages to host (multi-select checklist). + + Enter selects the recommended package only, matching the default flow. + """ + options = package_dir_options(entry) + if len(options) <= 1: + return options + default = {opt["target_directory"] for opt in options if opt["recommended"]} + labels = [] + for opt in options: + marker = " [recommended]" if opt["recommended"] else "" + labels.append((f"{opt['install_id']} -> {opt['target_directory']}{marker}", + opt["target_directory"])) + chosen = ask_checklist( + f"Which {entry['display_name']} packages should the server host?", + labels, default=default) + return [opt for opt in options if opt["target_directory"] in chosen] + + +def ask_package_task(install_id: str) -> str: + """Ask how to host a voice-design package: vdes or tts.""" + return ask_menu( + f"How should the '{install_id}' package be hosted?", + [ + ("design (vdes) - describe the voice with --instructions", + TASK_VDES), + ("tts - normal synthesis", TASK_TTS), + ], + default_index=1) + + def ask_families(catalog: List[dict]) -> List[str]: - """Show a numbered checklist and return the chosen family keys. + """Show a numbered table and return the chosen family keys. Input is comma/space-separated numbers; Enter alone selects the first - entry (the default Qwen3-TTS flow). At least one family is required. + entry. At least one family is required. """ + rows: List[Tuple[str, str]] = [] + for entry in catalog: + capabilities = ["tts"] + if "clone" in entry["tasks"]: + capabilities.append("cloning") + if "design" in entry["tasks"]: + capabilities.append("design") + name = entry["display_name"] + if name != entry["family"]: + name = f"{name} ({entry['family']})" + rows.append((name, ", ".join(capabilities))) + number_width = len(str(len(rows))) + name_width = max([len("Model family")] + [len(name) for name, _ in rows]) + tasks_width = max([len("Tasks")] + [len(tasks) for _, tasks in rows]) + header = (f"{'#'.ljust(number_width)} | " + f"{'Model family'.ljust(name_width)} | " + f"{'Tasks'.ljust(tasks_width)}") + divider = (f"{'-' * number_width}-+-" + f"{'-' * name_width}-+-" + f"{'-' * tasks_width}") print("Select TTS model families to host (comma-separated numbers,") - print("or press Enter for the default Qwen3-TTS flow):") - for number, entry in enumerate(catalog, 1): - marker = " [tested with this converter]" if entry["tested"] else "" - langs = entry["languages"] - lang_text = ", ".join(langs[:6]) + ("..." if len(langs) > 6 else "") - if entry["family"] == FAMILY_QWEN3_TTS: - caps = "built-in speakers + voice cloning" - elif entry["clone_capable"]: - caps = "voice cloning" - else: - caps = "TTS (no cloning)" - detail = f"({lang_text}; {caps})" if lang_text else f"({caps})" - print(f" {number}) {entry['display_name']}{marker} {detail}") + print("or press Enter for the first family):") + print(header) + print(divider) + for number, (name, tasks) in enumerate(rows, 1): + print(f"{str(number).ljust(number_width)} | " + f"{name.ljust(name_width)} | " + f"{tasks.ljust(tasks_width)}") while True: try: answer = input("Choice [1]: ").strip() @@ -447,13 +598,19 @@ def ask_families(catalog: List[dict]) -> List[str]: print(f"Please enter comma-separated numbers between 1 and {len(catalog)}.") -def build_model_entry(family: str, model_id: str, model_path: str) -> dict: - """Assemble one server.json model entry.""" +def build_model_entry(family: str, model_id: str, model_path: str, + task: str = TASK_TTS) -> dict: + """Assemble one server.json model entry. + + ``task`` defaults to "tts"; voice design packages are hosted with + "vdes" so the server runs its design session for speech requests + (audiobook.py then requires --instructions with that entry). + """ return { "id": model_id, "family": family, "path": model_path, - "task": "tts", + "task": task, "mode": "offline", } @@ -495,6 +652,27 @@ def transcribe_wav_dir(wav_files: list, whisper_model: str) -> Dict[str, str]: return transcripts +def read_prompt_text(prompt_path: Path) -> Dict[str, str]: + """Parse a prompt_text file into a stem -> transcript mapping. + + Lines are ``<name>|<transcript>``; blank lines are skipped and a line + without a ``|`` separator is treated as a name with an empty transcript. + Returns an empty mapping when the file does not exist. + """ + if not prompt_path.exists(): + return {} + mapping: Dict[str, str] = {} + for line in prompt_path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + if "|" in line: + name, _, text = line.partition("|") + else: + name, text = line, "" + mapping[name.strip()] = text + return mapping + + def write_prompt_text(wav_dir: Path, transcripts: Dict[str, str]) -> Path: """Write the voice_dir prompt_text mapping into WAV_DIR. @@ -527,6 +705,20 @@ def print_empty_transcript_warning(transcripts: Dict[str, str]) -> None: print(bar) +def _apply_port_sync(port: int, accepted: bool) -> None: + """Write the port into converter/config.py, or report when declined.""" + if accepted: + if update_config_api_url_port(port): + print(f"[OK] Updated AUDIOCPP_API_URL in {CONFIG_PATH}") + else: + print(f"[WARNING] Could not update {CONFIG_PATH}; edit " + "AUDIOCPP_API_URL by hand so audiobook.py uses the " + "new port") + else: + print("[WARNING] Left AUDIOCPP_API_URL unchanged; audiobook.py " + f"will still use port {config_port()}") + + def _ask_host_port_backend_lazy(args: argparse.Namespace, default_lazy: bool ) -> Tuple[str, int, str, bool]: @@ -536,62 +728,110 @@ def _ask_host_port_backend_lazy(args: argparse.Namespace, if port != config_port(): if ask_bool(f"Update AUDIOCPP_API_URL in converter/config.py to port " f"{port} so audiobook.py talks to this server", True): - if update_config_api_url_port(port): - print(f"[OK] Updated AUDIOCPP_API_URL in {CONFIG_PATH}") - else: - print(f"[WARNING] Could not update {CONFIG_PATH}; edit " - "AUDIOCPP_API_URL by hand so audiobook.py uses the " - "new port") + _apply_port_sync(port, True) else: - print("[WARNING] Left AUDIOCPP_API_URL unchanged; audiobook.py " - f"will still use port {config_port()}") + _apply_port_sync(port, False) backend = args.backend if args.backend else ask_backend() lazy_load = args.lazy_load or ask_bool( "Load models lazily (on first use instead of at startup)", default_lazy) return host, port, backend, lazy_load -def _collect_transcripts(args: argparse.Namespace, - include_clone: bool) -> Dict[str, str]: +def _decide_transcription(wav_files: list, existing: Dict[str, str], + prompt_exists: bool, force: bool, + confirm: Callable[[str, bool], bool]) -> dict: + """Decide which voices to transcribe; CONFIRM asks the plan questions. + + Returns a plan dict: {"mode": "all"|"missing"|"keep", "missing": [...]}. + """ + mode = "all" + missing: List[Path] = [] + if prompt_exists and not force: + missing = [wav for wav in wav_files + if not existing.get(wav.stem, "").strip()] + if not missing: + if confirm("All voices already transcribed in prompt_text. " + "Re-transcribe anyway?", False): + mode = "all" + else: + mode = "keep" + elif confirm("Existing transcription and new .wavs detected, " + "only transcribe new voices?", True): + mode = "missing" + else: + mode = "all" + return {"mode": mode, "missing": missing} + + +def _transcribe(args: argparse.Namespace, include_clone: bool, + plan: Optional[dict] = None + ) -> Tuple[Dict[str, str], bool]: """Transcribe the wav directory into a stem -> transcript mapping. - Returns the mapping (empty when no wavs were found or cloning is not - used by any selected family). Runs only when a cloning voice library is - needed; a run without any clone-capable family ignores the wav directory - entirely. + Returns the mapping and a flag indicating whether it should be written to + prompt_text (False when an existing, complete prompt_text is kept as-is). + When PLAN is given (pre-collected by the TUI) no further questions are + asked; otherwise the plan is decided with the line prompts. """ if not include_clone: print(f"[WARNING] Ignoring {args.input_dir}: no clone-capable family " "selected, so voice presets are not used") - return {} + return {}, False wav_files = find_wav_files(args.input_dir) if not wav_files: print(f"[WARNING] No .wav files found in {args.input_dir}; writing the " "config without a voice_dir") - return {} + return {}, False + + prompt_path = args.input_dir / PROMPT_TEXT_FILENAME + existing = read_prompt_text(prompt_path) if ( + prompt_path.exists() and not args.force) else {} + + if plan is None: + plan = _decide_transcription( + wav_files, existing, prompt_path.exists(), args.force, + lambda question, default: ask_bool(question, default)) + + if plan["mode"] == "keep": + print(f"[INFO] Kept existing {prompt_path}; all voices were " + "already transcribed, nothing new to transcribe") + return existing, False + if whisper_backend_available() is None: print("[WARNING] Neither faster_whisper nor whisper was found, so " "reference .wav files cannot be transcribed automatically and " "every transcript will be empty.") - print(' Did you remember to "conda activate qwen3-tts"? ' - "Transcripts must be added by hand (see the warning at the end).") - return transcribe_wav_dir(wav_files, args.whisper_model) + print(" Install whisper (or faster_whisper) in your " + "audiobook environment to transcribe automatically; otherwise " + "transcripts must be added by hand (see the warning at the end).") + + if plan["mode"] == "missing": + new_transcripts = transcribe_wav_dir(plan["missing"], args.whisper_model) + transcripts = dict(existing) + transcripts.update(new_transcripts) + else: + transcripts = transcribe_wav_dir(wav_files, args.whisper_model) + return transcripts, True -def _offer_config_model_id_sync(model_id: str) -> None: - """Offer to point converter/config.py at a single non-Qwen model entry. +def _offer_config_model_id_sync(model_id: str, + accepted: Optional[bool] = None) -> None: + """Offer to point converter/config.py at a single hosted model entry. The converter requests the model id configured in AUDIOCPP_MODEL_ID, and single-model servers use the same id for the clone entry, so both - ids are rewritten together. + ids are rewritten together. When ACCEPTED is None the user is asked + (line prompt); otherwise the given decision is applied. """ if config.AUDIOCPP_MODEL_ID == model_id \ and config.AUDIOCPP_CLONE_MODEL_ID == model_id: return - if ask_bool("Update AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID in " - f"converter/config.py to '{model_id}' so audiobook.py uses " - "this model", True): + if accepted is None: + accepted = ask_bool("Update AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID " + f"in converter/config.py to '{model_id}' so " + "audiobook.py uses this model", True) + if accepted: if update_config_model_ids(model_id, model_id): print(f"[OK] Updated the model ids in {CONFIG_PATH}") else: @@ -611,30 +851,398 @@ def _print_multi_model_model_id_note(entry_ids: List[str]) -> None: f"{', '.join(entry_ids)}") +def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]], + catalog_by_family: Dict[str, dict], + task_picker: Callable[[str], str], + id_picker: Callable[[str, str, str], str] + ) -> Tuple[List[dict], List[str], List[Tuple[str, str]], + List[str], bool]: + """Build server.json model entries from the selected families/packages. + + TASK_PICKER is called for each design package to choose vdes/tts; + ID_PICKER resolves a duplicate server entry id. Returns (model_entries, + entry_ids, install_guidance, design_entry_ids, include_clone). + """ + model_entries: List[dict] = [] + entry_ids: List[str] = [] + install_guidance: List[Tuple[str, str]] = [] + design_entry_ids: List[str] = [] + include_clone = False + for family in family_keys: + entry = catalog_by_family[family] + include_clone = include_clone or entry["clone_capable"] + for opt in chosen[family]: + task = task_picker(opt["install_id"]) if opt["design"] else TASK_TTS + base_id = (f"{entry['preferred_id']}-design" + if task == TASK_VDES else entry["preferred_id"]) + model_id = base_id + if model_id in entry_ids: + model_id = id_picker(entry["display_name"], opt["install_id"], + f"{base_id}-2") + entry_ids.append(model_id) + model_entries.append(build_model_entry( + family, model_id, f"models/{opt['target_directory']}", + task=task)) + install_guidance.append((entry["display_name"], opt["install_id"])) + if task == TASK_VDES: + design_entry_ids.append(model_id) + return (model_entries, entry_ids, install_guidance, + design_entry_ids, include_clone) + + +def _write_and_advise(wav_dir: Optional[Path], output_path: Path, + model_entries: List[dict], entry_ids: List[str], + install_guidance: List[Tuple[str, str]], + design_entry_ids: List[str], family_keys: List[str], + catalog_by_family: Dict[str, dict], host: str, port: int, + backend: str, lazy_load: bool, + transcripts: Dict[str, str], write_prompt: bool) -> None: + """Console phase shared by both UI modes: write files and print guidance.""" + voice_dir: Optional[str] = None + if transcripts: + if write_prompt: + prompt_path = wav_dir / PROMPT_TEXT_FILENAME + write_prompt_text(wav_dir, transcripts) + print(f"[OK] Wrote {prompt_path}") + voice_dir = str(wav_dir.resolve()) + + server_config = build_server_config( + host=host, port=port, backend=backend, lazy_load=lazy_load, + model_entries=model_entries, voice_dir=voice_dir) + + print("\nGenerated server.json:") + print(json.dumps(server_config, indent=2, ensure_ascii=False)) + + with output_path.open("w", encoding="utf-8") as handle: + json.dump(server_config, handle, indent=2, ensure_ascii=False) + handle.write("\n") + + print(f"\n[OK] Wrote {output_path} with {len(model_entries)} model " + f"entry/entries" + (f" and voice_dir '{voice_dir}'" if voice_dir else "")) + for display_name, install_id in install_guidance: + print(f"[INFO] Install {display_name} from the audio.cpp checkout: " + f"python3 tools/model_manager_v2.py install {install_id}") + if len(model_entries) > 1: + print("[INFO] Models load lazily and stay in memory until the server " + "exits; restart the server (or POST /v1/tasks/unload_models) " + "before switching to a large model to free VRAM.") + for family in family_keys: + if catalog_by_family[family]["clone_capable"]: + print(f"[INFO] {catalog_by_family[family]['display_name']} supports " + "voice cloning: run audiobook.py with --backend audiocpp " + "--voice <preset name>") + for design_id in design_entry_ids: + print(f"[INFO] Voice design entry '{design_id}' hosted with task " + "'vdes': convert with python audiobook.py --backend audiocpp " + f"--model {design_id} " + '--instructions "A warm adult female narrator"') + + +def _build_tree_families(catalog: List[dict]) -> List[dict]: + """Shape the catalog into the checkbox_tree widget's family list.""" + families: List[dict] = [] + for entry in catalog: + capabilities = ["tts"] + if "clone" in entry["tasks"]: + capabilities.append("cloning") + if "design" in entry["tasks"]: + capabilities.append("design") + name = entry["display_name"] + if name != entry["family"]: + name = f"{name} ({entry['family']})" + if entry["tested"]: + name = f"{name} [tested]" + options = [] + for opt in package_dir_options(entry): + label = opt["install_id"] + if opt["design"]: + label = f"{label} (voice design)" + options.append({ + "key": opt["target_directory"], + "label": label, + "recommended": opt["recommended"], + }) + families.append({ + "label": name, + "detail": ", ".join(capabilities), + "options": options, + }) + return families + + +def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser + ) -> Optional[dict]: + """Run every TUI screen; return the collected settings, or None to abort.""" + tui = _load_tui() + + # 1. audio.cpp checkout (flag, detected, or browsed). + audiocpp_dir = args.audiocpp_dir + if audiocpp_dir is None: + audiocpp_dir = detect_audiocpp_dir() + if audiocpp_dir is None: + audiocpp_dir = tui.browse_directory( + stdscr, "Locate your audio.cpp checkout", + validate=lambda p: None if (p / "model_specs").is_dir() + else "No model_specs/ directory here", + start=Path.cwd()) + audiocpp_dir = Path(audiocpp_dir).resolve() + if not audiocpp_dir.is_dir(): + raise _TuiError(f"audio.cpp checkout not found: {audiocpp_dir}") + try: + catalog = load_model_catalog(audiocpp_dir) + except NotADirectoryError as exc: + raise _TuiError(str(exc)) + if not catalog: + raise _TuiError(f"No TTS model families found in " + f"{audiocpp_dir}/model_specs; check the checkout is " + "up to date") + catalog_by_family = {entry["family"]: entry for entry in catalog} + + # 2. Output path + overwrite confirmation. + output_path = args.output if args.output is not None \ + else audiocpp_dir / "server.json" + if not args.force and output_path.exists() \ + and not tui.confirm(stdscr, + f"{output_path} already exists. Overwrite?", + default=True): + if args.output is None: + output_path = Path.cwd() / "server.json" + if output_path.exists() and not tui.confirm( + stdscr, f"{output_path} already exists. Overwrite?", + default=True): + return None + else: + return None + + # 3. Families and packages (flag or tree). + chosen: Dict[str, List[dict]] = {} + if args.families is not None: + requested = [f.strip() for f in args.families.split(",") if f.strip()] + unknown = [f for f in requested if f not in catalog_by_family] + if unknown: + raise _TuiError( + f"Unknown family in --families: {', '.join(unknown)}. " + f"Available: {', '.join(catalog_by_family)}") + family_keys: List[str] = [] + for family in requested: + if family not in family_keys: + family_keys.append(family) + chosen[family] = [opt for opt in package_dir_options( + catalog_by_family[family]) if opt["recommended"]] + else: + tree_families = _build_tree_families(catalog) + picked = tui.checkbox_tree( + stdscr, "Select TTS model families to host", + tree_families, expand_all=args.all_packages) + family_keys = [] + for family_index, option_key in picked: + family = catalog[family_index]["family"] + if family not in chosen: + chosen[family] = [] + family_keys.append(family) + chosen[family].append(option_key) + for family in list(chosen): + keyed = {opt["target_directory"]: opt + for opt in package_dir_options(catalog_by_family[family])} + chosen[family] = [keyed[key] for key in chosen[family]] + + # 4. Design task menus and duplicate-id renames. + def task_picker(install_id: str) -> str: + return tui.menu( + stdscr, f"How should the '{install_id}' package be hosted?", + [ + ("design (vdes) - describe the voice with --instructions", + TASK_VDES), + ("tts - normal synthesis", TASK_TTS), + ], default_index=0) + + def id_picker(display_name: str, install_id: str, default: str) -> str: + return tui.line_edit( + stdscr, + f"Server model id for {display_name} package '{install_id}'", + default) + + model_entries, entry_ids, install_guidance, design_entry_ids, include_clone = \ + _build_entries(family_keys, chosen, catalog_by_family, + task_picker, id_picker) + + # 5. Server settings. + host = args.host if args.host else tui.line_edit(stdscr, "Bind host", + DEFAULT_HOST) + if args.port is not None: + port = args.port + else: + port_text = tui.line_edit( + stdscr, "Port", str(config_port()), + validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535) + else "Enter a port number between 1 and 65535") + port = int(port_text) + sync_port: Optional[bool] = None + if port != config_port(): + sync_port = tui.confirm( + stdscr, f"Update AUDIOCPP_API_URL in converter/config.py to port " + f"{port} so audiobook.py talks to this server", default=True) + backend = args.backend if args.backend else tui.menu( + stdscr, "Which inference backend was audiocpp_server built for?", + [ + ("cuda - NVIDIA GPUs (fastest)", "cuda"), + ("vulkan - cross-vendor GPU", "vulkan"), + ("hip - AMD GPUs", "hip"), + ("cpu - no GPU required", "cpu"), + ], default_index=0) + default_lazy = len(model_entries) > 1 + lazy_load = args.lazy_load or tui.confirm( + stdscr, "Load models lazily (on first use instead of at startup)", + default=default_lazy) + + # 6. Wav directory (flag, browsed when cloning, else skipped). + if args.input_dir is not None: + wav_dir = args.input_dir + elif include_clone: + wav_dir = tui.browse_directory( + stdscr, "Directory with .wav voice cloning files", + start=Path.cwd()) + else: + wav_dir = None + + # 7. Transcription plan (questions only; transcription runs after). + plan: Optional[dict] = None + if include_clone and wav_dir is not None: + wav_files = find_wav_files(wav_dir) + if wav_files: + prompt_path = wav_dir / PROMPT_TEXT_FILENAME + existing = read_prompt_text(prompt_path) if ( + prompt_path.exists() and not args.force) else {} + plan = _decide_transcription( + wav_files, existing, prompt_path.exists(), args.force, + lambda question, default: tui.confirm(stdscr, question, default)) + + # 8. Single-model id sync decision. + sync_model_ids: Optional[bool] = None + if len(entry_ids) == 1 and not ( + config.AUDIOCPP_MODEL_ID == entry_ids[0] + and config.AUDIOCPP_CLONE_MODEL_ID == entry_ids[0]): + sync_model_ids = tui.confirm( + stdscr, "Update AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID in " + f"converter/config.py to '{entry_ids[0]}' so audiobook.py uses " + "this model", default=True) + + # 9. Summary and final confirmation. + summary_lines = [ + f"Output: {output_path}", + f"Server: {host}:{port} ({backend}, lazy_load={'on' if lazy_load else 'off'})", + f"Models: {', '.join(entry_ids)}", + ] + if wav_dir is not None: + summary_lines.append(f"Voices: {wav_dir}") + if not tui.confirm(stdscr, "Generate server.json?", default=True, + body=summary_lines): + return None + + return { + "audiocpp_dir": audiocpp_dir, + "catalog": catalog, + "catalog_by_family": catalog_by_family, + "output_path": output_path, + "family_keys": family_keys, + "chosen": chosen, + "model_entries": model_entries, + "entry_ids": entry_ids, + "install_guidance": install_guidance, + "design_entry_ids": design_entry_ids, + "include_clone": include_clone, + "host": host, + "port": port, + "backend": backend, + "lazy_load": lazy_load, + "sync_port": sync_port, + "sync_model_ids": sync_model_ids, + "wav_dir": wav_dir, + "plan": plan, + } + + +def _run_tui(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: + """Run the TUI wizard, then the shared console phase.""" + import curses + tui = _load_tui() + try: + settings = curses.wrapper(_wizard, args, parser) + except _TuiError as exc: + print(f"[ERROR] {exc}", file=sys.stderr) + return 2 + except tui.WizardCancelled: + print("\n[INFO] Cancelled; nothing was written") + return 1 + if settings is None: + print("[INFO] Aborted; existing server.json kept") + return 1 + + # Port sync (applied now that the terminal is back). + if settings["sync_port"] is True: + _apply_port_sync(settings["port"], True) + elif settings["sync_port"] is False: + _apply_port_sync(settings["port"], False) + + # Transcription (console; the questions were already answered in the TUI). + args.input_dir = settings["wav_dir"] + if settings["include_clone"]: + transcripts, write_prompt = _transcribe(args, True, plan=settings["plan"]) + elif args.input_dir is not None: + print(f"[WARNING] Ignoring {args.input_dir}: no clone-capable family " + "selected, so voice presets are not used") + transcripts, write_prompt = {}, False + else: + transcripts, write_prompt = {}, False + + _write_and_advise( + settings["wav_dir"], settings["output_path"], settings["model_entries"], + settings["entry_ids"], settings["install_guidance"], + settings["design_entry_ids"], settings["family_keys"], + settings["catalog_by_family"], settings["host"], settings["port"], + settings["backend"], settings["lazy_load"], transcripts, write_prompt) + + if len(settings["entry_ids"]) == 1: + _offer_config_model_id_sync(settings["entry_ids"][0], + settings["sync_model_ids"]) + elif len(settings["entry_ids"]) > 1: + _print_multi_model_model_id_note(settings["entry_ids"]) + print_empty_transcript_warning(transcripts) + return 0 + + def main() -> int: parser = argparse.ArgumentParser( description="Generate a server.json for the audio.cpp audiocpp_server " "hosting one or more TTS model families used by this converter.") - parser.add_argument("input_dir", type=resolve_wav_dir_arg, metavar="WAV_DIR", + parser.add_argument("--wavs", type=resolve_wav_dir_arg, default=None, + dest="input_dir", metavar="WAV_DIR", help="Directory with .wav reference files to publish as " - "a server-level voice_dir cloning library (required)") - parser.add_argument("--output", type=Path, default=Path("server.json"), + "a server-level voice_dir cloning library (asked " + "for when omitted)") + parser.add_argument("--output", type=Path, default=None, help="Output path for server.json (default: " - "server.json in the current directory)") - parser.add_argument("--audiocpp-dir", type=Path, default=None, + "server.json inside the audio.cpp checkout; if it " + "already exists you are asked [Y/n] to overwrite, " + "and answering 'n' writes server.json in the " + "current directory instead)") + parser.add_argument("--audiocpp-dir", type=normalize_dir_arg, default=None, help="Path to a local audio.cpp checkout containing a " "model_specs/ directory (default: detected from " "AUDIOCPP_DIR or an audio.cpp directory next to/above " "the current working directory; prompted otherwise)") parser.add_argument("--families", type=str, default=None, help="Comma-separated model families to host, as named " - "in the audio.cpp catalog (e.g. " - "qwen3_tts,higgs_audio_tts). Skips the family checklist") - parser.add_argument("--models", choices=MODEL_SELECTIONS, default=None, - help="Which Qwen3-TTS models to host: both (default), " - "custom (CustomVoice speakers only), or clone " - "(Base voice cloning only). Only valid when the " - "qwen3_tts family is selected") + "in the audio.cpp catalog (e.g. " + "qwen3_tts,higgs_audio_tts). Skips the family " + "checklist") + parser.add_argument("--all-packages", action="store_true", + help="Instead of hosting each family's recommended " + "package, offer a checklist of every installable " + "package (distinct target_directory) so several " + "packages of one family can be hosted at once. " + "In the TUI this pre-expands every family in the " + "tree (which always lists all packages)") parser.add_argument("--host", type=str, default=None, help="Bind host for the server (default: 127.0.0.1)") parser.add_argument("--port", type=int, default=None, @@ -652,14 +1260,38 @@ def main() -> int: parser.add_argument("--force", action="store_true", help="Overwrite the output file (and prompt_text) " "without prompting") + parser.add_argument("--notui", action="store_true", + help="Use the classic line prompts instead of the " + "full-screen TUI (automatic when curses is " + "unavailable or stdin/stdout is not a terminal)") args = parser.parse_args() + if args.input_dir is not None and not args.input_dir.is_dir(): + parser.error( + f"WAV directory not found: {args.input_dir}\n" + f" (resolved from the current working directory: " + f"{Path.cwd()})\n" + " --wavs must be a directory containing the .wav " + "reference files to use as voice cloning presets") + + if _tui_enabled(args): + return _run_tui(args, parser) + + # ---- Line-prompt flow (original behaviour). --------------------------- + + # Resolve the wav directory (flag, else prompt). + if args.input_dir is None: + answer = ask("Directory with .wav reference files", "") + args.input_dir = resolve_wav_dir_arg(answer) if answer else None + if args.input_dir is None: + parser.error("--wavs is required: a directory containing the .wav " + "reference files to use as voice cloning presets") if not args.input_dir.is_dir(): parser.error( f"WAV directory not found: {args.input_dir}\n" f" (resolved from the current working directory: " f"{Path.cwd()})\n" - " WAV_DIR must be a directory containing the .wav " + " --wavs must be a directory containing the .wav " "reference files to use as voice cloning presets") # Resolve the audio.cpp checkout and load its model catalog. @@ -667,7 +1299,10 @@ def main() -> int: if audiocpp_dir is None: audiocpp_dir = detect_audiocpp_dir() if audiocpp_dir is None: - audiocpp_dir = Path(ask("Path to your audio.cpp checkout", "") or "") + print("[INFO] Could not find an audio.cpp checkout next to or above " + "the current directory.") + answer = ask("Path to your audio.cpp checkout", "") + audiocpp_dir = normalize_dir_arg(answer) if answer else None if not audiocpp_dir: parser.error( "An audio.cpp checkout is required to read the model catalog. " @@ -686,10 +1321,23 @@ def main() -> int: f"No TTS model families found in {audiocpp_dir}/model_specs; " "check the checkout is up to date") - if args.output.exists() and not args.force \ - and not prompt_overwrite(args.output): - print("[INFO] Aborted; existing server.json kept") - return 1 + # Resolve the server.json output path. It defaults to the audio.cpp + # checkout; an existing file is overwritten only with confirmation, and a + # declined overwrite of the default location falls back to the current + # working directory. + output_path = args.output if args.output is not None \ + else audiocpp_dir / "server.json" + if not args.force and output_path.exists() \ + and not ask_bool(f"{output_path} already exists. Overwrite?", True): + if args.output is None: + output_path = Path.cwd() / "server.json" + if output_path.exists() and not ask_bool( + f"{output_path} already exists. Overwrite?", True): + print("[INFO] Aborted; existing server.json kept") + return 1 + else: + print("[INFO] Aborted; existing server.json kept") + return 1 # Select families. if args.families is not None: @@ -708,151 +1356,38 @@ def main() -> int: family_keys = ask_families(catalog) catalog_by_family = {entry["family"]: entry for entry in catalog} - is_qwen = FAMILY_QWEN3_TTS in family_keys - if not is_qwen and args.models is not None: - parser.error("--models only applies to the qwen3_tts family") - if is_qwen and args.models is not None and len(family_keys) > 1 \ - and args.models != "both": - parser.error( - "--models custom/clone selects Qwen3-TTS sub-entries and is only " - "valid when qwen3_tts is the sole selected family") - - print("[INFO] Model ids from converter/config.py:") - print(f" built-in speakers (CustomVoice): '{config.AUDIOCPP_MODEL_ID}'") - print(f" voice cloning (Base): '{config.AUDIOCPP_CLONE_MODEL_ID}'") - model_entries: List[dict] = [] - entry_ids: List[str] = [] - non_qwen_single_id: Optional[str] = None - - if is_qwen: - selection = args.models if args.models is not None else ask_models() - # When qwen3_tts is selected with other families, keep both entries so - # speaker mode and cloning are both available; custom/clone sub-choice - # is only honored when qwen3_tts is the sole family. - if len(family_keys) > 1 and args.models is None: - selection = "both" - include_custom = selection in ("both", "custom") - include_clone = selection in ("both", "clone") - - custom_voice_id = config.AUDIOCPP_MODEL_ID - clone_model_id = config.AUDIOCPP_CLONE_MODEL_ID - if include_custom and include_clone and custom_voice_id == clone_model_id: - print(f"[WARNING] AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID are " - f"both '{custom_voice_id}' in converter/config.py, but server " - "model ids must be unique.") - clone_model_id = ask_distinct_clone_id(custom_voice_id) - - custom_voice_path = base_path = None - if include_custom: - custom_voice_path = ask("Path to the Qwen3-TTS CustomVoice GGUF package", - DEFAULT_CUSTOM_VOICE_PATH) - model_entries.append(build_model_entry( - FAMILY_QWEN3_TTS, custom_voice_id, custom_voice_path)) - entry_ids.append(custom_voice_id) - if include_clone: - base_path = ask("Path to the Qwen3-TTS Base GGUF package", - DEFAULT_BASE_PATH) - model_entries.append(build_model_entry( - FAMILY_QWEN3_TTS, clone_model_id, base_path)) - entry_ids.append(clone_model_id) - qwen_include_clone = include_clone - else: - qwen_include_clone = False - - # Non-Qwen families: one entry each. + chosen: Dict[str, List[dict]] = {} for family in family_keys: - if family == FAMILY_QWEN3_TTS: - continue entry = catalog_by_family[family] - model_id = entry["preferred_id"] - # Ensure uniqueness against already-chosen ids. - if model_id in entry_ids: - model_id = ask(f"Server model id for {entry['display_name']}", - f"{model_id}-2") - model_path = entry["default_path"] - # For a single non-Qwen family, ask the path (matching the old flow); - # for several, use the catalog default to keep the prompt count sane. - if len(family_keys) == 1: - model_path = ask(f"Path to the {entry['display_name']} package", - model_path) - model_entries.append(build_model_entry(family, model_id, model_path)) - entry_ids.append(model_id) - if len(family_keys) == 1: - non_qwen_single_id = model_id - - # Whether any selected family can clone (drives voice_dir / wav transcription). - include_clone = qwen_include_clone or any( - catalog_by_family[f]["clone_capable"] - for f in family_keys if f != FAMILY_QWEN3_TTS) - - # Default to lazy loading only when hosting more than one family: a - # single-family server (including the Qwen3-TTS CustomVoice+Base pair) - # loads at startup as before, while a multi-family server avoids loading - # every model until it is actually used. - default_lazy = len(family_keys) > 1 - host, port, backend, lazy_load = _ask_host_port_backend_lazy(args, default_lazy) - - transcripts = _collect_transcripts(args, include_clone) - - voice_dir: Optional[str] = None - if transcripts: - prompt_path = args.input_dir / PROMPT_TEXT_FILENAME - if prompt_path.exists() and not args.force: - if not ask_bool(f"Overwrite existing {prompt_path}", True): - print(f"[INFO] Kept existing {prompt_path}; new transcripts " - "were not written") - else: - write_prompt_text(args.input_dir, transcripts) - print(f"[OK] Wrote {prompt_path}") + if args.all_packages: + chosen[family] = ask_package_dirs(entry) else: - write_prompt_text(args.input_dir, transcripts) - print(f"[OK] Wrote {prompt_path}") - voice_dir = str(args.input_dir.resolve()) - - server_config = build_server_config( - host=host, port=port, backend=backend, lazy_load=lazy_load, - model_entries=model_entries, voice_dir=voice_dir) + chosen[family] = [opt for opt in package_dir_options(entry) + if opt["recommended"]] + + model_entries, entry_ids, install_guidance, design_entry_ids, include_clone = \ + _build_entries(family_keys, chosen, catalog_by_family, + task_picker=lambda install_id: ask_package_task(install_id), + id_picker=lambda display_name, install_id, base_id: ask( + f"Server model id for {display_name} package " + f"'{install_id}'", f"{base_id}-2")) + + # Default to lazy loading when hosting more than one model entry: a + # single-entry server loads at startup, while a multi-entry server avoids + # loading every model until it is actually used. + default_lazy = len(model_entries) > 1 + host, port, backend, lazy_load = _ask_host_port_backend_lazy(args, default_lazy) - print("\nGenerated server.json:") - print(json.dumps(server_config, indent=2, ensure_ascii=False)) - if not ask_bool(f"\nWrite this to {args.output}", True): - print("[INFO] Aborted; nothing written") - return 1 + transcripts, write_prompt = _transcribe(args, include_clone) - with args.output.open("w", encoding="utf-8") as handle: - json.dump(server_config, handle, indent=2, ensure_ascii=False) - handle.write("\n") + _write_and_advise( + args.input_dir, output_path, model_entries, entry_ids, + install_guidance, design_entry_ids, family_keys, catalog_by_family, + host, port, backend, lazy_load, transcripts, write_prompt) - # Post-generation guidance. - print(f"\n[OK] Wrote {args.output} with {len(model_entries)} model entry/entries" - + (f" and voice_dir '{voice_dir}'" if voice_dir else "")) - for family in family_keys: - entry = catalog_by_family[family] - if family == FAMILY_QWEN3_TTS: - print("[INFO] Install the Qwen3-TTS packages from the audio.cpp " - "checkout:") - print(" python3 tools/model_manager_v2.py install " - "qwen3_tts_1_7b_customvoice_q8_0") - print(" python3 tools/model_manager_v2.py install " - "qwen3_tts_1_7b_base_q8_0") - else: - print(f"[INFO] Install {entry['display_name']} from the audio.cpp " - f"checkout: python3 tools/model_manager_v2.py install " - f"{entry['install_id']}") - if len(model_entries) > 1: - print("[INFO] Models load lazily and stay in memory until the server " - "exits; restart the server (or POST /v1/tasks/unload_models) " - "before switching to a large model to free VRAM.") - if family_keys != [FAMILY_QWEN3_TTS]: - for family in family_keys: - if family == FAMILY_QWEN3_TTS: - continue - entry = catalog_by_family[family] - print(f"[INFO] Clone-only family {entry['display_name']}: run " - "audiobook.py with --backend audiocpp --voice <preset name>") - if len(entry_ids) == 1 and non_qwen_single_id is not None: - _offer_config_model_id_sync(non_qwen_single_id) + if len(entry_ids) == 1: + _offer_config_model_id_sync(entry_ids[0]) elif len(entry_ids) > 1: _print_multi_model_model_id_note(entry_ids) print_empty_transcript_warning(transcripts) diff --git a/tools/tui.py b/tools/tui.py new file mode 100644 index 0000000..e1eda05 --- /dev/null +++ b/tools/tui.py @@ -0,0 +1,511 @@ +#!/usr/bin/env python3 +"""Minimal curses TUI widgets for the interactive tools. + +One screen per decision: a directory browser, an expandable checkbox +tree, a single-line text editor, a single-choice menu, and a yes/no +confirm. There is no framework — every widget is a function that runs +its own key loop on a curses window and returns the chosen value. + +Common key bindings: + + Up/Down (or k/j) move the cursor + Enter accept + Esc abort the whole wizard (raises WizardCancelled) + +On screens without typed text (menus, confirm, tree, browser) 'q' also +aborts; inside text editors it is an ordinary character. +""" + +import os +from pathlib import Path +from typing import Callable, List, Optional, Sequence, Tuple + +# Make Esc register quickly instead of pausing for an escape sequence. +os.environ.setdefault("ESCDELAY", "25") + + +class WizardCancelled(Exception): + """Raised when the user presses Esc to abort the wizard.""" + + +# --------------------------------------------------------------------------- +# Shared drawing helpers +# --------------------------------------------------------------------------- + +def _addstr(scr, y: int, x: int, text: str, attr: int = 0) -> None: + """addstr that ignores out-of-bounds and terminal-capability errors.""" + try: + scr.addstr(y, x, text, attr) + except Exception: + pass + + +def _fit(text: str, width: int) -> str: + """Truncate TEXT to WIDTH columns, appending '~' when cut.""" + if width < 1: + return "" + if len(text) <= width: + return text + return text[: max(0, width - 1)] + "~" + + +class Frame: + """A screen frame: title, scrolling body rows, message and footer. + + Widgets append styled body rows via mark(), call draw() after every + state change, and read keys through get_key()/edit_line(). + """ + + def __init__(self, scr, title: str, footer: str): + import curses + self.curses = curses + self.scr = scr + self.title = title + self.footer = footer + self.message = "" # transient status line + self.message_attr = None # None -> bold reverse video + self.rows: List[dict] = [] # {text, attr, indent} + self.scroll = 0 + self.cursor = 0 # highlighted row index + + def mark(self, text: str, attr: int = 0, indent: int = 0) -> None: + self.rows.append({"text": text, "attr": attr, "indent": indent}) + + def draw(self) -> None: + curses = self.curses + scr = self.scr + scr.erase() + height, width = scr.getmaxyx() + if height < 6 or width < 20: + _addstr(scr, 0, 0, _fit("Terminal too small", width - 1), + curses.A_BOLD) + scr.refresh() + return + top = 2 + visible = height - 3 - top + if visible < 1: + visible = 1 + # Keep the cursor inside the viewport. + if self.cursor < self.scroll: + self.scroll = self.cursor + elif self.cursor >= self.scroll + visible: + self.scroll = self.cursor - visible + 1 + if self.scroll + visible > len(self.rows): + self.scroll = max(0, len(self.rows) - visible) + scrolling = len(self.rows) > visible + indicator = f" {self.cursor + 1}/{len(self.rows)} " if scrolling else "" + title_width = width - 1 - (len(indicator) if indicator else 0) + _addstr(scr, 0, 0, _fit(self.title, title_width), + curses.A_BOLD | curses.A_UNDERLINE) + for index in range(self.scroll, + min(len(self.rows), self.scroll + visible)): + row = self.rows[index] + line = " " * row["indent"] + row["text"] + attr = row["attr"] + if index == self.cursor: + attr |= curses.A_REVERSE + _addstr(scr, top + index - self.scroll, 0, + _fit(line, width - 1), attr) + if indicator: + _addstr(scr, 0, max(0, width - len(indicator)), indicator, + curses.A_DIM) + if self.message: + attr = self.message_attr + if attr is None: + attr = curses.A_BOLD | curses.A_REVERSE + _addstr(scr, height - 2, 0, _fit(self.message, width - 1), attr) + _addstr(scr, height - 1, 0, _fit(self.footer, width - 1), curses.A_DIM) + scr.refresh() + + # -- key helpers ------------------------------------------------------ + + def get_key(self, cancel_keys: Sequence[int] = (27,)) -> int: + """Read one key; cancel keys and Ctrl-C raise WizardCancelled.""" + try: + key = self.scr.getch() + except KeyboardInterrupt: + raise WizardCancelled() from None + if key == 3: # Ctrl-C + raise WizardCancelled() + if key in cancel_keys: + raise WizardCancelled() + return key + + def edit_line(self, start: str, prompt: str = "" + ) -> Optional[str]: + """Run an inline editor on the message line. + + Returns the edited string on Enter, or None when the user backs + out with Esc (the caller decides what that means). + """ + curses = self.curses + text = start + while True: + height, width = self.scr.getmaxyx() + self.message = "" + self.draw() + room = max(1, width - 2 - len(prompt)) + shown = text if len(text) < room else ">" + text[-(room - 2):] + _addstr(self.scr, height - 2, 0, + _fit(f"{prompt}{shown}_", width - 1), curses.A_BOLD) + self.scr.refresh() + try: + key = self.scr.getch() + except KeyboardInterrupt: + raise WizardCancelled() from None + if key == 27: + return None + if key in (10, 13): # Enter + return text + if key in (curses.KEY_BACKSPACE, 8, 127): + text = text[:-1] + elif 32 <= key < 127: + text += chr(key) + + +# --------------------------------------------------------------------------- +# Widget: yes/no confirm +# --------------------------------------------------------------------------- + +def confirm(scr, question: str, default: bool = False, + body: Optional[Sequence[str]] = None) -> bool: + """Ask a yes/no QUESTION; Enter takes DEFAULT, Esc aborts. + + BODY lines are shown above the question (a summary, for example). + """ + frame = Frame(scr, question, + "y = yes n = no Enter = default Esc = cancel") + cancel = (27, ord("q")) + while True: + frame.rows = [] + for line in body or []: + frame.mark(line) + if body: + frame.mark("") + hint = "[Y/n]" if default else "[y/N]" + frame.mark(f"{question} {hint}") + frame.cursor = len(frame.rows) - 1 + frame.draw() + key = frame.get_key(cancel) + if key in (ord("y"), ord("Y")): + return True + if key in (ord("n"), ord("N")): + return False + if key in (10, 13): + return default + + +# --------------------------------------------------------------------------- +# Widget: single-choice menu +# --------------------------------------------------------------------------- + +def menu(scr, title: str, options: Sequence[tuple], default_index: int = 0): + """Show OPTIONS as (label, value) pairs; return the chosen value. + + The cursor starts on DEFAULT_INDEX; Enter returns the highlighted + option's value. + """ + frame = Frame(scr, title, + "Up/Down = move Enter = select Esc = cancel") + cancel = (27, ord("q")) + cursor = max(0, min(default_index, len(options) - 1)) + while True: + frame.rows = [] + for label, _ in options: + frame.mark(label) + frame.cursor = cursor + frame.draw() + curses = frame.curses + key = frame.get_key(cancel) + if key in (curses.KEY_UP, ord("k")): + cursor = (cursor - 1) % len(options) + elif key in (curses.KEY_DOWN, ord("j")): + cursor = (cursor + 1) % len(options) + elif key in (10, 13): + return options[cursor][1] + + +# --------------------------------------------------------------------------- +# Widget: single-line text editor +# --------------------------------------------------------------------------- + +def line_edit(scr, title: str, default: str, + validate: Optional[Callable[[str], Optional[str]]] = None + ) -> str: + """Edit one line of text, pre-filled with DEFAULT; Enter accepts. + + VALIDATE receives the entered string and returns an error message or + None; Enter on an invalid value shows the message and keeps editing. + Esc aborts the wizard ('q' is an ordinary character here). + """ + frame = Frame(scr, title, + "type to edit Backspace = erase Enter = accept " + "Esc = cancel") + text = default + error = "" + while True: + frame.rows = [] + frame.mark("") + frame.mark(f" {text}_") + frame.cursor = 1 + frame.message = error + frame.draw() + curses = frame.curses + key = frame.get_key() # Esc only; 'q' must stay typeable + if key in (10, 13): + if validate is None: + return text + error = validate(text) + if error is None: + return text + error = f"{error} (edit, then Enter)" + continue + if key in (curses.KEY_BACKSPACE, 8, 127): + text = text[:-1] + elif 32 <= key < 127: + text += chr(key) + + +# --------------------------------------------------------------------------- +# Widget: directory browser +# --------------------------------------------------------------------------- + +def _list_dirs(path: Path) -> List[Path]: + """Return the subdirectories of PATH, sorted, dot-dirs excluded.""" + try: + entries = [child for child in path.iterdir() + if child.is_dir() and not child.name.startswith(".")] + except OSError: + return [] + return sorted(entries, key=lambda child: child.name.lower()) + + +def browse_directory(scr, title: str, + validate: Optional[Callable[[Path], Optional[str]]] = None, + start: Optional[Path] = None + ) -> Path: + """Pick a directory; Enter accepts the directory being listed. + + Right (or l) descends into the highlighted entry, Left/Backspace/u + goes to the parent, and e edits the path directly. VALIDATE receives + the listed directory and returns an error message or None; Enter on + an invalid directory is refused with that message. Esc aborts the + wizard. + """ + footer = ("Up/Down = move Right = open Left = parent e = edit " + "path Enter = choose this directory Esc = cancel") + frame = Frame(scr, title, footer) + cancel = (27, ord("q")) + current = Path(start) if start is not None else Path.cwd() + try: + current = current.resolve() + except OSError: + current = Path.cwd() + cursor = 0 + + def validation_error() -> Optional[str]: + if validate is None: + return None + return validate(current) + + while True: + entries = _list_dirs(current) + cursor = max(0, min(cursor, max(0, len(entries) - 1))) + frame.rows = [] + frame.mark(f"Directory: {current}", frame.curses.A_BOLD) + error = validation_error() + if error is None: + frame.mark(" This directory is a valid choice. Press Enter.", + frame.curses.A_DIM) + else: + frame.mark(f" {error}", frame.curses.A_BOLD) + frame.mark("") + if not entries: + frame.mark(" (no subdirectories)") + for entry in entries: + frame.mark(f" {entry.name}/") + header = 3 # directory line, validity line, blank separator + frame.cursor = header + (cursor if entries else 0) + frame.message = "" + frame.draw() + curses = frame.curses + key = frame.get_key(cancel) + if key in (curses.KEY_UP, ord("k")): + cursor = max(0, cursor - 1) + elif key in (curses.KEY_DOWN, ord("j")): + if entries: + cursor = min(len(entries) - 1, cursor + 1) + elif key in (curses.KEY_RIGHT, ord("l")): + if entries: + current = entries[cursor] + cursor = 0 + elif key in (curses.KEY_LEFT, ord("h"), ord("u"), + curses.KEY_BACKSPACE, 8, 127): + parent = current.parent + if parent != current: + current = parent + cursor = 0 + elif key == ord("e"): + result = frame.edit_line("", prompt="path: ") + if result is not None: + candidate = Path(os.path.expanduser(result)) + if not candidate.is_absolute(): + candidate = current / candidate + try: + candidate = candidate.resolve() + except OSError: + pass + if candidate.is_dir(): + current = candidate + cursor = 0 + else: + frame.message = f"Not a directory: {candidate}" + frame.draw() + frame.get_key(cancel) + frame.get_key(cancel) + elif key in (10, 13): # Enter: accept the listed directory + error = validation_error() + if error is None: + return current + frame.message = f"{error} (keep browsing)" + frame.draw() + frame.get_key(cancel) + + +# --------------------------------------------------------------------------- +# Widget: expandable checkbox tree +# --------------------------------------------------------------------------- + +def checkbox_tree(scr, title: str, families: List[dict], + footer: Optional[str] = None, + expand_all: bool = False) -> List[Tuple[int, str]]: + """Pick model families and packages from an expandable tree. + + FAMILIES is a list of dicts (one per family) shaped like:: + + { + "label": "Qwen3-TTS (qwen3_tts)", + "detail": "tts, cloning, design", + "options": [ + {"key": "Base-GGUF", "label": "base", "recommended": True}, + {"key": "VoiceDesign-GGUF", "label": "voicedesign", + "recommended": False}, + ], + } + + Space on a family row checks its recommended option (or clears every + option when one is already checked); Space on an option row toggles + that option. Tab/Right expands or collapses the family under the + cursor. Enter returns the flat list of (family_index, option_key) + pairs for every checked option, in tree order; at least one checked + option is required. The first family's recommended option starts + checked (the prompt flow's default), and with EXPAND_ALL every + family starts expanded. + """ + footer = footer or ("Up/Down = move Tab/Right = expand Space = check " + "Enter = accept Esc = cancel") + frame = Frame(scr, title, footer) + cancel = (27, ord("q")) + expanded = {index for index in range(len(families))} if expand_all else set() + checked = set() # (family_index, option_key) + + if families: + expanded.add(0) + first = families[0]["options"] + for option in first: + if option.get("recommended"): + checked.add((0, option["key"])) + break + else: + if first: + checked.add((0, first[0]["key"])) + + def family_checked(index: int) -> bool: + return any(pair[0] == index for pair in checked) + + def accept() -> List[Tuple[int, str]]: + return [(index, option["key"]) + for index, family in enumerate(families) + for option in family["options"] + if (index, option["key"]) in checked] + + def visible_nodes() -> List[tuple]: + nodes: List[tuple] = [] # ("family", i) or ("option", i, key) + for index, family in enumerate(families): + nodes.append(("family", index)) + if index in expanded: + for option in family["options"]: + nodes.append(("option", index, option["key"])) + return nodes + + cursor = 0 + while True: + nodes = visible_nodes() + cursor = max(0, min(cursor, len(nodes) - 1)) + frame.rows = [] + for node in nodes: + if node[0] == "family": + index = node[1] + family = families[index] + mark = "x" if family_checked(index) else " " + arrow = "-" if index in expanded else "+" + attr = frame.curses.A_BOLD if family_checked(index) else 0 + frame.mark(f"[{mark}] {arrow} {family['label']}", attr) + else: + _, index, option_key = node + option = next(opt for opt in families[index]["options"] + if opt["key"] == option_key) + is_on = (index, option_key) in checked + mark = "x" if is_on else " " + note = " [recommended]" if option.get("recommended") else "" + frame.mark(f" [{mark}] {option['label']}{note}") + frame.cursor = cursor + node = nodes[cursor] + frame.message = families[node[1]].get("detail", "") + frame.message_attr = frame.curses.A_DIM + frame.draw() + curses = frame.curses + key = frame.get_key(cancel) + if key in (curses.KEY_UP, ord("k")): + cursor = (cursor - 1) % len(nodes) + elif key in (curses.KEY_DOWN, ord("j")): + cursor = (cursor + 1) % len(nodes) + elif key in (9, curses.KEY_RIGHT, ord("l")) and node[0] == "family": + index = node[1] + if index in expanded: + expanded.discard(index) + else: + expanded.add(index) + elif key == curses.KEY_LEFT and node[0] == "family": + expanded.discard(node[1]) + elif key == ord(" "): + if node[0] == "family": + index = node[1] + options = families[index]["options"] + if family_checked(index): + for option in options: + checked.discard((index, option["key"])) + else: + for option in options: + if option.get("recommended"): + checked.add((index, option["key"])) + break + else: + if options: + checked.add((index, options[0]["key"])) + expanded.add(index) + else: + _, index, option_key = node + if (index, option_key) in checked: + checked.discard((index, option_key)) + else: + checked.add((index, option_key)) + elif key in (10, 13): # Enter: accept the checked selection + selection = accept() + if selection: + return selection + frame.message = "Check at least one model package (Space)" + frame.message_attr = None + frame.draw() + frame.get_key(cancel) + frame.message_attr = frame.curses.A_DIM |