SimurgArray/src/simurg/cli/main.py

"""``simurg`` command-line interface (spec §2, plan §14).

Available: ``version``, ``inspect`` (P1), ``ingest`` (P2), ``normalize`` (P3), ``genotype`` (P4).
Later phases add the other commands of CLI-001.
"""

from __future__ import annotations

import json
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Annotated, Any, TypeVar

import numpy as np
import typer

from simurg import PRODUCT_NAME, __version__
from simurg.core.errors import ExitCode, SimurgError
from simurg.core.log import Reporter
from simurg.core.provenance import git_revision

app = typer.Typer(
    name="simurg",
    help=f"{PRODUCT_NAME}: analysis toolkit for Illumina Infinium genotyping BeadChip data.",
    no_args_is_help=True,
    add_completion=False,
    pretty_exceptions_enable=False,
    context_settings={"help_option_names": ["-h", "--help"]},
)

F = TypeVar("F", bound=Callable[..., Any])


def version_string() -> str:
    rev = git_revision()
    return f"{PRODUCT_NAME} {__version__}" + (f"+{rev}" if rev else "")


def _version_cb(value: bool) -> None:
    if value:
        typer.echo(version_string())
        raise typer.Exit(0)


@app.callback()
def main(
    version: Annotated[
        bool, typer.Option("--version", callback=_version_cb, is_eager=True, help="Print the version and exit.")
    ] = False,
) -> None:
    """Top-level options."""


def run_guarded(reporter: Reporter, fn: Callable[[], int]) -> None:
    """Run a command body mapping errors to catalogue codes and exit codes (CLI-014)."""
    reporter.open()
    code = ExitCode.OK
    try:
        code = ExitCode(fn())
    except SimurgError as e:
        reporter.error(e.code, e.message + (f"\n  hint: {e.hint}" if e.hint else ""), exc=e, file=e.path)
        code = e.exit_code
    except KeyboardInterrupt:
        reporter.error("E_INTERNAL", "interrupted")
        code = ExitCode.SIGINT
    except Exception as e:
        reporter.error(
            "E_INTERNAL", f"internal error: {type(e).__name__}: {e}. Rerun with --debug and report it.", exc=e
        )
        code = ExitCode.INTERNAL
    finally:
        reporter.close()
    raise typer.Exit(int(code))


# --------------------------------------------------------------------------------------------------------
@app.command("version")
def version_cmd() -> None:
    """Print the version."""
    typer.echo(version_string())


# --------------------------------------------------------------------------------------------------------
def _jsonable(o: Any) -> Any:
    if isinstance(o, np.generic):
        return o.item()
    if isinstance(o, np.ndarray):
        return o.tolist()
    if isinstance(o, Path):
        return str(o)
    if isinstance(o, bytes):
        return o.hex()
    if isinstance(o, float) and o != o:  # NaN
        return None
    return str(o)


def inspect_file(path: Path) -> dict[str, Any]:
    from simurg.formats.detect import detect_kind

    kind = detect_kind(path)
    info: dict[str, Any] = {"file": str(path), "kind": kind}
    if kind == "idat":
        from simurg.formats.idat import read_idat

        md = read_idat(path).metadata()
        md.pop("path", None)
        info.update(md)
    elif kind == "bpm":
        from simurg.formats.bpm import read_bpm

        info.update(read_bpm(path).summary())
    elif kind == "csv_manifest":
        from simurg.formats.csv_manifest import assay_format, read_csv_manifest

        m = read_csv_manifest(path, load_sequences=False)
        s = m.summary()
        s.pop("normalization_bins", None)
        s.pop("infinium_i_at", None)
        s.pop("infinium_i_gc", None)
        s["heading"] = m.heading
        s["assay_format"] = assay_format(m)
        info.update(s)
    elif kind == "egt":
        from simurg.formats.egt import read_egt

        info.update(read_egt(path).summary())
    elif kind == "gtc":
        from simurg.formats.gtc import read_gtc, validate_gtc

        g = read_gtc(path, check_write_complete=False)
        info.update(g.summary())
        info["problems"] = validate_gtc(g)
    elif kind == "samplesheet":
        from simurg.formats.samplesheet import parse_samplesheet

        sh = parse_samplesheet(path)
        info.update(
            {
                "rows": len(sh.rows),
                "columns": sh.columns,
                "key_columns": [sh.barcode_column, sh.position_column],
                "root_folder": str(sh.root_folder) if sh.root_folder else None,
                "header": sh.header,
                "has_sample_name": sh.has_sample_name,
                "warnings": [f"{c}: {m}" for c, m in sh.warnings],
            }
        )
    else:
        raise SimurgError("E_FILE_UNREADABLE", f"{path}: unrecognised file type", path=str(path))
    info.pop("path", None)
    return info


def _print_human(info: dict[str, Any]) -> None:
    typer.echo(f"== {info['file']}  ({info['kind']})")
    for k, v in info.items():
        if k in ("file", "kind"):
            continue
        if isinstance(v, list) and len(v) > 12:
            v = [*v[:12], f"… ({len(v)} total)"]
        if isinstance(v, (dict, list)):
            v = json.dumps(v, default=_jsonable, ensure_ascii=False)
        typer.echo(f"  {k}: {v}")


