PGx phenotypes: pgx phenotype|annotate (PGX-024..028, CLI-034)

CPIC lookups by allele function counts or activity values (duplications N x activity, *5/*0 = 0, total-score
fallback), Indeterminate when feasible solutions disagree or an allele is not in the guideline; per-sample
simurg.pgx/1 JSON and diplotype_annotations.tsv with PGX-026 metadata lines.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01814GKVzZVL3FuF3DeiZXhv

src/simurg/pgx/phenotype.py

@@ -0,0 +1,118 @@
+"""Diplotype → phenotype lookup — spec PGX-024, PGX-025. Pure functions, no I/O.
+
+[DB: CPIC gene_result_lookup — the diplotype's allele functions (or, for activity-score genes, allele activity
+values) as a count key, e.g. {"Normal function": 1, "No function": 1} or {"1.0": 1, "0.25": 1}.] A duplicated allele
+``*Nxk`` contributes k × its activity value (CPIC activity-score rule for gene duplications); a whole-gene deletion
+(``*5``, ``*0``) has activity 0 and "No function". When a key is absent from the table, activity-score genes fall back
+to the total score (the phenotype of the largest tabulated total ≤ the score); anything else is ``Indeterminate`` with
+the flag ``phenotype_not_in_guideline`` (PGX-024).
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass, field
+
+
+@dataclass
+class GuidelineTable:
+    gene: str
+    by_key: dict[str, tuple[str, str | None]]  # canonical key JSON → (phenotype, total activity)
+    by_total: list[tuple[float, str]]  # (total activity, phenotype), sorted
+    function: dict[str, str | None]  # allele → clinical function
+    activity: dict[str, float | None]  # allele → activity value
+
+    @property
+    def activity_based(self) -> bool:
+        return bool(self.by_total)
+
+
+@dataclass
+class PhenotypeResult:
+    phenotype: str | None
+    activity_score: float | None
+    flags: list[str] = field(default_factory=list)
+
+
+def _num(v: object) -> float | None:
+    try:
+        return float(str(v))
+    except (TypeError, ValueError):
+        return None
+
+
+def build_table(gene: str, rows: list[tuple[str, str | None, str]], alleles: list[tuple[str, str | None, str | None]]
+                ) -> GuidelineTable:  # fmt: skip
+    """``rows``: (lookup_key JSON, total activity, phenotype); ``alleles``: (allele, function, activity)."""
+    by_key: dict[str, tuple[str, str | None]] = {}
+    totals: dict[float, str] = {}
+    for key, total, pheno in rows:
+        k = json.dumps(json.loads(key), sort_keys=True)
+        by_key.setdefault(k, (pheno, total))
+        t = _num(total)
+        if t is not None:
+            totals.setdefault(t, pheno)
+    func = {a: f for a, f, _ in alleles}
+    act = {a: _num(v) for a, _, v in alleles}
+    return GuidelineTable(gene, by_key, sorted(totals.items()), func, act)
+
+
+def _split(allele: str) -> tuple[str, int]:
+    m = re.fullmatch(r"(.+?)x(\d+|N)", allele)
+    if not m:
+        return allele, 1
+    return m.group(1), (3 if m.group(2) == "N" else int(m.group(2)))
+
+
+def phenotype(t: GuidelineTable, haplotypes: tuple[str, ...]) -> PhenotypeResult:
+    """Phenotype of one diplotype (haplotypes as reported, e.g. ("*2x2", "*4") or ("*1", "*5"))."""
+    deletion = {"*5", "*0"}
+    if t.activity_based:
+        vals: list[float] = []
+        for h in haplotypes:
+            core, n = _split(h)
+            a = 0.0 if core in deletion else t.activity.get(core)
+            if a is None:
+                return PhenotypeResult("Indeterminate", None, ["phenotype_not_in_guideline"])
+            vals.append(a * n)
+        counts: dict[str, int] = {}
+        for v in vals:
+            s = f"{v:.1f}" if abs(v * 10 - round(v * 10)) < 1e-9 else f"{v:g}"
+            counts[s] = counts.get(s, 0) + 1
+        key = json.dumps(counts, sort_keys=True)
+        total = sum(vals)
+        hit = t.by_key.get(key)
+        if hit is not None:
+            return PhenotypeResult(hit[0], total)
+        below = [p for tot, p in t.by_total if tot <= total + 1e-9]
+        if below:
+            return PhenotypeResult(below[-1], total)
+        return PhenotypeResult("Indeterminate", total, ["phenotype_not_in_guideline"])
+    funcs: dict[str, int] = {}
+    for h in haplotypes:
+        core, n = _split(h)
+        f = "No function" if core in deletion else t.function.get(core)
+        if f is None:
+            return PhenotypeResult("Indeterminate", None, ["phenotype_not_in_guideline"])
+        if n > 1:
+            return PhenotypeResult("Indeterminate", None, ["phenotype_not_in_guideline"])  # no CPIC rule for xN here
+        funcs[f] = funcs.get(f, 0) + 1
+    hit = t.by_key.get(json.dumps(funcs, sort_keys=True))
+    if hit is None:
+        return PhenotypeResult("Indeterminate", None, ["phenotype_not_in_guideline"])
+    return PhenotypeResult(hit[0], None)
+
+
+def gene_phenotype(t: GuidelineTable, feasible: list[tuple[str, ...]]) -> tuple[PhenotypeResult, list[str]]:
+    """PGX-025: a phenotype only when every feasible solution agrees; otherwise ``Indeterminate`` plus the
+    per-solution phenotypes."""
+    res = [phenotype(t, s) for s in feasible]
+    if not res:
+        return PhenotypeResult(None, None, ["no_call"]), []
+    phenos = sorted({r.phenotype or "" for r in res})
+    if len(phenos) == 1:
+        acts = {r.activity_score for r in res}
+        return PhenotypeResult(res[0].phenotype, res[0].activity_score if len(acts) == 1 else None,
+                               sorted({f for r in res for f in r.flags})), []  # fmt: skip
+    return PhenotypeResult("Indeterminate", None, ["feasible_solutions_disagree"]), [r.phenotype or "" for r in res]