| """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"]
|
|
|
|
|
| _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: <runs-root>/../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: <runs-root>/../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"),
|
| 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: <runs-root>/../data/ears); dataset is <root>/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: <repo>/configs/data.")
|
| imp = sub.add_parser(
|
| "data-import",
|
| parents=[_common_parser()],
|
| help="External transcribed corpus -> latent dataset (nida.data.importer).",
|
| )
|
| imp.add_argument(
|
| "--source",
|
|
|
| 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: <runs-root>/../data/import-<source>).",
|
| )
|
| imp.add_argument(
|
| "--cache",
|
| type=Path,
|
| default=None,
|
| help="Local scratch (default: /content/nida_cache/import/<root name>).",
|
| )
|
| 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: <repo>/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)
|
|
|
| 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 '<wav>\t<transcript>' 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: <runs-root>/../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_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
|
|
|
|
|
|
|
|
|
| _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"
|
| )
|
|
|
|
|
| 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)
|
|
|
| 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}")
|
|
|
|
|
| if __name__ == "__main__":
|
| raise SystemExit(main())
|
|
|