@app.command("inspect")
def inspect_cmd(
    files: Annotated[list[Path], typer.Argument(help="IDAT, BPM, CSV manifest, EGT, GTC or sample-sheet files.")],
    as_json: Annotated[bool, typer.Option("--json", help="Machine-readable JSON output.")] = False,
    log_json: Annotated[bool, typer.Option("--log-json", "--json-log", help="JSON Lines log on stderr.")] = False,
    debug: Annotated[bool, typer.Option("--debug", help="Debug logging with tracebacks.")] = False,
    log_level: Annotated[str, typer.Option("--log-level", help="debug|info|warning|error")] = "info",
    quiet: Annotated[bool, typer.Option("--quiet", "--no-progress", help="Suppress info messages.")] = False,
) -> None:
    """Print the header / table of contents of array files (spec CLI-020)."""
    rep = Reporter(
        command="inspect", level="debug" if debug else log_level, json_stderr=log_json, debug=debug, quiet=quiet
    )

    def body() -> int:
        results = []
        failures = 0
        for f in files:
            if not f.exists():
                raise SimurgError("E_FILE_NOT_FOUND", f"{f}: no such file", path=str(f))
            try:
                info = inspect_file(f)
            except SimurgError as e:
                rep.error(e.code, e.message, file=str(f))
                failures += 1
                continue
            results.append(info)
            if not as_json:
                _print_human(info)
        if as_json:
            out = results[0] if len(results) == 1 else results
            sys.stdout.write(json.dumps(out, default=_jsonable, ensure_ascii=False, indent=2) + "\n")
        if failures and not results:
            return ExitCode.USAGE
        return ExitCode.PARTIAL if failures else ExitCode.OK

    run_guarded(rep, body)


# --------------------------------------------------------------------------------------------------------
@app.command("ingest")
def ingest_cmd(
    bpm: Annotated[Path, typer.Option("--bpm", help="BPM manifest (required).")],
    out: Annotated[Path, typer.Option("--out", help="Output folder (required).")],
    idats: Annotated[Path | None, typer.Option("--idats", help="Folder searched recursively for IDAT pairs.")] = None,
    sheet: Annotated[Path | None, typer.Option("--sheet", help="Sample sheet (CSV).")] = None,
    csv: Annotated[Path | None, typer.Option("--csv", help="CSV manifest, checked against the BPM.")] = None,
    threads: Annotated[int, typer.Option("--threads", min=0, max=1024, help="Worker processes (0 = auto).")] = 0,
    force: Annotated[bool, typer.Option("--force", help="Replace existing outputs.")] = False,
    resume: Annotated[bool, typer.Option("--resume", help="Keep outputs whose inputs are unchanged.")] = False,
    layout: Annotated[str, typer.Option("--layout", help="nested|flat")] = "nested",
    on_duplicate_idat: Annotated[str, typer.Option("--on-duplicate-idat", help="error|first|newest")] = "error",
    log_json: Annotated[bool, typer.Option("--log-json", "--json-log", help="JSON Lines log on stderr.")] = False,
    debug: Annotated[bool, typer.Option("--debug", help="Debug logging with tracebacks.")] = False,
    log_level: Annotated[str, typer.Option("--log-level", help="debug|info|warning|error")] = "info",
    quiet: Annotated[bool, typer.Option("--quiet", "--no-progress", help="Suppress info messages.")] = False,
) -> None:
    """IDAT pairs + BPM → canonical per-sample raw intensities (Parquet). Phase P2."""
    from simurg.pipeline.ingest import IngestOptions, run_ingest

    if layout not in ("nested", "flat"):
        raise typer.BadParameter("--layout must be nested or flat")
    if on_duplicate_idat not in ("error", "first", "newest"):
        raise typer.BadParameter("--on-duplicate-idat must be error, first or newest")
    rep = Reporter(
        command="ingest",
        level="debug" if debug else log_level,
        json_stderr=log_json,
        debug=debug,
        quiet=quiet,
        log_file=out / "logs" / "ingest.log",
    )
    opts = IngestOptions(
        out=out,
        bpm=bpm,
        idats=idats,
        sheet=sheet,
        csv=csv,
        threads=threads,
        force=force,
        resume=resume,
        layout=layout,
        on_duplicate_idat=on_duplicate_idat,
        argv=sys.argv[1:],
    )
    run_guarded(rep, lambda: run_ingest(opts, rep))


