PGx CN: scan-date batch warning (PCN-006), CN-consistent BAF evidence (PCN-010, --bafs-in-cn),

noise-outlier exclusion + robustness test (PCN-017/EDGE-026), platform from the CSV manifest (CLI-032)

Also: star-allele calls are checked for one manifest/build per run (PGX-011), duplications at the CN cap are
named xN (PGX-022), and tandem names attach supporting variants to their full copy.
This commit is contained in:
kemal committed 2026-09-29 17:11:34 +01:00
1 parent ed5e92b859
commit 1cb7b59559
8 files changed
+216 -25

No files matched your search

+47 -3
View File
@@ -57,7 +57,7 @@ def test_packaged_regions() -> None: # PCN-001
assert next(r for r in regs if r.gene == "GSTT1").chrom == "chr22_KI270879v1_alt"
def _dataset(tmp: Path, n_samples: int = 40) -> tuple[Path, Path, Path, Path]:
def _dataset(tmp: Path, n_samples: int = 40, n_outliers: int = 0) -> tuple[Path, Path, Path, Path]:
rng = np.random.default_rng(5)
loci = []
for c in ("1", "2"):
@@ -94,6 +94,16 @@ def _dataset(tmp: Path, n_samples: int = 40) -> tuple[Path, Path, Path, Path]:
cluster_file="c.egt") # fmt: skip
(gtcs / sid).mkdir(parents=True)
write_gtc(g, gtcs / sid / f"{sid}.gtc")
for j in range(n_outliers): # EDGE-026: noisy, shifted arrays that still pass the LRR SD gate
sid = f"2000000099{j:02d}_R01C01"
lrr = rng.normal(0.3, 0.19, n)
g = Gtc(num_snps=n, genotypes=np.ones(n, np.uint8), genotype_scores=np.full(n, 0.9, np.float32),
b_allele_freqs=np.zeros(n, np.float32), logr_ratios=lrr.astype(np.float32),
raw_x=np.full(n, 1000, np.uint16), raw_y=np.full(n, 1000, np.uint16),
base_calls=np.zeros((n, 2), np.uint8), gender=b"F", call_rate=0.995, logr_dev=0.19,
cluster_file="c.egt") # fmt: skip
(gtcs / sid).mkdir(parents=True)
write_gtc(g, gtcs / sid / f"{sid}.gtc")
(tmp / "truth.txt").write_text("\n".join(f"{k}\t{a}\t{b}" for k, (a, b) in truth.items()))
return bpm, fa, gtcs, regions
@@ -102,10 +112,11 @@ def test_pgx_cn_train_and_call(tmp_path: Path) -> None: # CLI-031/032, PCN-002/
bpm, fa, gtcs, regions = _dataset(tmp_path)
run = CliRunner()
r = run.invoke(app, ["pgx", "copy-number", "train", "--gtcs", str(gtcs), "--bpm", str(bpm), "--fasta", str(fa),
"--out", str(tmp_path / "model"), "--regions", str(regions), "--quiet"]) # fmt: skip
"--out", str(tmp_path / "model"), "--regions", str(regions), "--platform", "EX",
"--quiet"]) # fmt: skip
assert r.exit_code == 2 and "E_PGX_TRAIN_TOO_SMALL" in r.output # 40 < 96 (PCN-005)
r = run.invoke(app, ["pgx", "copy-number", "train", "--gtcs", str(gtcs), "--bpm", str(bpm), "--fasta", str(fa),
"--out", str(tmp_path / "model"), "--regions", str(regions), "--min-samples", "30",
"--out", str(tmp_path / "model"), "--regions", str(regions), "--min-samples", "30", "--platform", "EX",
"--min-passing", "30", "--quiet"]) # fmt: skip
assert r.exit_code == 0, r.output
model = tmp_path / "model" / "m.simurg-pgxcn"
@@ -143,3 +154,36 @@ def test_pgx_cn_train_and_call(tmp_path: Path) -> None: # CLI-031/032, PCN-002/
head = (out / "pgx_cn_sample_summary.tsv").read_text().splitlines()[0].split("\t")
assert head[:6] == ["sample_id", "sample_name", "sample_folder", "logr_dev", "passed_batch_qc",
"used_in_reference"] and head[-1] == "flags" # fmt: skip
def _calls(out: Path) -> dict[str, str]:
rows = [ln.split("\t") for ln in (out / "pgx_cn_sample_summary.tsv").read_text().splitlines()]
head = rows[0]
return {r[0]: ":".join(r[head.index("cn_GENEA")] + r[head.index("cn_GENEB")]) for r in rows[1:]}
def test_outlier_robustness(tmp_path: Path) -> None: # PCN-017 / EDGE-026: 10 % outliers leave other calls alone
run = CliRunner()
res = {}
for tag, n_out in (("clean", 0), ("outliers", 4)):
d = tmp_path / tag
d.mkdir()
bpm, fa, gtcs, regions = _dataset(d, 40, n_out)
r = run.invoke(app, ["pgx", "copy-number", "train", "--gtcs", str(gtcs), "--bpm", str(bpm), "--fasta", str(fa),
"--out", str(d / "model"), "--regions", str(regions), "--min-samples", "30",
"--min-passing", "30", "--platform", "EX", "--quiet"]) # fmt: skip
assert r.exit_code == 0, r.output
r = run.invoke(app, ["pgx", "copy-number", "call", "--gtcs", str(gtcs), "--model", str(d / "model" / "m.simurg-pgxcn"),
"--fasta", str(fa), "--out", str(d / "call"), "--quiet"]) # fmt: skip
assert r.exit_code == 0, r.output
res[tag] = _calls(d / "call")
assert (d / "call" / "pgx_cn_baf_evidence.tsv").exists() # PCN-010 output
same = sum(res["clean"][k] == res["outliers"][k] for k in res["clean"])
assert same >= 0.99 * len(res["clean"])
def test_platform_required(tmp_path: Path) -> None: # CLI-032: --platform auto needs a CSV manifest
bpm, fa, gtcs, regions = _dataset(tmp_path, 30)
r = CliRunner().invoke(app, ["pgx", "copy-number", "train", "--gtcs", str(gtcs), "--bpm", str(bpm), "--fasta",
str(fa), "--out", str(tmp_path / "m"), "--regions", str(regions), "--quiet"]) # fmt: skip
assert r.exit_code == 2 and "E_PGX_PLATFORM_UNKNOWN" in r.output