"""Canonical DeepGenopix run-name helpers. HF Jobs write to shared Hub namespaces. These helpers produce names that carry the architecture, source checkpoint, corpus, and mode so downstream jobs do not accidentally consume a generic or overwritten run. """ from __future__ import annotations import argparse import hashlib import re from pathlib import Path from deepgenopix.notebook_support import build_experiment_config _SAFE = re.compile(r"[^A-Za-z0-9._-]+") def slug(value: str) -> str: """Return a filesystem/HF-path-safe slug.""" cleaned = _SAFE.sub("-", value.strip()).strip("-._") return cleaned or "unnamed" def short_ref(value: str) -> str: """Return a stable short reference for a path or Hub URL.""" stem = slug(Path(value.rstrip("/")).stem or value) digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:8] return f"{stem}-{digest}" def train_run_id(preset: str) -> str: """Return the canonical architecture-specific classifier run id.""" return build_experiment_config(preset).run_id def quant_train_name(objective: str, source_checkpoint: str, reference_parquet: str) -> str: """Return a canonical QuantEncoder training output name.""" return "__".join( [ f"quantenc-{slug(objective)}", f"src-{short_ref(source_checkpoint)}", f"ref-{short_ref(reference_parquet)}", ] ) def quantify_name(encoder: str, reads: str, mode: str) -> str: """Return a canonical quantification output name.""" return "__".join( [ f"quant-{short_ref(encoder)}", f"reads-{short_ref(reads)}", f"mode-{slug(mode)}", ] ) def variant_name(quant_run_name: str, null_stats: str | None = None) -> str: """Return a canonical variant run name tied to the quant run it consumes.""" parts = [f"variant-{slug(quant_run_name)}"] if null_stats: parts.append(f"null-{short_ref(null_stats)}") return "__".join(parts) def main() -> int: parser = argparse.ArgumentParser(description="Generate canonical DeepGenopix run names") subparsers = parser.add_subparsers(dest="command", required=True) train = subparsers.add_parser("train") train.add_argument("--preset", required=True) quant_train = subparsers.add_parser("quant-train") quant_train.add_argument("--objective", required=True) quant_train.add_argument("--source-checkpoint", required=True) quant_train.add_argument("--reference-parquet", required=True) quantify = subparsers.add_parser("quantify") quantify.add_argument("--encoder", required=True) quantify.add_argument("--reads", required=True) quantify.add_argument("--mode", required=True) variant = subparsers.add_parser("variant") variant.add_argument("--quant-run-name", required=True) variant.add_argument("--null-stats", default=None) args = parser.parse_args() if args.command == "train": print(train_run_id(args.preset)) elif args.command == "quant-train": print(quant_train_name(args.objective, args.source_checkpoint, args.reference_parquet)) elif args.command == "quantify": print(quantify_name(args.encoder, args.reads, args.mode)) elif args.command == "variant": print(variant_name(args.quant_run_name, args.null_stats)) return 0 if __name__ == "__main__": raise SystemExit(main())