# --------------------------------------------------------------------------------------------------------
@app.command("normalize")
def normalize_cmd(
    indir: Annotated[Path, typer.Option("--in", help="Output folder of `simurg ingest`.")],
    norm_version: Annotated[str, typer.Option("--norm-version", help="1.2.0 (current) or 1.1.2 (older)")] = "1.2.0",
    threads: Annotated[int, typer.Option("--threads", min=0, max=1024, help="Worker processes (0 = auto).")] = 0,
    force: Annotated[bool, typer.Option("--force", help="Replace existing outputs.")] = False,
    log_json: Annotated[bool, typer.Option("--log-json", "--json-log", help="JSON Lines log on stderr.")] = False,
    debug: Annotated[bool, typer.Option("--debug", help="Debug logging with tracebacks.")] = False,
    quiet: Annotated[bool, typer.Option("--quiet", "--no-progress", help="Suppress info messages.")] = False,
) -> None:
    """Fit per-bin normalization transforms and write normalized X/Y, theta and R (phase P3)."""
    from simurg.pipeline.normalize import NormalizeOptions, run_normalize

    rep = Reporter(
        command="normalize", level="debug" if debug else "info", json_stderr=log_json, debug=debug, quiet=quiet
    )
    opts = NormalizeOptions(indir=indir, version=norm_version, threads=threads, force=force, argv=sys.argv[1:])
    run_guarded(rep, lambda: run_normalize(opts, rep))


# --------------------------------------------------------------------------------------------------------
@app.command("genotype")
def genotype_cmd(
    bpm: Annotated[Path, typer.Option("--bpm", help="BPM manifest (required).")],
    egt: Annotated[Path, typer.Option("--egt", help="EGT cluster file (required).")],
    out: Annotated[Path, typer.Option("--out", help="Output folder (required).")],
    idats: Annotated[Path | None, typer.Option("--idats", help="Folder searched recursively for IDAT pairs.")] = None,
    sheet: Annotated[Path | None, typer.Option("--sheet", help="Sample sheet (CSV).")] = None,
    csv: Annotated[Path | None, typer.Option("--csv", help="CSV manifest, checked against the BPM.")] = None,
    cutoff: Annotated[float, typer.Option("--cutoff", help="GenCall score below which a call is a no-call.")] = 0.15,
    preset: Annotated[
        str, typer.Option("--preset", help="current (default) or autocall-1.6.2 (legacy vendor behaviour)")
    ] = "current",
    norm_version: Annotated[
        str | None, typer.Option("--norm-version", help="1.2.0 or 1.1.2 (default: from --preset)")
    ] = None,
    sex_version: Annotated[
        int | None, typer.Option("--sex-version", min=1, max=2, help="Sex estimate rule 1 or 2 (default: preset)")
    ] = None,
    gtc_version: Annotated[
        int | None, typer.Option("--gtc-version", help="GTC format 3 or 5 (default: preset)")
    ] = None,
    autocall_date: Annotated[
        str | None, typer.Option("--autocall-date", help="Fixed value for GTC field 201 (reproducible output).")
    ] = None,
    threads: Annotated[int, typer.Option("--threads", min=0, max=1024, help="Worker processes (0 = auto).")] = 0,
    force: Annotated[bool, typer.Option("--force", help="Replace existing outputs.")] = False,
    layout: Annotated[str, typer.Option("--layout", help="nested|flat")] = "nested",
    on_duplicate_idat: Annotated[str, typer.Option("--on-duplicate-idat", help="error|first|newest")] = "error",
    log_json: Annotated[bool, typer.Option("--log-json", "--json-log", help="JSON Lines log on stderr.")] = False,
    debug: Annotated[bool, typer.Option("--debug", help="Debug logging with tracebacks.")] = False,
    quiet: Annotated[bool, typer.Option("--quiet", "--no-progress", help="Suppress info messages.")] = False,
) -> None:
    """IDAT pairs → GTC files and a sample summary (GenCall-equivalent calling; phase P4)."""
    from simurg.pipeline.genotype import GenotypeOptions, run_genotype

    if layout not in ("nested", "flat"):
        raise typer.BadParameter("--layout must be nested or flat")
    from simurg.genotype.sample import PRESETS

    if preset not in PRESETS:
        raise typer.BadParameter(f"--preset must be one of {sorted(PRESETS)}")
    if gtc_version not in (None, 3, 5):
        raise typer.BadParameter("--gtc-version must be 3 or 5")
    if norm_version not in (None, "1.2.0", "1.1.2"):
        raise typer.BadParameter("--norm-version must be 1.2.0 or 1.1.2")
    rep = Reporter(
        command="genotype",
        level="debug" if debug else "info",
        json_stderr=log_json,
        debug=debug,
        quiet=quiet,
        log_file=out / "logs" / "genotype.log",
    )
    opts = GenotypeOptions(
        out=out,
        bpm=bpm,
        egt=egt,
        idats=idats,
        sheet=sheet,
        csv=csv,
        threads=threads,
        force=force,
        layout=layout,
        on_duplicate_idat=on_duplicate_idat,
        cutoff=cutoff,
        preset=preset,
        norm_version=norm_version,
        sex_version=sex_version,
        gtc_version=gtc_version,
        autocall_date=autocall_date,
        argv=sys.argv[1:],
    )
    run_guarded(rep, lambda: run_genotype(opts, rep))


if __name__ == "__main__":  # pragma: no cover
    app()