"""Nida command-line interface. Every subcommand accepts the common run flags (``--runs-root``, ``--run-id``, ``--git-commit``) so the Colab runner can call any command the same way. """ from __future__ import annotations import argparse import json import logging import sys from collections.abc import Sequence from pathlib import Path from nida.envreport import collect_env, summarize_env from nida.runs import RunContext from nida.utils.io import atomic_write_json, atomic_write_text __all__ = ["main"] # src/nida/cli.py → repository root (editable installs, as on Colab). _REPO_DIR = Path(__file__).resolve().parents[2] def _common_parser() -> argparse.ArgumentParser: common = argparse.ArgumentParser(add_help=False) common.add_argument( "--runs-root", type=Path, default=None, help="Directory holding run folders. If omitted, nothing is written to disk.", ) common.add_argument("--run-id", default=None, help="Explicit run id (default: generated).") common.add_argument("--git-commit", default=None, help="Commit SHA of the code being run.") return common def _positive_int(text: str) -> int: """argparse type: an integer >= 1 (clear error otherwise).""" try: value = int(text) except ValueError: raise argparse.ArgumentTypeError(f"expected an integer >= 1, got {text!r}") from None if value < 1: raise argparse.ArgumentTypeError(f"expected an integer >= 1, got {value}") return value def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="nida", description="Nida expressive Turkish TTS.") sub = parser.add_subparsers(dest="command", required=True) sub.add_parser( "env-report", parents=[_common_parser()], help="Print (and optionally record) Python/torch/GPU facts.", ) bench = sub.add_parser( "bench-phase0", parents=[_common_parser()], help="Phase 0 benchmarks B1 (ASR) and B2 (VAE); see docs/benchmarks/B1-B2-tasarim.md.", ) bench.add_argument("--stage", default="all", help="A stage name or 'all'.") bench.add_argument( "--workspace", type=Path, default=None, help="Durable benchmark directory (default: /../bench/phase0).", ) bench.add_argument("--cache", type=Path, default=Path("/content/nida_cache")) bench.add_argument("--envs", type=Path, default=Path("/content/nida_envs")) bench.add_argument( "--repo", type=Path, default=_REPO_DIR, help="Repository root (for configs and envs)." ) bench.add_argument("--c-results", type=Path, default=None, help="B1 layer-C panel results CSV.") bench.add_argument( "--b2-panel-results", type=Path, default=None, help="B2 listening panel results CSV." ) bench.add_argument( "--retry-failed", action="store_true", help="Retry worker items that failed before." ) pilot = sub.add_parser( "data-pilot", parents=[_common_parser()], help="Phase 1 data pipeline pilot on YODAS (docs/plans/faz-1.md).", ) pilot.add_argument("--stage", default="all", help="A stage name or 'all'.") pilot.add_argument( "--root", type=Path, default=None, help="Durable pilot directory (default: /../data/pilot).", ) pilot.add_argument("--cache", type=Path, default=Path("/content/nida_cache/pilot")) pilot.add_argument("--envs", type=Path, default=Path("/content/nida_envs")) pilot.add_argument("--repo", type=Path, default=_REPO_DIR) pilot.add_argument( "--subset", choices=("tr000", "tr100"), # keep in sync with nida.data.pilot.YODAS_SUBSETS default="tr000", help=( "YODAS Turkish subset: tr000 (manual subtitles, 64 shards) or tr100 (automatic " "subtitles, 430 shards; videos that occur in tr000 are dropped)." ), ) pilot.add_argument( "--shards", default="00000002", help="Comma-separated shards of the chosen subset." ) pilot.add_argument("--max-hours", type=float, default=10.0) pilot.add_argument("--asr-primary", default="omniasr_llm_7b") pilot.add_argument("--asr-secondary", default="qwen3_asr_1_7b") pilot.add_argument("--retry-failed", action="store_true") pilot.add_argument( "--asr-batch-size", type=int, default=None, help="Override the ASR workers' batch size (bench sizes fit a 22 GB L4).", ) pilot.add_argument( "--parallel-workers", type=_positive_int, default=1, help=( "Run up to N of the independent GPU worker stages (sed, diar, both ASR systems, " "ctc, emotion) at the same time in 'all' (default 1 = one after another)." ), ) ears = sub.add_parser( "data-ears", parents=[_common_parser()], help="EARS -> latent dataset (ADR-0007; docs/03-data-pipeline.md).", ) ears.add_argument( "--root", type=Path, default=None, help="Durable directory (default: /../data/ears); dataset is /dataset.", ) ears.add_argument( "--speakers", default="all", help="Comma-separated speakers (p001,p002) or 'all'." ) ears.add_argument("--cache", type=Path, default=Path("/content/nida_cache/ears")) ears.add_argument("--envs", type=Path, default=Path("/content/nida_envs")) ears.add_argument("--repo", type=Path, default=_REPO_DIR) ears.add_argument("--seed", type=int, default=20261009, help="Speaker split seed.") ears.add_argument("--config-dir", type=Path, default=None, help="Default: /configs/data.") imp = sub.add_parser( "data-import", parents=[_common_parser()], help="External transcribed corpus -> latent dataset (nida.data.importer).", ) imp.add_argument( "--source", # keep in sync with nida.data.importer.sources.SOURCES choices=("alania", "tr_full", "wavdir", "issai_tsc", "mediaspeech_tr", "common_voice_tr"), required=True, help=( "alania (HF parquet), tr_full (HF data.jsonl + wavs), wavdir (--source-dir), " "issai_tsc (HF tar.gz), mediaspeech_tr (OpenSLR TR.tgz) or common_voice_tr " "(--source-dir: a Common Voice release folder or tarball you downloaded)." ), ) imp.add_argument( "--stage", default="all", help="select | audio | latents | ctc | timings | dataset | report | all.", ) imp.add_argument( "--root", type=Path, default=None, help="Durable directory (default: /../data/import-).", ) imp.add_argument( "--cache", type=Path, default=None, help="Local scratch (default: /content/nida_cache/import/).", ) imp.add_argument("--envs", type=Path, default=Path("/content/nida_envs")) imp.add_argument("--repo", type=Path, default=_REPO_DIR) imp.add_argument("--config-dir", type=Path, default=None, help="Default: /configs/data.") imp.add_argument("--max-hours", type=float, default=10.0, help="Stop selecting at this many h.") imp.add_argument( "--shard-offset", type=int, default=0, help="First work unit (parquet shard / row chunk); roots take disjoint ranges.", ) imp.add_argument( "--shards", type=_positive_int, default=None, help="Number of work units (default: rest)." ) imp.add_argument( "--source-dir", type=Path, default=None, help="wavdir: folder with manifest.jsonl; common_voice_tr: release folder or tarball.", ) imp.add_argument( "--revision", default=None, help=( "Hub commit sha to pin (default: current head, recorded); mediaspeech_tr: sha256 of " "TR.tgz (default: the one in configs/data/import_corpora.yaml)." ), ) imp.add_argument("--retry-failed", action="store_true") imp.add_argument( "--audio-threads", type=_positive_int, default=4, help="Threads converting raw audio to 16 kHz wavs (does not change the outputs).", ) train = sub.add_parser( "train", parents=[_common_parser()], help="Phase 2 flow-matching training on a latent dataset (docs/04-architecture.md).", ) train.add_argument("--config", type=Path, required=True, help="YAML under configs/train/.") train.add_argument( "--data", type=Path, required=True, action="append", help="Latent dataset directory; repeat the flag to train on the union of several.", ) train.add_argument( "--out", type=Path, required=True, help="Training workspace (checkpoints, metrics.jsonl)." ) train.add_argument( "--resume", action="store_true", help="Continue from the newest checkpoint in --out." ) train.add_argument("--device", default=None, help="cpu | cuda | auto (default: auto).") train.add_argument( "--init-from", type=Path, default=None, help="Fine-tune: start from this checkpoint's weights and latent stats (same model " "config, except dur_cond none -> emotion) when --out has no checkpoint yet; " "optimizer and schedule start fresh.", ) samples = sub.add_parser( "synth-samples", parents=[_common_parser()], help="Listening page: validation items re-spoken by a trained checkpoint.", ) samples.add_argument( "--checkpoint", type=Path, required=True, help="ckpt-*.pt or a training workspace." ) samples.add_argument("--data", type=Path, required=True, help="Latent dataset directory.") samples.add_argument("--out", type=Path, required=True, help="Output directory.") samples.add_argument("--n", type=int, default=12, help="Validation items to generate.") samples.add_argument("--steps", type=int, default=32, help="ODE steps.") samples.add_argument("--w-text", type=float, default=2.0) # Keeps 1.0 on purpose (sample CER stays comparable with earlier runs); `say` uses 2.0. samples.add_argument("--w-emo", type=float, default=1.0) samples.add_argument("--seed", type=int, default=0) samples.add_argument("--method", choices=["euler", "midpoint"], default="euler") samples.add_argument( "--sway", type=float, default=-1.0, help="Sway time-step warp (0 = uniform steps)." ) samples.add_argument("--envs", type=Path, default=Path("/content/nida_envs")) samples.add_argument("--repo", type=Path, default=_REPO_DIR) samples.add_argument("--device", default=None, help="cpu | cuda | auto (default: auto).") samples.add_argument( "--no-asr", action="store_true", help="Skip the Qwen3-ASR read-back (CER) check." ) say = sub.add_parser( "say", parents=[_common_parser()], help="Speak free text after a validation speaker prompt (aligned checkpoints).", ) say.add_argument("--checkpoint", type=Path, required=True, help="ckpt-*.pt or a workspace.") say.add_argument( "--data", type=Path, default=None, help="Latent dataset (prompt source); not needed with only --prompt-list prompts.", ) say.add_argument("--out", type=Path, required=True, help="Output directory.") text = say.add_mutually_exclusive_group(required=True) text.add_argument( "--text", help=( "Text to speak (sentences split on . ! ? …; long ones also at , ; :). Numbers, " "%%, TL, clock times, abbreviations and acronyms need not be written out: they are " "normalised to spoken Turkish first (--no-normalize disables this). " "'@mutlu:0.8' sets the emotion of the text after it; '[duraksama:0.3]' inserts " "exactly 0.3 s of silence (it ends the sentence); '[duraksama]' lets the model pause." ), ) text.add_argument("--text-file", type=Path, help="UTF-8 file with the text to speak.") say.add_argument( "--prompt-id", action="append", default=None, help="Validation segment (repeatable)." ) say.add_argument( "--n-prompts", type=int, default=3, help="Prompts to pick when no --prompt-id is given." ) say.add_argument( "--prompt-list", type=Path, default=None, help="TSV '\t' of prompts from outside the datasets.", ) say.add_argument( "--rank-prompts", type=int, default=0, help="Decode this many prompt candidates and keep the cleanest --n-prompts (0 = off).", ) say.add_argument("--steps", type=int, default=32, help="ODE steps.") say.add_argument("--w-text", type=float, default=3.0) say.add_argument("--w-emo", type=float, default=2.0) say.add_argument("--seed", type=int, default=0) say.add_argument("--method", choices=["euler", "midpoint"], default="euler") say.add_argument("--sway", type=float, default=-1.0) say.add_argument("--speed", type=float, default=1.0, help=">1 faster, <1 slower.") say.add_argument("--envs", type=Path, default=Path("/content/nida_envs")) say.add_argument("--repo", type=Path, default=_REPO_DIR) say.add_argument("--device", default=None, help="cpu | cuda | auto (default: auto).") say.add_argument("--no-asr", action="store_true", help="Skip the ASR read-back check.") say.add_argument( "--no-normalize", action="store_true", help=( "Use the text as written: no expansion of digits, symbols, abbreviations or " "acronyms (they must then be spelled out, or the text is rejected)." ), ) say.add_argument( "--strict-normalize", action="store_true", help=( "Stop on every text-normalisation issue (a guessed acronym, an ambiguous time or " "ordinal, ...); by default only issues that would be misread stop the run." ), ) say.add_argument( "--single-env", action="store_true", help=( "Run every worker (VAE, CTC, ASR) with this interpreter instead of building one " "venv per worker (a local install with all packages, e.g. a release user)." ), ) queue = sub.add_parser( "queue", parents=[_common_parser()], help="Run nida.cli commands listed in a file, one after another (nida.jobqueue).", ) queue.add_argument("--file", type=Path, required=True, help="Queue file (one job a line).") queue.add_argument( "--wait-minutes", type=float, default=0.0, help="Keep polling the file this long after the last job before returning.", ) queue.add_argument( "--bundle-dir", type=Path, default=None, help="Default: /../code." ) queue.add_argument("--log-dir", type=Path, default=Path("/content/nida_logs")) queue.add_argument("--repo", type=Path, default=_REPO_DIR) return parser def _env_report_md(metrics: dict[str, object]) -> str: def show(value: object) -> str: return "—" if value is None else str(value) lines = ["# env-report", ""] if metrics["gpu_name"] is None: lines.append("- device: CPU only") else: lines.append(f"- device: {metrics['gpu_name']} (count {metrics['gpu_count']})") for key in ( "gpu_memory_gb", "compute_capability", "bf16_supported", "torch_version", "cuda_version", "torchaudio_version", "python", "cpu_count", "ram_total_gb", "disk_free_gb", ): lines.append(f"- {key}: {show(metrics[key])}") return "\n".join(lines) + "\n" def _resolve_device(choice: str | None, cuda_available: bool) -> str: """``--device``: ``None`` or ``auto`` → cuda when available, else cpu; other values as is.""" if choice in (None, "auto"): return "cuda" if cuda_available else "cpu" return str(choice) def _cmd_env_report(args: argparse.Namespace) -> int: env = collect_env() print(json.dumps(env, ensure_ascii=False, indent=2, sort_keys=True)) if args.runs_root is None: return 0 with RunContext( args.runs_root, "env-report", run_id=args.run_id, git_commit=args.git_commit ) as run: atomic_write_json(run.dir / "env.json", env) metrics = summarize_env(env) atomic_write_json(run.dir / "metrics.json", metrics) atomic_write_text(run.dir / "report.md", _env_report_md(metrics)) print(f"run_id={run.run_id}") return 0 def _cmd_bench_phase0(args: argparse.Namespace) -> int: from nida.bench.phase0 import Phase0 from nida.bench.prepare import Workspace logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") workspace_root = args.workspace if workspace_root is None: if args.runs_root is None: raise SystemExit("bench-phase0 needs --workspace or --runs-root") workspace_root = args.runs_root.parent / "bench" / "phase0" phase0 = Phase0( workspace=Workspace(root=workspace_root, cache=args.cache), envs_root=args.envs, repo_dir=args.repo, sources_path=args.repo / "configs" / "sources.yaml", retry_failed=args.retry_failed, ) if args.runs_root is None: result = phase0.run( args.stage, c_results=args.c_results, b2_panel_results=args.b2_panel_results ) print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) return 0 with RunContext( args.runs_root, f"bench-phase0-{args.stage}", run_id=args.run_id, git_commit=args.git_commit ) as run: result = phase0.run( args.stage, c_results=args.c_results, b2_panel_results=args.b2_panel_results ) atomic_write_json(run.dir / "metrics.json", json.loads(json.dumps(result, default=str))) reports = [ (workspace_root / "reports" / name) for name in ("b1_report.md", "b2_report.md", "b2_panel_report.md") if (workspace_root / "reports" / name).exists() ] body = "\n\n".join(r.read_text(encoding="utf-8") for r in reports) atomic_write_text( run.dir / "report.md", f"# bench-phase0 ({args.stage})\n\nworkspace: {workspace_root}\n\n{body}\n", ) print(f"run_id={run.run_id}") return 0 def _cmd_data_pilot(args: argparse.Namespace) -> int: from nida.data.pilot import Pilot, PilotPlan logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") root = args.root if root is None: if args.runs_root is None: raise SystemExit("data-pilot needs --root or --runs-root") root = args.runs_root.parent / "data" / "pilot" pilot = Pilot( root=root, cache=args.cache, envs_root=args.envs, repo_dir=args.repo, sources_path=args.repo / "configs" / "sources.yaml", config_dir=args.repo / "configs" / "data", plan=PilotPlan( subset=args.subset, shards=args.shards, max_hours=args.max_hours, asr_primary=args.asr_primary, asr_secondary=args.asr_secondary, ), retry_failed=args.retry_failed, asr_batch_size=args.asr_batch_size, parallel_workers=args.parallel_workers, ) if args.runs_root is None: print(json.dumps(pilot.run(args.stage), ensure_ascii=False, indent=2, default=str)) return 0 with RunContext( args.runs_root, f"data-pilot-{args.stage}", run_id=args.run_id, git_commit=args.git_commit ) as run: result = pilot.run(args.stage) atomic_write_json(run.dir / "metrics.json", json.loads(json.dumps(result, default=str))) report = root / "reports" / "pilot.md" body = report.read_text(encoding="utf-8") if report.exists() else "" result_json = json.dumps(result, ensure_ascii=False, indent=2, default=str) lines = [ f"# data-pilot ({args.stage})", "", f"root: {root}", "", "```json", result_json, "```", "", body, ] atomic_write_text(run.dir / "report.md", "\n".join(lines) + "\n") print(f"run_id={run.run_id}") return 0 def _cmd_data_ears(args: argparse.Namespace) -> int: from nida.bench import fetch from nida.bench.b1 import _stream_runner from nida.data.ears import Ears, load_settings, parse_speakers logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") root = args.root if root is None: if args.runs_root is None: raise SystemExit("data-ears needs --root or --runs-root") root = args.runs_root.parent / "data" / "ears" config_dir = args.config_dir or args.repo / "configs" / "data" speakers = parse_speakers(args.speakers, load_settings(config_dir).ears.n_speakers) ears = Ears( root=root, cache=args.cache, envs_root=args.envs, repo_dir=args.repo, sources_path=args.repo / "configs" / "sources.yaml", config_dir=config_dir, seed=args.seed, runner=_stream_runner, opener=fetch.default_opener, ) if args.runs_root is None: print(json.dumps(ears.run(speakers), ensure_ascii=False, indent=2, default=str)) return 0 with RunContext( args.runs_root, "data-ears", run_id=args.run_id, git_commit=args.git_commit ) as run: result = ears.run(speakers) atomic_write_json(run.dir / "metrics.json", json.loads(json.dumps(result, default=str))) report = root / "report.md" body = report.read_text(encoding="utf-8") if report.exists() else "" result_json = json.dumps(result, ensure_ascii=False, indent=2, default=str) lines = ["# data-ears", "", f"root: {root}", "", "```json", result_json, "```", "", body] atomic_write_text(run.dir / "report.md", "\n".join(lines) + "\n") print(f"run_id={run.run_id}") return 0 def _cmd_data_import(args: argparse.Namespace) -> int: from nida.data.importer import Importer, ImportRequest logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") root = args.root if root is None: if args.runs_root is None: raise SystemExit("data-import needs --root or --runs-root") root = args.runs_root.parent / "data" / f"import-{args.source}" cache = args.cache or Path("/content/nida_cache/import") / root.name importer = Importer( root=root, cache=cache, envs_root=args.envs, repo_dir=args.repo, request=ImportRequest( source=args.source, max_hours=args.max_hours, unit_offset=args.shard_offset, unit_count=args.shards, revision=args.revision, ), config_dir=args.config_dir, wavdir=args.source_dir, retry_failed=args.retry_failed, audio_threads=args.audio_threads, ) if args.runs_root is None: print(json.dumps(importer.run(args.stage), ensure_ascii=False, indent=2, default=str)) return 0 with RunContext( args.runs_root, f"data-import-{args.stage}", run_id=args.run_id, git_commit=args.git_commit ) as run: result = importer.run(args.stage) atomic_write_json(run.dir / "metrics.json", json.loads(json.dumps(result, default=str))) report = root / "report.md" body = report.read_text(encoding="utf-8") if report.exists() else "" result_json = json.dumps(result, ensure_ascii=False, indent=2, default=str) lines = [ f"# data-import {args.source} ({args.stage})", "", f"root: {root}", "", "```json", result_json, "```", "", body, ] atomic_write_text(run.dir / "report.md", "\n".join(lines) + "\n") print(f"run_id={run.run_id}") return 0 #: Exit code of ``nida train`` when a SIGTERM stopped it after saving a checkpoint. EXIT_TRAIN_INTERRUPTED = 3 def _cmd_train(args: argparse.Namespace) -> int: from nida.train.runner import TrainOutcome, run_training logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") def execute() -> TrainOutcome: return run_training( args.config, args.data, args.out, resume=args.resume, device=args.device, git_commit=args.git_commit, repo_dir=_REPO_DIR, init_from=args.init_from, ) if args.runs_root is None: outcome = execute() atomic_write_json(args.out / "metrics.json", outcome.metrics) atomic_write_text(args.out / "report.md", outcome.report_md) else: with RunContext( args.runs_root, "train", run_id=args.run_id, git_commit=args.git_commit ) as run: outcome = execute() if outcome.result.stopped_by_signal: run.status.extra["interrupted"] = "sigterm" atomic_write_json(run.dir / "metrics.json", outcome.metrics) atomic_write_text(run.dir / "report.md", outcome.report_md) print(f"run_id={run.run_id}") return EXIT_TRAIN_INTERRUPTED if outcome.result.stopped_by_signal else 0 def _cmd_synth_samples(args: argparse.Namespace) -> int: import torch from nida.bench import envs as bench_envs from nida.bench.b1 import _stream_runner from nida.data.envs import data_env_for_worker from nida.infer.samples import SampleSettings, make_listening_page logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") settings = SampleSettings( n_items=args.n, steps=args.steps, w_text=args.w_text, w_emo=args.w_emo, seed=args.seed, method=args.method, sway=args.sway, ) device = _resolve_device(args.device, torch.cuda.is_available()) def execute() -> Path: python = bench_envs.ensure_env( data_env_for_worker("vae_decode"), args.envs, args.repo, uv=bench_envs.find_uv(bench_envs.subprocess_runner), ) asr_python = None if not args.no_asr: asr_python = bench_envs.ensure_env( bench_envs.env_for_worker("asr_qwen"), args.envs, args.repo, uv=bench_envs.find_uv(bench_envs.subprocess_runner), ) return make_listening_page( args.checkpoint, args.data, args.out, settings, decoder_python=python, runner=_stream_runner, device=torch.device(device), asr_python=asr_python, ) if args.runs_root is None: print(execute()) return 0 with RunContext( args.runs_root, "synth-samples", run_id=args.run_id, git_commit=args.git_commit ) as run: page = execute() settings_json = (args.out / "settings.json").read_text(encoding="utf-8") atomic_write_json(run.dir / "metrics.json", {"page": str(page)}) atomic_write_text( run.dir / "report.md", f"# synth-samples\n\npage: {page}\n\n```json\n{settings_json}\n```\n", ) print(f"run_id={run.run_id}") return 0 #: Package pins of the isolated worker envs that ``say --single-env`` replaces #: (``nida.data.envs`` / ``nida.bench.envs``): checked, never installed. _SINGLE_ENV_PINS = {"transformers": "4.57.6"} _SINGLE_ENV_ASR_PINS = {"qwen-asr": "0.0.6"} def _single_env_report(with_asr: bool) -> dict[str, object]: """Interpreter and package versions a ``--single-env`` run uses, with every difference from the worker envs' pins logged as a warning and listed (results may then differ from Colab runs).""" from importlib import metadata pins = {**_SINGLE_ENV_PINS, **(_SINGLE_ENV_ASR_PINS if with_asr else {})} versions: dict[str, str | None] = {} for name in ("torch", "torchaudio", *pins): try: versions[name] = metadata.version(name) except metadata.PackageNotFoundError: versions[name] = None mismatches = { name: {"pinned": pin, "installed": versions[name]} for name, pin in pins.items() if versions[name] != pin } for name, diff in mismatches.items(): logging.getLogger(__name__).warning( "say --single-env: %s %s installed, worker env pins %s", name, diff["installed"], diff["pinned"], ) return { "single_env": True, "python": sys.executable, "versions": versions, "pin_mismatches": mismatches, } def _cmd_say(args: argparse.Namespace) -> int: import torch from nida.bench import envs as bench_envs from nida.bench.b1 import _stream_runner from nida.data.envs import data_env_for_worker from nida.infer.say import ( SaySettings, make_say_page, pick_prompts, prompt_candidates, rank_prompts, ) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") settings = SaySettings( steps=args.steps, w_text=args.w_text, w_emo=args.w_emo, seed=args.seed, method=args.method, sway=args.sway, speed=args.speed, normalize_text=not args.no_normalize, strict_normalize=args.strict_normalize, ) text = args.text if args.text is not None else args.text_file.read_text(encoding="utf-8") device = _resolve_device(args.device, torch.cuda.is_available()) if args.data is None and (args.prompt_list is None or args.prompt_id or args.rank_prompts): raise SystemExit("say: --data is required unless every prompt comes from --prompt-list") if args.rank_prompts and (args.prompt_list is not None or args.prompt_id): logging.getLogger(__name__).warning( "say: --rank-prompts is ignored when --prompt-list or --prompt-id gives the prompts" ) # make_say_page reads data_dir only for dataset prompt ids, which the guard above rules # out when --data is missing. data_dir: Path = args.data if args.data is not None else args.out / "no-dataset" worker_device = "cuda" if torch.device(device).type == "cuda" else "cpu" if args.single_env: args.out.mkdir(parents=True, exist_ok=True) atomic_write_json(args.out / "environment.json", _single_env_report(not args.no_asr)) else: (args.out / "environment.json").unlink(missing_ok=True) # never a stale single-env report def env_python(spec: bench_envs.EnvSpec, uv: str | None) -> Path: if args.single_env: return Path(sys.executable) assert uv is not None return bench_envs.ensure_env(spec, args.envs, args.repo, uv=uv) def execute() -> Path: uv = None if args.single_env else bench_envs.find_uv(bench_envs.subprocess_runner) python = env_python(data_env_for_worker("vae_decode"), uv) external = [] if args.prompt_list is not None: from nida.data.registry import DATA_MODELS from nida.infer.external_prompt import build_external_prompts, read_prompt_list ctc_python = env_python(data_env_for_worker(DATA_MODELS["mpoyraz_ctc_tr"].worker), uv) external = build_external_prompts( read_prompt_list(args.prompt_list), args.out / "external", vae_python=env_python(data_env_for_worker("vae_encode"), uv), ctc_python=ctc_python, runner=_stream_runner, device=worker_device, ) if args.prompt_id: prompt_ids = args.prompt_id elif external: prompt_ids = [] elif args.rank_prompts > 0: ranked = rank_prompts( args.data, prompt_candidates(args.data, args.rank_prompts), args.out, decoder_python=python, runner=_stream_runner, device=worker_device, ) prompt_ids = [str(r["segment_id"]) for r in ranked[: args.n_prompts]] else: prompt_ids = pick_prompts(args.data, args.n_prompts) asr_python = None if not args.no_asr: asr_python = env_python(bench_envs.env_for_worker("asr_qwen"), uv) return make_say_page( args.checkpoint, data_dir, prompt_ids, text, args.out, settings, decoder_python=python, runner=_stream_runner, device=torch.device(device), asr_python=asr_python, external=external, ) if args.runs_root is None: print(execute()) return 0 with RunContext(args.runs_root, "say", run_id=args.run_id, git_commit=args.git_commit) as run: page = execute() settings_json = (args.out / "settings.json").read_text(encoding="utf-8") atomic_write_json(run.dir / "metrics.json", {"page": str(page)}) atomic_write_text( run.dir / "report.md", f"# say\n\npage: {page}\n\n```json\n{settings_json}\n```\n" ) print(f"run_id={run.run_id}") return 0 def _cmd_queue(args: argparse.Namespace) -> int: from nida.jobqueue import ( run_git, run_queue, state_path_for, stream_to_log, summary_markdown, ) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") if args.runs_root is None or args.git_commit is None: print("queue needs --runs-root and --git-commit (the Colab runner passes both)") return 2 bundle_dir = args.bundle_dir or args.runs_root.parent / "code" with RunContext(args.runs_root, "queue", run_id=args.run_id, git_commit=args.git_commit) as run: summary = run_queue( args.file, runs_root=args.runs_root, git_commit=args.git_commit, repo_dir=args.repo, bundle_dir=bundle_dir, log_dir=args.log_dir, runner=stream_to_log, git=run_git, wait_minutes=args.wait_minutes, ) state_file = state_path_for(args.file) state = json.loads(state_file.read_text(encoding="utf-8")) if state_file.exists() else {} atomic_write_json( run.dir / "metrics.json", { "jobs": len(summary.outcomes), "failed": len(summary.failed), "interrupted": summary.interrupted, }, ) atomic_write_text(run.dir / "report.md", summary_markdown(summary, state)) if summary.interrupted: run.status.extra["interrupted"] = str(summary.interrupted_key) print(f"run_id={run.run_id}") return EXIT_TRAIN_INTERRUPTED if summary.interrupted else 0 def main(argv: Sequence[str] | None = None) -> int: """Entry point. Returns a process exit code.""" args = _build_parser().parse_args(argv) if args.command == "env-report": return _cmd_env_report(args) if args.command == "bench-phase0": return _cmd_bench_phase0(args) if args.command == "data-pilot": return _cmd_data_pilot(args) if args.command == "data-ears": return _cmd_data_ears(args) if args.command == "data-import": return _cmd_data_import(args) if args.command == "train": return _cmd_train(args) if args.command == "synth-samples": return _cmd_synth_samples(args) if args.command == "say": return _cmd_say(args) if args.command == "queue": return _cmd_queue(args) raise AssertionError(f"unhandled command {args.command!r}") # argparse prevents this if __name__ == "__main__": raise SystemExit(main())