PGx CN: pgx copy-number train|call (CLI-031/032, PCN-001..017, IN-050), region BED v1, demo identities (K-028)
ci / test (3.12) (push) Failing after 8s
ci / test (3.13) (push) Failing after 10s

- simurg/pgx/cn.py: per-region CN mixture, mu_c = delta + gamma*log2((c+1)/3), free lower CN-0 component,
  Hardy-Weinberg-constrained weights over haplotype copy numbers, EM from a grid of label starts.
- pipeline/pgx_cn.py: model file (.simurg-pgxcn), batch rules (PCN-004), per-sample CN VCF + tbi, batch and
  sample summaries.
- The GSA v4 PGx demo arrays are 30 CEPH/1000G individuals (scripts/pgx/demo36_identity.tsv); against NYGC 30x
  SV RD_CN 224/240 region calls agree (GSTM1, UGT2B17 30/30), in-sample (docs/validation/V2_pgx_cn.md).

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01814GKVzZVL3FuF3DeiZXhv
This commit is contained in:
kemalandClaude Opus 5.5 committed 2026-09-29 13:16:19 +01:00
1 parent 8c425182c3
commit 58e70bc894
14 files changed
+1306 -1

No files matched your search

+72
View File
@@ -803,6 +803,78 @@ def cn_annotate_cmd(
run_guarded(rep, lambda: run_cn_annotate(opts, rep))
# --------------------------------------------------------------------------------------------------------
pgx_app = typer.Typer(help="Pharmacogenomics: targeted copy number, star alleles, phenotypes (v2).")
app.add_typer(pgx_app, name="pgx")
pgx_cn_app = typer.Typer(help="Targeted PGx copy number (PCN-001).")
pgx_app.add_typer(pgx_cn_app, name="copy-number")
@pgx_cn_app.command("train")
def pgx_cn_train_cmd(
gtcs: Annotated[Path, typer.Option("--gtcs", help="Folder searched recursively for training GTCs (required).")],
bpm: Annotated[Path, typer.Option("--bpm", help="BPM manifest (required).")],
fasta: Annotated[Path, typer.Option("--fasta", help="GRCh38 FASTA (required).")],
out: Annotated[Path, typer.Option("--out", help="Output folder.")] = Path("."),
mask: Annotated[Path | None, typer.Option("--mask", help="Probe mask TSV (IN-052).")] = None,
regions: Annotated[Path | None, typer.Option("--regions", help="Region BED (default: built-in v1).")] = None,
summary: Annotated[Path | None, typer.Option("--summary", help="sample_summary.tsv (assay control).")] = None,
platform: Annotated[str, typer.Option("--platform", help="auto|LCG|EX")] = "auto",
min_samples: Annotated[int, typer.Option("--min-samples", min=1)] = 96,
min_passing: Annotated[int, typer.Option("--min-passing", min=1)] = 90,
max_samples: Annotated[int, typer.Option("--max-samples", min=1)] = 3000,
train_lrr_sd_max: Annotated[float, typer.Option("--train-lrr-sd-max")] = 0.20,
train_call_rate_min: Annotated[float, typer.Option("--train-call-rate-min")] = 0.99,
train_assay_control_min: Annotated[float, typer.Option("--train-assay-control-min")] = 1.0,
force: ForceOpt = False,
log_json: LogJson = False,
debug: Debug = False,
quiet: Quiet = False,
) -> None:
"""Train a PGx CN model (CLI-032, PCN-005)."""
from simurg.pipeline.pgx_cn import PgxTrainOptions, run_train
if platform not in ("auto", "LCG", "EX"):
raise typer.BadParameter("--platform auto|LCG|EX")
rep = _reporter("pgx copy-number train", out, log_json, debug, quiet)
o = PgxTrainOptions(gtcs=gtcs, bpm=bpm, fasta=fasta, out=out, mask=mask, regions=regions, summary=summary,
platform=platform, min_samples=min_samples, min_passing=min_passing, max_samples=max_samples,
lrr_sd_max=train_lrr_sd_max, call_rate_min=train_call_rate_min,
assay_control_min=train_assay_control_min, force=force) # fmt: skip
run_guarded(rep, lambda: run_train(o, rep))
@pgx_cn_app.command("call")
def pgx_cn_call_cmd(
gtcs: Annotated[Path, typer.Option("--gtcs", help="Folder searched recursively for GTC files (required).")],
model: Annotated[Path, typer.Option("--model", help="PGx CN model (.simurg-pgxcn, required).")],
fasta: Annotated[Path, typer.Option("--fasta", help="GRCh38 FASTA with alt contigs (required).")],
out: Annotated[Path, typer.Option("--out", help="Output folder (required).")],
min_batch: Annotated[int, typer.Option("--min-batch", min=1)] = 24,
min_passing: Annotated[int, typer.Option("--min-passing", min=1)] = 22,
recommended_passing: Annotated[int, typer.Option("--recommended-passing", min=1)] = 96,
lrr_sd_max: Annotated[float, typer.Option("--lrr-sd-max")] = 0.20,
cluster_mismatch: Annotated[str, typer.Option("--cluster-mismatch", help="error|warn")] = "error",
no_bgzip: Annotated[bool, typer.Option("--no-bgzip")] = False,
force: ForceOpt = False,
log_json: LogJson = False,
debug: Debug = False,
quiet: Quiet = False,
) -> None:
"""Call targeted PGx copy number per region (CLI-031, PCN-009)."""
from simurg.pipeline.pgx_cn import PgxCallOptions, run_call
if cluster_mismatch not in ("error", "warn"):
raise typer.BadParameter("--cluster-mismatch error|warn")
if min_passing > min_batch:
raise typer.BadParameter("--min-passing must be <= --min-batch")
rep = _reporter("pgx copy-number call", out, log_json, debug, quiet)
o = PgxCallOptions(gtcs=gtcs, model=model, fasta=fasta, out=out, min_batch=min_batch, min_passing=min_passing,
recommended_passing=recommended_passing, lrr_sd_max=lrr_sd_max,
cluster_mismatch=cluster_mismatch, bgzip=not no_bgzip, force=force) # fmt: skip
run_guarded(rep, lambda: run_call(o, rep))
# --------------------------------------------------------------------------------------------------------
methyl_app = typer.Typer(help="Methylation arrays: control QC, detection, beta values (MET-001).")
app.add_typer(methyl_app, name="methyl")