diff options
| -rw-r--r-- | .gitignore | 4 | ||||
| -rw-r--r-- | README.md | 20 | ||||
| -rwxr-xr-x | audiobook.py | 229 | ||||
| -rw-r--r-- | backends/__init__.py | 114 | ||||
| -rwxr-xr-x | backends/audiocpp.py (renamed from tools/make_audiocpp_server_json.py) | 1223 | ||||
| -rw-r--r-- | backends/common.py | 231 | ||||
| -rwxr-xr-x | backends/faster.py | 398 | ||||
| -rw-r--r-- | backends/qwen.py | 254 | ||||
| -rw-r--r-- | docs/backend-faster.md | 6 | ||||
| -rw-r--r-- | docs/backend-qwen.md | 2 | ||||
| -rw-r--r-- | hub.py | 377 | ||||
| -rw-r--r-- | requirements.txt | 2 | ||||
| -rw-r--r-- | tests/test_backends.py | 91 | ||||
| -rw-r--r-- | tests/test_backends_audiocpp.py | 1062 | ||||
| -rw-r--r-- | tests/test_backends_faster.py (renamed from tests/test_make_faster_voices_json.py) | 67 | ||||
| -rw-r--r-- | tests/test_hub.py | 91 | ||||
| -rw-r--r-- | tests/test_make_audiocpp_server_json.py | 1743 | ||||
| -rw-r--r-- | tests/test_tui.py | 37 | ||||
| -rwxr-xr-x | tools/make_faster_voices_json.py | 121 | ||||
| -rw-r--r-- | tui.py (renamed from tools/tui.py) | 48 |
20 files changed, 3414 insertions, 2706 deletions
@@ -8,6 +8,10 @@ output/* input/* !input/.gitkeep +# Backend checkouts cloned by the setup wizards (backends.audiocpp / .faster) +/audio.cpp/ +/faster-qwen3-tts/ + *.epub input/*.txt *.m4b @@ -32,7 +32,23 @@ pip install -r requirements.txt Put your book files (epub, etc.) in the `input/` directory. The output goes to `output/`. -You need to install one of the following backends (see below for installation/usage) +## Quick start (TUI) + +Run the generator with no arguments in a terminal: + +```bash +python audiobook.py +``` + +A full-screen TUI opens and detects which TTS backends are already set up. From the menu you can: + +- **Convert books…** — process the `input/` directory with a ready backend (it reads the backend's `server.json` / `voices.json` so you pick the model and voice from menus), or +- **Set up a backend…** — clone, build, and configure a backend end-to-end (audio.cpp, qwen, faster), or +- **Modify a backend…** — regenerate its config (a new `server.json`, rebuild `voices.json`, change ports/speaker). + +Everything the TUI does can also be scripted with flags: `python audiobook.py --backend audiocpp --model higgs --voice narrator`, or `python -m backends.audiocpp --families higgs_audio_tts --clone --build-backend cuda`. + +You need one of the following backends (the TUI sets them up for you; manual steps below): | Backend | Description | | -------------------------------------------------------------------- | ------------------------------------------------------ | @@ -91,7 +107,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`, that will interactively make this file for you, including automatically transcribing `.wav` voices to clone with `whisper`. It runs as a colorful DOS-style 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). +The easiest way is the TUI: run `python audiobook.py`, choose **Set up a backend… → audio.cpp**, and it clones `audio.cpp` into `./audio.cpp` (or reuses an existing checkout), builds `audiocpp_server`, lets you pick model families/packages from an expandable checkbox tree (reading the checkout's `model_specs/`), transcribes `.wav` voices with `whisper`, writes `server.json` into the checkout, syncs `converter/config.py`, and prints the launch command. Run it directly with `python -m backends.audiocpp` (flags like `--wavs`, `--families`, `--build-backend`, `--clone` skip the corresponding screens for scripting). 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 { diff --git a/audiobook.py b/audiobook.py index 6ad056c..da67ac7 100755 --- a/audiobook.py +++ b/audiobook.py @@ -3,6 +3,8 @@ TTS Audiobook Generator Converts TXT, PDF and EPUB files into audiobooks using a local TTS server. +Run with no arguments in a terminal for the full TUI (set up backends, +process the input directory); pass flags to script a conversion directly. Edit converter/config.py to change voice and processing settings. """ @@ -35,17 +37,92 @@ from converter.tts import ( ) +def convert(backend: str = None, voice: str = None, clone: str = None, + transcription: str = None, no_transcription: bool = False, + language: str = None, speed: float = 1.0, single_file: bool = False, + output_format: str = None, debug: bool = False, chunk: bool = False, + model_id: str = None, instructions: str = None, + request_options: dict = None) -> int: + """Run one conversion pass with explicit options (used by the CLI and hub). + + Returns the process exit code (0 on success, 1 on failure, 130 on + Ctrl-C). BACKEND defaults to config.BACKEND, OUTPUT_FORMAT to + config.AUDIO_FORMAT. LANGUAGE is already-normalized where required. + """ + backend = backend or config.BACKEND + output_format = output_format or config.AUDIO_FORMAT + request_options = request_options or {} + setup_logging(debug=debug) + setup_directories() + + if backend == BACKEND_FASTER: + voice_mode = VOICE_MODE_CLONE + elif backend == BACKEND_AUDIOCPP: + voice_mode = VOICE_MODE_CLONE if voice else VOICE_MODE_CUSTOM + else: + voice_mode = VOICE_MODE_CLONE if clone else VOICE_MODE_CUSTOM + + book_files, planned = AudiobookConverter.preflight_overwrites( + backend=backend, voice=voice, voice_mode=voice_mode, + voice_clone_ref_audio=clone, output_format=output_format, + instructions=instructions, + ) + if not book_files: + print("[INFO] Nothing to convert. Add a .txt, .pdf, or .epub file " + "to the input folder and run again.") + return 0 + if not planned: + print("[INFO] Nothing to convert (all books skipped)") + return 0 + + try: + converter = AudiobookConverter( + voice_mode=voice_mode, voice_clone_ref_audio=clone, + voice_clone_ref_text=transcription, + skip_transcription=no_transcription, speed=speed, + single_file=single_file, output_format=output_format, + language=language, backend=backend, voice=voice, debug=debug, + chunk=chunk, model_id=model_id, instructions=instructions, + request_options=request_options, + ) + converter._book_files = book_files + converter._planned = planned + ok = converter.run() + except KeyboardInterrupt: + print("\n[WARNING] Shutdown requested by user") + return 130 + except Exception as exc: + print(f"[FATAL] Fatal error: {exc}") + traceback.print_exc() + return 1 + return 0 if ok else 1 + + def main() -> None: - """Entry point with argparse.""" + """Entry point: TUI hub with no args in a terminal, else argparse CLI.""" + # No arguments + interactive terminal -> the TUI hub (set up backends + # and process the input directory end-to-end). Anything else is the + # scriptable argparse CLI. + if not sys.argv[1:]: + try: + interactive = sys.stdin.isatty() and sys.stdout.isatty() + except (AttributeError, ValueError): + interactive = False + if interactive: + import hub + sys.exit(hub.run()) + # Non-interactive with no args: a default conversion run (cron/etc). + sys.exit(convert()) + parser = argparse.ArgumentParser( description="Convert books to audiobooks using a local TTS server", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: - # Use the default audio.cpp audiocpp_server (speaker mode - Vivian speaker, or a server-side voice) + # No arguments, in a terminal: the full TUI (set up backends, convert). python audiobook.py - # Use the audio.cpp audiocpp_server with a server-side voice preset + # Use the audio.cpp audiocpp_server (speaker mode - Vivian speaker, or a server-side voice) python audiobook.py --backend audiocpp --voice narrator # Use the audio.cpp audiocpp_server with a voice design model (task 'vdes') @@ -64,34 +141,23 @@ Examples: ) parser.add_argument( - "--clone", - type=str, - metavar="PATH", + "--clone", type=str, metavar="PATH", help=("Path to reference audio file for voice cloning (WAV format). " "Passing this flag switches the converter to voice clone mode.") ) - parser.add_argument( - "--transcription", - type=str, - default=None, + "--transcription", type=str, default=None, help=("Transcript of the reference audio for in-context cloning (recommended for " "highest quality). If omitted, a local Whisper backend is used if installed; " "otherwise the converter falls back to x-vector-only mode.") ) - parser.add_argument( - "--no-transcription", - action="store_true", + "--no-transcription", action="store_true", help=("Skip automatic transcription of the reference audio (use x-vector-only " "cloning). Ignored when --transcription is provided.") ) - parser.add_argument( - "--language", - type=str, - default=None, - metavar="LANG", + "--language", type=str, default=None, metavar="LANG", help=("Output language for the synthesized speech, e.g. English, Japanese, " "or Auto (language names and short codes like en/ja are accepted). " "With --backend audiocpp the language is adapted to the model " @@ -99,32 +165,22 @@ Examples: "omitted when the model detects the language itself. Defaults to " "the LANGUAGE setting in converter/config.py (English).") ) - parser.add_argument( - "--speed", - type=float, - default=1.0, + "--speed", type=float, default=1.0, help="Playback speed factor for the final audiobook (1.0 = normal). Pitch-preserving." ) - parser.add_argument( - "--format", - choices=list(AUDIO_FORMATS), - default=config.AUDIO_FORMAT, + "--format", choices=list(AUDIO_FORMATS), default=config.AUDIO_FORMAT, help=f"Output container format (default: {config.AUDIO_FORMAT}). m4b uses AAC audio." ) - parser.add_argument( - "--single-file", - action="store_true", + "--single-file", action="store_true", help=("Combine all chapters into a single audio file. By default books with " "chapters (e.g. EPUB) are converted to one file per chapter. " "Ignored for m4b, which is always a single file.") ) - parser.add_argument( - "--backend", - choices=[BACKEND_AUDIOCPP, BACKEND_QWEN, BACKEND_FASTER], + "--backend", choices=[BACKEND_AUDIOCPP, BACKEND_QWEN, BACKEND_FASTER], default=config.BACKEND, help=("TTS server to talk to: the Qwen3-TTS demo server (qwen), the " "faster-qwen3-tts OpenAI-compatible server (faster), or an " @@ -133,12 +189,8 @@ Examples: "and more. Defaults to the BACKEND setting in " "converter/config.py (audiocpp).") ) - parser.add_argument( - "--voice", - type=str, - default=None, - metavar="NAME", + "--voice", type=str, default=None, metavar="NAME", help=("Voice to request from a server-side voice configuration. faster: " "a key in the server's voices.json ('default' when it was started " "with --ref-audio). audiocpp: a voice_preset or voice_dir entry " @@ -147,44 +199,32 @@ Examples: "the qwen backend (use converter/config.py SPEAKER or --clone " "there).") ) - parser.add_argument( - "--debug", - action="store_true", + "--debug", action="store_true", help=("Troubleshooting mode: dump each chunk's raw audio and the exact text " "sent for it under the debug/ folder (organized per book and chapter), " "and log every TTS request and response to the console and log file.") ) - parser.add_argument( - "--chunk", - action="store_true", + "--chunk", action="store_true", help=("Force client-side chunking into CHUNK_SIZE-word requests (see " "converter/config.py). Only matters for --backend audiocpp, which " "otherwise sends each chapter as one request and lets the server " "chunk long text itself; the qwen and faster backends always " "chunk.") ) - parser.add_argument( - "--model", - type=str, - default=None, - metavar="ID", + "--model", type=str, default=None, metavar="ID", help=("audio.cpp server model entry id to use for this run " "(--backend audiocpp only). Overrides AUDIOCPP_MODEL_ID in " "converter/config.py, which is useful for a server hosting " "several lazily-loaded models: generate one server.json with " - "tools/make_audiocpp_server_json.py, then pick the model per " - "run with --model. Leave unset to use the config id, or to " - "auto-select when the server hosts exactly one entry.") + "backends.audiocpp, then pick the model per run with --model. " + "Leave unset to use the config id, or to auto-select when the " + "server hosts exactly one entry.") ) - parser.add_argument( - "--instructions", - type=str, - default=None, - metavar="TEXT", + "--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 " @@ -194,13 +234,8 @@ Examples: "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", + "--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, " @@ -290,68 +325,16 @@ Examples: parser.error(f"--option expects KEY=VALUE (got {item!r})") request_options[key.strip()] = value - setup_logging(debug=args.debug) - setup_directories() - - if args.backend == BACKEND_FASTER: - voice_mode = VOICE_MODE_CLONE - elif args.backend == BACKEND_AUDIOCPP: - voice_mode = VOICE_MODE_CLONE if args.voice else VOICE_MODE_CUSTOM - else: - voice_mode = VOICE_MODE_CLONE if args.clone else VOICE_MODE_CUSTOM - - # Ask every overwrite question up front, before spending time connecting - # to a TTS server: a user who declines (or has nothing to convert) never - # waits on a slow server handshake. Nothing in this step needs the server. - book_files, planned = AudiobookConverter.preflight_overwrites( - backend=args.backend, - voice=args.voice, - voice_mode=voice_mode, - voice_clone_ref_audio=args.clone, - output_format=args.format, - instructions=args.instructions, - ) - - if not book_files: - print("[INFO] Nothing to convert. Add a .txt, .pdf, or .epub file " - "to the input folder and run again.") - sys.exit(0) - - if not planned: - print("[INFO] Nothing to convert (all books skipped)") - sys.exit(0) - - try: - converter = AudiobookConverter( - voice_mode=voice_mode, - voice_clone_ref_audio=args.clone, - voice_clone_ref_text=args.transcription, - skip_transcription=args.no_transcription, - speed=args.speed, - single_file=args.single_file, - output_format=args.format, - language=args.language, - backend=args.backend, - voice=args.voice, - 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 - ok = converter.run() - except KeyboardInterrupt: - print("\n[WARNING] Shutdown requested by user") - sys.exit(130) - except Exception as exc: - print(f"[FATAL] Fatal error: {exc}") - traceback.print_exc() - sys.exit(1) - - sys.exit(0 if ok else 1) + sys.exit(convert( + backend=args.backend, voice=args.voice, clone=args.clone, + transcription=args.transcription, no_transcription=args.no_transcription, + language=args.language, speed=args.speed, single_file=args.single_file, + output_format=args.format, debug=args.debug, chunk=args.chunk, + model_id=args.model, instructions=args.instructions, + request_options=request_options, + )) if __name__ == "__main__": main() + diff --git a/backends/__init__.py b/backends/__init__.py new file mode 100644 index 0000000..9203143 --- /dev/null +++ b/backends/__init__.py @@ -0,0 +1,114 @@ +"""Registry of the TTS backends the audiobook generator can talk to. + +Each backend (audio.cpp, qwen, faster) lives in its own module and owns +its setup wizard, its status detection, and the launch command it prints +once configured. This package aggregates them into a single registry so +``audiobook.py``'s TUI hub and future tools can iterate backends without +hardcoding their names: ``backends.detect_all()`` reports which are set +up, and ``backends.REGISTRY`` drives the hub's setup/modify menus. + +Adding a backend: create ``backends/<name>.py`` exposing +``detect() -> BackendStatus``, ``run_tui() -> int`` and +``modify_actions: list[ModifyAction]``, then append a ``BackendInfo`` in +``_build_registry`` below. ``audiobook.py`` and the hub pick it up +automatically. +""" + +from dataclasses import dataclass, field +from typing import Callable, List, Optional + + +@dataclass +class BackendStatus: + """How far a backend is set up, plus the command to start it. + + INSTALLED means the backend itself is present (a cloned + built + checkout, or a pip package). CONFIGURED means the supporting files are + in place (a server.json / voices.json and a converter/config.py that + points at the right port). DETAILS are short status lines for the hub. + LAUNCH_HINT is the exact command the user runs to start the server. + """ + key: str + label: str + installed: bool + configured: bool + details: List[str] = field(default_factory=list) + launch_hint: str = "" + + @property + def ready(self) -> bool: + """True when the backend is installed and configured for use.""" + return self.installed and self.configured + + +@dataclass +class ModifyAction: + """A per-backend "modify" menu entry (e.g. "New server.json").""" + label: str + run: Callable[[], int] + + +@dataclass +class BackendInfo: + """One registry entry: identity, detector, setup wizard, modify menu.""" + key: str + label: str + detect: Callable[[], BackendStatus] + setup_tui: Callable[[], int] + modify_actions: List[ModifyAction] = field(default_factory=list) + + +REGISTRY: List[BackendInfo] = [] +_BY_KEY: dict = {} + + +def _build_registry() -> None: + """Import the backend modules and wire up REGISTRY (once).""" + if REGISTRY: + return + from . import audiocpp, faster, qwen + + REGISTRY.append(BackendInfo( + key="audiocpp", + label="audio.cpp", + detect=audiocpp.detect, + setup_tui=audiocpp.run_tui, + modify_actions=audiocpp.modify_actions, + )) + REGISTRY.append(BackendInfo( + key="qwen", + label="Qwen3-TTS (demo server)", + detect=qwen.detect, + setup_tui=qwen.run_tui, + modify_actions=qwen.modify_actions, + )) + REGISTRY.append(BackendInfo( + key="faster", + label="faster-qwen3-tts", + detect=faster.detect, + setup_tui=faster.run_tui, + modify_actions=faster.modify_actions, + )) + for info in REGISTRY: + _BY_KEY[info.key] = info + + +def get(key: str) -> Optional[BackendInfo]: + """Return the registry entry for KEY, or None.""" + _build_registry() + return _BY_KEY.get(key) + + +def detect_all() -> List[BackendStatus]: + """Detect every registered backend's status, in registry order.""" + _build_registry() + return [info.detect() for info in REGISTRY] + + +def detect(key: str) -> Optional[BackendStatus]: + """Detect a single backend by key.""" + info = get(key) + return info.detect() if info is not None else None + + +_build_registry() diff --git a/tools/make_audiocpp_server_json.py b/backends/audiocpp.py index 3446428..b401366 100755 --- a/tools/make_audiocpp_server_json.py +++ b/backends/audiocpp.py @@ -1,85 +1,31 @@ #!/usr/bin/env python3 -"""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, 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 +"""Set up the audio.cpp TTS backend for the audiobook generator. + +This does the whole audio.cpp setup end-to-end as a full-screen DOS-style +TUI: locate or clone an audio.cpp checkout into ``./audio.cpp``, optionally +build ``audiocpp_server``, pick model families/packages from the checkout's +``model_specs`` catalog, transcribe reference .wav voices, write +``server.json``, sync ``converter/config.py``, download the models, and +print the exact command to start the server. It is driven by +``audiobook.py``'s TUI hub (``backends.REGISTRY``) but can also be run +directly for scripting — every value has a flag, and a non-interactive run +with all flags supplied never opens the TUI. + +The converter 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 colorful full-screen TUI (curses): every -screen is a centered DOS-style dialog on a black desktop — a file -browser for the audio.cpp checkout and the .wav directory, an -expandable checkbox tree of model families and their installable -packages, centered single-question screens for the server settings, -and Yes/No buttons for every confirmation. In the checkout browser, -pressing Enter (or Right) on a subdirectory named ``audio.cpp`` that -already contains ``model_specs/`` picks it directly, skipping the -``[ Use this directory ]`` step; pressing Esc on the overwrite -confirmation then returns to the browser inside that checkout (with -the auto-pick disabled), instead of aborting the wizard. Esc on any -other wizard screen falls back to the previous screen group (only the -first screen, the checkout browser, exits on Esc). 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". All families are treated equally and listed alphabetically. - -The .wav directory browser (and the prompt default) starts in the single -directory that directly contains .wav files across the audio.cpp checkout -and the tts-audiobook-generator root, if exactly one exists; the -generator's ``output/`` directory is never offered. - -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 [--wavs WAV_DIR] - [--output PATH] [--audiocpp-dir PATH] [--families FAM1,FAM2] + python -m backends.audiocpp [--wavs WAV_DIR] [--output PATH] + [--audiocpp-dir PATH] [--clone] [--families FAM1,FAM2] [--all-packages] [--host HOST] [--port PORT] - [--backend {cuda,vulkan,hip,cpu}] [--lazy-load] - [--whisper-model NAME] [--force] [--notui] - ---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 (in the TUI, Esc on that prompt returns to the -checkout browser rather than aborting). After a successful run the -console output is the written file plus one copy-pasteable -model_manager_v2.py install command per hosted model; you are also -asked whether to run those downloads automatically. - ---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. A leading ``~`` in a -path argument or prompt answer is expanded. - ---backend is the inference backend audiocpp_server was built for. When the -checkout contains a build directory (``build/<platform>-<backend>-<type>`` -with a built ``bin/audiocpp_server``), that backend is auto-detected, -selected by default and marked ``[auto-detected]`` in the menu. + [--build-backend {cuda,vulkan,hip,cpu}] [--backend {cuda,vulkan,hip,cpu}] + [--lazy-load] [--whisper-model NAME] [--force] + [--download] [--no-sync-port] [--no-sync-model-ids] + +With no flags and a terminal, the TUI wizard runs. Without a terminal +(or with all flags supplied), it runs non-interactively from the flags; +any missing required value is a hard error with a remediation hint. """ import argparse @@ -92,29 +38,40 @@ import urllib.parse from pathlib import Path from typing import Callable, Dict, List, Optional, Set, Tuple -# Allow running from any working directory. +# Allow running directly (python backends/audiocpp.py) from any cwd. sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +import tui +from backends import BackendStatus, ModifyAction +from backends import common +from backends.common import ( + CONFIG_PATH, + PROMPT_TEXT_FILENAME, + TTS_ROOT, + detect_wav_dir, + find_wav_files, + normalize_dir_arg, + read_prompt_text, + resolve_wav_dir_arg, + wav_dir_info as _wav_dir_info, + wav_dir_preview as _wav_dir_preview, + write_prompt_text, +) from converter import config from converter.tts import transcribe_reference_audio, whisper_backend_available DEFAULT_HOST = "127.0.0.1" FALLBACK_PORT = 8080 -CONFIG_PATH = Path(__file__).resolve().parent.parent / "converter" / "config.py" - -# The tts-audiobook-generator checkout root (where audiobook.py lives), used -# to default the .wav directory browser. The audio.cpp checkout is detected -# separately (see detect_audiocpp_dir). -TTS_ROOT = Path(__file__).resolve().parent.parent -# Output directory of tts-audiobook-generator; never offered as a .wav source. -TTS_OUTPUT_DIR = "output" BACKENDS = ("cuda", "vulkan", "hip", "cpu") -PROMPT_TEXT_FILENAME = "prompt_text" TASK_TTS = "tts" TASK_VDES = "vdes" +# audio.cpp is cloned into a sibling directory of the audiobook generator. +AUDIOCPP_DIR_NAME = "audio.cpp" +AUDIOCPP_GIT_URL = "https://github.com/0xShug0/audio.cpp" + # Sentinel returned by tui.confirm (via its cancel_value) when the user # presses Esc on an overwrite prompt to go back to the checkout browser # instead of aborting the wizard. @@ -154,137 +111,18 @@ class _TuiError(Exception): """ -def _curses_importable() -> bool: - """Return True when the curses module can be imported.""" +def _interactive() -> bool: + """True when the TUI wizard can run (curses importable + tty).""" 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 - paths are always validated against the current working directory. - """ - cleaned = value.strip() - if len(cleaned) >= 2 and cleaned[0] == cleaned[-1] and cleaned[0] in "\"'": - cleaned = cleaned[1:-1] - return Path(os.path.expanduser(cleaned)).resolve() - - -def 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( - (path for path in input_dir.iterdir() - if path.is_file() and path.suffix.lower() == ".wav"), - key=lambda path: path.name.lower(), - ) - - -def _count_wavs(directory: Path) -> int: - """Count the .wav files in DIRECTORY (0 when it cannot be read).""" - try: - return sum(1 for path in directory.iterdir() - if path.is_file() and path.suffix.lower() == ".wav") - except OSError: - return 0 - - -def detect_wav_dir(audiocpp_dir: Path, tts_root: Path) -> Optional[Path]: - """Find a unique directory that directly contains .wav files. - - Looks shallowly (the root itself and its immediate subdirectories) in - both the audio.cpp checkout and the tts-audiobook-generator root (where - audiobook.py lives), since clone reference .wavs commonly live in either. - The tts-audiobook-generator ``output/`` directory is excluded. When - exactly one candidate is found it is returned (as a starting directory - for the .wav browser); when none or several are found None is returned - so the caller falls back to its default start location. - """ - candidates: List[Path] = [] - seen: Set[Path] = set() - - def consider(directory: Path) -> None: - try: - resolved = directory.resolve() - except OSError: - return - if resolved in seen: - return - seen.add(resolved) - if _count_wavs(directory) > 0: - candidates.append(directory) - - for root in (audiocpp_dir, tts_root): - if not root.is_dir(): - continue - consider(root) - try: - children = sorted(root.iterdir(), key=lambda p: p.name.lower()) - except OSError: - continue - for child in children: - if not child.is_dir() or child.name.startswith("."): - continue - # Exclude the tts-audiobook-generator output directory. - if root == tts_root and child.name == TTS_OUTPUT_DIR: - continue - consider(child) - - if len(candidates) == 1: - return candidates[0] - return None - - -def _wav_dir_info(directory: Path) -> Tuple[str, str]: - """TUI status describing the directory listed in the wav browser.""" - count = _count_wavs(directory) - if count: - wavs = ".wav" if count == 1 else ".wavs" - return (f"{count} {wavs} found in this directory. Press Enter.", - "ok") - return ("No .wav files found in this directory", "warn") - - -def _wav_dir_preview(directory: Path) -> Tuple[str, str]: - """TUI status describing a highlighted subdirectory in the wav browser.""" - count = _count_wavs(directory) - if count: - wavs = ".wav" if count == 1 else ".wavs" - return (f"{count} {wavs}", "ok") - return ("no .wav files", "info") - - def _resolve_audiocpp_root(directory: Path) -> Optional[Path]: """Return the audio.cpp checkout root for DIRECTORY, or None. @@ -332,101 +170,6 @@ def _checkout_auto_select(entry: Path) -> Optional[Path]: return None -def ask(prompt: str, default: Optional[str] = None) -> Optional[str]: - """Prompt for a free-text value with a default; EOF returns the default.""" - suffix = f" [{default}]" if default is not None else "" - try: - answer = input(f"{prompt}{suffix}: ").strip() - except EOFError: - return default - return answer or default - - -def ask_bool(prompt: str, default: bool = False) -> bool: - """Prompt for a yes/no answer; Enter or EOF accepts the default.""" - suffix = " [Y/n]" if default else " [y/N]" - while True: - try: - answer = input(f"{prompt}{suffix}: ").strip().lower() - except EOFError: - return default - if not answer: - return default - if answer in ("y", "yes"): - return True - if answer in ("n", "no"): - return False - print("Please answer 'y' or 'n'.") - - -def ask_port(default: int) -> int: - """Prompt for a port number; Enter or EOF accepts the default.""" - while True: - try: - answer = input(f"Port [{default}]: ").strip() - except EOFError: - return default - if not answer: - return default - try: - value = int(answer) - except ValueError: - value = None - if value is not None and 1 <= value <= 65535: - return value - print("Please enter a port number between 1 and 65535.") - - -def ask_menu(title: str, options: list, default_index: int = 1) -> str: - """Show a numbered menu and return the chosen option's value.""" - print(title) - for number, (label, _) in enumerate(options, 1): - print(f" {number}) {label}") - while True: - try: - answer = input(f"Choice [{default_index}]: ").strip() - except EOFError: - return options[default_index - 1][1] - if not answer: - return options[default_index - 1][1] - if answer.isdigit() and 1 <= int(answer) <= len(options): - return options[int(answer) - 1][1] - print(f"Please enter a number between 1 and {len(options)}.") - - -def ask_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)}.") - - # Backend display order, with short descriptions. The backend name is padded # so the descriptions' dashes line up in the menu. _BACKEND_DESCRIPTIONS = ( @@ -459,13 +202,6 @@ def _backend_options(detected: Optional[str] = None return options, default_index -def ask_backend(detected: Optional[str] = None) -> str: - options, default_index = _backend_options(detected) - return ask_menu( - "Which inference backend was audiocpp_server built for?", - options, default_index=default_index + 1) - - def config_port() -> int: """Return the port of AUDIOCPP_API_URL in converter/config.py.""" try: @@ -733,100 +469,6 @@ def package_dir_options(entry: dict) -> List[dict]: 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 table and return the chosen family keys. - - Input is comma/space-separated numbers; Enter alone selects the first - 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 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() - except EOFError: - return [catalog[0]["family"]] - if not answer: - return [catalog[0]["family"]] - 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(catalog): - indices.append(int(part)) - else: - valid = False - break - if valid and indices: - chosen: List[str] = [] - seen = set() - for index in indices: - family = catalog[index - 1]["family"] - if family not in seen: - seen.add(family) - chosen.append(family) - return chosen - print(f"Please enter comma-separated numbers between 1 and {len(catalog)}.") - - def build_model_entry(family: str, model_id: str, model_path: str, task: str = TASK_TTS) -> dict: """Assemble one server.json model entry. @@ -881,40 +523,6 @@ 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. - - One ``<basename-without-extension>|<transcript>`` line per voice. - Returns the path of the written file. - """ - prompt_path = wav_dir / PROMPT_TEXT_FILENAME - lines = [f"{name}|{text}" for name, text in transcripts.items()] - prompt_path.write_text("\n".join(lines) + "\n", encoding="utf-8") - return prompt_path - - def print_empty_transcript_warning(transcripts: Dict[str, str]) -> None: """Print a loud, final warning for voices whose transcript is empty.""" empty = sorted(name for name, text in transcripts.items() if not text) @@ -946,30 +554,6 @@ def _apply_port_sync(port: int, accepted: bool) -> None: f"will still use port {config_port()}") -def _ask_host_port_backend_lazy(args: argparse.Namespace, - default_lazy: bool, - detected_backend: Optional[str] = None - ) -> Tuple[str, int, str, bool]: - """Ask for (or take from flags) the shared server settings. - - DETECTED_BACKEND (from detect_backend) is offered as the default backend - selection when --backend is not given. - """ - host = args.host if args.host else ask("Bind host", DEFAULT_HOST) - port = args.port if args.port is not None else ask_port(config_port()) - if port != config_port(): - if ask_bool(f"Update AUDIOCPP_API_URL in converter/config.py to port " - f"{port} so audiobook.py talks to this server", True): - _apply_port_sync(port, True) - else: - _apply_port_sync(port, False) - backend = args.backend if args.backend else \ - ask_backend(detected_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 _decide_transcription(wav_files: list, existing: Dict[str, str], prompt_exists: bool, force: bool, confirm: Callable[[str, bool], bool]) -> dict: @@ -1000,15 +584,14 @@ def _decide_transcription(wav_files: list, existing: Dict[str, str], def _transcribe(args: argparse.Namespace, include_clone: bool, - plan: Optional[dict] = None - ) -> Tuple[Dict[str, str], bool]: + plan: dict) -> Tuple[Dict[str, str], bool]: """Transcribe the wav directory into a stem -> transcript mapping. 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 and the prompt_text mapping it already read is reused; otherwise - the plan is decided with the line prompts. + PLAN is always pre-collected — by the TUI (via _decide_transcription and + its confirm callbacks) or by _flag_plan for a non-interactive run — so no + questions are asked here. """ if not include_clone: print(f"[WARNING] Ignoring {args.input_dir}: no clone-capable family " @@ -1022,14 +605,7 @@ def _transcribe(args: argparse.Namespace, include_clone: bool, return {}, False prompt_path = args.input_dir / PROMPT_TEXT_FILENAME - if plan is None: - 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: ask_bool(question, default)) - else: - existing = plan.get("existing") or {} + existing = plan.get("existing") or {} if plan else {} if plan["mode"] == "keep": print(f"[INFO] Kept existing {prompt_path}; all voices were " @@ -1053,22 +629,36 @@ def _transcribe(args: argparse.Namespace, include_clone: bool, return transcripts, True -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. +def _flag_plan(wav_files: list, prompt_path: Path, force: bool) -> dict: + """Build a transcription plan for a non-interactive (flag-only) run. + + With --force everything is re-transcribed; otherwise an existing + prompt_text is reused and only voices with an empty transcript are + re-transcribed, mirroring what the TUI confirms interactively. + """ + if prompt_path.exists() and not force: + existing = read_prompt_text(prompt_path) + missing = [wav for wav in wav_files + if not existing.get(wav.stem, "").strip()] + if not missing: + return {"mode": "keep", "missing": [], "existing": existing} + return {"mode": "missing", "missing": missing, "existing": existing} + return {"mode": "all", "missing": [], "existing": {}} + + +def _offer_config_model_id_sync(model_id: str, accepted: Optional[bool]) -> None: + """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. When ACCEPTED is None the user is asked - (line prompt); otherwise the given decision is applied. + ids are rewritten together. ACCEPTED is True/False (apply/skip the + rewrite) or None when no single-entry sync applies (nothing to do). """ if config.AUDIOCPP_MODEL_ID == model_id \ and config.AUDIOCPP_CLONE_MODEL_ID == model_id: return 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) + return if accepted: if not update_config_model_ids(model_id, model_id): print(f"[WARNING] Could not update {CONFIG_PATH}; edit " @@ -1249,7 +839,6 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser the first screen (the audio.cpp checkout browser) Esc aborts the whole wizard as before. """ - tui = _load_tui() def ask_confirm(question: str, default: bool) -> bool: result = tui.confirm(stdscr, question, default=default, @@ -1276,26 +865,61 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser auto_accept = True browser_start: Path = Path.cwd() force_browse = False + + def do_browse(): + return tui.browse_directory( + stdscr, "Select your audio.cpp directory", + validate=lambda p: None if _resolve_audiocpp_root(p) + else "No model_specs/ directory here", + info=_audiocpp_root_status, + preview=_audiocpp_root_preview, + help_lines=["The root folder of your audio.cpp " + "checkout;", + "it is the one that contains " + "model_specs/"], + start=browser_start, + auto_select=_checkout_auto_select if auto_accept + else None) + while True: audiocpp_dir = args.audiocpp_dir - if audiocpp_dir is None: - audiocpp_dir = detect_audiocpp_dir() + if audiocpp_dir is None and not force_browse: + audiocpp_dir = find_local_checkout() if force_browse: audiocpp_dir = None if audiocpp_dir is None: - audiocpp_dir = tui.browse_directory( - stdscr, "Select your audio.cpp directory", - validate=lambda p: None if _resolve_audiocpp_root(p) - else "No model_specs/ directory here", - info=_audiocpp_root_status, - preview=_audiocpp_root_preview, - help_lines=["The root folder of your audio.cpp " - "checkout;", - "it is the one that contains " - "model_specs/"], - start=browser_start, - auto_select=_checkout_auto_select if auto_accept - else None) + if force_browse: + # Esc on an overwrite confirmation came back here: go + # straight back into the browser inside the previously + # accepted checkout (auto-accept disabled). + audiocpp_dir = do_browse() + else: + # No checkout found anywhere: offer to clone one into + # ./audio.cpp or browse for an existing checkout. Esc + # on this first menu aborts the wizard. + choice = tui.menu( + stdscr, "No audio.cpp checkout found", + [(f"Clone into ./{AUDIOCPP_DIR_NAME} " + f"(from {AUDIOCPP_GIT_URL})", "clone"), + ("Browse for an existing checkout", "browse")], + help_lines=[ + "audio.cpp hosts the TTS model families " + "this generator uses.", + "Clone it into the project directory, or " + "point at an existing checkout."]) + if choice == "clone": + target = TTS_ROOT / AUDIOCPP_DIR_NAME + with tui.suspend(stdscr): + rc = common.git_clone(AUDIOCPP_GIT_URL, + target) + if rc != 0: + raise _TuiError( + f"git clone failed (exit {rc}). Clone " + f"audio.cpp manually: git clone " + f"{AUDIOCPP_GIT_URL} {target}") + audiocpp_dir = target + else: + audiocpp_dir = do_browse() audiocpp_dir = Path(audiocpp_dir).resolve() if not audiocpp_dir.is_dir(): raise _TuiError(f"audio.cpp checkout not found: " @@ -1470,11 +1094,18 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser if sync_port is _GO_BACK: step = 2 continue - if args.backend: + if args.build_backend: + backend = args.build_backend + build = detected_backend is None + elif args.backend: backend = args.backend + build = False + elif detected_backend is not None: + # Already built: use the detected backend, no menu, no build. + backend = detected_backend + build = False else: - backend_options, backend_default = \ - _backend_options(detected_backend) + backend_options, backend_default = _backend_options(None) backend = tui.menu( stdscr, "Which inference backend was audiocpp_server " "built for?", backend_options, @@ -1482,6 +1113,15 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser if backend is _GO_BACK: step = 2 continue + # Not built for any backend yet: offer to build it now. The + # build itself runs in the console tail after the wizard. + build = tui.confirm( + stdscr, f"audiocpp_server is not built for {backend}. " + f"Build it now (runs scripts/build_*)?", + default=True, cancel_value=_GO_BACK) + if build is _GO_BACK: + step = 2 + continue default_lazy = len(model_entries) > 1 if args.lazy_load: lazy_load = True @@ -1576,6 +1216,7 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser "host": host, "port": port, "backend": backend, + "build": build, "lazy_load": lazy_load, "sync_port": sync_port, "sync_model_ids": sync_model_ids, @@ -1585,25 +1226,134 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser } -def _run_tui(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: - """Run the TUI wizard, then the shared console phase.""" - import curses - tui = _load_tui() +def find_local_checkout() -> Optional[Path]: + """Best-effort location of an audio.cpp checkout with model_specs. + + Checks the AUDIOCPP_DIR environment variable, then ``./audio.cpp`` inside + the tts-audiobook-generator root, then an ``audio.cpp`` directory in or + above the current working directory. Returns the path only when it + contains a ``model_specs`` directory. + """ + candidates: List[Path] = [] + env_dir = os.environ.get("AUDIOCPP_DIR") + if env_dir: + candidates.append(Path(os.path.expanduser(env_dir))) + candidates.append(TTS_ROOT / AUDIOCPP_DIR_NAME) + cwd = Path.cwd() + candidates.append(cwd / AUDIOCPP_DIR_NAME) + candidates.append(cwd.parent / AUDIOCPP_DIR_NAME) + candidates.append(cwd.parent.parent / AUDIOCPP_DIR_NAME) + for candidate in candidates: + try: + resolved = candidate.resolve() + except OSError: + continue + if (resolved / "model_specs").is_dir(): + return resolved + return None + + +def find_audiocpp_server_bin(audiocpp_dir: Path) -> Optional[Path]: + """Return the built audiocpp_server binary, or None when not built. + + Scans ``audiocpp_dir/build/*`` for a build directory containing + ``bin/audiocpp_server`` (``.exe`` allowed on Windows). When several + builds exist the first (alphabetical) is returned. + """ + build_root = audiocpp_dir / "build" + if not build_root.is_dir(): + return None 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 + build_dirs = sorted(build_root.iterdir(), + key=lambda p: p.name.lower()) + except OSError: + return None + for build_dir in build_dirs: + if not build_dir.is_dir(): + continue + for name in ("audiocpp_server", "audiocpp_server.exe"): + server = build_dir / "bin" / name + if server.exists(): + return server + return None + + +def find_build_script(audiocpp_dir: Path) -> Optional[Path]: + """Return the audio.cpp build helper script to run, or None. + + Prefers ``scripts/build_linux.sh``; otherwise the first + ``scripts/build_*.sh`` it finds. (Windows ``.bat`` scripts are not run + automatically — build manually there.) + """ + scripts = audiocpp_dir / "scripts" + if not scripts.is_dir(): + return None + preferred = scripts / "build_linux.sh" + if preferred.exists(): + return preferred try: - curses.curs_set(1) # restore the text cursor hidden by the TUI - except curses.error: - pass - if settings is None: - print("[INFO] Aborted; existing server.json kept") + candidates = sorted(scripts.glob("build_*.sh"), + key=lambda p: p.name.lower()) + except OSError: + return None + return candidates[0] if candidates else None + + +def build_audiocpp(audiocpp_dir: Path, backend: str) -> int: + """Build audiocpp_server for BACKEND, streaming output to the console. + + Returns the build script's exit code (non-zero when the script is + missing). Run from a console context (after the TUI wizard returns, or + inside ``tui.suspend``). + """ + script = find_build_script(audiocpp_dir) + if script is None: + print(f"[ERROR] No build script found in {audiocpp_dir}/scripts; " + "build audiocpp_server manually (see the audio.cpp README)") return 1 + print(f"[INFO] Building audiocpp_server for {backend} " + f"({script} --backend {backend} --target audiocpp_server)...") + return common.run_console_subprocess( + ["sh", str(script), "--backend", backend, "--target", + "audiocpp_server"], + cwd=audiocpp_dir) + + +def _print_launch_hint(audiocpp_dir: Path, output_path: Path) -> None: + """Print the exact command to start the server (or build guidance).""" + binary = find_audiocpp_server_bin(audiocpp_dir) + print() + if binary is not None: + print("Start the server with:") + print(f" {binary} --config {output_path}") + else: + print("[INFO] audiocpp_server binary not found. Build it first, e.g.:") + script = find_build_script(audiocpp_dir) + if script is not None: + print(f" sh {script} --backend <cuda|vulkan|hip|cpu> " + "--target audiocpp_server") + print(f" then run: ./build/<platform>-<backend>-release/bin/" + f"audiocpp_server --config {output_path}") + + +def _execute(settings: dict, args: argparse.Namespace) -> int: + """Shared console tail: build, sync, transcribe, write, install, advise. + + Runs after the TUI wizard returns (or after _collect_from_flags for a + non-interactive run): the terminal is plain, so subprocess output and + transcription progress appear normally. + """ + audiocpp_dir = settings["audiocpp_dir"] + + # Build audiocpp_server first (the longest step), when requested. + if settings.get("build"): + rc = build_audiocpp(audiocpp_dir, settings["backend"]) + if rc != 0: + print(f"[WARNING] build exited with code {rc}; the server.json " + "was still written — build audiocpp_server manually before " + "starting it") + else: + print("[OK] build complete") # Port sync (applied now that the terminal is back). if settings["sync_port"] is True: @@ -1611,9 +1361,9 @@ def _run_tui(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: elif settings["sync_port"] is False: _apply_port_sync(settings["port"], False) - # Transcription (console; the questions were already answered in the TUI). + # Transcription (console; the questions were already answered). args.input_dir = settings["wav_dir"] - if settings["include_clone"]: + if settings["include_clone"] and args.input_dir is not None: 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 " @@ -1623,7 +1373,7 @@ def _run_tui(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: transcripts, write_prompt = {}, False _write_and_advise( - settings["audiocpp_dir"], settings["wav_dir"], settings["output_path"], + audiocpp_dir, settings["wav_dir"], settings["output_path"], settings["model_entries"], settings["install_guidance"], settings["host"], settings["port"], settings["backend"], settings["lazy_load"], transcripts, write_prompt) @@ -1632,52 +1382,244 @@ def _run_tui(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: _offer_config_model_id_sync(settings["entry_ids"][0], settings["sync_model_ids"]) print_empty_transcript_warning(transcripts) - _install_models(settings["audiocpp_dir"], settings["install_guidance"], + _install_models(audiocpp_dir, settings["install_guidance"], settings["download"]) + _print_launch_hint(audiocpp_dir, settings["output_path"]) return 0 -def main() -> int: +def run_tui(args: Optional[argparse.Namespace] = None, + parser: Optional[argparse.ArgumentParser] = None) -> int: + """Run the audio.cpp setup wizard end-to-end. + + With no ARGS (the hub's call) a default namespace is built so the full + wizard runs. Called from ``main`` after argparse when the terminal is + interactive. Returns the process exit code. + """ + import curses + if args is None: + parser = build_parser() + args = parser.parse_args([]) + if args.input_dir is not None and not args.input_dir.is_dir(): + print(f"[ERROR] --wavs not found: {args.input_dir}", + file=sys.stderr) + return 2 + 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 + try: + curses.curs_set(1) # restore the text cursor hidden by the TUI + except curses.error: + pass + if settings is None: + print("[INFO] Aborted; existing server.json kept") + return 1 + return _execute(settings, args) + + +def _collect_from_flags(args: argparse.Namespace, + parser: argparse.ArgumentParser) -> Optional[dict]: + """Build the settings dict from flags for a non-interactive run. + + Every required value must come from a flag (there are no prompts in a + non-interactive run); a missing one is a hard ``parser.error``. Returns + the settings dict, or None when the user declined an overwrite (the + default-location fallback then also exists). + """ + # Checkout: --audiocpp-dir, else a local checkout, else --clone clones one. + audiocpp_dir = args.audiocpp_dir + if audiocpp_dir is None: + audiocpp_dir = find_local_checkout() + if audiocpp_dir is None and args.clone: + target = TTS_ROOT / AUDIOCPP_DIR_NAME + rc = common.git_clone(AUDIOCPP_GIT_URL, target) + if rc != 0: + parser.error(f"git clone failed (exit {rc}); clone audio.cpp " + f"manually: git clone {AUDIOCPP_GIT_URL} {target}") + audiocpp_dir = target + if audiocpp_dir is None: + parser.error( + "An audio.cpp checkout is required. Pass --audiocpp-dir PATH, " + "or --clone to clone ./audio.cpp, or run without flags for the " + "TUI wizard.") + audiocpp_dir = Path(audiocpp_dir).resolve() + if not audiocpp_dir.is_dir(): + parser.error(f"audio.cpp checkout not found: {audiocpp_dir}") + root = _resolve_audiocpp_root(audiocpp_dir) + if root is None: + parser.error(f"{audiocpp_dir} has no model_specs/ directory; point " + "--audiocpp-dir at the root of an audio.cpp checkout") + audiocpp_dir = root + try: + catalog = load_model_catalog(audiocpp_dir) + except NotADirectoryError as exc: + parser.error(str(exc)) + if not catalog: + parser.error( + f"No TTS model families found in {audiocpp_dir}/model_specs; " + "check the checkout is up to date") + catalog_by_family = {entry["family"]: entry for entry in catalog} + + # Families: required from --families in a non-interactive run. + if args.families is None: + parser.error("--families is required in a non-interactive run (or run " + "without flags for the TUI wizard)") + 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: + parser.error( + f"Unknown family in --families: {', '.join(unknown)}. " + f"Available: {', '.join(catalog_by_family)}") + family_keys: List[str] = [] + for fam in requested: + if fam not in family_keys: + family_keys.append(fam) + + chosen: Dict[str, List[dict]] = {} + for family in family_keys: + opts = package_dir_options(catalog_by_family[family]) + if args.all_packages: + chosen[family] = opts + else: + chosen[family] = [opt for opt in opts if opt["recommended"]] + + # Non-interactive pickers: design packages default to vdes, dup ids get -2. + def task_picker(install_id: str) -> str: + return TASK_VDES + + def id_picker(display_name: str, install_id: str, default: str) -> str: + return default + + model_entries, entry_ids, install_guidance, design_entry_ids, include_clone = \ + _build_entries(family_keys, chosen, catalog_by_family, + task_picker, id_picker) + + # Server settings. + host = args.host or DEFAULT_HOST + detected_backend = detect_backend(audiocpp_dir) + if args.build_backend: + backend = args.build_backend + build = detected_backend is None + elif args.backend: + backend = args.backend + build = False + elif detected_backend is not None: + backend = detected_backend + build = False + else: + backend = "cuda" + build = False + port = args.port if args.port is not None else config_port() + lazy_load = args.lazy_load if args.lazy_load else (len(model_entries) > 1) + + # Output path / overwrite (decline falls back to cwd, then aborts). + output_path = args.output if args.output is not None \ + else audiocpp_dir / "server.json" + if output_path.exists() and not args.force: + if args.output is None: + output_path = Path.cwd() / "server.json" + if output_path.exists() and not args.force: + print("[INFO] Aborted; existing server.json kept") + return None + else: + print("[INFO] Aborted; existing server.json kept") + return None + + # Config sync decisions (auto-apply unless explicitly declined). + sync_port: Optional[bool] = None + if port != config_port(): + sync_port = not args.no_sync_port + 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 = not args.no_sync_model_ids + + # Wav dir + transcription plan. + wav_dir = args.input_dir + 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 + plan = _flag_plan(wav_files, prompt_path, args.force) + + 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, + "build": build, + "lazy_load": lazy_load, + "sync_port": sync_port, + "sync_model_ids": sync_model_ids, + "wav_dir": wav_dir, + "plan": plan, + "download": args.download, + } + + +def build_parser() -> argparse.ArgumentParser: + """The audio.cpp setup CLI (also used to build a default namespace).""" 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.") + description="Set up the audio.cpp TTS backend: clone/build, pick " + "models, write server.json, and sync converter/config.py.") 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 (asked " - "for when omitted)") + "for when omitted in the TUI)") parser.add_argument("--output", type=Path, default=None, help="Output path for server.json (default: " - "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)") + "server.json inside the audio.cpp checkout; an " + "existing file is overwritten only with --force " + "or a TUI confirm)") 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)") + "AUDIOCPP_DIR or ./audio.cpp; in the TUI you can " + "clone one instead)") + parser.add_argument("--clone", action="store_true", + help="Non-interactive: clone audio.cpp into " + "./audio.cpp when no checkout is found") 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") + "in the audio.cpp catalog (e.g. " + "qwen3_tts,higgs_audio_tts). Required in a " + "non-interactive run; skips the family tree in " + "the TUI") 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)") + help="Host every installable package of each selected " + "family (distinct target_directory) instead of " + "only the recommended one. Voice-design packages " + "are hosted with task 'vdes'") parser.add_argument("--host", type=str, default=None, help="Bind host for the server (default: 127.0.0.1)") parser.add_argument("--port", type=int, default=None, help="Port for the server (default: the port in " "AUDIOCPP_API_URL from converter/config.py)") parser.add_argument("--backend", choices=BACKENDS, default=None, - help="Inference backend audiocpp_server was built " - "for (default: auto-detected from the checkout's " + help="Inference backend recorded in server.json " + "(default: auto-detected from the checkout's " "build/ directory, else cuda)") + parser.add_argument("--build-backend", choices=BACKENDS, default=None, + help="Build audiocpp_server for this backend when it " + "is not built yet, and use it in server.json") parser.add_argument("--lazy-load", action="store_true", help="Load models on first use instead of at startup " "(default: on when more than one model is hosted)") @@ -1687,10 +1629,59 @@ 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)") + parser.add_argument("--download", action="store_true", + help="Run model_manager_v2.py install for each hosted " + "model automatically (default: print the commands " + "only)") + parser.add_argument("--no-sync-port", action="store_true", + help="Do not rewrite AUDIOCPP_API_URL in " + "converter/config.py when --port differs") + parser.add_argument("--no-sync-model-ids", action="store_true", + help="Do not rewrite AUDIOCPP_MODEL_ID/" + "AUDIOCPP_CLONE_MODEL_ID for a single-entry server") + return parser + + +def detect() -> BackendStatus: + """Detect how far audio.cpp is set up, plus the command to start it.""" + checkout = find_local_checkout() + details: List[str] = [] + launch = "" + if checkout is None: + return BackendStatus("audiocpp", "audio.cpp", installed=False, + configured=False, + details=["not cloned — run setup to clone " + "./audio.cpp"]) + details.append(f"checkout: {checkout}") + binary = find_audiocpp_server_bin(checkout) + built = binary is not None + if built: + details.append(f"built: {binary}") + else: + details.append("not built — run setup to build audiocpp_server") + server_json = checkout / "server.json" + configured = server_json.exists() + if configured: + details.append(f"config: {server_json}") + launch = (f"{binary} --config {server_json}" + if built else + f"./build/<platform>-<backend>-release/bin/" + f"audiocpp_server --config {server_json}") + else: + details.append("no server.json — run setup to configure models") + return BackendStatus("audiocpp", "audio.cpp", installed=built, + configured=configured, details=details, + launch_hint=launch) + + +modify_actions: List[ModifyAction] = [ + ModifyAction("Reconfigure audio.cpp (models, voices, server.json)", + run_tui), +] + + +def main() -> int: + parser = build_parser() args = parser.parse_args() if args.input_dir is not None and not args.input_dir.is_dir(): @@ -1701,140 +1692,14 @@ def main() -> int: " --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). The prompt default is - # the unique directory that directly contains .wav files across the - # audio.cpp checkout (best-effort detected here) and the - # tts-audiobook-generator root, so the user usually just presses Enter. - if args.input_dir is None: - tentative_checkout = args.audiocpp_dir or detect_audiocpp_dir() - wav_start = detect_wav_dir(tentative_checkout, TTS_ROOT) \ - if tentative_checkout is not None else None - default = str(wav_start) if wav_start is not None else "" - answer = ask("Directory with .wav reference files", default) - 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" - " --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. - audiocpp_dir = args.audiocpp_dir - if audiocpp_dir is None: - audiocpp_dir = detect_audiocpp_dir() - if audiocpp_dir is None: - 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. " - "Clone one with `git clone https://github.com/0xShug0/audio.cpp` " - "and pass --audiocpp-dir PATH (or set the AUDIOCPP_DIR environment " - "variable)") - audiocpp_dir = audiocpp_dir.resolve() - if not audiocpp_dir.is_dir(): - parser.error(f"audio.cpp checkout not found: {audiocpp_dir}") - root = _resolve_audiocpp_root(audiocpp_dir) - if root is None: - parser.error(f"{audiocpp_dir} has no model_specs/ directory; point " - "--audiocpp-dir at the root of an audio.cpp checkout") - audiocpp_dir = root - try: - catalog = load_model_catalog(audiocpp_dir) - except NotADirectoryError as exc: - parser.error(str(exc)) - if not catalog: - parser.error( - f"No TTS model families found in {audiocpp_dir}/model_specs; " - "check the checkout is up to date") - - # 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: - requested = [f.strip() for f in args.families.split(",") if f.strip()] - catalog_families = {entry["family"] for entry in catalog} - unknown = [f for f in requested if f not in catalog_families] - if unknown: - parser.error( - f"Unknown family in --families: {', '.join(unknown)}. " - f"Available: {', '.join(entry['family'] for entry in catalog)}") - family_keys: List[str] = [] - for fam in requested: - if fam not in family_keys: - family_keys.append(fam) - else: - family_keys = ask_families(catalog) - - catalog_by_family = {entry["family"]: entry for entry in catalog} - - chosen: Dict[str, List[dict]] = {} - for family in family_keys: - entry = catalog_by_family[family] - if args.all_packages: - chosen[family] = ask_package_dirs(entry) - else: - 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 - detected_backend = detect_backend(audiocpp_dir) - host, port, backend, lazy_load = _ask_host_port_backend_lazy( - args, default_lazy, detected_backend) - - transcripts, write_prompt = _transcribe(args, include_clone) - - _write_and_advise( - audiocpp_dir, args.input_dir, output_path, model_entries, - install_guidance, host, port, backend, lazy_load, transcripts, - write_prompt) - - if len(entry_ids) == 1: - _offer_config_model_id_sync(entry_ids[0]) - print_empty_transcript_warning(transcripts) + if _interactive(): + return run_tui(args, parser) - download = _decide_download( - audiocpp_dir, lambda question, default: ask_bool(question, default)) - _install_models(audiocpp_dir, install_guidance, download) - return 0 + # Non-interactive (no terminal, or all flags supplied): flag-only path. + settings = _collect_from_flags(args, parser) + if settings is None: + return 1 + return _execute(settings, args) if __name__ == "__main__": diff --git a/backends/common.py b/backends/common.py new file mode 100644 index 0000000..2c6437f --- /dev/null +++ b/backends/common.py @@ -0,0 +1,231 @@ +"""Shared helpers for the backend setup wizards. + +Every TTS backend setup wizard (audio.cpp, qwen, faster) lives in its own +module under ``backends``; this module holds the pieces more than one of +them needs: .wav discovery, path normalization, and the regex edit that +keeps ``converter/config.py`` in sync with the choices made in a wizard. +It deliberately imports nothing from the other backend modules (or the +TUI) so it can be reused without pulling curses into a non-interactive +run. +""" + +import os +import re +import urllib.parse +from pathlib import Path +from typing import Dict, List, Optional, Set, Tuple + +# The tts-audiobook-generator checkout root (where audiobook.py lives). +# Backend checkouts are cloned into subdirectories of this root +# (./audio.cpp, ./faster-qwen3-tts) so a single tree holds everything. +TTS_ROOT = Path(__file__).resolve().parent.parent + +# converter/config.py — rewritten in place by update_config_value so the +# converter picks up the host/port/voice a wizard configured. +CONFIG_PATH = TTS_ROOT / "converter" / "config.py" + +# Output directory of tts-audiobook-generator; never offered as a .wav +# source by detect_wav_dir. +TTS_OUTPUT_DIR = "output" + +# The voice-transcript mapping file audio.cpp reads from its voice_dir. +# (The faster backend uses voices.json instead; see backends.faster.) +PROMPT_TEXT_FILENAME = "prompt_text" + + +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 + paths are always validated against the current working directory. + """ + cleaned = value.strip() + if len(cleaned) >= 2 and cleaned[0] == cleaned[-1] and cleaned[0] in "\"'": + cleaned = cleaned[1:-1] + return Path(os.path.expanduser(cleaned)).resolve() + + +def 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[Path]: + """Return the .wav files in INPUT_DIR, sorted alphabetically by name.""" + return sorted( + (path for path in input_dir.iterdir() + if path.is_file() and path.suffix.lower() == ".wav"), + key=lambda path: path.name.lower(), + ) + + +def count_wavs(directory: Path) -> int: + """Count the .wav files in DIRECTORY (0 when it cannot be read).""" + try: + return sum(1 for path in directory.iterdir() + if path.is_file() and path.suffix.lower() == ".wav") + except OSError: + return 0 + + +def detect_wav_dir(audiocpp_dir: Path, tts_root: Path) -> Optional[Path]: + """Find a unique directory that directly contains .wav files. + + Looks shallowly (the root itself and its immediate subdirectories) in + both the audio.cpp checkout and the tts-audiobook-generator root (where + audiobook.py lives), since clone reference .wavs commonly live in + either. The tts-audiobook-generator ``output/`` directory is excluded. + When exactly one candidate is found it is returned (as a starting + directory for the .wav browser); when none or several are found None is + returned so the caller falls back to its default start location. + """ + candidates: List[Path] = [] + seen: Set[Path] = set() + + def consider(directory: Path) -> None: + try: + resolved = directory.resolve() + except OSError: + return + if resolved in seen: + return + seen.add(resolved) + if count_wavs(directory) > 0: + candidates.append(directory) + + for root in (audiocpp_dir, tts_root): + if not root.is_dir(): + continue + consider(root) + try: + children = sorted(root.iterdir(), key=lambda p: p.name.lower()) + except OSError: + continue + for child in children: + if not child.is_dir() or child.name.startswith("."): + continue + if root == tts_root and child.name == TTS_OUTPUT_DIR: + continue + consider(child) + + if len(candidates) == 1: + return candidates[0] + return None + + +def wav_dir_info(directory: Path) -> Tuple[str, str]: + """TUI status describing the directory listed in the wav browser.""" + count = count_wavs(directory) + if count: + wavs = ".wav" if count == 1 else ".wavs" + return (f"{count} {wavs} found in this directory. Press Enter.", + "ok") + return ("No .wav files found in this directory", "warn") + + +def wav_dir_preview(directory: Path) -> Tuple[str, str]: + """TUI status describing a highlighted subdirectory in the wav browser.""" + count = count_wavs(directory) + if count: + wavs = ".wav" if count == 1 else ".wavs" + return (f"{count} {wavs}", "ok") + return ("no .wav files", "info") + + +def url_with_port(url: str, port: int) -> str: + """Return URL with its port replaced/inserted as PORT.""" + parts = urllib.parse.urlsplit(url) + host = parts.hostname or "127.0.0.1" + return urllib.parse.urlunsplit( + (parts.scheme or "http", f"{host}:{port}", parts.path, "", "")) + + +def update_config_value(key: str, value: str, + config_path: Optional[Path] = None) -> bool: + """Rewrite a ``KEY = "value"`` line in converter/config.py. + + Only the quoted literal is replaced; surrounding lines and the trailing + comment are preserved. Returns True when the file was changed. Used by + the qwen and faster wizards to keep their API URL / voice / speaker + settings in sync with the converter. + """ + path = Path(config_path) if config_path is not None else CONFIG_PATH + try: + text = path.read_text(encoding="utf-8") + except OSError: + return False + match = re.search(r'(?m)^(\s*' + re.escape(key) + r'\s*=\s*")([^"]*)(")', + text) + if not match or match.group(2) == value: + return False + text = text[:match.start(2)] + value + text[match.end(2):] + try: + path.write_text(text, encoding="utf-8") + except OSError: + return False + return True + + +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. + + One ``<basename-without-extension>|<transcript>`` line per voice. + Returns the path of the written file. + """ + prompt_path = wav_dir / PROMPT_TEXT_FILENAME + lines = [f"{name}|{text}" for name, text in transcripts.items()] + prompt_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + return prompt_path + + +def run_console_subprocess(argv: List[str], cwd: Optional[Path] = None) -> int: + """Run a subprocess whose output streams to the plain console. + + Used inside ``tui.suspend`` for clone/build/pip steps: the caller has + already left curses mode, so the child inherits the real terminal and + its output appears normally. Returns the process exit code. + """ + import subprocess + try: + result = subprocess.run(argv, cwd=str(cwd) if cwd is not None else None) + except OSError as exc: + print(f"[ERROR] Could not run {' '.join(argv)}: {exc}") + return 1 + return result.returncode + + +def git_clone(url: str, target: Path) -> int: + """Clone URL into TARGET, streaming to the console. Returns exit code.""" + print(f"[INFO] Cloning {url} into {target}...") + return run_console_subprocess(["git", "clone", url, str(target)]) + + +def pip_install(packages: List[str]) -> int: + """pip install PACKAGES (into the current environment). Returns exit code.""" + print(f"[INFO] pip install {' '.join(packages)}...") + import sys + return run_console_subprocess([sys.executable, "-m", "pip", "install", + *packages]) diff --git a/backends/faster.py b/backends/faster.py new file mode 100755 index 0000000..4a2cc6f --- /dev/null +++ b/backends/faster.py @@ -0,0 +1,398 @@ +#!/usr/bin/env python3 +"""Set up the faster-qwen3-tts backend for the audiobook generator. + +faster-qwen3-tts is an OpenAI-compatible Qwen3-TTS server with CUDA-graph +inference (NVIDIA GPU required). It always uses voice cloning, with the +reference voice configured on the server through a ``voices.json``. This +module sets the whole backend up end-to-end as a TUI: pip-install the +package, clone the repo (for ``examples/openai_server.py``), build a +``voices.json`` from a directory of .wav references (transcribed with +Whisper), sync ``converter/config.py``, and print the launch command. It is +driven by ``audiobook.py``'s hub but can also be run directly with flags. + +Usage: + python -m backends.faster [--wavs WAV_DIR] [--output PATH] + [--language LANG] [--whisper-model NAME] [--force] + [--port PORT] [--voice NAME] [--skip-install] [--skip-clone] +""" + +import argparse +import importlib.util +import json +import sys +from pathlib import Path +from typing import List, Optional + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import tui +from backends import BackendStatus, ModifyAction +from backends import common +from backends.common import TTS_ROOT, find_wav_files, normalize_dir_arg +from converter import config +from converter.tts import normalize_language, transcribe_reference_audio, \ + whisper_backend_available + +FASTER_DIR_NAME = "faster-qwen3-tts" +FASTER_GIT_URL = "https://github.com/andimarafioti/faster-qwen3-tts" +FASTER_PIP_PKG = "faster-qwen3-tts[demo]" +WHISPER_MODELS = ("tiny", "base", "small", "medium", "large-v3") + + +def _checkout() -> Path: + return TTS_ROOT / FASTER_DIR_NAME + + +def _is_installed() -> bool: + return importlib.util.find_spec("faster_qwen3_tts") is not None + + +def _is_cloned() -> bool: + return (_checkout() / "examples" / "openai_server.py").is_file() + + +def _config_port() -> int: + import urllib.parse + try: + return urllib.parse.urlsplit(config.FASTER_API_URL).port or 8000 + except ValueError: + return 8000 + + +def build_voices(wav_files: list, language: str, whisper_model: str) -> dict: + """Transcribe each wav file and build the voices mapping.""" + voices = {} + for wav_file in wav_files: + name = wav_file.stem + print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...") + text = transcribe_reference_audio(str(wav_file), model_name=whisper_model) + if text: + print(f"[OK] {name}: {text}") + else: + print(f"[WARNING] No transcript for '{name}'; the faster backend " + "strongly recommends an accurate transcript — consider " + "editing voices.json by hand before starting the server") + voices[name] = { + "ref_audio": str(wav_file.resolve()), + "ref_text": text or "", + "language": language, + } + return voices + + +def _write_voices_json(output_path: Path, wav_dir: Path, language: str, + whisper_model: str, force: bool) -> Optional[dict]: + """Transcribe the wav dir and write voices.json; return the voices dict.""" + wav_files = find_wav_files(wav_dir) + if not wav_files: + print(f"[ERROR] No .wav files found in {wav_dir}") + return None + if whisper_backend_available() is None: + print("[WARNING] Neither faster_whisper nor whisper was found, so " + "transcripts will be empty — install one or edit voices.json " + "by hand.") + voices = build_voices(wav_files, language, whisper_model) + with output_path.open("w", encoding="utf-8") as handle: + json.dump(voices, handle, indent=4, ensure_ascii=False) + handle.write("\n") + print(f"[OK] Wrote {output_path} with {len(voices)} voice(s): " + f"{', '.join(voices)}") + return voices + + +def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: + """Linear TUI wizard collecting every faster-setup decision.""" + _GO_BACK = object() + + def confirm(question: str, default: bool = True) -> Optional[bool]: + res = tui.confirm(stdscr, question, default=default, + cancel_value=_GO_BACK) + return None if res is _GO_BACK else res + + # Step 0: pip install (if not installed and not skipped). + do_install = False + if not _is_installed() and not args.skip_install: + choice = confirm("faster-qwen3-tts is not installed. " + "pip install it now?", default=True) + if choice is None: + return None + do_install = choice + + # Step 1: clone (if not cloned and not skipped). + do_clone = False + if not _is_cloned() and not args.skip_clone: + choice = confirm(f"faster-qwen3-tts repo not cloned. Clone it into " + f"./{FASTER_DIR_NAME}?", default=True) + if choice is None: + return None + do_clone = choice + + # Step 2: voices.json — wav dir, language, whisper model, output path. + wav_dir = args.input_dir + if wav_dir is None: + wav_dir = tui.browse_directory( + stdscr, "Select the directory with your .wav voices", + info=common.wav_dir_info, preview=common.wav_dir_preview, + start=Path.cwd()) + language = args.language + if language is None: + lang_text = tui.line_edit( + stdscr, "Language", config.LANGUAGE, + validate=lambda s: None if _try_language(s) + else "Unknown language (e.g. English, en)", + help_lines=["Language for every voice, as passed to the TTS " + "model (names or short codes accepted)"]) + language = lang_text + whisper_model = args.whisper_model + if whisper_model is None: + whisper_model = tui.menu( + stdscr, "Whisper model for transcription", + [(m, m) for m in WHISPER_MODELS], + default_index=WHISPER_MODELS.index("base")) + output_path = args.output + if output_path is None: + # Default into the cloned checkout; fall back to the wav directory + # when the checkout is not present (so a flag-only run still works). + output_path = (_checkout() / "voices.json") if _is_cloned() \ + else (wav_dir / "voices.json") + if output_path.exists() and not args.force: + choice = confirm(f"{output_path} already exists. Overwrite?", + default=True) + if choice is None or choice is False: + # Fall back to a path in the current directory. + output_path = Path.cwd() / "voices.json" + + # Step 3: port + default voice. + port = args.port + if port is None: + port_text = tui.line_edit( + stdscr, "Server 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) + + return { + "do_install": do_install, + "do_clone": do_clone, + "wav_dir": wav_dir, + "language": language, + "whisper_model": whisper_model, + "output_path": output_path, + "port": port, + "force": args.force, + } + + +def _try_language(value: str) -> bool: + try: + normalize_language(value) + return True + except ValueError: + return False + + +def _execute(settings: dict) -> int: + """Console tail: install, clone, write voices.json, sync, advise.""" + if settings["do_install"]: + rc = common.pip_install([FASTER_PIP_PKG]) + if rc != 0: + print(f"[WARNING] pip install failed (exit {rc}); install " + f"{FASTER_PIP_PKG} manually") + else: + print("[OK] faster-qwen3-tts installed") + + if settings["do_clone"]: + rc = common.git_clone(FASTER_GIT_URL, _checkout()) + if rc != 0: + print(f"[WARNING] git clone failed (exit {rc}); clone manually: " + f"git clone {FASTER_GIT_URL} {_checkout()}") + else: + print(f"[OK] cloned into {_checkout()}") + + voices = _write_voices_json(settings["output_path"], settings["wav_dir"], + settings["language"], settings["whisper_model"], + settings["force"]) + if voices is None: + return 1 + + # Sync converter/config.py port + default voice. + port = settings["port"] + new_url = common.url_with_port(config.FASTER_API_URL, port) + if new_url != config.FASTER_API_URL: + if common.update_config_value("FASTER_API_URL", new_url): + print(f"[OK] Updated FASTER_API_URL to {new_url}") + else: + print("[WARNING] Could not update FASTER_API_URL; edit " + "converter/config.py by hand") + default_voice = next(iter(voices)) + if default_voice != config.FASTER_VOICE: + if common.update_config_value("FASTER_VOICE", default_voice): + print(f"[OK] Updated FASTER_VOICE to {default_voice}") + else: + print("[WARNING] Could not update FASTER_VOICE; edit " + "converter/config.py by hand") + + _print_launch_hint(settings["output_path"], port) + return 0 + + +def _print_launch_hint(voices_path: Path, port: int) -> None: + print() + if _is_cloned(): + print("Start the server with:") + print(f" python {_checkout()}/examples/openai_server.py " + f"--voices {voices_path} --port {port}") + else: + print("[INFO] Clone faster-qwen3-tts to get examples/openai_server.py,") + print(f" then run it with --voices {voices_path} --port {port}") + + +def run_tui(args: Optional[argparse.Namespace] = None) -> int: + """Run the faster setup wizard end-to-end.""" + import curses + if args is None: + args = build_parser().parse_args([]) + try: + settings = curses.wrapper(_wizard, args) + except tui.WizardCancelled: + print("\n[INFO] Cancelled; nothing was written") + return 1 + try: + curses.curs_set(1) + except curses.error: + pass + if settings is None: + print("[INFO] Aborted") + return 1 + return _execute(settings) + + +def _collect_from_flags(args: argparse.Namespace, + parser: argparse.ArgumentParser) -> Optional[dict]: + """Build the settings dict from flags for a non-interactive run.""" + if args.input_dir is None: + parser.error("--wavs is required in a non-interactive run (or run " + "without flags for the TUI wizard)") + if not args.input_dir.is_dir(): + parser.error(f"WAV directory not found: {args.input_dir}") + try: + language = normalize_language(args.language or config.LANGUAGE) + except ValueError as exc: + parser.error(str(exc)) + output_path = args.output if args.output is not None \ + else ((_checkout() / "voices.json") if _is_cloned() + else (args.input_dir / "voices.json")) + if output_path.exists() and not args.force: + print("[INFO] Aborted; existing voices.json kept") + return None + return { + "do_install": (not _is_installed()) and not args.skip_install, + "do_clone": (not _is_cloned()) and not args.skip_clone, + "wav_dir": args.input_dir, + "language": language, + "whisper_model": args.whisper_model or "base", + "output_path": output_path, + "port": args.port if args.port is not None else _config_port(), + "force": args.force, + } + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Set up the faster-qwen3-tts backend: pip install, clone, " + "build voices.json, and sync converter/config.py.") + parser.add_argument("input_dir", type=normalize_dir_arg, nargs="?", + default=None, metavar="WAV_DIR", + help="Directory with .wav reference files (required in " + "a non-interactive run; browsed for in the TUI)") + parser.add_argument("--output", type=Path, default=None, + help="Output path for voices.json (default: " + "./faster-qwen3-tts/voices.json, or " + "WAV_DIR/voices.json when not cloned)") + parser.add_argument("--language", type=str, default=None, + help="Language for all voices (default: English; " + "names and short codes accepted)") + parser.add_argument("--whisper-model", type=str, default=None, + choices=WHISPER_MODELS, + help="Whisper model size for transcription " + "(default: base)") + parser.add_argument("--force", action="store_true", + help="Overwrite an existing voices.json without " + "prompting") + parser.add_argument("--port", type=int, default=None, + help="Server port to record in converter/config.py " + "(default: the port in FASTER_API_URL)") + parser.add_argument("--skip-install", action="store_true", + help="Do not pip install faster-qwen3-tts[demo]") + parser.add_argument("--skip-clone", action="store_true", + help="Do not clone the faster-qwen3-tts repo") + return parser + + +def detect() -> BackendStatus: + """Detect how far faster-qwen3-tts is set up, plus the launch command.""" + installed = _is_installed() + cloned = _is_cloned() + voices_json = _checkout() / "voices.json" + configured = installed and cloned and voices_json.exists() + details: List[str] = [] + details.append("pip: installed" if installed else + "not installed — run setup to pip install") + details.append(f"checkout: {_checkout()}" if cloned else + f"not cloned — run setup to clone ./{FASTER_DIR_NAME}") + details.append(f"voices: {voices_json}" if voices_json.exists() else + "no voices.json — run setup to create one") + launch = "" + if cloned and voices_json.exists(): + launch = (f"python {_checkout()}/examples/openai_server.py " + f"--voices {voices_json} --port {_config_port()}") + return BackendStatus("faster", "faster-qwen3-tts", + installed=installed and cloned, + configured=configured, details=details, + launch_hint=launch) + + +def _run_voices_only_tui() -> int: + """Rebuild voices.json via the TUI (the "modify" action). + + Runs the same wizard but skips the pip/clone prerequisites so it goes + straight to picking the .wav directory and writing voices.json. + """ + args = build_parser().parse_args([]) + args.skip_install = True + args.skip_clone = True + return run_tui(args) + + +modify_actions: List[ModifyAction] = [ + ModifyAction("Rebuild voices.json", _run_voices_only_tui), + ModifyAction("Reconfigure faster-qwen3-tts", run_tui), +] + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + + if _interactive(): + return run_tui(args) + + settings = _collect_from_flags(args, parser) + if settings is None: + return 1 + return _execute(settings) + + +def _interactive() -> bool: + try: + import curses # noqa: F401 + except ImportError: + return False + try: + return sys.stdin.isatty() and sys.stdout.isatty() + except (AttributeError, ValueError): + return False + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backends/qwen.py b/backends/qwen.py new file mode 100644 index 0000000..48e1804 --- /dev/null +++ b/backends/qwen.py @@ -0,0 +1,254 @@ +#!/usr/bin/env python3 +"""Set up the Qwen3-TTS demo backend for the audiobook generator. + +qwen-tts is a pip package providing the ``qwen-tts-demo`` server, which +hosts the Qwen3-TTS CustomVoice (built-in speakers) and Base (voice +cloning) models on separate ports. This module sets it up end-to-end as a +TUI: pip-install the package, configure the two ports and the built-in +speaker in ``converter/config.py``, and print the launch commands. It is +driven by ``audiobook.py``'s hub but can also be run directly with flags. + +Usage: + python -m backends.qwen [--port-custom PORT] [--port-clone PORT] + [--speaker NAME] [--skip-install] +""" + +import argparse +import importlib.util +import shutil +import sys +from pathlib import Path +from typing import List, Optional + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import tui +from backends import BackendStatus, ModifyAction +from backends import common +from converter import config + +QWEN_PIP_PKG = "qwen-tts" +QWEN_CUSTOMVOICE_MODEL = "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice" +QWEN_BASE_MODEL = "Qwen/Qwen3-TTS-12Hz-1.7B-Base" +DEFAULT_CUSTOM_PORT = 7860 +DEFAULT_CLONE_PORT = 7861 + +# Built-in CustomVoice speakers (see converter/config.py SPEAKER). +QWEN_SPEAKERS = ("Vivian", "Serena", "Uncle_Fu", "Dylan", "Eric", "Ryan", + "Aiden", "Ono_Anna", "Sohee") + + +def _is_installed() -> bool: + if shutil.which("qwen-tts-demo"): + return True + return importlib.util.find_spec("qwen_tts") is not None + + +def _config_port(url: str, fallback: int) -> int: + import urllib.parse + try: + return urllib.parse.urlsplit(url).port or fallback + except ValueError: + return fallback + + +def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: + """Linear TUI wizard collecting every qwen-setup decision.""" + _GO_BACK = object() + + def confirm(question: str, default: bool = True) -> Optional[bool]: + res = tui.confirm(stdscr, question, default=default, + cancel_value=_GO_BACK) + return None if res is _GO_BACK else res + + # Step 0: pip install (if not installed and not skipped). + do_install = False + if not _is_installed() and not args.skip_install: + choice = confirm("qwen-tts is not installed. pip install it now?", + default=True) + if choice is None: + return None + do_install = choice + + # Step 1: ports. + custom_port = args.port_custom + if custom_port is None: + port_text = tui.line_edit( + stdscr, "CustomVoice (built-in speaker) port", + str(_config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT)), + validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535) + else "Enter a port number between 1 and 65535", + help_lines=["The port for qwen-tts-demo CustomVoice (speaker mode)"]) + custom_port = int(port_text) + clone_port = args.port_clone + if clone_port is None: + port_text = tui.line_edit( + stdscr, "Base (voice clone) port", + str(_config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT)), + validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535) + else "Enter a port number between 1 and 65535", + help_lines=["The port for qwen-tts-demo Base (voice cloning)"]) + clone_port = int(port_text) + + # Step 2: built-in speaker. + speaker = args.speaker + if speaker is None: + speaker = tui.menu( + stdscr, "Built-in CustomVoice speaker", + [(s, s) for s in QWEN_SPEAKERS], + default_index=max(0, QWEN_SPEAKERS.index(config.SPEAKER) + if config.SPEAKER in QWEN_SPEAKERS else 0), + help_lines=["Used by audiobook.py --backend qwen without --clone"]) + + return { + "do_install": do_install, + "custom_port": custom_port, + "clone_port": clone_port, + "speaker": speaker, + } + + +def _execute(settings: dict) -> int: + """Console tail: install, sync config, advise.""" + if settings["do_install"]: + rc = common.pip_install([QWEN_PIP_PKG]) + if rc != 0: + print(f"[WARNING] pip install failed (exit {rc}); install " + f"{QWEN_PIP_PKG} manually") + else: + print(f"[OK] {QWEN_PIP_PKG} installed") + + custom_url = common.url_with_port(config.QWEN_API_URL, settings["custom_port"]) + if custom_url != config.QWEN_API_URL: + if common.update_config_value("QWEN_API_URL", custom_url): + print(f"[OK] Updated QWEN_API_URL to {custom_url}") + else: + print("[WARNING] Could not update QWEN_API_URL; edit " + "converter/config.py by hand") + clone_url = common.url_with_port(config.CLONE_API_URL, settings["clone_port"]) + if clone_url != config.CLONE_API_URL: + if common.update_config_value("CLONE_API_URL", clone_url): + print(f"[OK] Updated CLONE_API_URL to {clone_url}") + else: + print("[WARNING] Could not update CLONE_API_URL; edit " + "converter/config.py by hand") + if settings["speaker"] != config.SPEAKER: + if common.update_config_value("SPEAKER", settings["speaker"]): + print(f"[OK] Updated SPEAKER to {settings['speaker']}") + else: + print("[WARNING] Could not update SPEAKER; edit " + "converter/config.py by hand") + + _print_launch_hint(settings["custom_port"], settings["clone_port"]) + return 0 + + +def _print_launch_hint(custom_port: int, clone_port: int) -> None: + print() + print("Start the servers (in separate terminals):") + print(f" qwen-tts-demo {QWEN_CUSTOMVOICE_MODEL} --ip 127.0.0.1 " + f"--port {custom_port}") + print(f" qwen-tts-demo {QWEN_BASE_MODEL} --ip 127.0.0.1 " + f"--port {clone_port}") + print("Then run: python audiobook.py --backend qwen") + + +def run_tui(args: Optional[argparse.Namespace] = None) -> int: + """Run the qwen setup wizard end-to-end.""" + import curses + if args is None: + args = build_parser().parse_args([]) + try: + settings = curses.wrapper(_wizard, args) + except tui.WizardCancelled: + print("\n[INFO] Cancelled; nothing was written") + return 1 + try: + curses.curs_set(1) + except curses.error: + pass + if settings is None: + print("[INFO] Aborted") + return 1 + return _execute(settings) + + +def _collect_from_flags(args: argparse.Namespace, + parser: argparse.ArgumentParser) -> dict: + return { + "do_install": (not _is_installed()) and not args.skip_install, + "custom_port": args.port_custom if args.port_custom is not None + else _config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT), + "clone_port": args.port_clone if args.port_clone is not None + else _config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT), + "speaker": args.speaker or config.SPEAKER, + } + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Set up the Qwen3-TTS demo backend: pip install, " + "configure ports/speaker, and print launch commands.") + parser.add_argument("--port-custom", type=int, default=None, + help="CustomVoice (speaker) port (default: " + f"{DEFAULT_CUSTOM_PORT})") + parser.add_argument("--port-clone", type=int, default=None, + help="Base (voice clone) port (default: " + f"{DEFAULT_CLONE_PORT})") + parser.add_argument("--speaker", type=str, default=None, + choices=QWEN_SPEAKERS, + help="Built-in CustomVoice speaker (default: " + f"{config.SPEAKER})") + parser.add_argument("--skip-install", action="store_true", + help="Do not pip install qwen-tts") + return parser + + +def detect() -> BackendStatus: + """Detect whether qwen-tts is installed, plus the launch commands.""" + installed = _is_installed() + custom_port = _config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT) + clone_port = _config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT) + details: List[str] = [] + details.append("pip: installed" if installed else + "not installed — run setup to pip install qwen-tts") + details.append(f"CustomVoice port: {custom_port}") + details.append(f"Base (clone) port: {clone_port}") + details.append(f"speaker: {config.SPEAKER}") + launch = (f"qwen-tts-demo {QWEN_CUSTOMVOICE_MODEL} --ip 127.0.0.1 " + f"--port {custom_port} ; qwen-tts-demo {QWEN_BASE_MODEL} " + f"--ip 127.0.0.1 --port {clone_port}") + return BackendStatus("qwen", "Qwen3-TTS (demo server)", + installed=installed, configured=installed, + details=details, launch_hint=launch) + + +modify_actions: List[ModifyAction] = [ + ModifyAction("Reconfigure Qwen3-TTS (ports/speaker)", run_tui), +] + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + + if _interactive(): + return run_tui(args) + + settings = _collect_from_flags(args, parser) + return _execute(settings) + + +def _interactive() -> bool: + try: + import curses # noqa: F401 + except ImportError: + return False + try: + return sys.stdin.isatty() and sys.stdout.isatty() + except (AttributeError, ValueError): + return False + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/docs/backend-faster.md b/docs/backend-faster.md index 83614c7..40b10f7 100644 --- a/docs/backend-faster.md +++ b/docs/backend-faster.md @@ -10,16 +10,16 @@ pip install -U qwen-tts pip install "faster-qwen3-tts[demo]" ``` -**This backend always uses voice cloning**. The reference voice and language are configured on the **server**, not through the converter. The server does not transcribe reference audio itself, so do it manually or use the `tools/make_faster_voices_json.py` helper (see below). +**This backend always uses voice cloning**. The reference voice and language are configured on the **server**, not through the converter. The server does not transcribe reference audio itself, so do it manually or use the `backends.faster` setup wizard (see below). -The pip package does not include the server script, so clone the repository: +The pip package does not include the server script, so clone the repository (the `backends.faster` wizard does this for you into `./faster-qwen3-tts`): ```bash git clone https://github.com/andimarafioti/faster-qwen3-tts cd faster-qwen3-tts ``` -Create a `voices.json` mapping names to reference configurations (.wav to clone, transcript, language). Optionally run `python ./tools/make_faster_voices_json.py path/to/clone/wavs` to automatically create a `voices.json` using whisper to automatically transcribe the test audio. +Create a `voices.json` mapping names to reference configurations (.wav to clone, transcript, language). The TUI setup (`python audiobook.py` → **Set up a backend… → faster-qwen3-tts**, or `python -m backends.faster path/to/clone/wavs`) pip-installs the package, clones the repo, transcribes the `.wav` files with `whisper`, and writes `voices.json` for you. ```json { diff --git a/docs/backend-qwen.md b/docs/backend-qwen.md index 34078e3..b564149 100644 --- a/docs/backend-qwen.md +++ b/docs/backend-qwen.md @@ -1,5 +1,7 @@ # Backend Option 2: Qwen3-TTS +The TUI sets this up: run `python audiobook.py` → **Set up a backend… → Qwen3-TTS**, or `python -m backends.qwen`. It pip-installs `qwen-tts` and configures the two ports and built-in speaker in `converter/config.py`, then prints the launch commands. Manual steps: + Install qwen-tts with pip: ```bash @@ -0,0 +1,377 @@ +#!/usr/bin/env python3 +"""The TUI main menu for the audiobook generator (run via ``audiobook.py``). + +The hub is the single entry point for the whole workflow: it detects which +backends are already set up and offers to convert the input directory with +one of them, set up a new backend, or modify/reconfigure an existing one. +Each backend's setup wizard runs in its own curses session, so the hub +collects a "command" inside its own wrapper, returns to the plain terminal, +and then dispatches — no nested curses sessions. + +Esc on the main menu quits the hub. Esc inside a sub-menu falls back to the +main menu. +""" + +import json +import sys +from pathlib import Path +from typing import Optional + +import tui +import audiobook +from backends import REGISTRY, detect_all, get +from backends import audiocpp as audiocpp_backend +from backends import faster as faster_backend +from converter import config +from converter.converter import AUDIO_FORMATS +from converter.tts import AUDIOCPP_FAMILY_QWEN3_TTS, BACKEND_AUDIOCPP, \ + BACKEND_FASTER, BACKEND_QWEN + +_GO_BACK = object() + + +def run() -> int: + """Run the hub menu loop until the user quits. Returns exit code.""" + import curses + while True: + try: + command = curses.wrapper(_hub_menu) + except tui.WizardCancelled: + return 0 + except KeyboardInterrupt: + return 130 + if command is None: + return 0 + kind = command[0] + if kind == "quit": + return 0 + if kind == "setup": + info = get(command[1]) + if info is not None: + info.setup_tui() + elif kind == "modify": + info = get(command[1]) + if info is not None and command[2] < len(info.modify_actions): + info.modify_actions[command[2]].run() + elif kind == "convert": + _run_conversion(command[1], command[2]) + + +def _hub_menu(stdscr) -> Optional[tuple]: + """Show the main menu; return a command tuple, or None to quit.""" + while True: + statuses = detect_all() + summary = ["Backend status:"] + for st in statuses: + mark = "ready" if st.ready else ( + "installed" if st.installed else "not set up") + summary.append(f" {st.label}: {mark}") + choice = tui.menu( + stdscr, "tts-audiobook-generator", + [("Convert books...", "convert"), + ("Set up a backend...", "setup"), + ("Modify a backend...", "modify"), + ("Quit", "quit")], + help_lines=summary) + if choice is None or choice == "quit": + return None + if choice == "convert": + cmd = _convert_menu(stdscr, statuses) + if cmd is not None: + return cmd + elif choice == "setup": + cmd = _setup_menu(stdscr, statuses) + if cmd is not None: + return cmd + elif choice == "modify": + cmd = _modify_menu(stdscr, statuses) + if cmd is not None: + return cmd + + +def _setup_menu(stdscr, statuses) -> Optional[tuple]: + """Pick a backend to set up. Returns ("setup", key) or None to go back.""" + options = [(f"{info.label} ({_status_mark(info.key, statuses)})", + info.key) for info in REGISTRY] + choice = tui.menu(stdscr, "Set up a backend", options, + back_value=_GO_BACK, + help_lines=["Clone/build/install a backend so you can " + "convert with it."]) + if choice is _GO_BACK or choice is None: + return None + return ("setup", choice) + + +def _modify_menu(stdscr, statuses) -> Optional[tuple]: + """Pick an installed backend and one of its modify actions.""" + installed = [info for info in REGISTRY + if _status_mark(info.key, statuses) != "not set up"] + if not installed: + tui.flash(stdscr, "No backend is set up yet — use 'Set up a backend' first.") + return None + options = [(info.label, info.key) for info in installed] + key = tui.menu(stdscr, "Modify a backend", options, back_value=_GO_BACK) + if key is _GO_BACK or key is None: + return None + info = get(key) + actions = info.modify_actions + choice = tui.menu( + stdscr, f"Modify {info.label}", + [(action.label, index) for index, action in enumerate(actions)], + back_value=_GO_BACK) + if choice is _GO_BACK or choice is None: + return None + return ("modify", key, choice) + + +def _status_mark(key: str, statuses) -> str: + for st in statuses: + if st.key == key: + return "ready" if st.ready else ( + "installed" if st.installed else "not set up") + return "not set up" + + +def _convert_menu(stdscr, statuses) -> Optional[tuple]: + """Pick a ready backend and collect per-backend run settings.""" + ready = [st for st in statuses if st.ready] + options = [(f"{st.label}", st.key) for st in ready] + if not ready: + choice = tui.menu( + stdscr, "No backend is ready", + [("Set up a backend...", "__setup__")], + help_lines=["Set up a backend (clone/build/configure) before " + "converting."]) + if choice == "__setup__": + return _setup_menu(stdscr, statuses) + return None + options.append(("Set up a backend...", "__setup__")) + key = tui.menu(stdscr, "Convert books with...", options, + back_value=_GO_BACK) + if key is _GO_BACK or key is None: + return None + if key == "__setup__": + return _setup_menu(stdscr, statuses) + if key == BACKEND_AUDIOCPP: + return _convert_audiocpp(stdscr, statuses) + if key == BACKEND_QWEN: + return _convert_qwen(stdscr) + if key == BACKEND_FASTER: + return _convert_faster(stdscr) + return None + + +def _convert_audiocpp(stdscr, statuses) -> Optional[tuple]: + """Collect audio.cpp run settings by reading ./audio.cpp/server.json.""" + checkout = audiocpp_backend.find_local_checkout() + server_json = checkout / "server.json" if checkout else None + if not server_json or not server_json.exists(): + tui.flash(stdscr, "No server.json found in the audio.cpp checkout. " + "Run 'Set up a backend' first.") + return None + try: + data = json.loads(server_json.read_text(encoding="utf-8")) + except (OSError, ValueError): + tui.flash(stdscr, f"Could not read {server_json}.") + return None + models = data.get("models") or [] + if not models: + tui.flash(stdscr, "No model entries in server.json. Reconfigure " + "audio.cpp first.") + return None + model_options = [(f"{m.get('id')} ({m.get('family')}, {m.get('task', 'tts')})", + m.get("id")) for m in models] + model_id = tui.menu(stdscr, "Select the audio.cpp model to use", + model_options, back_value=_GO_BACK) + if model_id is _GO_BACK or model_id is None: + return None + entry = next((m for m in models if m.get("id") == model_id), {}) + family = entry.get("family") + task = entry.get("task", "tts") + + # Voice: optional for qwen3_tts (built-in speaker), required otherwise. + voice = None + voice_dir = data.get("voice_dir") + voices = _list_voices(voice_dir) if voice_dir else [] + if task == "vdes": + # Voice design: no voice, instructions required. + pass + elif family == AUDIOCPP_FAMILY_QWEN3_TTS: + # Speaker mode available; voice optional. + if voices: + opts = [("(built-in speaker)", None)] + [(v, v) for v in voices] + voice = tui.menu(stdscr, "Voice", opts, back_value=_GO_BACK) + if voice is _GO_BACK: + return None + else: + voice = None + else: + if not voices: + tui.flash(stdscr, f"This model needs a --voice but voice_dir " + f"{voice_dir} has no .wav voices. Reconfigure " + "audio.cpp or add voices.") + return None + voice = tui.menu(stdscr, "Select the voice to clone", [(v, v) for v in voices], + back_value=_GO_BACK) + if voice is _GO_BACK or voice is None: + return None + + # Instructions: required for vdes, optional otherwise. + instructions = None + if task == "vdes": + instructions = tui.line_edit( + stdscr, "Voice design instructions (required for this model)", + config.AUDIOCPP_INSTRUCTIONS, + validate=lambda s: None if s.strip() + else "Describe the voice, e.g. 'A warm female narrator'", + back_value=_GO_BACK) + if instructions is _GO_BACK: + return None + else: + instructions = tui.line_edit( + stdscr, "Style instructions (optional, blank for none)", + config.AUDIOCPP_INSTRUCTIONS, back_value=_GO_BACK) + if instructions is _GO_BACK: + return None + if not instructions.strip(): + instructions = None + + common_kw = _common_options(stdscr) + if common_kw is None: + return None + return ("convert", BACKEND_AUDIOCPP, { + "model_id": model_id, "voice": voice, "instructions": instructions, + **common_kw, + }) + + +def _convert_qwen(stdscr) -> Optional[tuple]: + """Collect qwen run settings: built-in speaker or clone a .wav.""" + mode = tui.menu( + stdscr, "Qwen3-TTS mode", + [("Custom voice (built-in speaker)", "custom"), + ("Voice clone from a .wav file", "clone")], + back_value=_GO_BACK, + help_lines=[f"Speaker: {config.SPEAKER} (change it via Modify Qwen)"]) + if mode is _GO_BACK or mode is None: + return None + clone = None + if mode == "clone": + clone = tui.line_edit( + stdscr, "Path to a reference .wav (10-15s is ideal)", + "", + validate=lambda s: None if (s and Path(s).is_file() + and s.lower().endswith(".wav")) + else "Enter the path to an existing .wav file", + back_value=_GO_BACK) + if clone is _GO_BACK: + return None + common_kw = _common_options(stdscr) + if common_kw is None: + return None + return ("convert", BACKEND_QWEN, {"clone": clone, **common_kw}) + + +def _convert_faster(stdscr) -> Optional[tuple]: + """Collect faster run settings: pick a voice from voices.json.""" + checkout = faster_backend._checkout() + voices_json = checkout / "voices.json" + if not voices_json.exists(): + tui.flash(stdscr, f"No voices.json at {voices_json}. Run 'Set up a " + "backend' for faster first.") + return None + try: + voices = json.loads(voices_json.read_text(encoding="utf-8")) + except (OSError, ValueError): + tui.flash(stdscr, f"Could not read {voices_json}.") + return None + if not voices: + tui.flash(stdscr, "voices.json has no voices. Reconfigure faster.") + return None + default = config.FASTER_VOICE if config.FASTER_VOICE in voices else \ + next(iter(voices)) + voice = tui.menu( + stdscr, "Select the voice to clone", + [(k, k) for k in voices], + default_index=list(voices).index(default), back_value=_GO_BACK) + if voice is _GO_BACK or voice is None: + return None + common_kw = _common_options(stdscr) + if common_kw is None: + return None + return ("convert", BACKEND_FASTER, {"voice": voice, **common_kw}) + + +def _common_options(stdscr) -> Optional[dict]: + """Collect output format, speed, single-file, chunk, debug.""" + fmt_options = [(f, f) for f in AUDIO_FORMATS] + fmt_default = AUDIO_FORMATS.index(config.AUDIO_FORMAT) \ + if config.AUDIO_FORMAT in AUDIO_FORMATS else 0 + output_format = tui.menu(stdscr, "Output format", fmt_options, + default_index=fmt_default, back_value=_GO_BACK) + if output_format is _GO_BACK or output_format is None: + return None + speed_text = tui.line_edit( + stdscr, "Playback speed (1.0 = normal)", "1.0", + validate=lambda s: None if (_is_float(s) and float(s) > 0) + else "Enter a positive number, e.g. 1.0", + back_value=_GO_BACK) + if speed_text is _GO_BACK: + return None + single_file = tui.confirm(stdscr, "Combine all chapters into one file?", + default=False, cancel_value=_GO_BACK) + if single_file is _GO_BACK: + return None + chunk = tui.confirm(stdscr, "Force client-side chunking (--chunk)?", + default=False, cancel_value=_GO_BACK) + if chunk is _GO_BACK: + return None + debug = tui.confirm(stdscr, "Debug mode (dump per-chunk audio/text)?", + default=False, cancel_value=_GO_BACK) + if debug is _GO_BACK: + return None + return { + "output_format": output_format, + "speed": float(speed_text), + "single_file": single_file, + "chunk": chunk, + "debug": debug, + } + + +def _run_conversion(backend: str, kwargs: dict) -> None: + """Run a conversion in the plain console (after the TUI returns).""" + status = next((s for s in detect_all() if s.key == backend), None) + if status is not None and not status.ready: + print(f"[WARNING] {status.label} is not fully set up.") + if status is not None and status.launch_hint: + print("[INFO] Make sure the server is running. Start it with:") + print(f" {status.launch_hint}") + audiobook.convert(backend=backend, **kwargs) + + +def _list_voices(voice_dir: str) -> list: + """Return sorted .wav stems in VOICE_DIR (best-effort).""" + try: + path = Path(voice_dir) + if not path.is_dir(): + return [] + return sorted( + (p.stem for p in path.iterdir() + if p.is_file() and p.suffix.lower() == ".wav"), + key=str.lower, + ) + except OSError: + return [] + + +def _is_float(value: str) -> bool: + try: + float(value) + return True + except ValueError: + return False + + +if __name__ == "__main__": + sys.exit(run()) diff --git a/requirements.txt b/requirements.txt index 1747be4..4e23dcf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,7 +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 +# windows-curses>=2.3 # Windows only: enables the TUI (audiobook.py hub + backends.* wizards) # Audio processing # Note: ffmpeg is required to concatenate and encode the final audiobook. diff --git a/tests/test_backends.py b/tests/test_backends.py new file mode 100644 index 0000000..4017cd4 --- /dev/null +++ b/tests/test_backends.py @@ -0,0 +1,91 @@ +"""Tests for the backends package registry and detection aggregation.""" + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from backends import REGISTRY, detect_all, get + + +class RegistryTests(unittest.TestCase): + def test_registry_has_the_three_backends(self): + keys = [info.key for info in REGISTRY] + self.assertEqual(keys, ["audiocpp", "qwen", "faster"]) + + def test_every_entry_has_detect_and_setup_tui(self): + for info in REGISTRY: + self.assertTrue(callable(info.detect), info.key) + self.assertTrue(callable(info.setup_tui), info.key) + self.assertIsInstance(info.modify_actions, list) + for action in info.modify_actions: + self.assertTrue(callable(action.run)) + + def test_get_returns_entry_by_key(self): + self.assertIs(get("audiocpp").key, "audiocpp") + self.assertIsNone(get("nonexistent")) + + +class DetectAllTests(unittest.TestCase): + def test_detect_all_returns_one_status_per_backend(self): + statuses = detect_all() + self.assertEqual([s.key for s in statuses], + ["audiocpp", "qwen", "faster"]) + for s in statuses: + self.assertIn(s.key, ("audiocpp", "qwen", "faster")) + # ready requires both installed and configured; on a clean + # machine none are ready. + if s.ready: + self.assertTrue(s.installed and s.configured) + + def test_audiocpp_status_when_cloned_built_configured(self): + with tempfile.TemporaryDirectory() as td: + root = Path(td) + checkout = root / "audio.cpp" + checkout.mkdir() + (checkout / "model_specs").mkdir() + (checkout / "build" / "linux-cuda-release" / "bin").mkdir( + parents=True) + (checkout / "build" / "linux-cuda-release" / "bin" + / "audiocpp_server").write_bytes(b"x") + (checkout / "server.json").write_text('{"models":[]}', + encoding="utf-8") + from backends import audiocpp + with patch.object(audiocpp, "find_local_checkout", + return_value=checkout): + status = audiocpp.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertTrue(status.ready) + self.assertIn("audiocpp_server", status.launch_hint) + + def test_qwen_status_reflects_install(self): + from backends import qwen + with patch.object(qwen, "_is_installed", return_value=True): + status = qwen.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertIn("qwen-tts-demo", status.launch_hint) + with patch.object(qwen, "_is_installed", return_value=False): + status = qwen.detect() + self.assertFalse(status.installed) + self.assertFalse(status.configured) + + def test_faster_status_reflects_install_clone_voices(self): + from backends import faster + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) / "faster-qwen3-tts" + (checkout / "examples").mkdir(parents=True) + (checkout / "examples" / "openai_server.py").write_text("x") + (checkout / "voices.json").write_text('{"default":{}}', + encoding="utf-8") + with patch.object(faster, "_is_installed", return_value=True), \ + patch.object(faster, "_checkout", return_value=checkout): + status = faster.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertIn("openai_server.py", status.launch_hint) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_backends_audiocpp.py b/tests/test_backends_audiocpp.py new file mode 100644 index 0000000..9882ce1 --- /dev/null +++ b/tests/test_backends_audiocpp.py @@ -0,0 +1,1062 @@ +"""Tests for the audio.cpp backend setup module (backends/audiocpp.py).""" + +import io +import json +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest.mock import MagicMock, patch + +from converter import config +from backends import audiocpp as make_server + +FAKE_CONFIG = ( + 'LANGUAGE = "English"\n' + "\n" + 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' + "\n" + "CHUNK_SIZE = 250\n" +) + +FAKE_CONFIG_WITH_MODEL_IDS = ( + 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' + "\n" + 'AUDIOCPP_MODEL_ID = "qwen" # server entry for speaker mode\n' + 'AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"\n' +) + + +def _write_spec(checkout: Path, family: str, *, display_name=None, + tasks=("tts", "clone"), languages=("en",), packages=None, + category="tts"): + """Write a minimal model_specs/<family>.json into a fake checkout.""" + specs = checkout / "model_specs" + specs.mkdir(parents=True, exist_ok=True) + if packages is None: + packages = [{ + "id": f"{family}_q8_0", "default": True, "format": "gguf", + "target_directory": f"{family}-GGUF", + }] + spec = { + "family": family, + "display_name": display_name or family, + "category": category, + "tasks": list(tasks), + "languages": list(languages), + "packages": packages, + } + (specs / f"{family}.json").write_text(json.dumps(spec), encoding="utf-8") + return spec + + +def _make_checkout(tmp: Path) -> Path: + """Create a fake audio.cpp checkout with a realistic model_specs set.""" + checkout = tmp / "audio.cpp" + checkout.mkdir() + _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"}, + {"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=[{ + "id": "higgs_audio_tts_4b_q8_0", "default": True, + "format": "gguf", + "target_directory": "Higgs-Audio-v3-TTS-4B-GGUF", + }]) + _write_spec(checkout, "voxcpm2", display_name="VoxCPM2-2B", + languages=("en", "zh"), + packages=[{ + "id": "voxcpm2_q8_0", "default": True, "format": "gguf", + "target_directory": "VoxCPM2-GGUF", + }]) + _write_spec(checkout, "index_tts2", display_name="IndexTTS-2", + languages=("zh", "en"), + packages=[{ + "id": "index_tts2_q8_0", "default": True, "format": "gguf", + "target_directory": "IndexTTS2-GGUF", + }]) + _write_spec(checkout, "pocket_tts", display_name="PocketTTS-100M", + tasks=("tts", "clone"), languages=("en", "de"), + packages=[{ + "id": "pocket_tts_q8_0", "default": True, "format": "gguf", + "target_directory": "PocketTTS-GGUF", + }]) + _write_spec(checkout, "supertonic", display_name="Supertonic 3", + tasks=("tts",), languages=("en", "ko"), + packages=[{ + "id": "supertonic_q8_0", "default": True, "format": "gguf", + "target_directory": "Supertonic-GGUF", + }]) + # An ASR family that must be filtered out. + _write_spec(checkout, "qwen3_asr", display_name="Qwen3-ASR", + tasks=("asr",), category="asr") + # A TTS family with no installable packages (must be skipped). + _write_spec(checkout, "empty_tts", display_name="Empty TTS", + tasks=("tts",), packages=[]) + return checkout + + +class FindWavFilesTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _touch(self, name): + path = self.folder / name + path.write_bytes(b"x") + return path + + def test_finds_only_wavs_case_insensitive(self): + self._touch("b.wav") + self._touch("a.WAV") + self._touch("notes.txt") + (self.folder / "sub").mkdir() + (self.folder / "sub" / "c.wav").write_bytes(b"x") + names = [path.name for path in make_server.find_wav_files(self.folder)] + self.assertEqual(names, ["a.WAV", "b.wav"]) + + def test_sorted_alphabetically_case_insensitive(self): + for name in ("Zed.wav", "alpha.wav", "Beta.wav"): + self._touch(name) + names = [path.name for path in make_server.find_wav_files(self.folder)] + self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) + + def test_empty_directory_returns_empty_list(self): + self.assertEqual(make_server.find_wav_files(self.folder), []) + + +class DetectWavDirTests(unittest.TestCase): + """Shallow .wav-directory discovery across the two checkout roots.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + self.audiocpp = self.root / "audio.cpp" + self.tts_root = self.root / "tts-audiobook-generator" + self.audiocpp.mkdir() + self.tts_root.mkdir() + + def tearDown(self): + self._td.cleanup() + + def _wav_dir(self, where, name="voices"): + directory = where / name + directory.mkdir(parents=True, exist_ok=True) + (directory / "voice.wav").write_bytes(b"x") + return directory + + def test_unique_wav_dir_in_tts_root_returned(self): + found = self._wav_dir(self.tts_root, "voices") + self.assertEqual(make_server.detect_wav_dir(self.audiocpp, + self.tts_root), + found) + + def test_unique_wav_dir_in_audiocpp_root_returned(self): + found = self._wav_dir(self.audiocpp, "reference") + self.assertEqual(make_server.detect_wav_dir(self.audiocpp, + self.tts_root), + found) + + def test_root_itself_containing_wavs_returned(self): + (self.tts_root / "direct.wav").write_bytes(b"x") + self.assertEqual(make_server.detect_wav_dir(self.audiocpp, + self.tts_root), + self.tts_root) + + def test_multiple_wav_dirs_returns_none(self): + self._wav_dir(self.tts_root, "one") + self._wav_dir(self.audiocpp, "two") + self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.tts_root)) + + def test_output_dir_of_tts_root_excluded(self): + self._wav_dir(self.tts_root, "output") + self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.tts_root)) + + def test_no_wavs_returns_none(self): + self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.tts_root)) + + def test_nested_wav_dir_not_seen(self): + nested = self.tts_root / "outer" / "inner" + nested.mkdir(parents=True) + (nested / "voice.wav").write_bytes(b"x") + self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.tts_root)) + + +class ConfigPortTests(unittest.TestCase): + def test_port_parsed_from_config_url(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:8080"): + self.assertEqual(make_server.config_port(), 8080) + + def test_missing_port_falls_back(self): + with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1"): + self.assertEqual(make_server.config_port(), + make_server.FALLBACK_PORT) + + def test_invalid_url_falls_back(self): + with patch.object(config, "AUDIOCPP_API_URL", "not a url"): + self.assertEqual(make_server.config_port(), + make_server.FALLBACK_PORT) + + def test_url_with_port_replaces_port(self): + self.assertEqual( + make_server._url_with_port("http://127.0.0.1:8080", 9000), + "http://127.0.0.1:9000") + + def test_url_without_port_adds_port(self): + self.assertEqual( + make_server._url_with_port("http://localhost", 8080), + "http://localhost:8080") + + +class UpdateConfigPortTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.config_path = Path(self._tmp.name) / "config.py" + self.config_path.write_text(FAKE_CONFIG, encoding="utf-8") + + def tearDown(self): + self._tmp.cleanup() + + def test_rewrites_port_preserving_comment(self): + changed = make_server.update_config_api_url_port( + 8080, config_path=self.config_path) + self.assertTrue(changed) + text = self.config_path.read_text(encoding="utf-8") + self.assertIn( + 'AUDIOCPP_API_URL = "http://127.0.0.1:8080" # audio.cpp audiocpp_server', + text) + self.assertIn('LANGUAGE = "English"', text) + self.assertIn("CHUNK_SIZE = 250", text) + + def test_returns_false_when_no_url_line(self): + path = Path(self._tmp.name) / "other.py" + path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") + self.assertFalse(make_server.update_config_api_url_port( + 8080, config_path=path)) + + def test_returns_false_when_port_unchanged(self): + self.assertFalse(make_server.update_config_api_url_port( + 9999, config_path=self.config_path)) + self.assertEqual(self.config_path.read_text(encoding="utf-8"), + FAKE_CONFIG) + + def test_returns_false_when_file_missing(self): + self.assertFalse(make_server.update_config_api_url_port( + 8080, config_path=Path(self._tmp.name) / "nope.py")) + + +class UpdateConfigModelIdsTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.config_path = Path(self._tmp.name) / "config.py" + self.config_path.write_text(FAKE_CONFIG_WITH_MODEL_IDS, + encoding="utf-8") + + def tearDown(self): + self._tmp.cleanup() + + def test_rewrites_both_ids_preserving_lines(self): + changed = make_server.update_config_model_ids( + "higgs", "higgs", config_path=self.config_path) + self.assertTrue(changed) + text = self.config_path.read_text(encoding="utf-8") + self.assertIn('AUDIOCPP_MODEL_ID = "higgs" # server entry for speaker mode', + text) + self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "higgs"', text) + self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:9999"', text) + + def test_clone_id_optional(self): + changed = make_server.update_config_model_ids( + "voxcpm2", config_path=self.config_path) + self.assertTrue(changed) + text = self.config_path.read_text(encoding="utf-8") + self.assertIn('AUDIOCPP_MODEL_ID = "voxcpm2"', text) + self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"', text) + + def test_returns_false_when_ids_unchanged(self): + changed = make_server.update_config_model_ids( + "qwen", "qwen-clone", config_path=self.config_path) + self.assertFalse(changed) + self.assertEqual(self.config_path.read_text(encoding="utf-8"), + FAKE_CONFIG_WITH_MODEL_IDS) + + def test_returns_false_when_lines_missing(self): + path = Path(self._tmp.name) / "other.py" + path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") + self.assertFalse(make_server.update_config_model_ids( + "higgs", "higgs", config_path=path)) + + def test_returns_false_when_file_missing(self): + self.assertFalse(make_server.update_config_model_ids( + "higgs", "higgs", + config_path=Path(self._tmp.name) / "nope.py")) + + +class ResolveWavDirArgTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_resolves_to_absolute(self): + self.assertEqual(make_server.resolve_wav_dir_arg(str(self.folder)), + self.folder.resolve()) + + def test_strips_surrounding_quotes(self): + quoted = f'"{self.folder}"' + self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.folder.resolve()) + + def test_strips_single_quotes(self): + quoted = f"'{self.folder}'" + self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.folder.resolve()) + + def test_strips_whitespace(self): + self.assertEqual(make_server.resolve_wav_dir_arg(f" {self.folder} "), + self.folder.resolve()) + + def test_expands_tilde(self): + with patch.object(make_server.os.path, "expanduser", + return_value=str(self.folder)) as mock_expand: + result = make_server.resolve_wav_dir_arg("~/voices") + mock_expand.assert_called_once_with("~/voices") + self.assertEqual(result, self.folder.resolve()) + + def test_trailing_slash_preserved_as_dir(self): + self.assertEqual(make_server.resolve_wav_dir_arg(f"{self.folder}/"), + self.folder.resolve()) + + +class 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 CheckoutAutoSelectTests(unittest.TestCase): + """TUI browser auto-accept callback for an audio.cpp checkout.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + + def tearDown(self): + self._td.cleanup() + + def test_accepts_audio_cpp_containing_model_specs(self): + checkout = self.root / "audio.cpp" + checkout.mkdir() + (checkout / "model_specs").mkdir() + self.assertEqual(make_server._checkout_auto_select(checkout), + checkout) + + def test_rejects_audio_cpp_without_model_specs(self): + checkout = self.root / "audio.cpp" + checkout.mkdir() + self.assertIsNone(make_server._checkout_auto_select(checkout)) + + def test_rejects_other_name_even_with_model_specs(self): + other = self.root / "not-audiocpp" + other.mkdir() + (other / "model_specs").mkdir() + self.assertIsNone(make_server._checkout_auto_select(other)) + + def test_rejects_plain_directory(self): + plain = self.root / "somewhere" + plain.mkdir() + self.assertIsNone(make_server._checkout_auto_select(plain)) + + +class DefaultModelIdTests(unittest.TestCase): + def test_preferred_ids_for_tested_families(self): + self.assertEqual(make_server.default_model_id("qwen3_tts"), "qwen") + self.assertEqual(make_server.default_model_id("higgs_audio_tts"), "higgs") + self.assertEqual(make_server.default_model_id("voxcpm2"), "voxcpm2") + self.assertEqual(make_server.default_model_id("index_tts2"), "indextts2") + + def test_derived_id_strips_trailing_tts_and_underscores(self): + self.assertEqual(make_server.default_model_id("pocket_tts"), "pocket") + self.assertEqual(make_server.default_model_id("dots_tts"), "dots") + self.assertEqual(make_server.default_model_id("moss_tts_local"), + "mossttslocal") + + +class LoadModelCatalogTests(unittest.TestCase): + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = _make_checkout(Path(self._td.name)) + + def tearDown(self): + self._td.cleanup() + + def test_includes_tts_families_excludes_asr(self): + catalog = make_server.load_model_catalog(self.checkout) + families = [entry["family"] for entry in catalog] + self.assertIn("qwen3_tts", families) + self.assertIn("higgs_audio_tts", families) + self.assertIn("pocket_tts", families) + self.assertIn("supertonic", families) + self.assertNotIn("qwen3_asr", families) + + def test_skips_families_with_no_packages(self): + catalog = make_server.load_model_catalog(self.checkout) + self.assertNotIn("empty_tts", + [entry["family"] for entry in catalog]) + + def test_families_sorted_alphabetically_by_display_name(self): + catalog = make_server.load_model_catalog(self.checkout) + names = [entry["display_name"].lower() for entry in catalog] + self.assertEqual(names, sorted(names)) + self.assertNotIn("tested", catalog[0]) + self.assertNotIn("TESTED_FAMILIES", dir(make_server)) + + def test_default_package_and_target_directory_resolved(self): + catalog = make_server.load_model_catalog(self.checkout) + by_family = {entry["family"]: entry for entry in catalog} + higgs = by_family["higgs_audio_tts"] + self.assertEqual(higgs["install_id"], "higgs_audio_tts_4b_q8_0") + self.assertEqual(higgs["default_path"], + "models/Higgs-Audio-v3-TTS-4B-GGUF") + + def test_picks_first_gguf_when_no_default_flag(self): + _write_spec(self.checkout, "voxcpm2", display_name="VoxCPM2-2B", + packages=[ + {"id": "voxcpm2_bf16", "format": "gguf", + "target_directory": "VoxCPM2-GGUF"}, + {"id": "voxcpm2_q8_0", "format": "gguf", + "target_directory": "VoxCPM2-GGUF"}, + ]) + catalog = make_server.load_model_catalog(self.checkout) + by_family = {entry["family"]: entry for entry in catalog} + self.assertEqual(by_family["voxcpm2"]["install_id"], "voxcpm2_bf16") + + def test_clone_capability_from_tasks(self): + catalog = make_server.load_model_catalog(self.checkout) + by_family = {entry["family"]: entry for entry in catalog} + self.assertTrue(by_family["higgs_audio_tts"]["clone_capable"]) + self.assertFalse(by_family["supertonic"]["clone_capable"]) + + def test_missing_model_specs_dir_raises(self): + empty = Path(self._td.name) / "empty" + empty.mkdir() + with self.assertRaises(NotADirectoryError): + make_server.load_model_catalog(empty) + + +class DetectBackendTests(unittest.TestCase): + """Backend detection from audio.cpp build directory names.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = Path(self._td.name) / "audio.cpp" + self.checkout.mkdir() + + def tearDown(self): + self._td.cleanup() + + def _build(self, name, binary="audiocpp_server"): + build_dir = self.checkout / "build" / name + bin_dir = build_dir / "bin" + bin_dir.mkdir(parents=True) + (bin_dir / binary).write_bytes(b"x") + return build_dir + + def test_no_build_dir_returns_none(self): + self.assertIsNone(make_server.detect_backend(self.checkout)) + + def test_unique_linux_backend_detected(self): + self._build("linux-cuda-release") + self.assertEqual(make_server.detect_backend(self.checkout), "cuda") + + def test_windows_exe_backend_detected(self): + self._build("windows-vulkan-debug", binary="audiocpp_server.exe") + self.assertEqual(make_server.detect_backend(self.checkout), "vulkan") + + def test_hip_backend_detected(self): + self._build("linux-hip-release") + self.assertEqual(make_server.detect_backend(self.checkout), "hip") + + def test_cpu_backend_detected(self): + self._build("linux-cpu-release") + self.assertEqual(make_server.detect_backend(self.checkout), "cpu") + + def test_metal_maps_to_cpu(self): + self._build("macos-metal-release") + self.assertEqual(make_server.detect_backend(self.checkout), "cpu") + + def test_multiple_backends_returns_none(self): + self._build("linux-cuda-release") + self._build("linux-cpu-release") + self.assertIsNone(make_server.detect_backend(self.checkout)) + + def test_multiple_builds_same_backend_detected(self): + self._build("linux-cuda-release") + self._build("windows-cuda-debug") + self.assertEqual(make_server.detect_backend(self.checkout), "cuda") + + def test_build_dir_without_binary_ignored(self): + (self.checkout / "build" / "linux-cuda-release").mkdir(parents=True) + self.assertIsNone(make_server.detect_backend(self.checkout)) + + def test_non_matching_build_dir_name_ignored(self): + self._build("linux-mybuild-release") + self.assertIsNone(make_server.detect_backend(self.checkout)) + + +class BackendOptionsTests(unittest.TestCase): + """Aligned backend menu labels and the [auto-detected] default.""" + + def test_options_have_aligned_dashes(self): + options, default_index = make_server._backend_options() + dash_columns = {label.index(" - ") for label, _ in options} + self.assertEqual(len(dash_columns), 1) + self.assertEqual(default_index, 0) + + def test_detected_backend_marked_and_defaulted(self): + options, default_index = make_server._backend_options("vulkan") + labels = [label for label, _ in options] + self.assertEqual(default_index, labels.index(next( + label for label, value in options + if value == "vulkan" and label.endswith("[auto-detected]")))) + self.assertTrue(labels[default_index].endswith("[auto-detected]")) + self.assertEqual(options[default_index][1], "vulkan") + + def test_unknown_detected_backend_is_ignored(self): + options, default_index = make_server._backend_options("opencl") + self.assertEqual(default_index, 0) + self.assertFalse(any("[auto-detected]" in label + for label, _ in options)) + + def test_labels_keep_backend_values(self): + options, _ = make_server._backend_options() + self.assertEqual([value for _, value in options], + list(make_server.BACKENDS)) + + +class BuildServerConfigTests(unittest.TestCase): + def test_single_entry_without_voice_dir(self): + entry = make_server.build_model_entry( + "higgs_audio_tts", "higgs", "models/Higgs-GGUF") + cfg = make_server.build_server_config( + "127.0.0.1", 8080, "cuda", False, [entry]) + self.assertEqual(cfg["host"], "127.0.0.1") + self.assertEqual(cfg["port"], 8080) + self.assertEqual(cfg["backend"], "cuda") + self.assertFalse(cfg["lazy_load"]) + self.assertEqual(cfg["models"], [entry]) + self.assertNotIn("voice_dir", cfg) + + def test_voice_dir_added_when_given(self): + entry = make_server.build_model_entry("voxcpm2", "voxcpm2", "models/V") + cfg = make_server.build_server_config( + "0.0.0.0", 9000, "cpu", True, [entry], + voice_dir="/abs/voices") + self.assertTrue(cfg["lazy_load"]) + self.assertEqual(cfg["voice_dir"], "/abs/voices") + + def test_model_entry_shape(self): + entry = make_server.build_model_entry("index_tts2", "indextts2", "p") + self.assertEqual(entry["id"], "indextts2") + self.assertEqual(entry["family"], "index_tts2") + self.assertEqual(entry["path"], "p") + 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 InstallModelsTests(unittest.TestCase): + """Printing or auto-running the model install commands.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = Path(self._td.name) / "audio.cpp" + self.checkout.mkdir() + self.manager = self.checkout / "tools" / "model_manager_v2.py" + self.manager.parent.mkdir() + self.manager.write_text("#!/usr/bin/env python3\n", encoding="utf-8") + self.guidance = [("Higgs Audio v3 TTS 4B", "higgs_audio_tts_4b_q8_0"), + ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0"), + ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0")] + + def tearDown(self): + self._td.cleanup() + + def test_declined_download_prints_commands_deduped(self): + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(make_server.subprocess, "run") as run: + make_server._install_models(self.checkout, self.guidance, + download=False) + out = buf.getvalue() + self.assertEqual(out.count("install higgs_audio_tts_4b_q8_0"), 1) + self.assertEqual(out.count("install qwen3_tts_1_7b_base_q8_0"), 1) + run.assert_not_called() + + def test_accepted_download_runs_each_command(self): + with patch.object(make_server.subprocess, "run", + return_value=MagicMock(returncode=0)) as run: + make_server._install_models(self.checkout, self.guidance, + download=True) + self.assertEqual(run.call_count, 2) + commands = [call[0][0] for call in run.call_args_list] + self.assertEqual(commands[0], + [sys.executable, str(self.manager), "install", + "higgs_audio_tts_4b_q8_0"]) + self.assertEqual(commands[1], + [sys.executable, str(self.manager), "install", + "qwen3_tts_1_7b_base_q8_0"]) + for call in run.call_args_list: + self.assertEqual(call[1]["cwd"], str(self.checkout)) + + def test_missing_manager_falls_back_to_printing(self): + self.manager.unlink() + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(make_server.subprocess, "run") as run: + make_server._install_models(self.checkout, self.guidance, + download=True) + self.assertIn("install higgs_audio_tts_4b_q8_0", buf.getvalue()) + run.assert_not_called() + + def test_failed_install_reports_warning_and_continues(self): + results = iter([MagicMock(returncode=1), MagicMock(returncode=0)]) + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(make_server.subprocess, "run", + side_effect=lambda *a, **k: next(results)) as run: + make_server._install_models(self.checkout, self.guidance, + download=True) + self.assertEqual(run.call_count, 2) + self.assertIn("exited with code 1", buf.getvalue()) + + def test_decide_download_skips_prompt_without_manager(self): + self.manager.unlink() + confirm = MagicMock() + self.assertFalse(make_server._decide_download(self.checkout, confirm)) + confirm.assert_not_called() + + def test_decide_download_asks_when_manager_present(self): + confirm = MagicMock(return_value=True) + self.assertTrue(make_server._decide_download(self.checkout, confirm)) + confirm.assert_called_once() + + +class TranscribeWavDirTests(unittest.TestCase): + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.folder = Path(self._td.name) + self.narrator = self.folder / "narrator.wav" + self.narrator.write_bytes(b"x") + self.other = self.folder / "other.wav" + self.other.write_bytes(b"x") + + def tearDown(self): + self._td.cleanup() + + def test_transcribes_to_stem_map_with_absolute_paths(self): + transcripts = {str(self.narrator): "First.", + str(self.other): "Second."} + with patch.object(make_server, "transcribe_reference_audio", + side_effect=lambda path, model_name="base": + transcripts[path]): + result = make_server.transcribe_wav_dir( + [self.narrator, self.other], "base") + self.assertEqual(list(result), ["narrator", "other"]) + self.assertEqual(result["narrator"], "First.") + + def test_failed_transcription_keeps_empty_string(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value=None): + result = make_server.transcribe_wav_dir([self.narrator], "base") + self.assertEqual(result["narrator"], "") + + def test_whisper_model_name_passed_through(self): + with patch.object(make_server, "transcribe_reference_audio", + return_value="text") as mock_transcribe: + make_server.transcribe_wav_dir([self.narrator], "large-v3") + self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], + "large-v3") + + def test_write_prompt_text_format(self): + path = make_server.write_prompt_text( + self.folder, {"narrator": "Hello.", "other": "World."}) + self.assertEqual(path, self.folder / make_server.PROMPT_TEXT_FILENAME) + text = path.read_text(encoding="utf-8") + self.assertIn("narrator|Hello.", text) + self.assertIn("other|World.", text) + + +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"]) + 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 FindAudiocppServerBinTests(unittest.TestCase): + """Locating the built audiocpp_server binary.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = Path(self._td.name) / "audio.cpp" + self.checkout.mkdir() + + def tearDown(self): + self._td.cleanup() + + def _build(self, name, binary="audiocpp_server"): + bin_dir = self.checkout / "build" / name / "bin" + bin_dir.mkdir(parents=True) + (bin_dir / binary).write_bytes(b"x") + + def test_no_build_dir_returns_none(self): + self.assertIsNone(make_server.find_audiocpp_server_bin(self.checkout)) + + def test_finds_built_binary(self): + self._build("linux-cuda-release") + self.assertEqual( + make_server.find_audiocpp_server_bin(self.checkout), + self.checkout / "build" / "linux-cuda-release" / "bin" + / "audiocpp_server") + + def test_finds_windows_exe(self): + self._build("windows-vulkan-debug", binary="audiocpp_server.exe") + self.assertEqual( + make_server.find_audiocpp_server_bin(self.checkout).name, + "audiocpp_server.exe") + + def test_build_dir_without_binary_returns_none(self): + (self.checkout / "build" / "linux-cuda-release" / "bin").mkdir( + parents=True) + self.assertIsNone(make_server.find_audiocpp_server_bin(self.checkout)) + + +class BuildAudiocppTests(unittest.TestCase): + """Running the audio.cpp build helper script.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.checkout = Path(self._td.name) / "audio.cpp" + self.checkout.mkdir() + self.scripts = self.checkout / "scripts" + self.scripts.mkdir() + (self.scripts / "build_linux.sh").write_text("#!/bin/sh\n", + encoding="utf-8") + + def tearDown(self): + self._td.cleanup() + + def test_runs_build_script_with_backend_and_target(self): + with patch.object(make_server.common, "run_console_subprocess", + return_value=0) as run: + rc = make_server.build_audiocpp(self.checkout, "cuda") + self.assertEqual(rc, 0) + argv = run.call_args[0][0] + self.assertEqual(argv[:3], ["sh", str(self.scripts / "build_linux.sh"), + "--backend"]) + self.assertIn("cuda", argv) + self.assertIn("--target", argv) + self.assertIn("audiocpp_server", argv) + self.assertEqual(run.call_args[1]["cwd"], self.checkout) + + def test_missing_script_returns_nonzero(self): + for f in self.scripts.iterdir(): + f.unlink() + rc = make_server.build_audiocpp(self.checkout, "cuda") + self.assertNotEqual(rc, 0) + + +class AudiocppDetectTests(unittest.TestCase): + """backends.audiocpp.detect() status reporting.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + self.checkout = _make_checkout(self.root) + + def tearDown(self): + self._td.cleanup() + + def test_not_cloned(self): + with patch.object(make_server, "find_local_checkout", return_value=None): + status = make_server.detect() + self.assertFalse(status.installed) + self.assertFalse(status.configured) + self.assertIn("not cloned", status.details[0]) + + def test_cloned_not_built_not_configured(self): + with patch.object(make_server, "find_local_checkout", + return_value=self.checkout), \ + patch.object(make_server, "find_audiocpp_server_bin", + return_value=None): + status = make_server.detect() + self.assertFalse(status.installed) + self.assertFalse(status.configured) + self.assertEqual(status.launch_hint, "") + + def test_built_and_configured_ready(self): + binary = self.checkout / "build" / "linux-cuda-release" / "bin" \ + / "audiocpp_server" + binary.parent.mkdir(parents=True) + binary.write_bytes(b"x") + server_json = self.checkout / "server.json" + server_json.write_text('{"models":[]}', encoding="utf-8") + with patch.object(make_server, "find_local_checkout", + return_value=self.checkout): + status = make_server.detect() + self.assertTrue(status.installed) + self.assertTrue(status.configured) + self.assertIn(str(binary), status.launch_hint) + self.assertIn(str(server_json), status.launch_hint) + + +class NonInteractiveMainTests(unittest.TestCase): + """The flag-only (non-TUI) path through main(), end to end.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.root = Path(self._td.name) + self.folder = self.root / "wavs" + self.folder.mkdir() + self.output = self.root / "server.json" + self.checkout = _make_checkout(self.root) + # Isolate config.py rewrites so no test touches the real one. + self.fake_config = self.root / "config.py" + self.fake_config.write_text(FAKE_CONFIG, encoding="utf-8") + patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) + patcher.start() + self.addCleanup(patcher.stop) + # Tests run without a tty -> main() takes the non-interactive path. + patcher = patch.object(make_server, "_interactive", return_value=False) + patcher.start() + self.addCleanup(patcher.stop) + + def tearDown(self): + self._td.cleanup() + + def _run(self, argv, transcribe=None, whisper="faster_whisper"): + argv = ["backends/audiocpp.py"] + argv + transcribe_effect = transcribe if transcribe is not None \ + else MagicMock() + with patch.object(sys, "argv", argv), \ + patch.object(make_server, "transcribe_reference_audio", + side_effect=transcribe_effect), \ + patch.object(make_server, "whisper_backend_available", + return_value=whisper): + return make_server.main() + + def _args(self, *extra): + return ["--wavs", str(self.folder), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout)] + list(extra) + + def test_default_run_hosts_recommended_entry(self): + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--no-sync-model-ids")) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["host"], "127.0.0.1") + self.assertEqual(data["port"], make_server.config_port()) + self.assertEqual(data["backend"], "cuda") + self.assertFalse(data["lazy_load"]) + self.assertEqual([m["id"] for m in data["models"]], ["higgs"]) + self.assertNotIn("voice_dir", data) + + def test_port_sync_accepted_updates_config(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:9999"): + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--port", "8080", + "--no-sync-model-ids")) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:8080"', + self.fake_config.read_text(encoding="utf-8")) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(data["port"], 8080) + + def test_port_sync_declined_keeps_config(self): + with patch.object(config, "AUDIOCPP_API_URL", + "http://127.0.0.1:9999"): + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--port", "8080", + "--no-sync-port", "--no-sync-model-ids")) + self.assertEqual(exit_code, 0) + self.assertIn('"http://127.0.0.1:9999"', + self.fake_config.read_text(encoding="utf-8")) + + def test_model_id_sync_accepted_updates_config(self): + self.fake_config.write_text(FAKE_CONFIG_WITH_MODEL_IDS, + encoding="utf-8") + exit_code = self._run(self._args("--families", "higgs_audio_tts")) + self.assertEqual(exit_code, 0) + 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_multi_family_lazy_with_voice_dir(self): + (self.folder / "narrator.wav").write_bytes(b"x") + exit_code = self._run( + self._args("--families", "qwen3_tts,higgs_audio_tts", + "--no-sync-model-ids"), + transcribe=lambda path, model_name="base": "a transcript") + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual([m["id"] for m in data["models"]], ["qwen", "higgs"]) + self.assertTrue(data["lazy_load"]) + 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|a transcript", prompt) + + def test_force_overwrites_existing_output(self): + self.output.write_text('{"old": true}', encoding="utf-8") + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--force", + "--no-sync-model-ids")) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + self.assertEqual(len(data["models"]), 1) + + def test_existing_output_declined_keeps_file(self): + self.output.write_text('{"old": true}', encoding="utf-8") + exit_code = self._run( + self._args("--families", "higgs_audio_tts", "--no-sync-model-ids")) + self.assertEqual(exit_code, 1) + self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), + {"old": True}) + + def test_all_packages_hosts_design_as_vdes(self): + exit_code = self._run( + self._args("--families", "qwen3_tts", "--all-packages", + "--no-sync-model-ids")) + self.assertEqual(exit_code, 0) + data = json.loads(self.output.read_text(encoding="utf-8")) + by_id = {m["id"]: m for m in data["models"]} + self.assertIn("qwen-design", by_id) + self.assertEqual(by_id["qwen-design"]["task"], "vdes") + # The non-design packages are hosted with task "tts". + self.assertTrue(any(m["id"] in ("qwen", "qwen-2") and m["task"] == "tts" + for m in data["models"])) + + def test_unknown_family_rejected(self): + with self.assertRaises(SystemExit) as ctx: + self._run(self._args("--families", "not_a_family", + "--no-sync-model-ids")) + self.assertEqual(ctx.exception.code, 2) + + def test_missing_checkout_rejected(self): + with patch.object(make_server, "find_local_checkout", + return_value=None), \ + self.assertRaises(SystemExit) as ctx: + self._run(["--families", "higgs_audio_tts", "--output", + str(self.output), "--no-sync-model-ids"]) + self.assertEqual(ctx.exception.code, 2) + + def test_missing_wav_dir_rejected(self): + missing = self.root / "nope" + with self.assertRaises(SystemExit) as ctx: + self._run(["--wavs", str(missing), "--output", str(self.output), + "--audiocpp-dir", str(self.checkout), + "--families", "higgs_audio_tts", "--no-sync-model-ids"]) + self.assertEqual(ctx.exception.code, 2) + + def test_families_required_in_noninteractive_run(self): + with self.assertRaises(SystemExit) as ctx: + self._run(self._args("--no-sync-model-ids")) + self.assertEqual(ctx.exception.code, 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_make_faster_voices_json.py b/tests/test_backends_faster.py index f26e645..641f6ee 100644 --- a/tests/test_make_faster_voices_json.py +++ b/tests/test_backends_faster.py @@ -1,4 +1,4 @@ -"""Tests for the faster-qwen3-tts voices.json generator tool.""" +"""Tests for the faster-qwen3-tts backend setup module (backends/faster.py).""" import json import sys @@ -7,7 +7,7 @@ import unittest from pathlib import Path from unittest.mock import patch -from tools import make_faster_voices_json as make_voices +from backends import faster as make_voices class FindWavFilesTests(unittest.TestCase): @@ -81,24 +81,35 @@ class BuildVoicesTests(unittest.TestCase): class MainTests(unittest.TestCase): + """The flag-only (non-TUI) path through main(), end to end.""" + def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.folder = Path(self._tmp.name) (self.folder / "narrator.wav").write_bytes(b"x") (self.folder / "alpha.wav").write_bytes(b"x") self.output = self.folder / "voices.json" + # Avoid touching the real converter/config.py and pip/git. + patcher = patch.object(make_voices.common, "update_config_value", + return_value=False) + patcher.start() + self.addCleanup(patcher.stop) + patcher = patch.object(make_voices, "_interactive", return_value=False) + patcher.start() + self.addCleanup(patcher.stop) def tearDown(self): self._tmp.cleanup() def _run(self, argv): - with patch.object(sys, "argv", ["make_voices.py"] + argv): + with patch.object(sys, "argv", ["backends/faster.py"] + argv), \ + patch.object(make_voices, "transcribe_reference_audio", + return_value="hello"): return make_voices.main() def test_writes_json_with_alphabetical_voice_order(self): - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"): - exit_code = self._run([str(self.folder)]) + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(list(data), ["alpha", "narrator"]) @@ -107,59 +118,51 @@ class MainTests(unittest.TestCase): def test_custom_output_path(self): custom = Path(self._tmp.name) / "custom.json" - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"): - self._run([str(self.folder), "--output", str(custom)]) + exit_code = self._run([str(self.folder), "--output", str(custom), + "--skip-install", "--skip-clone"]) + self.assertEqual(exit_code, 0) self.assertTrue(custom.exists()) self.assertFalse(self.output.exists()) def test_invalid_language_errors_before_work(self): with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe: with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder), "--language", "klingon"]) + self._run([str(self.folder), "--output", str(self.output), + "--language", "klingon", "--skip-install", + "--skip-clone"]) self.assertEqual(ctx.exception.code, 2) mock_transcribe.assert_not_called() def test_missing_input_dir_errors(self): with self.assertRaises(SystemExit) as ctx: - self._run([str(self.folder / "nope")]) + self._run([str(self.folder / "nope"), "--output", str(self.output), + "--skip-install", "--skip-clone"]) self.assertEqual(ctx.exception.code, 2) - def test_no_wav_files_errors(self): + def test_no_wav_files_returns_error(self): empty = Path(tempfile.mkdtemp()) try: - with self.assertRaises(SystemExit) as ctx: - self._run([str(empty)]) - self.assertEqual(ctx.exception.code, 2) + exit_code = self._run([str(empty), "--output", + str(empty / "voices.json"), + "--skip-install", "--skip-clone"]) + self.assertEqual(exit_code, 1) finally: empty.rmdir() def test_existing_output_declined_keeps_file(self): self.output.write_text('{"old": true}', encoding="utf-8") - with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe, \ - patch("builtins.input", return_value="n"): - exit_code = self._run([str(self.folder)]) + with patch.object(make_voices, "transcribe_reference_audio") as mock_transcribe: + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 1) mock_transcribe.assert_not_called() self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), {"old": True}) - def test_existing_output_accepted_overwrites(self): - self.output.write_text('{"old": true}', encoding="utf-8") - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"), \ - patch("builtins.input", return_value="y"): - exit_code = self._run([str(self.folder)]) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(list(data), ["alpha", "narrator"]) - def test_force_overwrites_without_prompt(self): self.output.write_text('{"old": true}', encoding="utf-8") - with patch.object(make_voices, "transcribe_reference_audio", - return_value="hello"), \ - patch("builtins.input", side_effect=AssertionError("prompted")): - exit_code = self._run([str(self.folder), "--force"]) + exit_code = self._run([str(self.folder), "--output", str(self.output), + "--force", "--skip-install", "--skip-clone"]) self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(list(data), ["alpha", "narrator"]) diff --git a/tests/test_hub.py b/tests/test_hub.py new file mode 100644 index 0000000..5f6d992 --- /dev/null +++ b/tests/test_hub.py @@ -0,0 +1,91 @@ +"""Tests for the TUI hub (hub.py) menu and helpers. + +The hub drives the same curses widgets as tui.py, so these tests reuse the +fake curses/screen from test_tui to run the menu without a terminal. +""" + +import unittest +from pathlib import Path +from unittest.mock import patch + +import hub +import tui +from tests.test_tui import FakeCurses, FakeScreen + + +class HubHelperTests(unittest.TestCase): + """Pure helpers in hub.py (no curses).""" + + def test_is_float(self): + self.assertTrue(hub._is_float("1.0")) + self.assertTrue(hub._is_float("2")) + self.assertFalse(hub._is_float("abc")) + self.assertFalse(hub._is_float("")) + + def test_list_voices_from_dir(self): + with __import__("tempfile").TemporaryDirectory() as td: + d = Path(td) + (d / "Narrator.wav").write_bytes(b"x") + (d / "Alpha.WAV").write_bytes(b"x") + (d / "notes.txt").write_bytes(b"x") + voices = hub._list_voices(str(d)) + # Stems preserve case; sorting is case-insensitive. + self.assertEqual(voices, ["Alpha", "Narrator"]) + + def test_list_voices_missing_dir(self): + self.assertEqual(hub._list_voices("/no/such/dir"), []) + + def test_status_mark(self): + from backends import BackendStatus + ready = BackendStatus("k", "l", installed=True, configured=True) + half = BackendStatus("k", "l", installed=True, configured=False) + none = BackendStatus("k", "l", installed=False, configured=False) + self.assertEqual(hub._status_mark("k", [ready]), "ready") + self.assertEqual(hub._status_mark("k", [half]), "installed") + self.assertEqual(hub._status_mark("k", [none]), "not set up") + self.assertEqual(hub._status_mark("missing", []), "not set up") + + +class HubMenuTests(unittest.TestCase): + """Drive _hub_menu with a fake screen (no terminal).""" + + def setUp(self): + tui._THEME.clear() + self.curses = FakeCurses() + from unittest.mock import patch as _patch + self._patcher = _patch.dict("sys.modules", {"curses": self.curses}) + self._patcher.start() + self.addCleanup(self._patcher.stop) + self.addCleanup(tui._THEME.clear) + + def test_quit_returns_none(self): + # Main menu: move to "Quit" (4th option, index 3) and press Enter. + screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_DOWN, 10]) + with patch.object(hub, "detect_all", return_value=[]): + result = hub._hub_menu(screen) + self.assertIsNone(result) + + def test_convert_with_no_ready_backend_offers_setup(self): + # Convert -> "Set up a backend..." is the only entry -> Enter selects + # it -> setup menu lists 3 backends; press Esc to go back -> convert + # returns None -> main menu loops. Then quit (Down x3 + Enter). + from backends import BackendInfo, BackendStatus + none = BackendStatus("k", "l", installed=False, configured=False) + infos = [BackendInfo("audiocpp", "audio.cpp", lambda: none, + lambda: 0), + BackendInfo("qwen", "Qwen", lambda: none, lambda: 0), + BackendInfo("faster", "faster", lambda: none, lambda: 0)] + with patch.object(hub, "detect_all", return_value=[none, none, none]), \ + patch.object(hub, "REGISTRY", infos): + # Convert(Enter), setup-entry(Enter), Esc on setup menu, + # back at main menu -> Down x3 -> Enter (Quit). + screen = FakeScreen(keys=[10, 10, 27, + FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_DOWN, 10]) + result = hub._hub_menu(screen) + self.assertIsNone(result) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_make_audiocpp_server_json.py b/tests/test_make_audiocpp_server_json.py deleted file mode 100644 index 39d32a8..0000000 --- a/tests/test_make_audiocpp_server_json.py +++ /dev/null @@ -1,1743 +0,0 @@ -"""Tests for the audio.cpp server.json generator tool.""" - -import argparse -import io -import json -import sys -import tempfile -import unittest -from contextlib import redirect_stdout -from pathlib import Path -from unittest.mock import MagicMock, patch - -from converter import config -from tools import make_audiocpp_server_json as make_server -from tools import tui - -FAKE_CONFIG = ( - 'LANGUAGE = "English"\n' - "\n" - 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' - "\n" - "CHUNK_SIZE = 250\n" -) - -FAKE_CONFIG_WITH_MODEL_IDS = ( - 'AUDIOCPP_API_URL = "http://127.0.0.1:9999" # audio.cpp audiocpp_server\n' - "\n" - 'AUDIOCPP_MODEL_ID = "qwen" # server entry for speaker mode\n' - 'AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"\n' -) - - -def _write_spec(checkout: Path, family: str, *, display_name=None, - tasks=("tts", "clone"), languages=("en",), packages=None, - category="tts"): - """Write a minimal model_specs/<family>.json into a fake checkout.""" - specs = checkout / "model_specs" - specs.mkdir(parents=True, exist_ok=True) - if packages is None: - packages = [{ - "id": f"{family}_q8_0", "default": True, "format": "gguf", - "target_directory": f"{family}-GGUF", - }] - spec = { - "family": family, - "display_name": display_name or family, - "category": category, - "tasks": list(tasks), - "languages": list(languages), - "packages": packages, - } - (specs / f"{family}.json").write_text(json.dumps(spec), encoding="utf-8") - return spec - - -def _make_checkout(tmp: Path) -> Path: - """Create a fake audio.cpp checkout with a realistic model_specs set.""" - checkout = tmp / "audio.cpp" - checkout.mkdir() - _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"}, - {"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=[{ - "id": "higgs_audio_tts_4b_q8_0", "default": True, - "format": "gguf", - "target_directory": "Higgs-Audio-v3-TTS-4B-GGUF", - }]) - _write_spec(checkout, "voxcpm2", display_name="VoxCPM2-2B", - languages=("en", "zh"), - packages=[{ - "id": "voxcpm2_q8_0", "default": True, "format": "gguf", - "target_directory": "VoxCPM2-GGUF", - }]) - _write_spec(checkout, "index_tts2", display_name="IndexTTS-2", - languages=("zh", "en"), - packages=[{ - "id": "index_tts2_q8_0", "default": True, "format": "gguf", - "target_directory": "IndexTTS2-GGUF", - }]) - _write_spec(checkout, "pocket_tts", display_name="PocketTTS-100M", - tasks=("tts", "clone"), languages=("en", "de"), - packages=[{ - "id": "pocket_tts_q8_0", "default": True, "format": "gguf", - "target_directory": "PocketTTS-GGUF", - }]) - _write_spec(checkout, "supertonic", display_name="Supertonic 3", - tasks=("tts",), languages=("en", "ko"), - packages=[{ - "id": "supertonic_q8_0", "default": True, "format": "gguf", - "target_directory": "Supertonic-GGUF", - }]) - # An ASR family that must be filtered out. - _write_spec(checkout, "qwen3_asr", display_name="Qwen3-ASR", - tasks=("asr",), category="asr") - # A TTS family with no installable packages (must be skipped). - _write_spec(checkout, "empty_tts", display_name="Empty TTS", - tasks=("tts",), packages=[]) - return checkout - - -class FindWavFilesTests(unittest.TestCase): - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - - def tearDown(self): - self._tmp.cleanup() - - def _touch(self, name): - path = self.folder / name - path.write_bytes(b"x") - return path - - def test_finds_only_wavs_case_insensitive(self): - self._touch("b.wav") - self._touch("a.WAV") - self._touch("notes.txt") - (self.folder / "sub").mkdir() - (self.folder / "sub" / "c.wav").write_bytes(b"x") - names = [path.name for path in make_server.find_wav_files(self.folder)] - self.assertEqual(names, ["a.WAV", "b.wav"]) - - def test_sorted_alphabetically_case_insensitive(self): - for name in ("Zed.wav", "alpha.wav", "Beta.wav"): - self._touch(name) - names = [path.name for path in make_server.find_wav_files(self.folder)] - self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) - - def test_empty_directory_returns_empty_list(self): - self.assertEqual(make_server.find_wav_files(self.folder), []) - - -class DetectWavDirTests(unittest.TestCase): - """Shallow .wav-directory discovery across the two checkout roots.""" - - def setUp(self): - self._td = tempfile.TemporaryDirectory() - self.root = Path(self._td.name) - self.audiocpp = self.root / "audio.cpp" - self.tts_root = self.root / "tts-audiobook-generator" - self.audiocpp.mkdir() - self.tts_root.mkdir() - - def tearDown(self): - self._td.cleanup() - - def _wav_dir(self, where, name="voices"): - directory = where / name - directory.mkdir(parents=True, exist_ok=True) - (directory / "voice.wav").write_bytes(b"x") - return directory - - def test_unique_wav_dir_in_tts_root_returned(self): - found = self._wav_dir(self.tts_root, "voices") - self.assertEqual(make_server.detect_wav_dir(self.audiocpp, - self.tts_root), - found) - - def test_unique_wav_dir_in_audiocpp_root_returned(self): - found = self._wav_dir(self.audiocpp, "reference") - self.assertEqual(make_server.detect_wav_dir(self.audiocpp, - self.tts_root), - found) - - def test_root_itself_containing_wavs_returned(self): - (self.tts_root / "direct.wav").write_bytes(b"x") - self.assertEqual(make_server.detect_wav_dir(self.audiocpp, - self.tts_root), - self.tts_root) - - def test_multiple_wav_dirs_returns_none(self): - self._wav_dir(self.tts_root, "one") - self._wav_dir(self.audiocpp, "two") - self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, - self.tts_root)) - - def test_output_dir_of_tts_root_excluded(self): - self._wav_dir(self.tts_root, "output") - self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, - self.tts_root)) - - def test_no_wavs_returns_none(self): - self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, - self.tts_root)) - - def test_nested_wav_dir_not_seen(self): - # Shallow search only: a wav dir two levels deep is not a candidate. - nested = self.tts_root / "outer" / "inner" - nested.mkdir(parents=True) - (nested / "voice.wav").write_bytes(b"x") - self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, - self.tts_root)) - - -class ConfigPortTests(unittest.TestCase): - def test_port_parsed_from_config_url(self): - with patch.object(config, "AUDIOCPP_API_URL", - "http://127.0.0.1:8080"): - self.assertEqual(make_server.config_port(), 8080) - - def test_missing_port_falls_back(self): - with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1"): - self.assertEqual(make_server.config_port(), - make_server.FALLBACK_PORT) - - def test_invalid_url_falls_back(self): - with patch.object(config, "AUDIOCPP_API_URL", "not a url"): - self.assertEqual(make_server.config_port(), - make_server.FALLBACK_PORT) - - def test_url_with_port_replaces_port(self): - self.assertEqual( - make_server._url_with_port("http://127.0.0.1:8080", 9000), - "http://127.0.0.1:9000") - - def test_url_without_port_adds_port(self): - self.assertEqual( - make_server._url_with_port("http://localhost", 8080), - "http://localhost:8080") - - -class UpdateConfigPortTests(unittest.TestCase): - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.config_path = Path(self._tmp.name) / "config.py" - self.config_path.write_text(FAKE_CONFIG, encoding="utf-8") - - def tearDown(self): - self._tmp.cleanup() - - def test_rewrites_port_preserving_comment(self): - changed = make_server.update_config_api_url_port( - 8080, config_path=self.config_path) - self.assertTrue(changed) - text = self.config_path.read_text(encoding="utf-8") - self.assertIn( - 'AUDIOCPP_API_URL = "http://127.0.0.1:8080" # audio.cpp audiocpp_server', - text) - self.assertIn('LANGUAGE = "English"', text) - self.assertIn("CHUNK_SIZE = 250", text) - - def test_returns_false_when_no_url_line(self): - path = Path(self._tmp.name) / "other.py" - path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") - self.assertFalse(make_server.update_config_api_url_port( - 8080, config_path=path)) - - def test_returns_false_when_port_unchanged(self): - self.assertFalse(make_server.update_config_api_url_port( - 9999, config_path=self.config_path)) - self.assertEqual(self.config_path.read_text(encoding="utf-8"), - FAKE_CONFIG) - - def test_returns_false_when_file_missing(self): - self.assertFalse(make_server.update_config_api_url_port( - 8080, config_path=Path(self._tmp.name) / "nope.py")) - - -class UpdateConfigModelIdsTests(unittest.TestCase): - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.config_path = Path(self._tmp.name) / "config.py" - self.config_path.write_text(FAKE_CONFIG_WITH_MODEL_IDS, - encoding="utf-8") - - def tearDown(self): - self._tmp.cleanup() - - def test_rewrites_both_ids_preserving_lines(self): - changed = make_server.update_config_model_ids( - "higgs", "higgs", config_path=self.config_path) - self.assertTrue(changed) - text = self.config_path.read_text(encoding="utf-8") - self.assertIn('AUDIOCPP_MODEL_ID = "higgs" # server entry for speaker mode', - text) - self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "higgs"', text) - self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:9999"', text) - - def test_clone_id_optional(self): - changed = make_server.update_config_model_ids( - "voxcpm2", config_path=self.config_path) - self.assertTrue(changed) - text = self.config_path.read_text(encoding="utf-8") - self.assertIn('AUDIOCPP_MODEL_ID = "voxcpm2"', text) - self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"', text) - - def test_returns_false_when_ids_unchanged(self): - changed = make_server.update_config_model_ids( - "qwen", "qwen-clone", config_path=self.config_path) - self.assertFalse(changed) - self.assertEqual(self.config_path.read_text(encoding="utf-8"), - FAKE_CONFIG_WITH_MODEL_IDS) - - def test_returns_false_when_lines_missing(self): - path = Path(self._tmp.name) / "other.py" - path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") - self.assertFalse(make_server.update_config_model_ids( - "higgs", "higgs", config_path=path)) - - def test_returns_false_when_file_missing(self): - self.assertFalse(make_server.update_config_model_ids( - "higgs", "higgs", - config_path=Path(self._tmp.name) / "nope.py")) - - -class ResolveWavDirArgTests(unittest.TestCase): - """Path normalization for the required WAV_DIR argument.""" - - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - - def tearDown(self): - self._tmp.cleanup() - - def test_resolves_to_absolute(self): - self.assertEqual(make_server.resolve_wav_dir_arg(str(self.folder)), - self.folder.resolve()) - - def test_strips_surrounding_quotes(self): - quoted = f'"{self.folder}"' - self.assertEqual(make_server.resolve_wav_dir_arg(quoted), - self.folder.resolve()) - - def test_strips_single_quotes(self): - quoted = f"'{self.folder}'" - self.assertEqual(make_server.resolve_wav_dir_arg(quoted), - self.folder.resolve()) - - def test_strips_whitespace(self): - self.assertEqual(make_server.resolve_wav_dir_arg(f" {self.folder} "), - self.folder.resolve()) - - def test_expands_tilde(self): - with patch.object(make_server.os.path, "expanduser", - return_value=str(self.folder)) as mock_expand: - result = make_server.resolve_wav_dir_arg("~/voices") - mock_expand.assert_called_once_with("~/voices") - self.assertEqual(result, self.folder.resolve()) - - def test_trailing_slash_preserved_as_dir(self): - self.assertEqual(make_server.resolve_wav_dir_arg(f"{self.folder}/"), - self.folder.resolve()) - - -class 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 CheckoutAutoSelectTests(unittest.TestCase): - """TUI browser auto-accept callback for an audio.cpp checkout.""" - - def setUp(self): - self._td = tempfile.TemporaryDirectory() - self.root = Path(self._td.name) - - def tearDown(self): - self._td.cleanup() - - def test_accepts_audio_cpp_containing_model_specs(self): - checkout = self.root / "audio.cpp" - checkout.mkdir() - (checkout / "model_specs").mkdir() - self.assertEqual(make_server._checkout_auto_select(checkout), - checkout) - - def test_rejects_audio_cpp_without_model_specs(self): - checkout = self.root / "audio.cpp" - checkout.mkdir() - self.assertIsNone(make_server._checkout_auto_select(checkout)) - - def test_rejects_other_name_even_with_model_specs(self): - other = self.root / "not-audiocpp" - other.mkdir() - (other / "model_specs").mkdir() - self.assertIsNone(make_server._checkout_auto_select(other)) - - def test_rejects_plain_directory(self): - plain = self.root / "somewhere" - plain.mkdir() - self.assertIsNone(make_server._checkout_auto_select(plain)) - - -class DefaultModelIdTests(unittest.TestCase): - def test_preferred_ids_for_tested_families(self): - self.assertEqual(make_server.default_model_id("qwen3_tts"), "qwen") - self.assertEqual(make_server.default_model_id("higgs_audio_tts"), "higgs") - self.assertEqual(make_server.default_model_id("voxcpm2"), "voxcpm2") - self.assertEqual(make_server.default_model_id("index_tts2"), "indextts2") - - def test_derived_id_strips_trailing_tts_and_underscores(self): - self.assertEqual(make_server.default_model_id("pocket_tts"), "pocket") - self.assertEqual(make_server.default_model_id("dots_tts"), "dots") - # Families without a _tts suffix just drop underscores. - self.assertEqual(make_server.default_model_id("moss_tts_local"), - "mossttslocal") - - -class LoadModelCatalogTests(unittest.TestCase): - def setUp(self): - self._tmp = list(tempfile._mkdtemp() and 0 for _ in range(0)) # noqa - self._td = tempfile.TemporaryDirectory() - self.checkout = _make_checkout(Path(self._td.name)) - - def tearDown(self): - self._td.cleanup() - - def test_includes_tts_families_excludes_asr(self): - catalog = make_server.load_model_catalog(self.checkout) - families = [entry["family"] for entry in catalog] - self.assertIn("qwen3_tts", families) - self.assertIn("higgs_audio_tts", families) - self.assertIn("pocket_tts", families) - self.assertIn("supertonic", families) - self.assertNotIn("qwen3_asr", families) - - def test_skips_families_with_no_packages(self): - catalog = make_server.load_model_catalog(self.checkout) - self.assertNotIn("empty_tts", - [entry["family"] for entry in catalog]) - - def test_families_sorted_alphabetically_by_display_name(self): - catalog = make_server.load_model_catalog(self.checkout) - names = [entry["display_name"].lower() for entry in catalog] - self.assertEqual(names, sorted(names)) - # No family is marked "tested" anymore; all are treated equally. - self.assertNotIn("tested", catalog[0]) - self.assertNotIn("TESTED_FAMILIES", dir(make_server)) - - def test_default_package_and_target_directory_resolved(self): - catalog = make_server.load_model_catalog(self.checkout) - by_family = {entry["family"]: entry for entry in catalog} - higgs = by_family["higgs_audio_tts"] - self.assertEqual(higgs["install_id"], "higgs_audio_tts_4b_q8_0") - self.assertEqual(higgs["default_path"], - "models/Higgs-Audio-v3-TTS-4B-GGUF") - - def test_picks_first_gguf_when_no_default_flag(self): - # Rewrite the voxcpm2 spec so no package is flagged default. - _write_spec(self.checkout, "voxcpm2", display_name="VoxCPM2-2B", - packages=[ - {"id": "voxcpm2_bf16", "format": "gguf", - "target_directory": "VoxCPM2-GGUF"}, - {"id": "voxcpm2_q8_0", "format": "gguf", - "target_directory": "VoxCPM2-GGUF"}, - ]) - catalog = make_server.load_model_catalog(self.checkout) - by_family = {entry["family"]: entry for entry in catalog} - # No default:true -> first gguf package wins. - self.assertEqual(by_family["voxcpm2"]["install_id"], "voxcpm2_bf16") - - def test_clone_capability_from_tasks(self): - catalog = make_server.load_model_catalog(self.checkout) - by_family = {entry["family"]: entry for entry in catalog} - self.assertTrue(by_family["higgs_audio_tts"]["clone_capable"]) - self.assertFalse(by_family["supertonic"]["clone_capable"]) - - def test_missing_model_specs_dir_raises(self): - empty = Path(self._td.name) / "empty" - empty.mkdir() - with self.assertRaises(NotADirectoryError): - make_server.load_model_catalog(empty) - - -class DetectBackendTests(unittest.TestCase): - """Backend detection from audio.cpp build directory names.""" - - def setUp(self): - self._td = tempfile.TemporaryDirectory() - self.checkout = Path(self._td.name) / "audio.cpp" - self.checkout.mkdir() - - def tearDown(self): - self._td.cleanup() - - def _build(self, name, binary="audiocpp_server"): - build_dir = self.checkout / "build" / name - bin_dir = build_dir / "bin" - bin_dir.mkdir(parents=True) - (bin_dir / binary).write_bytes(b"x") - return build_dir - - def test_no_build_dir_returns_none(self): - self.assertIsNone(make_server.detect_backend(self.checkout)) - - def test_unique_linux_backend_detected(self): - self._build("linux-cuda-release") - self.assertEqual(make_server.detect_backend(self.checkout), "cuda") - - def test_windows_exe_backend_detected(self): - self._build("windows-vulkan-debug", binary="audiocpp_server.exe") - self.assertEqual(make_server.detect_backend(self.checkout), "vulkan") - - def test_hip_backend_detected(self): - self._build("linux-hip-release") - self.assertEqual(make_server.detect_backend(self.checkout), "hip") - - def test_cpu_backend_detected(self): - self._build("linux-cpu-release") - self.assertEqual(make_server.detect_backend(self.checkout), "cpu") - - def test_metal_maps_to_cpu(self): - self._build("macos-metal-release") - self.assertEqual(make_server.detect_backend(self.checkout), "cpu") - - def test_multiple_backends_returns_none(self): - self._build("linux-cuda-release") - self._build("linux-cpu-release") - self.assertIsNone(make_server.detect_backend(self.checkout)) - - def test_multiple_builds_same_backend_detected(self): - self._build("linux-cuda-release") - self._build("windows-cuda-debug") - self.assertEqual(make_server.detect_backend(self.checkout), "cuda") - - def test_build_dir_without_binary_ignored(self): - (self.checkout / "build" / "linux-cuda-release").mkdir(parents=True) - self.assertIsNone(make_server.detect_backend(self.checkout)) - - def test_non_matching_build_dir_name_ignored(self): - self._build("linux-mybuild-release") - self.assertIsNone(make_server.detect_backend(self.checkout)) - - -class BackendOptionsTests(unittest.TestCase): - """Aligned backend menu labels and the [auto-detected] default.""" - - def test_options_have_aligned_dashes(self): - options, default_index = make_server._backend_options() - dash_columns = {label.index(" - ") for label, _ in options} - self.assertEqual(len(dash_columns), 1) - self.assertEqual(default_index, 0) - - def test_detected_backend_marked_and_defaulted(self): - options, default_index = make_server._backend_options("vulkan") - labels = [label for label, _ in options] - self.assertEqual(default_index, labels.index(next( - label for label, value in options - if value == "vulkan" and label.endswith("[auto-detected]")))) - self.assertTrue(labels[default_index].endswith("[auto-detected]")) - self.assertEqual(options[default_index][1], "vulkan") - - def test_unknown_detected_backend_is_ignored(self): - options, default_index = make_server._backend_options("opencl") - self.assertEqual(default_index, 0) - self.assertFalse(any("[auto-detected]" in label - for label, _ in options)) - - def test_labels_keep_backend_values(self): - options, _ = make_server._backend_options() - self.assertEqual([value for _, value in options], - list(make_server.BACKENDS)) - - -class AskFamiliesTests(unittest.TestCase): - def setUp(self): - self._td = tempfile.TemporaryDirectory() - self.checkout = _make_checkout(Path(self._td.name)) - self.catalog = make_server.load_model_catalog(self.checkout) - - def tearDown(self): - self._td.cleanup() - - def _ids(self): - return [entry["family"] for entry in self.catalog] - - def test_enter_selects_first_family(self): - with patch("builtins.input", side_effect=[""]): - self.assertEqual(make_server.ask_families(self.catalog), - [self.catalog[0]["family"]]) - - def test_eof_selects_first_family(self): - with patch("builtins.input", side_effect=EOFError): - self.assertEqual(make_server.ask_families(self.catalog), - [self.catalog[0]["family"]]) - - def test_comma_separated_numbers(self): - # 1 and 3 (higgs_audio_tts and pocket_tts in alphabetical order). - with patch("builtins.input", side_effect=["1,3"]): - chosen = make_server.ask_families(self.catalog) - self.assertEqual(chosen, ["higgs_audio_tts", "pocket_tts"]) - - def test_space_separated_numbers(self): - with patch("builtins.input", side_effect=["2 4"]): - chosen = make_server.ask_families(self.catalog) - self.assertEqual(chosen, ["index_tts2", "qwen3_tts"]) - - def test_dedupes_repeated_choices(self): - with patch("builtins.input", side_effect=["1,1,2"]): - chosen = make_server.ask_families(self.catalog) - self.assertEqual(chosen, ["higgs_audio_tts", "index_tts2"]) - - def test_invalid_input_reprompts(self): - with patch("builtins.input", side_effect=["foo", "0", "2"]): - chosen = make_server.ask_families(self.catalog) - self.assertEqual(chosen, ["index_tts2"]) - - -class BuildServerConfigTests(unittest.TestCase): - def test_single_entry_without_voice_dir(self): - entry = make_server.build_model_entry( - "higgs_audio_tts", "higgs", "models/Higgs-GGUF") - cfg = make_server.build_server_config( - "127.0.0.1", 8080, "cuda", False, [entry]) - self.assertEqual(cfg["host"], "127.0.0.1") - self.assertEqual(cfg["port"], 8080) - self.assertEqual(cfg["backend"], "cuda") - self.assertFalse(cfg["lazy_load"]) - self.assertEqual(cfg["models"], [entry]) - self.assertNotIn("voice_dir", cfg) - - def test_voice_dir_added_when_given(self): - entry = make_server.build_model_entry("voxcpm2", "voxcpm2", "models/V") - cfg = make_server.build_server_config( - "0.0.0.0", 9000, "cpu", True, [entry], - voice_dir="/abs/voices") - self.assertTrue(cfg["lazy_load"]) - self.assertEqual(cfg["voice_dir"], "/abs/voices") - - def test_model_entry_shape(self): - entry = make_server.build_model_entry("index_tts2", "indextts2", "p") - self.assertEqual(entry["id"], "indextts2") - self.assertEqual(entry["family"], "index_tts2") - self.assertEqual(entry["path"], "p") - 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 InstallModelsTests(unittest.TestCase): - """Printing or auto-running the model install commands.""" - - def setUp(self): - self._td = tempfile.TemporaryDirectory() - self.checkout = Path(self._td.name) / "audio.cpp" - self.checkout.mkdir() - self.manager = self.checkout / "tools" / "model_manager_v2.py" - self.manager.parent.mkdir() - self.manager.write_text("#!/usr/bin/env python3\n", encoding="utf-8") - self.guidance = [("Higgs Audio v3 TTS 4B", "higgs_audio_tts_4b_q8_0"), - ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0"), - ("Qwen3-TTS", "qwen3_tts_1_7b_base_q8_0")] - - def tearDown(self): - self._td.cleanup() - - def test_declined_download_prints_commands_deduped(self): - buf = io.StringIO() - with redirect_stdout(buf), \ - patch.object(make_server.subprocess, "run") as run: - make_server._install_models(self.checkout, self.guidance, - download=False) - out = buf.getvalue() - self.assertEqual(out.count("install higgs_audio_tts_4b_q8_0"), 1) - self.assertEqual(out.count("install qwen3_tts_1_7b_base_q8_0"), 1) - run.assert_not_called() - - def test_accepted_download_runs_each_command(self): - with patch.object(make_server.subprocess, "run", - return_value=MagicMock(returncode=0)) as run: - make_server._install_models(self.checkout, self.guidance, - download=True) - self.assertEqual(run.call_count, 2) - commands = [call[0][0] for call in run.call_args_list] - self.assertEqual(commands[0], - [sys.executable, str(self.manager), "install", - "higgs_audio_tts_4b_q8_0"]) - self.assertEqual(commands[1], - [sys.executable, str(self.manager), "install", - "qwen3_tts_1_7b_base_q8_0"]) - for call in run.call_args_list: - self.assertEqual(call[1]["cwd"], str(self.checkout)) - - def test_missing_manager_falls_back_to_printing(self): - self.manager.unlink() - buf = io.StringIO() - with redirect_stdout(buf), \ - patch.object(make_server.subprocess, "run") as run: - make_server._install_models(self.checkout, self.guidance, - download=True) - self.assertIn("install higgs_audio_tts_4b_q8_0", buf.getvalue()) - run.assert_not_called() - - def test_failed_install_reports_warning_and_continues(self): - results = iter([MagicMock(returncode=1), MagicMock(returncode=0)]) - buf = io.StringIO() - with redirect_stdout(buf), \ - patch.object(make_server.subprocess, "run", - side_effect=lambda *a, **k: next(results)) as run: - make_server._install_models(self.checkout, self.guidance, - download=True) - self.assertEqual(run.call_count, 2) - self.assertIn("exited with code 1", buf.getvalue()) - - def test_decide_download_skips_prompt_without_manager(self): - self.manager.unlink() - confirm = MagicMock() - self.assertFalse(make_server._decide_download(self.checkout, confirm)) - confirm.assert_not_called() - - def test_decide_download_asks_when_manager_present(self): - confirm = MagicMock(return_value=True) - self.assertTrue(make_server._decide_download(self.checkout, confirm)) - confirm.assert_called_once() - - -class TranscribeWavDirTests(unittest.TestCase): - def setUp(self): - self._td = tempfile.TemporaryDirectory() - self.folder = Path(self._td.name) - self.narrator = self.folder / "narrator.wav" - self.narrator.write_bytes(b"x") - self.other = self.folder / "other.wav" - self.other.write_bytes(b"x") - - def tearDown(self): - self._td.cleanup() - - def test_transcribes_to_stem_map_with_absolute_paths(self): - transcripts = {str(self.narrator): "First.", - str(self.other): "Second."} - with patch.object(make_server, "transcribe_reference_audio", - side_effect=lambda path, model_name="base": - transcripts[path]): - result = make_server.transcribe_wav_dir( - [self.narrator, self.other], "base") - self.assertEqual(list(result), ["narrator", "other"]) - self.assertEqual(result["narrator"], "First.") - - def test_failed_transcription_keeps_empty_string(self): - with patch.object(make_server, "transcribe_reference_audio", - return_value=None): - result = make_server.transcribe_wav_dir([self.narrator], "base") - self.assertEqual(result["narrator"], "") - - def test_whisper_model_name_passed_through(self): - with patch.object(make_server, "transcribe_reference_audio", - return_value="text") as mock_transcribe: - make_server.transcribe_wav_dir([self.narrator], "large-v3") - self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], - "large-v3") - - def test_write_prompt_text_format(self): - path = make_server.write_prompt_text( - self.folder, {"narrator": "Hello.", "other": "World."}) - self.assertEqual(path, self.folder / make_server.PROMPT_TEXT_FILENAME) - text = path.read_text(encoding="utf-8") - # One "name|transcript" line per voice, in insertion order. - self.assertIn("narrator|Hello.", text) - self.assertIn("other|World.", text) - - -class PromptHelperTests(unittest.TestCase): - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.folder = Path(self._tmp.name) - - def tearDown(self): - self._tmp.cleanup() - - def test_ask_port_reprompts_until_valid(self): - with patch("builtins.input", side_effect=["abc", "8081"]): - self.assertEqual(make_server.ask_port(8080), 8081) - - def test_ask_port_eof_returns_default(self): - with patch("builtins.input", side_effect=EOFError): - self.assertEqual(make_server.ask_port(8080), 8080) - - def test_ask_menu_reprompts_until_valid(self): - options = [("One", "one"), ("Two", "two")] - with patch("builtins.input", side_effect=["9", "2"]): - self.assertEqual( - make_server.ask_menu("Pick:", options, default_index=1), - "two") - - def test_ask_menu_eof_returns_default(self): - options = [("One", "one"), ("Two", "two")] - with patch("builtins.input", side_effect=EOFError): - self.assertEqual( - make_server.ask_menu("Pick:", options, default_index=1), - "one") - - -class 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.""" - - def setUp(self): - self._td = tempfile.TemporaryDirectory() - self.root = Path(self._td.name) - self.folder = self.root / "wavs" - self.folder.mkdir() - self.output = self.root / "server.json" - self.checkout = _make_checkout(self.root) - # Isolate the config.py rewrite target so no test can ever - # modify the repository's real converter/config.py. - self.fake_config = self.root / "config.py" - self.fake_config.write_text(FAKE_CONFIG, encoding="utf-8") - patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) - patcher.start() - self.addCleanup(patcher.stop) - # 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() - - def _run(self, argv, inputs=None, transcribe=None, whisper="faster_whisper"): - argv = ["make_audiocpp_server_json.py"] + argv - input_effect = inputs if inputs is not None else EOFError - transcribe_effect = transcribe if transcribe is not None else MagicMock() - with patch.object(sys, "argv", argv), \ - patch("builtins.input", side_effect=input_effect), \ - patch.object(make_server, "transcribe_reference_audio", - side_effect=transcribe_effect), \ - patch.object(make_server, "whisper_backend_available", - 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 single-family flow and shared server settings.""" - - def _args(self, *extra): - return ["--wavs", str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.checkout)] + list(extra) - - 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) - - def test_missing_audiocpp_dir_errors(self): - with self.assertRaises(SystemExit) as ctx: - self._run(self._args("--audiocpp-dir", str(self.root / "nope")), - inputs=[]) - self.assertEqual(ctx.exception.code, 2) - - def test_empty_audiocpp_dir_prompted_errors(self): - # No --audiocpp-dir and EOF at the prompt -> hard error. - buf = io.StringIO() - with patch.object(sys, "argv", - ["make_audiocpp_server_json.py", - "--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_wav_prompt_defaults_to_detected_dir(self): - # No --wavs: the prompt default is the unique .wav directory detected - # across the checkouts; pressing Enter accepts it. - (self.folder / "narrator.wav").write_bytes(b"x") - argv = ["make_audiocpp_server_json.py", - "--output", str(self.output), - "--audiocpp-dir", str(self.checkout)] - # wav(Enter -> default), family, host, port, backend, lazy, sync(y) - inputs = ["", "", "", "", "", "", "y"] - with patch.object(sys, "argv", argv), \ - patch("builtins.input", side_effect=inputs), \ - patch.object(make_server, "detect_wav_dir", - return_value=self.folder), \ - patch.object(make_server, "transcribe_reference_audio", - side_effect=lambda path, model_name="base": "t"), \ - patch.object(make_server, "whisper_backend_available", - return_value="faster_whisper"): - exit_code = make_server.main() - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - - 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 -> one entry, lazy defaults to False. - self.assertFalse(data["lazy_load"]) - # The first family alphabetically is Higgs Audio v3 TTS 4B. - self.assertEqual([model["id"] for model in data["models"]], ["higgs"]) - self.assertEqual( - [model["path"] for model in data["models"]], - ["models/Higgs-Audio-v3-TTS-4B-GGUF"]) - self.assertEqual(data["models"][0]["task"], "tts") - # voice_dir only when wavs are present; this run has none. - self.assertNotIn("voice_dir", data) - - def test_eof_uses_all_defaults(self): - 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"]), 1) - - 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. - # 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) - self.assertIn('"http://127.0.0.1:8080"', - self.fake_config.read_text(encoding="utf-8")) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["port"], 8080) - - def test_port_sync_declined_keeps_config(self): - with patch.object(config, "AUDIOCPP_API_URL", - "http://127.0.0.1:9999"): - inputs = ["", "", "n", "", "", "n"] - exit_code = self._run( - self._args("--port", "8080"), inputs=inputs) - self.assertEqual(exit_code, 0) - self.assertIn('"http://127.0.0.1:9999"', - self.fake_config.read_text(encoding="utf-8")) - - def test_matching_port_does_not_prompt_for_sync(self): - # config_port() is 8080 (real config); default port matches -> no sync. - inputs = self._defaults() - exit_code = self._run(self._args(), inputs=inputs) - self.assertEqual(exit_code, 0) - self.assertEqual(self.fake_config.read_text(encoding="utf-8"), - FAKE_CONFIG) - - 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") - exit_code = self._run(self._args(), inputs=["n"]) - self.assertEqual(exit_code, 1) - self.assertEqual(json.loads(self.output.read_text(encoding="utf-8")), - {"old": True}) - - def test_existing_output_accepted_overwrites(self): - self.output.write_text('{"old": true}', encoding="utf-8") - inputs = ["y"] + self._defaults() - exit_code = self._run(self._args(), inputs=inputs) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 1) - - def test_force_overwrites_without_prompt(self): - self.output.write_text('{"old": true}', encoding="utf-8") - inputs = self._defaults() - 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"]), 1) - - def test_flags_skip_prompts(self): - # --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", - "--host", "0.0.0.0", "--port", "9000", - "--backend", "cpu", "--lazy-load"), - inputs=["y", "y"]) - self.assertEqual(exit_code, 0) - self.assertIn('"http://127.0.0.1:9000"', - self.fake_config.read_text(encoding="utf-8")) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["host"], "0.0.0.0") - self.assertEqual(data["port"], 9000) - self.assertEqual(data["backend"], "cpu") - self.assertTrue(data["lazy_load"]) - - def test_final_output_is_wrote_plus_install_commands(self): - # Two families -> two entries; the console output ends with the Wrote - # line and one full-path install command per model, nothing else. - code, out = self._run_capturing( - self._args("--families", "qwen3_tts,higgs_audio_tts"), - inputs=["", "", "", ""]) - self.assertEqual(code, 0) - self.assertIn(f"Wrote {self.output.resolve()} with 2 entries.", out) - manager = self.checkout / "tools" / "model_manager_v2.py" - self.assertIn(f"python {manager} install qwen3_tts_1_7b_base_q8_0", - out) - self.assertIn(f"python {manager} install higgs_audio_tts_4b_q8_0", out) - # The generated JSON and the old [INFO] notes are no longer echoed. - self.assertNotIn("[INFO]", out) - self.assertNotIn('"models"', out) - - 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(["--wavs", str(missing), "--output", str(self.output), - "--audiocpp-dir", str(self.checkout)], - inputs=self._defaults()) - self.assertEqual(ctx.exception.code, 2) - shown = "".join(call[0][0] for call in mock_stderr.write.call_args_list) - self.assertIn(f"WAV directory not found: {missing.resolve()}", shown) - self.assertIn("directory containing the .wav", shown) - - 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_detected_backend_selected_by_default(self): - # A built backend in the checkout's build/ dir makes that backend the - # default; pressing Enter on the backend prompt accepts it. - build_dir = self.checkout / "build" / "linux-vulkan-release" / "bin" - build_dir.mkdir(parents=True) - (build_dir / "audiocpp_server").write_bytes(b"x") - # host, port, backend(Enter -> detected vulkan), lazy, sync(y) - inputs = ["", "", "", "", "", "y"] - code, out = self._run_capturing(self._args(), inputs=inputs) - self.assertEqual(code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["backend"], "vulkan") - - def test_download_accepted_runs_install_commands(self): - manager = self.checkout / "tools" / "model_manager_v2.py" - manager.parent.mkdir(parents=True, exist_ok=True) - manager.write_text("#!/usr/bin/env python3\n", encoding="utf-8") - argv = ["make_audiocpp_server_json.py"] + self._args( - "--families", "qwen3_tts") - with patch.object(sys, "argv", argv), \ - patch("builtins.input", side_effect=["", "", "", "", "y"]), \ - patch.object(make_server, "transcribe_reference_audio"), \ - patch.object(make_server, "whisper_backend_available", - return_value="faster_whisper"), \ - patch.object(make_server.subprocess, "run", - return_value=MagicMock(returncode=0)) as run: - code = make_server.main() - self.assertEqual(code, 0) - run.assert_called_once_with( - [sys.executable, str(manager), "install", - "qwen3_tts_1_7b_base_q8_0"], - cwd=str(self.checkout)) - - 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") - # packages(3=VoiceDesign), task(design default Enter), host, - # port, backend, lazy, sync(y) - inputs = ["3", "", "", "", "", "", "y"] - code, out = self._run_capturing( - self._args("--families", "qwen3_tts", "--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) - # 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. - # packages(2=CustomVoice), host, port, backend, lazy, sync(y) - inputs = ["2", "", "", "", "", "y"] - code, _ = self._run_capturing( - self._args("--families", "qwen3_tts", "--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. - # packages(1,3), task(design default Enter), host, port, backend, lazy - inputs = ["1,3", "", "", "", "", ""] - code, _ = self._run_capturing( - self._args("--families", "qwen3_tts", "--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): - """The --families flow for clone-only model families.""" - - def setUp(self): - super().setUp() - # These tests exercise AUDIOCPP_MODEL_ID rewriting, so the fake - # config must contain the model id lines to rewrite. - self.fake_config.write_text(FAKE_CONFIG_WITH_MODEL_IDS, - encoding="utf-8") - - def _args(self, family, *extra): - 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 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") - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(len(data["models"]), 1) - entry = data["models"][0] - self.assertEqual(entry["id"], "higgs") - self.assertEqual(entry["family"], "higgs_audio_tts") - self.assertEqual(entry["path"], "models/Higgs-Audio-v3-TTS-4B-GGUF") - self.assertEqual(entry["task"], "tts") - self.assertEqual(entry["mode"], "offline") - # Voice presets live in the server-level voice_dir, not per entry. - self.assertNotIn("voice_presets", entry) - 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|a transcript", prompt) - # 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") - # host, port, backend, lazy, sync(n) - inputs = ["", "", "", "", "n"] - exit_code = self._run( - self._args("voxcpm2"), inputs=inputs, - transcribe=lambda path, model_name="base": "t") - self.assertEqual(exit_code, 0) - text = self.fake_config.read_text(encoding="utf-8") - self.assertIn('AUDIOCPP_MODEL_ID = "qwen"', text) - self.assertIn('AUDIOCPP_CLONE_MODEL_ID = "qwen-clone"', text) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertEqual(data["models"][0]["family"], "voxcpm2") - - def test_no_wavs_warns_and_omits_voice_dir(self): - buf = io.StringIO() - # host, port, backend, lazy, sync(y) - inputs = ["", "", "", "", "y"] - with patch.object(sys, "argv", - ["make_audiocpp_server_json.py", - "--wavs", str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.checkout), - "--families", "index_tts2"]), \ - 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() - self.assertEqual(code, 0) - out = buf.getvalue() - self.assertIn("No .wav files found", out) - self.assertIn("model_manager_v2.py install index_tts2_q8_0", out) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertNotIn("voice_dir", data) - - def test_unknown_family_rejected(self): - with self.assertRaises(SystemExit) as ctx: - self._run(self._args("not_a_family"), inputs=[]) - self.assertEqual(ctx.exception.code, 2) - - -class MultiFamilyMainTests(_MainTestBase): - """Hosting several families in one server.json.""" - - def _args(self, *extra): - 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; 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, - transcribe=lambda path, model_name="base": "a transcript") - 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", "higgs"]) - # Two entries -> lazy defaults to True. - self.assertTrue(data["lazy_load"]) - self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - 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 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) - self.assertEqual(exit_code, 0) - data = json.loads(self.output.read_text(encoding="utf-8")) - by_id = {model["id"]: model for model in data["models"]} - self.assertEqual(by_id["higgs"]["path"], - "models/Higgs-Audio-v3-TTS-4B-GGUF") - self.assertEqual(by_id["voxcpm2"]["path"], "models/VoxCPM2-GGUF") - # 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. - # host, port, backend, lazy, sync(n) - inputs = ["", "", "", "", "n"] - with patch.object(sys, "argv", - ["make_audiocpp_server_json.py", - "--wavs", str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.checkout), - "--families", "supertonic"]), \ - 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() - self.assertEqual(code, 0) - out = buf.getvalue() - self.assertIn("no clone-capable family selected", out) - data = json.loads(self.output.read_text(encoding="utf-8")) - self.assertNotIn("voice_dir", data) - 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 ["--wavs", str(self.folder), "--output", str(self.output), - "--audiocpp-dir", str(self.checkout)] + list(extra) - - def _run_capturing(self, argv, inputs, transcribe, whisper): - 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=whisper), \ - redirect_stdout(buf): - code = make_server.main() - return code, buf.getvalue() - - 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") - inputs = self._defaults() - code, out = self._run_capturing( - self._args(), inputs=inputs, - transcribe=lambda path, model_name="base": None, - whisper="faster_whisper") - self.assertEqual(code, 0) - self.assertIn("MANUAL TRANSCRIPTION REQUIRED", out) - self.assertIn("narrator", out) - self.assertIn("alpha", out) - self.assertIn("prompt_text", out) - - def test_missing_whisper_backend_prints_install_warning(self): - (self.folder / "narrator.wav").write_bytes(b"x") - 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("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 = self._defaults() - code, out = self._run_capturing( - self._args(), inputs=inputs, - transcribe=lambda path, model_name="base": "a real transcript", - whisper="faster_whisper") - self.assertEqual(code, 0) - 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 WizardBackNavigationTests(unittest.TestCase): - """Esc-driven back navigation in the _wizard step state machine. - - The tui widget module is mocked so the steps can be driven directly; - Esc is simulated by having the mocked widgets return the _GO_BACK - sentinel (what the real widgets return for Esc via back_value). - """ - - def _settings(self, tmp, confirm_sequence): - checkout = _make_checkout(tmp) - manager = checkout / "tools" / "model_manager_v2.py" - manager.parent.mkdir(parents=True, exist_ok=True) - manager.write_text("#!/usr/bin/env python3\n", encoding="utf-8") - wav_dir = tmp / "wavs" - wav_dir.mkdir() - tui_mock = MagicMock() - tui_mock.menu.return_value = "cuda" - tui_mock.line_edit.side_effect = ["127.0.0.1", "8080"] - tui_mock.browse_directory.return_value = wav_dir - tui_mock.confirm.side_effect = confirm_sequence - args = argparse.Namespace( - audiocpp_dir=checkout, families="qwen3_tts", - all_packages=False, host=None, port=None, backend=None, - lazy_load=False, output=None, force=False, input_dir=None, - whisper_model="base") - with patch.object(make_server, "_load_tui", - return_value=tui_mock): - return make_server._wizard(None, args, None), tui_mock, checkout - - def test_esc_on_download_prompt_returns_to_previous_step(self): - tmp = tempfile.TemporaryDirectory() - self.addCleanup(tmp.cleanup) - # confirms: lazy(True), download Esc(_GO_BACK), download accept(True). - settings, tui_mock, _ = self._settings( - Path(tmp.name), [True, make_server._GO_BACK, True]) - self.assertIsNotNone(settings) - self.assertEqual(settings["backend"], "cuda") - downloads = [call[0][1] for call in tui_mock.confirm.call_args_list - if call[0][1] == "Automatically download the selected " - "models with model_manager_v2.py now?"] - self.assertEqual(len(downloads), 2, - "Esc on the download prompt must re-show it after " - "going back") - - def test_esc_back_to_wav_browser_rebrowses(self): - tmp = tempfile.TemporaryDirectory() - self.addCleanup(tmp.cleanup) - root = Path(tmp.name) - checkout = _make_checkout(root) - wav_dir = root / "wavs" - wav_dir.mkdir() - (wav_dir / "narrator.wav").write_bytes(b"x") - prompt = wav_dir / make_server.PROMPT_TEXT_FILENAME - prompt.write_text("narrator|already transcribed.\n", encoding="utf-8") - tui_mock = MagicMock() - tui_mock.menu.return_value = "cuda" - tui_mock.line_edit.side_effect = ["127.0.0.1", "8080"] - # 1st browse, then re-browse after Esc backs from the transcription - # plan, then browse again only if we re-reached step 4 once more. - tui_mock.browse_directory.side_effect = [wav_dir, wav_dir] - # lazy(True), transcription Esc(_GO_BACK), lazy(True again), - # transcription(True), download(True). - tui_mock.confirm.side_effect = [ - True, make_server._GO_BACK, True, True, True] - args = argparse.Namespace( - audiocpp_dir=checkout, families="qwen3_tts", - all_packages=False, host=None, port=None, backend=None, - lazy_load=False, output=None, force=False, input_dir=None, - whisper_model="base") - with patch.object(make_server, "_load_tui", return_value=tui_mock): - settings = make_server._wizard(None, args, None) - self.assertIsNotNone(settings) - self.assertEqual(tui_mock.browse_directory.call_count, 2, - "Esc on the transcription plan must re-open the " - "wav browser") - - def test_esc_on_first_step_aborts_wizard(self): - tmp = tempfile.TemporaryDirectory() - self.addCleanup(tmp.cleanup) - checkout = _make_checkout(Path(tmp.name)) - tui_mock = MagicMock() - tui_mock.browse_directory.side_effect = tui.WizardCancelled - args = argparse.Namespace( - audiocpp_dir=None, families="qwen3_tts", - all_packages=False, host=None, port=None, backend=None, - lazy_load=False, output=None, force=False, input_dir=None, - whisper_model="base") - with patch.object(make_server, "_load_tui", return_value=tui_mock), \ - self.assertRaises(tui.WizardCancelled): - make_server._wizard(None, args, None) - - def test_wav_browser_starts_in_detected_wav_dir(self): - tmp = tempfile.TemporaryDirectory() - self.addCleanup(tmp.cleanup) - root = Path(tmp.name) - checkout = _make_checkout(root) - # The only .wav directory across the checkout (and the real - # TTS_ROOT, which has none) is voices/ inside the checkout. - voices = checkout / "voices" - voices.mkdir() - (voices / "narrator.wav").write_bytes(b"x") - wav_dir = root / "wavs" - wav_dir.mkdir() - tui_mock = MagicMock() - tui_mock.menu.return_value = "cuda" - tui_mock.line_edit.side_effect = ["127.0.0.1", "8080"] - tui_mock.browse_directory.return_value = wav_dir - tui_mock.confirm.side_effect = [True, True] - args = argparse.Namespace( - audiocpp_dir=checkout, families="qwen3_tts", - all_packages=False, host=None, port=None, backend=None, - lazy_load=False, output=None, force=False, input_dir=None, - whisper_model="base") - with patch.object(make_server, "_load_tui", return_value=tui_mock): - make_server._wizard(None, args, None) - start = tui_mock.browse_directory.call_args[1].get("start") - self.assertEqual(start, voices.resolve()) - - -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_tui.py b/tests/test_tui.py index 94fb638..ba6f99f 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -1,4 +1,4 @@ -"""Tests for the DOS-style curses TUI widgets in tools/tui.py. +"""Tests for the DOS-style curses TUI widgets in tui.py. The widget module imports curses lazily, so these tests swap the curses module for a small fake (patched into sys.modules) and drive @@ -14,7 +14,7 @@ import unittest from pathlib import Path from unittest.mock import patch -from tools import tui +import tui class FakeCurses: @@ -73,6 +73,9 @@ class FakeCurses: def curs_set(self, visibility): pass + def endwin(self): + pass + class FakeScreen: """Recording curses window; getch() replays scripted keys.""" @@ -105,6 +108,9 @@ class FakeScreen: def hline(self, y, x, ch, n, attr=0): pass + def redrawwin(self): + pass + def getch(self): if not self.keys: raise AssertionError("the script ran out of keys") @@ -469,5 +475,32 @@ class CheckboxTreeTests(TuiTestCase): back_value=marker) +class SuspendTests(TuiTestCase): + """tui.suspend leaves curses, runs code, then repaints.""" + + def test_suspend_runs_block_and_restores(self): + ran = [] + with tui.suspend(self.screen): + ran.append("inside") + self.assertEqual(ran, ["inside"]) + + def test_suspend_always_restores_on_exception(self): + class Boom(Exception): + pass + with self.assertRaises(Boom): + with tui.suspend(self.screen): + raise Boom() + + +class FlashTests(TuiTestCase): + """tui.flash shows a notice until any key is pressed.""" + + def test_notice_dismissed_by_any_key(self): + screen = FakeScreen(keys=[10]) + # Should return (None) after consuming one key; not raise. + tui.flash(screen, "a notice", kind="warn") + self.assertEqual(screen.keys, []) + + if __name__ == "__main__": unittest.main() diff --git a/tools/make_faster_voices_json.py b/tools/make_faster_voices_json.py deleted file mode 100755 index 2e00f72..0000000 --- a/tools/make_faster_voices_json.py +++ /dev/null @@ -1,121 +0,0 @@ -#!/usr/bin/env python3 -"""Generate a voices.json file for the faster-qwen3-tts server. - -Scans a directory for .wav files, transcribes each with a local Whisper -backend (faster_whisper or whisper), and writes a voices.json - -Usage: - python tools/make_faster_voices_json.py INPUT_DIR [--output PATH] - [--language LANG] - [--whisper-model NAME] [--force] - -The output can be passed to the faster server: - python examples/openai_server.py --voices voices.json --port 8000 -""" - -import argparse -import json -import sys -from pathlib import Path - -# Allow running from any working directory. -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) - -from converter.tts import normalize_language, transcribe_reference_audio - - -def find_wav_files(input_dir: Path) -> list: - """Return the .wav files in INPUT_DIR, sorted alphabetically by name.""" - return sorted( - (path for path in input_dir.iterdir() - if path.is_file() and path.suffix.lower() == ".wav"), - key=lambda path: path.name.lower(), - ) - - -def prompt_overwrite(output_path: Path) -> bool: - """Ask whether to overwrite an existing output file.""" - while True: - try: - answer = input(f"{output_path} already exists. Overwrite? (y/n): ").strip().lower() - except EOFError: - print("\n[WARNING] No interactive input available; keeping existing file") - return False - if answer in ("y", "yes"): - return True - if answer in ("n", "no"): - return False - print("Please answer 'y' or 'n'.") - - -def build_voices(wav_files: list, language: str, whisper_model: str) -> dict: - """Transcribe each wav file and build the voices mapping.""" - voices = {} - for wav_file in wav_files: - name = wav_file.stem - print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...") - text = transcribe_reference_audio(str(wav_file), model_name=whisper_model) - if text: - print(f"[OK] {name}: {text}") - else: - print(f"[WARNING] No transcript for '{name}'; the faster backend " - "strongly recommends an accurate transcript — consider editing " - "voices.json by hand before starting the server") - voices[name] = { - "ref_audio": str(wav_file.resolve()), - "ref_text": text or "", - "language": language, - } - return voices - - -def main() -> int: - parser = argparse.ArgumentParser( - description="Generate a voices.json for the faster-qwen3-tts server " - "from a directory of .wav reference files.") - parser.add_argument("input_dir", type=Path, - help="Directory containing .wav reference audio files") - parser.add_argument("--output", type=Path, default=None, - help="Output path for voices.json " - "(default: INPUT_DIR/voices.json)") - parser.add_argument("--language", type=str, default="English", - help="Language for all voices, as passed to the TTS model " - "(default: English; names and short codes accepted)") - parser.add_argument("--whisper-model", type=str, default="base", - help="Whisper model size for transcription " - "(default: base)") - parser.add_argument("--force", action="store_true", - help="Overwrite the output file without prompting") - args = parser.parse_args() - - try: - language = normalize_language(args.language) - except ValueError as exc: - parser.error(str(exc)) - - if not args.input_dir.is_dir(): - parser.error(f"Input directory not found: {args.input_dir}") - - wav_files = find_wav_files(args.input_dir) - if not wav_files: - parser.error(f"No .wav files found in {args.input_dir}") - - output_path = args.output if args.output is not None \ - else args.input_dir / "voices.json" - if output_path.exists() and not args.force and not prompt_overwrite(output_path): - print("[INFO] Aborted; existing voices.json kept") - return 1 - - voices = build_voices(wav_files, language, args.whisper_model) - - with output_path.open("w", encoding="utf-8") as handle: - json.dump(voices, handle, indent=4, ensure_ascii=False) - handle.write("\n") - - print(f"[OK] Wrote {output_path} with {len(voices)} voice(s): " - f"{', '.join(voices)}") - return 0 - - -if __name__ == "__main__": - sys.exit(main()) @@ -31,6 +31,7 @@ When the terminal has no color support the theme degrades to bold/reverse/dim. """ +import contextlib import os import textwrap from pathlib import Path @@ -44,6 +45,53 @@ class WizardCancelled(Exception): """Raised when the user presses Esc to abort the wizard.""" +@contextlib.contextmanager +def suspend(scr): + """Temporarily leave curses to run plain-console code. + + Long-running steps that stream output to the terminal (cloning a + repository, building, pip-installing, transcribing) cannot share the + curses screen, so the wizard suspends curses for the duration of the + step and repaints the current screen afterward. ``scr`` is the curses + window returned to the wrapper callback. + """ + import curses + try: + curses.endwin() + except curses.error: + pass + try: + yield + finally: + try: + scr.redrawwin() + scr.refresh() + except Exception: + pass + try: + curses.curs_set(0) + except curses.error: + pass + + +def flash(scr, text: str, kind: str = "warn") -> None: + """Show a one-line notice until any key is pressed, then return. + + Used by the hub for "not set up yet"-style messages. KIND is a theme + key (warn/err/ok/info). Esc dismisses the notice (it does not abort). + """ + frame = Frame(scr, "Notice", "Press any key to continue Esc = back") + frame.mark(text, frame.theme.get(kind, frame.theme["body"])) + frame.cursor = None + frame.draw() + try: + key = scr.getch() + except KeyboardInterrupt: + raise WizardCancelled() from None + if key == 3: # Ctrl-C still aborts + raise WizardCancelled() + + # Esc and 'q' both abort on screens without typed text ('q' is an # ordinary character inside text editors). _CANCEL_KEYS = (27, ord("q")) |