"""``python -m qc run --spec --r1 R1.fastq.gz --r2 R2.fastq.gz [...]``""" from __future__ import annotations import argparse import json import sys from pathlib import Path from qc.engine import run_qc from qc.rust_engine import RustEngineUnavailable _ICON = {"pass": "PASS", "warn": "WARN", "fail": "FAIL"} def _print_report(r: dict) -> None: bar = "=" * 74 print(bar) print(f"SEQCOLYTE QC — {r['assay']} ({r['platform']}) spec={r['spec_id']} overall: {r['overall'].upper()}") print(bar) p = r["profile"] print(f"reads: {p['n_pairs']:,} pairs R1 modal {p['r1_len']['modal']} bp R2 modal {p['r2_len']['modal']} bp\n") print("checks:") for f in r["findings"]: af = f" [{f['affected_fraction']:.1%} of reads]" if f["affected_fraction"] is not None else "" print(f" {_ICON.get(f['verdict'], '?'):4} {f['title']:<52} {f['value']} {f['unit']} (want {f['threshold']}){af}") plan = r["plan"] print(f"\ndiagnosis ({plan.get('method', '?')}):") print(f" root cause : {plan['root_cause']}") print(f" {plan['diagnosis']}") if plan.get("ranked"): print(" ranked findings:") for x in plan["ranked"]: print(f" - [{x['severity']:>6}] {x['check_id']}: {x['why']}") if plan.get("llm_error"): print(f" (LLM ranking unavailable: {plan['llm_error']} — used deterministic fallback)") if r.get("eval"): e = r["eval"] c = e["confusion"] print(f"\neval vs ground-truth labels: precision={e['precision']} recall={e['recall']} f1={e['f1']}") print(f" predicted {e['predicted_affected']:,} affected / {e['true_affected']:,} truly affected of {e['n']:,} " f"(tp={c['tp']} fp={c['fp']} fn={c['fn']} tn={c['tn']})") print(bar) def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(prog="qc", description="Protocol-aware sequencing QC") sub = ap.add_subparsers(dest="cmd", required=True) rp = sub.add_parser("run", help="run QC on a FASTQ pair against a spec") rp.add_argument("--spec", required=True) rp.add_argument("--r1", required=True) rp.add_argument("--r2", required=True) rp.add_argument("--whitelist", default=None, help="cell-barcode whitelist (enables the whitelist check)") rp.add_argument("--labels", default=None, help="ground-truth labels TSV (enables the eval)") rp.add_argument("--no-llm", action="store_true", dest="no_llm", help="skip Claude ranking; deterministic only") rp.add_argument("--model", default="claude-opus-4-8") rp.add_argument("--max-reads", type=int, default=None, dest="max_reads") rp.add_argument("--json-out", default=None, dest="json_out") args = ap.parse_args(argv) try: report = run_qc(args.spec, args.r1, args.r2, whitelist=args.whitelist, labels=args.labels, use_llm=not args.no_llm, model=args.model, max_reads=args.max_reads) except RustEngineUnavailable as exc: print(f"error: {exc}\nbuild the compute core first: make rust", file=sys.stderr) return 2 _print_report(report) if args.json_out: Path(args.json_out).write_text(json.dumps(report, indent=2) + "\n") return 0 if __name__ == "__main__": raise SystemExit(main())