scripts/run_names.py

"""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())