import argparse import contextlib import importlib import io import json import math import os import re import shutil import sys import tempfile import wave from pathlib import Path EXIT_SUCCESS = 0 EXIT_INPUT_ERROR = 1 EXIT_MISSING_RESOURCE = 2 EXIT_RUNTIME_UNAVAILABLE = 3 EXIT_INFERENCE_ERROR = 4 REQUIRED_MODEL_FILES = ( "config.yaml", "bpe.model", "gpt.pth", "s2mel.pth", "wav2vec2bert_stats.pt", "feat1.pt", "feat2.pt", ) REQUIRED_MODEL_DIRS = ( "qwen0.6bemo4-merge", ) REQUIRED_AUX_MODEL_FILES = ( "hf_cache/semantic_codec_model.safetensors", "hf_cache/campplus_cn_common.bin", "hf_cache/bigvgan/config.json", "hf_cache/bigvgan/bigvgan_generator.pt", ) REQUIRED_AUX_MODEL_DIRS = ( "hf_cache/w2v-bert-2.0", ) MODEL_REPO_ID = "IndexTeam/IndexTTS-2" REQUIRED_PACKAGES = ("torch", "torchaudio", "indextts") PERSISTED_CONFIG_KEYS = ( "model_dir", "default_device", "use_fp16", "use_deepspeed", "use_cuda_kernel", "use_accel", "use_torch_compile", ) class InputValidationError(ValueError): pass class BatchFileError(ValueError): def __init__(self, message, exit_code): super().__init__(message) self.exit_code = exit_code class ConcatFileError(ValueError): def __init__(self, message, exit_code): super().__init__(message) self.exit_code = exit_code class ConcatExecutionError(RuntimeError): def __init__(self, message, cleanup_error=None): super().__init__(message) self.cleanup_error = cleanup_error def main(argv=None, tts_factory=None, stdin=None): parser = _build_parser() args = parser.parse_args(argv) if args.command == "init": return _run_init(args) if args.command == "config": return _run_config(args) if args.command == "download": return _run_download(args) if args.command == "check": return _run_check(args) if args.command == "synth": return _run_synth(args, tts_factory=tts_factory, stdin=stdin) if args.command == "batch": return _run_batch(args, tts_factory=tts_factory) if args.command == "concat": return _run_concat(args) parser.print_help(sys.stderr) return EXIT_INPUT_ERROR def _build_parser(): parser = argparse.ArgumentParser(prog="indextts2", description="IndexTTS2 command line") subparsers = parser.add_subparsers(dest="command") init = subparsers.add_parser( "init", help="Create persistent IndexTTS2 CLI state without downloading model resources", ) init.add_argument( "--model-dir", default=None, help="Persist a model resource directory", ) config = subparsers.add_parser( "config", help="Show or update persistent IndexTTS2 CLI configuration", ) config_subparsers = config.add_subparsers(dest="config_command") config_subparsers.add_parser("path", help="Print the persistent configuration file path") config_subparsers.add_parser("get", help="Print the current persistent configuration") config_set = config_subparsers.add_parser("set", help="Persist one configuration value") config_set.add_argument("key", choices=PERSISTED_CONFIG_KEYS) config_set.add_argument("value") download = subparsers.add_parser( "download", help="Download IndexTTS2 model resources", ) download.add_argument( "--source", choices=("huggingface", "modelscope", "auto"), default="auto", help="Model download source (default: auto-detect based on network)", ) download.add_argument( "--model-dir", default=None, help="Path to the IndexTTS2 model resource directory", ) download.add_argument( "--no-save", action="store_true", help="Do not persist --model-dir after a successful download", ) check = subparsers.add_parser( "check", help="Check local IndexTTS2 prerequisites without loading model weights", ) check.add_argument( "--model-dir", default=None, help="Path to the IndexTTS2 model directory", ) check.add_argument( "--device", default=None, help="Required runtime device, e.g. cpu, cuda, cuda:0, mps or xpu", ) batch = subparsers.add_parser( "batch", help="Validate a batch file and run batch synthesis", ) batch.add_argument( "--batch-file", required=True, help="Path to the JSON Lines batch file", ) batch.add_argument( "--model-dir", default=None, help="Path to the IndexTTS2 model directory", ) batch.add_argument( "--dry-run", action="store_true", help="Validate the batch file without loading model weights", ) batch.add_argument("--force", action="store_true", help="Overwrite output if it exists") batch.add_argument( "--output-dir", help="Directory for automatically named independent WAV outputs", ) batch.add_argument( "--output-prefix", help="Filename prefix for automatically named independent WAV outputs", ) batch.add_argument("--concat", action="store_true", help="Generate one concatenated batch output") batch.add_argument("--output", help="Path to write concatenated batch WAV audio") batch.add_argument("--keep-temp", action="store_true", help="Keep internal batch concat temporary files") batch.add_argument("--device", default=None, help="Runtime device") batch.add_argument("--fp16", action=argparse.BooleanOptionalAction, default=None, help="Use FP16 inference") batch.add_argument("--deepspeed", action=argparse.BooleanOptionalAction, default=None, help="Use DeepSpeed") batch.add_argument("--cuda-kernel", action=argparse.BooleanOptionalAction, default=None, help="Use CUDA kernel") batch.add_argument("--accel", action=argparse.BooleanOptionalAction, default=None, help="Use GPT2 acceleration engine") batch.add_argument("--torch-compile", action=argparse.BooleanOptionalAction, default=None, help="Use torch.compile for s2mel optimization") batch.add_argument("--verbose", action="store_true", help="Show verbose inference output") batch.add_argument("--voice", help="Default speaker reference audio for every batch task") batch.add_argument("--emotion-audio", help="Default emotion reference audio for every batch task") batch.add_argument("--emotion-text", help="Default emotion description text for every batch task") batch.add_argument("--emotion-vector", help="Default comma-separated 8-dimensional emotion vector") batch.add_argument( "--emotion-weight", default="1.0", help="Default emotion weight mapped to IndexTTS2 emo_alpha", ) concat = subparsers.add_parser( "concat", help="Validate an audio concat file", ) concat.add_argument( "--concat-file", required=True, help="Path to the JSON Lines concat file", ) concat.add_argument("--output", required=True, help="Path to write concatenated WAV audio") concat.add_argument("--force", action="store_true", help="Overwrite output if it exists") concat.add_argument( "--dry-run", action="store_true", help="Validate the concat file without creating output audio", ) synth = subparsers.add_parser( "synth", help="Synthesize one text input with IndexTTS2", ) synth.add_argument("--text", help="Text to synthesize") synth.add_argument("--text-file", help="UTF-8 text file to synthesize") synth.add_argument("--stdin", action="store_true", help="Read text from standard input") synth.add_argument("--voice", help="Path to the speaker reference audio") synth.add_argument("--emotion-audio", help="Path to the emotion reference audio") synth.add_argument("--emotion-text", help="Emotion description text") synth.add_argument("--emotion-vector", help="Comma-separated 8-dimensional emotion vector") synth.add_argument( "--emotion-weight", default="1.0", help="Emotion weight mapped to IndexTTS2 emo_alpha", ) synth.add_argument("--output", help="Path to write generated audio") synth.add_argument("--force", action="store_true", help="Overwrite output if it exists") synth.add_argument( "--model-dir", default=None, help="Path to the IndexTTS2 model directory", ) synth.add_argument("--device", default=None, help="Runtime device") synth.add_argument("--fp16", action=argparse.BooleanOptionalAction, default=None, help="Use FP16 inference") synth.add_argument("--deepspeed", action=argparse.BooleanOptionalAction, default=None, help="Use DeepSpeed") synth.add_argument("--cuda-kernel", action=argparse.BooleanOptionalAction, default=None, help="Use CUDA kernel") synth.add_argument("--accel", action=argparse.BooleanOptionalAction, default=None, help="Use GPT2 acceleration engine") synth.add_argument("--torch-compile", action=argparse.BooleanOptionalAction, default=None, help="Use torch.compile for s2mel optimization") synth.add_argument("--verbose", action="store_true", help="Show verbose inference output") return parser def _run_init(args): config = _load_persisted_config() if args.model_dir is not None: config["model_dir"] = _normalize_persisted_path(args.model_dir) elif not config.get("model_dir"): config["model_dir"] = _default_model_dir().as_posix() _ensure_user_state(config) print(f"Config: {_config_path()}") print(f"Model directory: {Path(config['model_dir'])}") return EXIT_SUCCESS def _run_config(args): if args.config_command == "path": print(_config_path()) return EXIT_SUCCESS if args.config_command == "get": print(_format_persisted_config(_load_persisted_config()), end="") return EXIT_SUCCESS if args.config_command == "set": config = _load_persisted_config() if args.key == "model_dir": value = _normalize_persisted_path(args.value) config[args.key] = value _config_path().parent.mkdir(parents=True, exist_ok=True) _save_persisted_config(config) print(f"{args.key} = {Path(value)}") return EXIT_SUCCESS if args.key == "default_device": config[args.key] = args.value _config_path().parent.mkdir(parents=True, exist_ok=True) _save_persisted_config(config) print(f"{args.key} = {args.value}") return EXIT_SUCCESS if args.key in {"use_fp16", "use_deepspeed", "use_cuda_kernel", "use_accel", "use_torch_compile"}: value = _parse_config_bool(args.value) if value is None: print(f"ERROR: {args.key} must be true or false", file=sys.stderr) return EXIT_INPUT_ERROR config[args.key] = value _config_path().parent.mkdir(parents=True, exist_ok=True) _save_persisted_config(config) print(f"{args.key} = {str(value).lower()}") return EXIT_SUCCESS print("ERROR: config requires a subcommand: path, get or set", file=sys.stderr) return EXIT_INPUT_ERROR def _run_download(args): model_dir = _resolve_model_dir(args.model_dir) model_dir.mkdir(parents=True, exist_ok=True) try: _download_model_resources(args.source, model_dir) except (ImportError, OSError) as exc: print(f"ERROR: runtime unavailable for {args.source} download source: {exc}", file=sys.stderr) print(f"Install download support with: pip install {_download_support_package(args.source)}", file=sys.stderr) return EXIT_RUNTIME_UNAVAILABLE missing_exit_code = _report_missing_model_resources(model_dir) if missing_exit_code is not None: return EXIT_MISSING_RESOURCE if args.model_dir is not None and not args.no_save: config = _load_persisted_config() config["model_dir"] = _normalize_persisted_path(args.model_dir) _config_path().parent.mkdir(parents=True, exist_ok=True) _save_persisted_config(config) print(f"Downloaded model resources to: {model_dir}") return EXIT_SUCCESS def _download_model_resources(source, model_dir): if source == "auto": from indextts.utils.model_download import snapshot_download snapshot_download(MODEL_REPO_ID, local_dir=str(model_dir)) elif source == "modelscope": from indextts.utils.model_download import _snapshot_from_modelscope _snapshot_from_modelscope(MODEL_REPO_ID, str(model_dir)) else: from huggingface_hub import snapshot_download snapshot_download(repo_id=MODEL_REPO_ID, local_dir=str(model_dir)) if _missing_primary_model_resources(model_dir): return from indextts.utils.model_download import ensure_models_available ensure_models_available(str(model_dir)) def _download_support_package(source): if source == "auto": return "huggingface_hub modelscope" if source == "huggingface": return "huggingface_hub" return "modelscope" def _ensure_user_state(config=None): if config is None: config = _load_persisted_config() model_dir = Path(config.get("model_dir") or _default_model_dir()) config["model_dir"] = model_dir.as_posix() _config_path().parent.mkdir(parents=True, exist_ok=True) model_dir.mkdir(parents=True, exist_ok=True) _save_persisted_config(config) def _resolve_model_dir(model_dir_arg=None): if model_dir_arg is not None: return Path(model_dir_arg).expanduser().resolve(strict=False) env_model_dir = os.environ.get("INDEXTTS2_MODEL_DIR") if env_model_dir: return Path(env_model_dir).expanduser().resolve(strict=False) config = _load_persisted_config() if config.get("model_dir"): return Path(config["model_dir"]).expanduser().resolve(strict=False) return _default_model_dir().resolve(strict=False) def _resolve_runtime_options(args): config = _load_persisted_config() return argparse.Namespace( device=args.device if args.device is not None else config.get("default_device"), fp16=args.fp16 if args.fp16 is not None else bool(config.get("use_fp16", False)), deepspeed=args.deepspeed if args.deepspeed is not None else bool(config.get("use_deepspeed", False)), cuda_kernel=args.cuda_kernel if args.cuda_kernel is not None else bool(config.get("use_cuda_kernel", False)), accel=args.accel if args.accel is not None else bool(config.get("use_accel", False)), torch_compile=args.torch_compile if args.torch_compile is not None else bool(config.get("use_torch_compile", False)), ) def _validate_optional_dependencies(runtime): """Fail fast when acceleration flags are set but optional dependencies are missing.""" if runtime.accel: try: importlib.import_module("flash_attn") except ImportError: print( "ERROR: --accel requires flash-attn, which is not installed. " "Install it with: uv sync --extra accel", file=sys.stderr, ) return EXIT_RUNTIME_UNAVAILABLE if runtime.torch_compile: try: importlib.import_module("triton") except ImportError: print( "ERROR: --torch-compile requires triton, which is not installed. " "Install it with: uv sync --extra torch_compile", file=sys.stderr, ) return EXIT_RUNTIME_UNAVAILABLE return None def _load_persisted_config(): path = _config_path() if not path.is_file(): return {} config = {} for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = [part.strip() for part in line.split("=", 1)] if key not in PERSISTED_CONFIG_KEYS: continue if value in {"true", "false"}: config[key] = value == "true" elif value.startswith('"') and value.endswith('"'): config[key] = _unquote_toml_string(value) return config def _save_persisted_config(config): _config_path().write_text(_format_persisted_config(config), encoding="utf-8") def _format_persisted_config(config): lines = [] for key in PERSISTED_CONFIG_KEYS: if key not in config or config[key] is None: continue value = config[key] if isinstance(value, bool): lines.append(f"{key} = {str(value).lower()}") else: lines.append(f'{key} = "{_quote_toml_string(str(value))}"') return "\n".join(lines) + ("\n" if lines else "") def _quote_toml_string(value): return value.replace("\\", "\\\\").replace('"', '\\"') def _unquote_toml_string(value): inner = value[1:-1] return re.sub(r'\\(["\\])', r"\1", inner) def _normalize_persisted_path(path_value): path = Path(path_value).expanduser() if not path.is_absolute(): path = Path.cwd() / path return path.resolve(strict=False).as_posix() def _parse_config_bool(value): normalized = value.strip().lower() if normalized == "true": return True if normalized == "false": return False return None def _config_path(): if sys.platform == "win32": root = os.environ.get("APPDATA") base = Path(root) if root else Path.home() / "AppData" / "Roaming" return base / "IndexTTS" / "config.toml" if sys.platform == "darwin": return Path.home() / "Library" / "Application Support" / "IndexTTS" / "config.toml" root = os.environ.get("XDG_CONFIG_HOME") base = Path(root) if root else Path.home() / ".config" return base / "indextts" / "config.toml" def _default_model_dir(): if sys.platform == "win32": root = os.environ.get("LOCALAPPDATA") base = Path(root) if root else Path.home() / "AppData" / "Local" return base / "IndexTTS" / "models" / "IndexTTS-2" if sys.platform == "darwin": return Path.home() / "Library" / "Application Support" / "IndexTTS" / "models" / "IndexTTS-2" root = os.environ.get("XDG_DATA_HOME") base = Path(root) if root else Path.home() / ".local" / "share" return base / "indextts" / "models" / "IndexTTS-2" def _run_synth(args, tts_factory=None, stdin=None): if _text_source_count(args) != 1: print("ERROR: provide exactly one text source: --text, --text-file or --stdin", file=sys.stderr) return EXIT_INPUT_ERROR if args.text_file and not Path(args.text_file).is_file(): print(f"ERROR: text file does not exist: {args.text_file}", file=sys.stderr) return EXIT_MISSING_RESOURCE text = _read_synth_text(args, stdin) if not text: print("ERROR: text is empty", file=sys.stderr) return EXIT_INPUT_ERROR if not args.voice: print("ERROR: --voice is required", file=sys.stderr) return EXIT_MISSING_RESOURCE voice_path = Path(args.voice) if not voice_path.is_file(): print(f"ERROR: voice reference audio does not exist: {voice_path}", file=sys.stderr) return EXIT_MISSING_RESOURCE emotion_conflict_error = _emotion_conflict_error(args) if emotion_conflict_error is not None: print(emotion_conflict_error, file=sys.stderr) return EXIT_INPUT_ERROR emotion_vector = None if args.emotion_vector is not None: try: emotion_vector = _parse_emotion_vector(args.emotion_vector) except InputValidationError as exc: print(f"ERROR: {exc}", file=sys.stderr) return EXIT_INPUT_ERROR if args.emotion_text is not None and not args.emotion_text.strip(): print("ERROR: --emotion-text must not be empty", file=sys.stderr) return EXIT_INPUT_ERROR emotion_path = Path(args.emotion_audio) if args.emotion_audio is not None else None if emotion_path is not None and not emotion_path.is_file(): print(f"ERROR: emotion reference audio does not exist: {emotion_path}", file=sys.stderr) return EXIT_MISSING_RESOURCE try: emotion_weight = float(args.emotion_weight) except ValueError: print(f"ERROR: --emotion-weight must be a float: {args.emotion_weight}", file=sys.stderr) return EXIT_INPUT_ERROR if not args.output: print("ERROR: --output is required", file=sys.stderr) return EXIT_INPUT_ERROR output_path = Path(args.output) if output_path.exists() and not args.force: print(f"ERROR: output file already exists: {output_path}", file=sys.stderr) return EXIT_INPUT_ERROR _ensure_user_state() model_dir = _resolve_model_dir(args.model_dir) runtime = _resolve_runtime_options(args) missing_exit_code = _report_missing_model_resources(model_dir) if missing_exit_code is not None: return EXIT_MISSING_RESOURCE output_path.parent.mkdir(parents=True, exist_ok=True) missing_dep_exit = _validate_optional_dependencies(runtime) if missing_dep_exit is not None: return missing_dep_exit if tts_factory is None: try: tts_factory = _load_indextts2(model_dir) except (ImportError, OSError) as exc: print(f"ERROR: runtime unavailable: {exc}", file=sys.stderr) return EXIT_RUNTIME_UNAVAILABLE try: with _synth_stdout_context(args.verbose): tts = tts_factory( cfg_path=str(model_dir / "config.yaml"), model_dir=str(model_dir), use_fp16=runtime.fp16, device=runtime.device, use_cuda_kernel=runtime.cuda_kernel, use_deepspeed=runtime.deepspeed, use_accel=runtime.accel, use_torch_compile=runtime.torch_compile, ) infer_kwargs = { "spk_audio_prompt": str(voice_path), "text": text, "output_path": str(output_path), "verbose": args.verbose, } if emotion_path is not None: infer_kwargs["emo_audio_prompt"] = str(emotion_path) infer_kwargs["emo_alpha"] = emotion_weight if args.emotion_text is not None: infer_kwargs["use_emo_text"] = True infer_kwargs["emo_text"] = args.emotion_text infer_kwargs["emo_alpha"] = emotion_weight if emotion_vector is not None: infer_kwargs["emo_vector"] = emotion_vector infer_kwargs["emo_alpha"] = emotion_weight tts.infer( **infer_kwargs, ) except Exception as exc: print(f"ERROR: inference failed: {exc}", file=sys.stderr) return EXIT_INFERENCE_ERROR print(f"Generated: {output_path}") return EXIT_SUCCESS def _run_batch(args, tts_factory=None): try: defaults = _validate_batch_defaults(args) output_config = _validate_batch_output_config(args) tasks = _load_batch_tasks( Path(args.batch_file), force=args.force, defaults=defaults, output_config=output_config, ) except BatchFileError as exc: print(f"ERROR: {exc}", file=sys.stderr) return exc.exit_code _ensure_user_state() model_dir = _resolve_model_dir(args.model_dir) runtime = _resolve_runtime_options(args) missing_exit_code = _report_missing_model_resources(model_dir) if missing_exit_code is not None: return EXIT_MISSING_RESOURCE if args.dry_run: if output_config["mode"] == "concat": print(f"Batch concat OK: {len(tasks)} tasks") else: print(f"Batch file OK: {len(tasks)} tasks") return EXIT_SUCCESS missing_dep_exit = _validate_optional_dependencies(runtime) if missing_dep_exit is not None: return missing_dep_exit if tts_factory is None: try: tts_factory = _load_indextts2(model_dir) except (ImportError, OSError) as exc: print(f"ERROR: runtime unavailable: {exc}", file=sys.stderr) return EXIT_RUNTIME_UNAVAILABLE verbose = getattr(args, "verbose", False) try: with _synth_stdout_context(verbose): tts = tts_factory( cfg_path=str(model_dir / "config.yaml"), model_dir=str(model_dir), use_fp16=runtime.fp16, device=runtime.device, use_cuda_kernel=runtime.cuda_kernel, use_deepspeed=runtime.deepspeed, use_accel=runtime.accel, use_torch_compile=runtime.torch_compile, ) except Exception as exc: print(f"ERROR: inference failed: {exc}", file=sys.stderr) return EXIT_INFERENCE_ERROR if output_config["mode"] == "concat": return _run_batch_concat(args, tasks, tts, verbose, output_config["output_path"]) for task in tasks: output_path = task["output_path"] output_path.parent.mkdir(parents=True, exist_ok=True) try: with _synth_stdout_context(verbose): infer_kwargs = { "spk_audio_prompt": str(task["voice_path"]), "text": task["text"], "output_path": str(output_path), "verbose": verbose, } infer_kwargs.update(task["emotion_kwargs"]) tts.infer(**infer_kwargs) except Exception as exc: print(f"ERROR: batch file line {task['line_number']} inference failed: {exc}", file=sys.stderr) return EXIT_INFERENCE_ERROR print(f"Generated: {output_path}") print(f"Batch complete: {len(tasks)} tasks generated") return EXIT_SUCCESS def _run_batch_concat(args, tasks, tts, verbose, output_path): temp_dir = None try: output_path.parent.mkdir(parents=True, exist_ok=True) temp_dir = _create_batch_concat_temp_dir(output_path) segments = [] expected_format = None expected_format_line = None for index, task in enumerate(tasks, start=1): segment_path = temp_dir / f"{index:04d}.wav" try: with _synth_stdout_context(verbose): infer_kwargs = { "spk_audio_prompt": str(task["voice_path"]), "text": task["text"], "output_path": str(segment_path), "verbose": verbose, } infer_kwargs.update(task["emotion_kwargs"]) tts.infer(**infer_kwargs) except Exception as exc: print(f"ERROR: batch file line {task['line_number']} inference failed: {exc}", file=sys.stderr) _handle_batch_concat_temp_dir_after_failure(temp_dir, keep_temp=args.keep_temp) return EXIT_INFERENCE_ERROR try: audio_format = _read_concat_wav_format(segment_path, task["line_number"]) except ConcatFileError as exc: print(f"ERROR: batch file line {task['line_number']} inference failed: {exc}", file=sys.stderr) _handle_batch_concat_temp_dir_after_failure(temp_dir, keep_temp=args.keep_temp) return EXIT_INFERENCE_ERROR if expected_format is None: expected_format = audio_format expected_format_line = task["line_number"] elif audio_format != expected_format: print( f"ERROR: batch file line {task['line_number']} inference failed: " f"generated WAV format does not match baseline line {expected_format_line}", file=sys.stderr, ) _handle_batch_concat_temp_dir_after_failure(temp_dir, keep_temp=args.keep_temp) return EXIT_INFERENCE_ERROR segments.append( { "line_number": task["line_number"], "audio_path": segment_path, "silence_after_ms": task["silence_after_ms"], "format": audio_format, } ) try: _concatenate_wav_segments(segments, output_path) except ConcatExecutionError as exc: print(f"ERROR: concat failed: {exc}", file=sys.stderr) if exc.cleanup_error is not None: print(f"WARNING: cleanup failed: {exc.cleanup_error}", file=sys.stderr) _handle_batch_concat_temp_dir_after_failure(temp_dir, keep_temp=args.keep_temp) return EXIT_INFERENCE_ERROR if args.keep_temp: print(f"Generated: {output_path}") print(f"Temp dir: {temp_dir}") else: cleanup_error = _cleanup_batch_concat_temp_dir(temp_dir) if cleanup_error is not None: print(f"ERROR: cleanup failed: {cleanup_error}", file=sys.stderr) return EXIT_INFERENCE_ERROR print(f"Generated: {output_path}") return EXIT_SUCCESS except Exception as exc: print(f"ERROR: batch concat failed: {exc}", file=sys.stderr) _handle_batch_concat_temp_dir_after_failure(temp_dir, keep_temp=args.keep_temp) return EXIT_INFERENCE_ERROR def _create_batch_concat_temp_dir(output_path): return Path(tempfile.mkdtemp(prefix=f".{output_path.name}.", dir=output_path.parent)) def _cleanup_batch_concat_temp_dir(temp_dir): if temp_dir is None: return None try: shutil.rmtree(temp_dir) except OSError as exc: return exc return None def _handle_batch_concat_temp_dir_after_failure(temp_dir, keep_temp=False): if keep_temp: if temp_dir is not None: print(f"Temp dir: {temp_dir}", file=sys.stderr) return cleanup_error = _cleanup_batch_concat_temp_dir(temp_dir) if cleanup_error is not None: print(f"WARNING: cleanup failed: {cleanup_error}", file=sys.stderr) def _run_concat(args): try: output_path = _resolve_command_path(args.output) segments = _load_concat_segments( _resolve_command_path(args.concat_file), output_path, force=args.force, ) except ConcatFileError as exc: print(f"ERROR: {exc}", file=sys.stderr) return exc.exit_code if not args.dry_run: try: _concatenate_wav_segments(segments, output_path) except ConcatExecutionError as exc: print(f"ERROR: concat failed: {exc}", file=sys.stderr) if exc.cleanup_error is not None: print(f"WARNING: cleanup failed: {exc.cleanup_error}", file=sys.stderr) return EXIT_INFERENCE_ERROR print(f"Generated: {output_path}") return EXIT_SUCCESS print(f"Concat file OK: {len(segments)} segments") return EXIT_SUCCESS def _text_source_count(args): return sum((args.text is not None, args.text_file is not None, args.stdin)) def _emotion_source_count(args): return sum( ( args.emotion_audio is not None, args.emotion_text is not None, args.emotion_vector is not None, ) ) def _emotion_conflict_error(args): if _emotion_source_count(args) <= 1: return None if args.emotion_vector is None and args.emotion_audio is not None and args.emotion_text is not None: return "ERROR: --emotion-audio and --emotion-text are mutually exclusive" return "ERROR: --emotion-vector, --emotion-audio and --emotion-text are mutually exclusive" def _read_synth_text(args, stdin): if args.stdin: source = sys.stdin if stdin is None else stdin return source.read().strip() if args.text_file: return Path(args.text_file).read_text(encoding="utf-8").strip() return args.text.strip() def _validate_batch_defaults(args): emotion_conflict_error = _emotion_conflict_error(args) if emotion_conflict_error is not None: raise BatchFileError(_strip_error_prefix(emotion_conflict_error), EXIT_INPUT_ERROR) try: emotion_weight = _parse_emotion_weight(args.emotion_weight, "--emotion-weight") except InputValidationError as exc: raise BatchFileError(str(exc), EXIT_INPUT_ERROR) from exc voice_path = None if args.voice is not None: voice_path = Path(args.voice) if not voice_path.is_file(): raise BatchFileError(f"voice reference audio does not exist: {voice_path}", EXIT_MISSING_RESOURCE) emotion_source = None if args.emotion_audio is not None: emotion_path = Path(args.emotion_audio) if not emotion_path.is_file(): raise BatchFileError( f"emotion reference audio does not exist: {emotion_path}", EXIT_MISSING_RESOURCE, ) emotion_source = ("emotion_audio", emotion_path) elif args.emotion_text is not None: if not args.emotion_text.strip(): raise BatchFileError("--emotion-text must not be empty", EXIT_INPUT_ERROR) emotion_source = ("emotion_text", args.emotion_text) elif args.emotion_vector is not None: try: emotion_source = ("emotion_vector", _parse_emotion_vector(args.emotion_vector)) except InputValidationError as exc: raise BatchFileError(str(exc), EXIT_INPUT_ERROR) from exc return { "voice_path": voice_path, "emotion_source": emotion_source, "emotion_weight": emotion_weight, } def _validate_batch_output_config(args): if args.concat: if args.output_dir is not None: raise BatchFileError("--concat cannot be used with --output-dir", EXIT_INPUT_ERROR) if args.output_prefix is not None: raise BatchFileError("--concat cannot be used with --output-prefix", EXIT_INPUT_ERROR) if args.output is None: raise BatchFileError("--output is required with --concat", EXIT_INPUT_ERROR) output_path = _resolve_command_path(args.output) if not _has_wav_extension(output_path): raise BatchFileError(f"--output must be a .wav file: {output_path}", EXIT_INPUT_ERROR) _reject_batch_auto_output_parent_conflicts(output_path) return {"mode": "concat", "output_path": output_path} if args.output is not None: raise BatchFileError("--output is only valid with --concat", EXIT_INPUT_ERROR) if args.keep_temp: raise BatchFileError("--keep-temp requires --concat", EXIT_INPUT_ERROR) if args.output_prefix is not None and args.output_dir is None: raise BatchFileError("--output-prefix requires --output-dir", EXIT_INPUT_ERROR) if args.output_prefix is not None: _validate_batch_output_prefix(args.output_prefix) if args.output_dir is None: return {"mode": "row"} return { "mode": "auto", "output_dir": _resolve_command_path(args.output_dir), "output_prefix": args.output_prefix, } def _validate_batch_output_prefix(output_prefix): if "/" in output_prefix or "\\" in output_prefix: raise BatchFileError("--output-prefix must not contain path separators", EXIT_INPUT_ERROR) prefix_path = Path(output_prefix) if prefix_path.suffix: raise BatchFileError("--output-prefix must not include a file extension", EXIT_INPUT_ERROR) if not output_prefix.strip(): raise BatchFileError("--output-prefix must not be empty", EXIT_INPUT_ERROR) def _strip_error_prefix(message): prefix = "ERROR: " if message.startswith(prefix): return message[len(prefix) :] return message def _load_batch_tasks(batch_file, force=False, defaults=None, output_config=None): if not batch_file.is_file(): raise BatchFileError(f"batch file does not exist: {batch_file}", EXIT_MISSING_RESOURCE) if defaults is None: defaults = {"voice_path": None, "emotion_source": None, "emotion_weight": 1.0} if output_config is None: output_config = {"mode": "row"} batch_dir = batch_file.parent tasks = [] outputs = {} allowed_fields = { "output", "text", "text_file", "voice", "emotion_audio", "emotion_text", "emotion_vector", "emotion_weight", "silence_after_ms", } for line_number, raw_line in enumerate(batch_file.read_text(encoding="utf-8").splitlines(), start=1): if not raw_line.strip(): continue text_path = None try: task = json.loads(raw_line) except json.JSONDecodeError as exc: raise BatchFileError(f"batch file line {line_number} is not valid JSON: {exc.msg}", EXIT_INPUT_ERROR) from exc if not isinstance(task, dict): raise BatchFileError( f"batch file line {line_number} must be a JSON object", EXIT_INPUT_ERROR, ) unknown_fields = sorted(set(task) - allowed_fields) if unknown_fields: unknown = ", ".join(unknown_fields) raise BatchFileError( f"batch file line {line_number} has unknown fields: {unknown}", EXIT_INPUT_ERROR, ) if "silence_after_ms" in task and output_config["mode"] != "concat": raise BatchFileError( f"batch file line {line_number} field 'silence_after_ms' is only valid with --concat", EXIT_INPUT_ERROR, ) silence_after_ms = _parse_batch_silence_after_ms(task, line_number) text_source_count = sum(key in task for key in ("text", "text_file")) if text_source_count != 1: raise BatchFileError( f"batch file line {line_number} must provide exactly one text source: text or text_file", EXIT_INPUT_ERROR, ) if "text" in task: if not isinstance(task["text"], str): raise BatchFileError( f"batch file line {line_number} field 'text' must be a string", EXIT_INPUT_ERROR, ) text = task["text"].strip() if not text: raise BatchFileError(f"batch file line {line_number} text is empty", EXIT_INPUT_ERROR) else: text_file = _require_batch_string(task, "text_file", line_number) text_path = _resolve_batch_path(batch_dir, text_file) if not text_path.is_file(): raise BatchFileError( f"batch file line {line_number} text file does not exist: {text_path}", EXIT_MISSING_RESOURCE, ) text = text_path.read_text(encoding="utf-8").strip() if not text: raise BatchFileError(f"batch file line {line_number} text is empty", EXIT_INPUT_ERROR) if "voice" in task: voice_path = _resolve_batch_path(batch_dir, _require_batch_string(task, "voice", line_number)) else: voice_path = defaults["voice_path"] if voice_path is None: raise BatchFileError(f"batch file line {line_number} missing required field: voice", EXIT_INPUT_ERROR) if not voice_path.is_file(): raise BatchFileError( f"batch file line {line_number} voice reference audio does not exist: {voice_path}", EXIT_MISSING_RESOURCE, ) emotion_kwargs = _batch_emotion_kwargs(task, batch_dir, line_number, defaults) output_path = _batch_task_output_path( task, batch_dir, line_number, len(tasks) + 1, output_config, ) if output_config["mode"] == "auto": _reject_batch_auto_output_input_conflicts( output_path, line_number, _batch_task_protected_input_paths(batch_file, text_path, voice_path, emotion_kwargs), ) _reject_batch_auto_output_parent_conflicts(output_path) if output_config["mode"] == "concat": _reject_batch_auto_output_input_conflicts( output_path, line_number, _batch_task_protected_input_paths(batch_file, text_path, voice_path, emotion_kwargs), ) else: output_key = str(output_path.resolve(strict=False)) if output_key in outputs: raise BatchFileError( f"batch file line {line_number} has duplicate output path: {output_path}", EXIT_INPUT_ERROR, ) outputs[output_key] = line_number if output_path.exists() and not force: raise BatchFileError( f"batch file line {line_number} output file already exists: {output_path}", EXIT_INPUT_ERROR, ) tasks.append( { "line_number": line_number, "text": text, "voice_path": voice_path, "output_path": output_path, "emotion_kwargs": emotion_kwargs, "silence_after_ms": silence_after_ms, } ) if output_config["mode"] == "concat" and not tasks: _reject_batch_concat_output_manifest_conflict(output_config["output_path"], batch_file) if output_config["mode"] == "concat" and output_config["output_path"].exists() and not force: raise BatchFileError(f"output file already exists: {output_config['output_path']}", EXIT_INPUT_ERROR) return tasks def _batch_task_protected_input_paths(batch_file, text_path, voice_path, emotion_kwargs): protected_paths = [batch_file, voice_path] if text_path is not None: protected_paths.append(text_path) emotion_path = emotion_kwargs.get("emo_audio_prompt") if emotion_path is not None: protected_paths.append(Path(emotion_path)) return protected_paths def _reject_batch_concat_output_manifest_conflict(output_path, batch_file): if _normalized_path_key(output_path) == _normalized_path_key(batch_file): raise BatchFileError( f"generated output conflicts with protected input path: {batch_file}", EXIT_INPUT_ERROR, ) def _reject_batch_auto_output_input_conflicts(output_path, line_number, protected_paths): output_key = _normalized_path_key(output_path) for protected_path in protected_paths: if output_key == _normalized_path_key(protected_path): raise BatchFileError( f"batch file line {line_number} generated output conflicts with protected input path: {protected_path}", EXIT_INPUT_ERROR, ) def _reject_batch_auto_output_parent_conflicts(output_path): parent = output_path.parent existing_parent = parent while not existing_parent.exists(): if existing_parent.parent == existing_parent: break existing_parent = existing_parent.parent if existing_parent.exists() and not existing_parent.is_dir(): raise BatchFileError( f"output parent path cannot be created because a file exists: {existing_parent}", EXIT_INPUT_ERROR, ) def _batch_task_output_path(task, batch_dir, line_number, task_number, output_config): output_value = task.get("output") if output_config["mode"] == "row": if output_value is None: raise BatchFileError(f"batch file line {line_number} missing required field: output", EXIT_INPUT_ERROR) return _resolve_batch_path(batch_dir, _require_batch_string(task, "output", line_number)) if output_config["mode"] == "concat": if output_value is not None: raise BatchFileError( f"batch file line {line_number} field 'output' is not allowed with --concat", EXIT_INPUT_ERROR, ) return output_config["output_path"] if output_value is not None: raise BatchFileError( f"batch file line {line_number} field 'output' is not allowed with --output-dir", EXIT_INPUT_ERROR, ) return output_config["output_dir"] / _auto_batch_output_name(task_number, output_config["output_prefix"]) def _auto_batch_output_name(task_number, output_prefix): stem = f"{task_number:04d}" if output_prefix: stem = f"{output_prefix}-{stem}" return f"{stem}.wav" def _batch_emotion_kwargs(task, batch_dir, line_number, defaults): row_source_fields = [ field_name for field_name in ("emotion_audio", "emotion_text", "emotion_vector") if field_name in task ] if len(row_source_fields) > 1: raise BatchFileError( f"batch file line {line_number} emotion_audio, emotion_text and emotion_vector are mutually exclusive", EXIT_INPUT_ERROR, ) if "emotion_weight" in task: try: emotion_weight = _parse_emotion_weight( task["emotion_weight"], f"batch file line {line_number} field 'emotion_weight'", ) except InputValidationError as exc: raise BatchFileError(str(exc), EXIT_INPUT_ERROR) from exc else: emotion_weight = defaults["emotion_weight"] if row_source_fields: source = _parse_batch_emotion_source(task, row_source_fields[0], batch_dir, line_number) else: source = defaults["emotion_source"] if source is None: if "emotion_weight" in task: raise BatchFileError( f"batch file line {line_number} field 'emotion_weight' requires an emotion source", EXIT_INPUT_ERROR, ) return {} source_name, source_value = source if source_name == "emotion_audio": return {"emo_audio_prompt": str(source_value), "emo_alpha": emotion_weight} if source_name == "emotion_text": return {"use_emo_text": True, "emo_text": source_value, "emo_alpha": emotion_weight} return {"emo_vector": source_value, "emo_alpha": emotion_weight} def _parse_batch_emotion_source(task, field_name, batch_dir, line_number): if field_name == "emotion_audio": emotion_path = _resolve_batch_path(batch_dir, _require_batch_string(task, field_name, line_number)) if not emotion_path.is_file(): raise BatchFileError( f"batch file line {line_number} emotion reference audio does not exist: {emotion_path}", EXIT_MISSING_RESOURCE, ) return ("emotion_audio", emotion_path) if field_name == "emotion_text": emotion_text = _require_batch_string(task, field_name, line_number) return ("emotion_text", emotion_text) try: emotion_vector = _parse_emotion_vector( task[field_name], f"batch file line {line_number} field 'emotion_vector'", ) except InputValidationError as exc: raise BatchFileError(str(exc), EXIT_INPUT_ERROR) from exc return ("emotion_vector", emotion_vector) def _parse_batch_silence_after_ms(task, line_number): if "silence_after_ms" not in task: return 0 value = task["silence_after_ms"] if isinstance(value, bool) or not isinstance(value, int): raise BatchFileError( f"batch file line {line_number} field 'silence_after_ms' must be a non-negative integer", EXIT_INPUT_ERROR, ) if value < 0: raise BatchFileError( f"batch file line {line_number} field 'silence_after_ms' must be a non-negative integer", EXIT_INPUT_ERROR, ) return value def _require_batch_string(task, field_name, line_number): value = task[field_name] if not isinstance(value, str): raise BatchFileError( f"batch file line {line_number} field '{field_name}' must be a string", EXIT_INPUT_ERROR, ) if not value.strip(): raise BatchFileError( f"batch file line {line_number} field '{field_name}' must not be empty", EXIT_INPUT_ERROR, ) return value def _resolve_batch_path(batch_dir, path_value): path = Path(path_value) if not path.is_absolute(): path = batch_dir / path return path def _load_concat_segments(concat_file, output_path, force=False): if not concat_file.is_file(): raise ConcatFileError(f"concat file does not exist: {concat_file}", EXIT_MISSING_RESOURCE) if _normalized_path_key(output_path) == _normalized_path_key(concat_file): raise ConcatFileError("--output must not be the same path as --concat-file", EXIT_INPUT_ERROR) if not _has_wav_extension(output_path): raise ConcatFileError(f"--output must be a .wav file: {output_path}", EXIT_INPUT_ERROR) _reject_concat_output_parent_conflicts(output_path) concat_dir = concat_file.parent segments = [] expected_format = None expected_format_line = None allowed_fields = {"audio", "silence_after_ms"} for line_number, raw_line in enumerate(concat_file.read_text(encoding="utf-8").splitlines(), start=1): if not raw_line.strip(): continue try: segment = json.loads(raw_line) except json.JSONDecodeError as exc: raise ConcatFileError( f"concat file line {line_number} is not valid JSON: {exc.msg}", EXIT_INPUT_ERROR, ) from exc if not isinstance(segment, dict): raise ConcatFileError( f"concat file line {line_number} must be a JSON object", EXIT_INPUT_ERROR, ) unknown_fields = sorted(set(segment) - allowed_fields) if unknown_fields: unknown = ", ".join(unknown_fields) raise ConcatFileError( f"concat file line {line_number} has unknown fields: {unknown}", EXIT_INPUT_ERROR, ) audio_path = _resolve_concat_audio_path(concat_dir, _require_concat_string(segment, "audio", line_number)) if not _has_wav_extension(audio_path): raise ConcatFileError( f"concat file line {line_number} field 'audio' must be a .wav file: {audio_path}", EXIT_INPUT_ERROR, ) silence_after_ms = _parse_concat_silence_after_ms(segment, line_number) audio_format = _read_concat_wav_format(audio_path, line_number) if expected_format is None: expected_format = audio_format expected_format_line = line_number elif audio_format != expected_format: raise ConcatFileError( f"concat file line {line_number} WAV format does not match baseline line {expected_format_line}", EXIT_INPUT_ERROR, ) segments.append( { "line_number": line_number, "audio_path": audio_path, "silence_after_ms": silence_after_ms, "format": audio_format, } ) if not segments: raise ConcatFileError("concat file must contain at least one segment", EXIT_INPUT_ERROR) _reject_concat_input_conflicts(output_path, segments) _reject_concat_output_file_conflicts(output_path, force=force) return segments def _concatenate_wav_segments(segments, output_path): temp_path = None try: output_path.parent.mkdir(parents=True, exist_ok=True) temp_path = _create_concat_temp_path(output_path) _write_concat_wav(temp_path, segments) os.replace(temp_path, output_path) temp_path = None except Exception as exc: cleanup_error = None if temp_path is not None: cleanup_error = _cleanup_concat_temp_file(temp_path) raise ConcatExecutionError(str(exc), cleanup_error=cleanup_error) from exc def _create_concat_temp_path(output_path): with tempfile.NamedTemporaryFile( prefix=f".{output_path.name}.", suffix=".wav", dir=output_path.parent, delete=False, ) as temp_file: return Path(temp_file.name) def _write_concat_wav(temp_path, segments): frame_rate, channels, sample_width = segments[0]["format"] with wave.open(str(temp_path), "wb") as output_wav: output_wav.setnchannels(channels) output_wav.setsampwidth(sample_width) output_wav.setframerate(frame_rate) for segment in segments: with wave.open(str(segment["audio_path"]), "rb") as input_wav: output_wav.writeframes(input_wav.readframes(input_wav.getnframes())) silence_frames = frame_rate * segment["silence_after_ms"] // 1000 if silence_frames: output_wav.writeframes(b"\0" * channels * sample_width * silence_frames) def _cleanup_concat_temp_file(temp_path): try: temp_path.unlink(missing_ok=True) except OSError as exc: return exc return None def _resolve_command_path(path_value): path = Path(path_value) if not path.is_absolute(): path = Path.cwd() / path return path def _resolve_concat_audio_path(concat_dir, path_value): path = Path(path_value) if not path.is_absolute(): path = concat_dir / path return path def _has_wav_extension(path): return path.suffix.lower() == ".wav" def _normalized_path_key(path): return str(path.resolve(strict=False)).casefold() def _reject_concat_output_parent_conflicts(output_path): parent = output_path.parent existing_parent = parent while not existing_parent.exists(): if existing_parent.parent == existing_parent: break existing_parent = existing_parent.parent if existing_parent.exists() and not existing_parent.is_dir(): raise ConcatFileError( f"output parent path cannot be created because a file exists: {existing_parent}", EXIT_INPUT_ERROR, ) def _reject_concat_output_file_conflicts(output_path, force=False): if output_path.exists() and not force: raise ConcatFileError(f"output file already exists: {output_path}", EXIT_INPUT_ERROR) def _reject_concat_input_conflicts(output_path, segments): output_key = _normalized_path_key(output_path) for segment in segments: if output_key == _normalized_path_key(segment["audio_path"]): raise ConcatFileError( f"concat file line {segment['line_number']} audio conflicts with --output: {segment['audio_path']}", EXIT_INPUT_ERROR, ) def _require_concat_string(segment, field_name, line_number): if field_name not in segment: raise ConcatFileError(f"concat file line {line_number} missing required field: {field_name}", EXIT_INPUT_ERROR) value = segment[field_name] if not isinstance(value, str): raise ConcatFileError( f"concat file line {line_number} field '{field_name}' must be a string", EXIT_INPUT_ERROR, ) if not value.strip(): raise ConcatFileError( f"concat file line {line_number} field '{field_name}' must not be empty", EXIT_INPUT_ERROR, ) return value def _parse_concat_silence_after_ms(segment, line_number): if "silence_after_ms" not in segment: return 0 value = segment["silence_after_ms"] if isinstance(value, bool) or not isinstance(value, int): raise ConcatFileError( f"concat file line {line_number} field 'silence_after_ms' must be a non-negative integer", EXIT_INPUT_ERROR, ) if value < 0: raise ConcatFileError( f"concat file line {line_number} field 'silence_after_ms' must be a non-negative integer", EXIT_INPUT_ERROR, ) return value def _read_concat_wav_format(audio_path, line_number): if not audio_path.is_file(): raise ConcatFileError( f"concat file line {line_number} audio file does not exist: {audio_path}", EXIT_MISSING_RESOURCE, ) try: with wave.open(str(audio_path), "rb") as wav_file: audio_format = ( wav_file.getframerate(), wav_file.getnchannels(), wav_file.getsampwidth(), ) frame_count = wav_file.getnframes() except (wave.Error, EOFError, OSError) as exc: raise ConcatFileError( f"concat file line {line_number} audio file is not a readable WAV: {audio_path}", EXIT_INPUT_ERROR, ) from exc if frame_count <= 0: raise ConcatFileError( f"concat file line {line_number} audio file is empty: {audio_path}", EXIT_INPUT_ERROR, ) return audio_format def _parse_emotion_vector(value, label="--emotion-vector"): if isinstance(value, str): value = value.strip() if not value: raise InputValidationError(f"{label} must not be empty") if value.startswith("[") and value.endswith("]"): value = value[1:-1] if not value.strip(): raise InputValidationError(f"{label} must not be empty") parts = [part.strip() for part in value.split(",")] elif isinstance(value, list): if not value: raise InputValidationError(f"{label} must not be empty") if any(isinstance(part, bool) for part in value): raise InputValidationError(f"{label} entries must be numeric") parts = value else: raise InputValidationError(f"{label} must be a string or JSON array") try: emotion_vector = [float(part) for part in parts] except (TypeError, ValueError) as exc: raise InputValidationError(f"{label} entries must be numeric") from exc if len(emotion_vector) != 8: raise InputValidationError(f"{label} must contain exactly 8 values; got {len(emotion_vector)}") out_of_range = [item for item in emotion_vector if not math.isfinite(item) or item < 0.0 or item > 1.0] if out_of_range: raise InputValidationError(f"{label} values must be between 0.0 and 1.0") vector_sum = sum(emotion_vector) if vector_sum > 0.8: raise InputValidationError(f"{label} sum must be <= 0.8; got {vector_sum:g}") return emotion_vector def _parse_emotion_weight(value, label): if isinstance(value, bool): raise InputValidationError(f"{label} must be a float: {value}") try: return float(value) except (TypeError, ValueError) as exc: raise InputValidationError(f"{label} must be a float: {value}") from exc def _load_indextts2(model_dir=None): if model_dir is not None: _configure_hf_cache(model_dir) from indextts.infer_v2 import IndexTTS2 return IndexTTS2 def _configure_hf_cache(model_dir): os.environ["HF_HUB_CACHE"] = str(Path(model_dir) / "hf_cache") def _synth_stdout_context(verbose): if verbose: return contextlib.nullcontext() return contextlib.redirect_stdout(io.StringIO()) def _run_check(args): _ensure_user_state() model_dir = _resolve_model_dir(args.model_dir) missing_exit_code = _report_missing_model_resources(model_dir) if missing_exit_code is not None: return EXIT_MISSING_RESOURCE imports = _import_required_packages() if imports.missing: missing = ", ".join(imports.missing) print(f"ERROR: missing required Python packages: {missing}", file=sys.stderr) return EXIT_RUNTIME_UNAVAILABLE devices = _detect_devices(imports.torch) if args.device and not _is_requested_device_available(imports.torch, devices, args.device): print(f"ERROR: requested device is not available: {args.device}", file=sys.stderr) return EXIT_RUNTIME_UNAVAILABLE print(f"Checking model directory: {model_dir}") print(f"OK: model directory {model_dir}") print("OK: required model files") print("OK: python packages") for device in ("cuda", "xpu", "mps", "cpu"): status = "available" if devices[device] else "unavailable" print(f"{device}: {status}") return EXIT_SUCCESS def _report_missing_model_resources(model_dir): missing_files = _missing_model_files(model_dir) if missing_files is None: print(f"ERROR: model directory does not exist: {model_dir}", file=sys.stderr) _print_model_resource_help(model_dir, "model directory does not exist") return EXIT_MISSING_RESOURCE if missing_files: missing = ", ".join(missing_files) print(f"ERROR: missing required model files: {missing}", file=sys.stderr) _print_model_resource_help(model_dir, missing) return EXIT_MISSING_RESOURCE return None def _print_model_resource_help(model_dir, missing_summary): print(f"Model directory: {model_dir}", file=sys.stderr) print(f"Missing resources: {missing_summary}", file=sys.stderr) print("Download with HuggingFace:", file=sys.stderr) print(f' huggingface-cli download {MODEL_REPO_ID} --local-dir "{model_dir}"', file=sys.stderr) print("Download with ModelScope:", file=sys.stderr) print(f' modelscope download --model {MODEL_REPO_ID} --local_dir "{model_dir}"', file=sys.stderr) print("Persist a different model resource directory:", file=sys.stderr) print(f" indextts2 config set model_dir {model_dir}", file=sys.stderr) print("Hint: rerun indextts2 download or choose a different model resource directory.", file=sys.stderr) def _missing_model_files(model_dir): if not model_dir.is_dir(): return None missing_files = _missing_primary_model_resources(model_dir) missing_aux_files = [ filename for filename in REQUIRED_AUX_MODEL_FILES if not _model_resource_path(model_dir, filename).is_file() ] missing_aux_dirs = [ dirname for dirname in REQUIRED_AUX_MODEL_DIRS if not _model_resource_path(model_dir, dirname).is_dir() ] return missing_files + missing_aux_files + missing_aux_dirs def _missing_primary_model_resources(model_dir): missing_files = [filename for filename in REQUIRED_MODEL_FILES if not (model_dir / filename).is_file()] missing_dirs = [dirname for dirname in REQUIRED_MODEL_DIRS if not (model_dir / dirname).is_dir()] return missing_files + missing_dirs def _model_resource_path(model_dir, relative_path): return model_dir.joinpath(*relative_path.split("/")) def _import_required_packages(): missing = [] imported = {} for package in REQUIRED_PACKAGES: try: imported[package] = importlib.import_module(package) except (ImportError, OSError): missing.append(package) return argparse.Namespace(missing=missing, torch=imported.get("torch")) def _detect_devices(torch_module): return { "cuda": _is_available(torch_module, "cuda"), "xpu": _is_available(torch_module, "xpu"), "mps": _is_mps_available(torch_module), "cpu": True, } def _is_available(torch_module, name): device_backend = getattr(torch_module, name, None) is_available = getattr(device_backend, "is_available", None) return bool(is_available and is_available()) def _is_mps_available(torch_module): backends = getattr(torch_module, "backends", None) mps = getattr(backends, "mps", None) is_available = getattr(mps, "is_available", None) return bool(is_available and is_available()) def _is_requested_device_available(torch_module, devices, device): family = _device_family(device) if not devices.get(family, False): return False if family in {"cuda", "xpu"}: index = _device_index(device, family) if index is None: return True return _indexed_device_available(torch_module, family, index) return device == family def _device_index(device, family): if device == family: return None prefix = f"{family}:" if not device.startswith(prefix): return -1 try: return int(device[len(prefix) :]) except ValueError: return -1 def _indexed_device_available(torch_module, family, index): if index < 0: return False device_backend = getattr(torch_module, family, None) device_count = getattr(device_backend, "device_count", None) if device_count is None: return False return index < device_count() def _device_family(device): if device == "cuda" or device.startswith("cuda:"): return "cuda" if device == "xpu" or device.startswith("xpu:"): return "xpu" if device == "mps": return "mps" if device == "cpu": return "cpu" return device if __name__ == "__main__": raise SystemExit(main())