PGx validation on GSE165845: WGS-derived star truth, 2-fold masks, array-resolvable concordance

- scripts/pgx/wgs_star_truth.py: star alleles from the 1000G 30x phased panel with the DB definitions
  (wildcard-aware haplotype assignment; checked against GeT-RM on the demo CEPH individuals).
- validate_star_getrm.py: --all-genes, --individuals, --coverage (projection onto what the array can see);
  replicate_concordance.py.
- 120 individuals x 3 GSA v1 arrays: array-resolvable 1637/1710 (95.7 %), top 1223/1800 (chip lacks many
  defining variants), GeT-RM subset 28/35, replicates 99.2 % without the failed array (V2_pgx_star.md, K-029).

scripts/pgx/validate_star_getrm.py

@@ -79,13 +79,42 @@ def main() -> int:
     ap.add_argument("--truth", required=True, type=Path)
     ap.add_argument("--identity", required=True, type=Path)
     ap.add_argument("--out", required=True, type=Path)
+    ap.add_argument("--all-genes", action="store_true",
+                    help="truth with the same definitions (wgs_star_truth.py): no legacy-name exclusions")  # fmt: skip
+    ap.add_argument("--individuals", type=Path, help="only these individuals (one per line)")
+    ap.add_argument("--coverage", type=Path, help="pgx_coverage.tsv of the array: also report array-resolvable "
+                    "concordance (both diplotypes projected onto the alleles the array can tell apart)")  # fmt: skip
     a = ap.parse_args()
+    if a.all_genes:
+        LEGACY_NOMENCLATURE.clear()
+    only = set(a.individuals.read_text().split()) if a.individuals else None
+    proj: dict[tuple[str, str], str] = {}
+    refs: dict[str, str] = {}
+    if a.coverage:
+        with a.coverage.open() as fh:
+            for r in csv.DictReader((ln for ln in fh if not ln.startswith("#")), delimiter="\t"):
+                if r["allele_kind"] == "reference":
+                    refs[r["gene"]] = r["allele"]
+                st = r["status"]
+                proj[(r["gene"], r["allele"])] = st.split(":", 1)[1] if st.startswith("equivalent_to:") else (
+                    r["allele"] if st == "callable" else "")  # fmt: skip
+
+    def project(gene: str, allele: str) -> str:
+        core = re.sub(r"x\d+$|xN$", "", allele.split("+")[-1])
+        v = proj.get((gene, core), core)
+        if v:
+            return v
+        ref = refs.get(gene, core)  # not covered: indistinguishable from the reference's class
+        return proj.get((gene, ref)) or ref
+
     with a.identity.open() as fh:
         ident = {r["sample_id"]: r["individual"] for r in csv.DictReader((ln for ln in fh if not ln.startswith("#")),
                                                                            delimiter="\t")}  # fmt: skip
     truth: dict[tuple[str, str], str] = {}
     with a.truth.open() as fh:
         for r in csv.DictReader(fh, delimiter="\t"):
+            if r.get("flag") == "incomplete" or (only is not None and r["individual"] not in only):
+                continue
             truth[(r["individual"], r["gene"])] = r["diplotype"]
     calls = json.loads(a.calls.read_text())
     per_gene: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
@@ -135,11 +164,21 @@ def main() -> int:
                                   if len(e.split("/")) >= 2)  # fmt: skip
             st["top"] += ok_top
             st["equivalent"] += ok_eq
+            if a.coverage and len(top) == 2 and len(opts) == 1 and all(len(x) == 1 for x in opts[0]):
+                tt = sorted(project(g["gene"], next(iter(x))) for x in opts[0])
+                cc = sorted(project(g["gene"], x) for x in top)
+                st["array_resolvable_n"] += 1
+                st["array_resolvable"] += tt == cc
             if not ok_top:
                 detail.append({"individual": ind, "gene": g["gene"], "truth": t, "call": g["genotype"],
                                "within_equivalents": ok_eq})  # fmt: skip
     comparable = {k: v for k, v in per_gene.items() if k not in LEGACY_NOMENCLATURE}
+    tot_ar = [
+        sum(v.get("array_resolvable", 0) for v in comparable.values()),
+        sum(v.get("array_resolvable_n", 0) for v in comparable.values()),
+    ]
     summary = {"individuals": len(seen), "genes": {k: dict(v) for k, v in sorted(per_gene.items())},
+               "array_resolvable_total": {"concordant": tot_ar[0], "n": tot_ar[1]},
                "legacy_nomenclature_genes": sorted(LEGACY_NOMENCLATURE),
                "total": {k: sum(v.get(k, 0) for v in comparable.values()) for k in ("n", "top", "equivalent")},
                "discordant": detail}  # fmt: skip
@@ -147,8 +186,12 @@ def main() -> int:
     for g, v in sorted(per_gene.items()):
         print(f"{g:8s} top {v['top']}/{v['n']}  within equivalents {v['equivalent']}/{v['n']}"
               + (f"  (not comparable {v['not_comparable']})" if v.get("not_comparable") else "")
+              + (f"  array-resolvable {v.get('array_resolvable', 0)}/{v.get('array_resolvable_n', 0)}"
+                 if a.coverage else "")
               + ("  [legacy nomenclature]" if g in LEGACY_NOMENCLATURE else ""))  # fmt: skip
     print("total (excluding legacy-nomenclature genes)", summary["total"])
+    if a.coverage:
+        print("array-resolvable total", summary["array_resolvable_total"])
     return 0