P6 complete: optional VCF fields, SNV+indel merging, auxiliary MNV loci, PLINK 2 .pgen

- gtc-to-vcf --info-fields ASSAY_TYPE,NORM_BIN,GENTRAIN,CLUSTER_SEP,ILMN_STRAND,PROBE_IDS (--egt) and
  --extra-format X,Y,NORMX,NORMY,THETA,R, in per-sample VCFs and the --merge BCF (OUT-012, plan §8.3);
  equal to gtc2vcf's values (THETA/R missing where both normalized intensities are 0, K-023)
- --merge-snv-indel (OUT-021, EDGE-013): alleles extended to the longest REF; SNV probes read the base at POS
- --auxiliary-loci (OUT-035, EDGE-015): rules (a)-(e), E_AUX_INVALID; rule (d) uses the last plus-strand base
  for minus-strand designs (spec amended, as GTCtoVCF)
- export plink --format bed|pgen|both: PLINK 2 fixed-width .pgen/.pvar/.psam; plink2 reads it back to a .bed
  byte-identical to ours
- cohort store keeps raw/normalized intensities for the BCF path; tests (unit + gtc2vcf oracle); docs
This commit is contained in:
kemal committed 2026-09-28 14:29:42 +01:00
1 parent d3add008f7
commit 1001bf951e
13 files changed
+611 -50

No files matched your search

+61
View File
@@ -232,3 +232,64 @@ def test_ploidy_rules() -> None: # OUT-018, EDGE-038
def test_plink_codes() -> None: # OUT-043
assert [bed_code(g, 1) for g in ("0/0", "0/1", "1/1", "./.", "1", "0")] == [3, 2, 0, 1, 0, 3]
assert bed_code("0/2", 1) == 3 and bed_code("0/2", 2) == 2 # split multi-allelic record
# ------------------------------------------------------------------------------------------------ P6 options
def test_merge_snv_indel_edge_013() -> None: # OUT-021
snv = ProbeSite(0, "snv", "1", 5, "A", ("A", "G"), "snv", 2, "1")
dele = ProbeSite(1, "del", "1", 5, "ACT", ("ACT", "A"), "indel", 1, "1")
sep = group_sites([snv, dele], {"1": 0})
assert len(sep) == 2 # default: separate records
(s,) = group_sites([snv, dele], {"1": 0}, merge_snv_indel=True)
assert s.ref == "ACT" and s.alts == ["GCT", "A"] and s.id == "snv;del"
ext_snv, ext_del = s.probes
assert ext_del.alleles == ("ACT", "A")
assert ext_snv.alleles == ("ACT", "GCT")
# SNV probe homozygous A (sees the anchor base of the deletion too), deletion probe het → REF/DEL
assert gt_text(genotype(s, (AA, AB), 2), 2) == "0/2"
# SNV het A/G, deletion hom REF → REF/SNV
assert gt_text(genotype(s, (AB, AA), 2), 2) == "0/1"
# incompatible REFs (not prefixes of each other) stay separate
other = ProbeSite(2, "x", "1", 5, "GT", ("GT", "G"), "indel", 1, "1")
assert len(group_sites([snv, other], {"1": 0}, merge_snv_indel=True)) == 2
def _aux_probe(alleles: tuple[str, str] = ("A", "C"), idx: int = 0, name: str = "mnv") -> ProbeSite:
return ProbeSite(idx, name, "1", 5, "A", alleles, "snv", 2, "1")
def test_auxiliary_loci_rules() -> None: # OUT-035, EDGE-015
from simurg.harmonize.sites import AuxRecord, apply_auxiliary
genome = "TTTTATGCC" # positions 5-6 = AT
f = fetcher(genome)
plus = lambda i: "+" # noqa: E731
ok = AuxRecord("mnv", "1", 5, "AT", ["CG"])
probes, errs = apply_auxiliary([_aux_probe()], [ok], f, lambda c: c, plus)
assert not errs and probes[0].ref == "AT" and probes[0].alleles == ("AT", "CG") and probes[0].reason == "aux_locus"
cases = {
"(a)": AuxRecord("mnv", "1", 5, "AT", ["CG", "GG"]),
"(b)": AuxRecord("mnv", "1", 5, "AT", ["C"]),
"(d)": AuxRecord("mnv", "1", 5, "AT", ["TG"]),
"(e)": AuxRecord("mnv", "1", 5, "AG", ["CG"]),
}
for rule, rec in cases.items():
_, errs = apply_auxiliary([_aux_probe()], [rec], f, lambda c: c, plus)
assert len(errs) == 1 and rule in errs[0], (rule, errs)
dup = [_aux_probe(), _aux_probe(idx=1, name="twin")]
_, errs = apply_auxiliary(dup, [ok], f, lambda c: c, plus)
assert "(c)" in errs[0]
_, errs = apply_auxiliary([_aux_probe()], [AuxRecord("nope", "1", 5, "AT", ["CG"])], f, lambda c: c, plus)
assert "not a probe" in errs[0]
# minus-strand design: the interrogated base is the MNV's last base on the plus strand
probes, errs = apply_auxiliary([_aux_probe(("T", "G"))], [AuxRecord("mnv", "1", 5, "AT", ["CG"])], f,
lambda c: c, lambda i: "-") # fmt: skip
assert not errs and probes[0].alleles == ("AT", "CG")
def test_pgen_codes() -> None: # plan §8.4
from simurg.pipeline.export import PGEN_FROM_BED, _pack
# PLINK 1 (A1 = ALT): 0 hom ALT, 1 missing, 2 het, 3 hom REF → PLINK 2: 2, 3, 1, 0
assert list(PGEN_FROM_BED) == [2, 3, 1, 0]
assert _pack([0, 1, 2, 3, 0], 2) == bytes([0b11100100, 0b00000000])