"""``simurg pgx star-allele call`` — star alleles from SNV VCFs, PGx DB and PGx CN VCFs (spec PGX-011 … PGX-021, CLI-033). I/O and orchestration; the scoring lives in ``simurg.pgx.star``. Scope of this version: genes whose result basis is ``star_allele`` in the DB. Named-haplotype (G6PD), variant-based and tagging genes, sub-allele collapsing and hybrid CYP2D6 structures follow (K-029). """ from __future__ import annotations import contextlib import gzip import json import math import sqlite3 from collections import defaultdict from dataclasses import dataclass, field from pathlib import Path from typing import Any import numpy as np import numpy.typing as npt from simurg import __version__ from simurg.core.errors import ExitCode, SimurgError from simurg.core.log import Reporter from simurg.pgx.star import ( CALLED, MASKED, NO_CALL, NOT_COVERED, GeneModel, Observations, Solution, call_gene, cyp2d6_structure, ) MAX_EDIT_DISTANCE = 3 # PGX-019 CN_REGION = {"CYP2D6": "CYP2D6.p5"} # PGX-013: total gene copies from the 5′ region CN_SOLUTION_REGIONS = {"CYP2D6": ("CYP2D6.p5", "CYP2D6.intron.2", "CYP2D6.exon.9")} # PGX-027 copy_number_solution CN_ONLY_GENES = {"GSTM1": "GSTM1", "GSTT1": "GSTT1", "UGT2B17": "UGT2B17"} # PGX-006: *0 from whole-gene CN ALL_BASES = ("star_allele", "haplotype_named", "variant_based", "tagging_variant") TAG_CAVEAT = "tag reliability varies by population" DELETION_ALLELE = {"CYP2D6": "*5"} @dataclass class StarOptions: vcfs: list[Path] db: Path out: Path population: str = "global" max_candidates: int = 10 feasible_ratio: float = 0.5 low_gs: float = 0.15 no_cn: bool = False allow_foreign_vcf: bool = False genes: list[str] = field(default_factory=list) force: bool = False def load_models( db: Path, population: str, genes: list[str] | None = None, bases: tuple[str, ...] = ("star_allele",) ) -> dict[str, GeneModel]: con = sqlite3.connect(db) q = f"SELECT gene, result_basis FROM genes WHERE result_basis IN ({','.join('?' * len(bases))}) ORDER BY gene" out = {} for gene, basis in con.execute(q, bases).fetchall(): if genes and gene not in genes: continue alle = con.execute("SELECT allele, kind FROM alleles WHERE gene = ? AND defined = 1 AND structural = 0 " "ORDER BY allele", (gene,)).fetchall() # fmt: skip names = [a for a, _ in alle] ref = next((a for a, k in alle if k == "reference"), names[0] if names else "") rows = con.execute("SELECT allele, chrom, pos, ref, alt, hgvs, rsid, location_id FROM allele_variants " "WHERE gene = ?", (gene,)).fetchall() # fmt: skip keys = sorted({(r[1], r[2], r[3], r[4]) for r in rows}, key=lambda k: (k[0], k[1], k[2], k[3])) kidx = {k: i for i, k in enumerate(keys)} ids: list[tuple[str, str | None]] = [("", None)] * len(keys) loc_keys: dict[int, list[int]] = defaultdict(list) carries = np.zeros((len(names), len(keys)), dtype=np.int8) hidx = {n: i for i, n in enumerate(names)} for allele, chrom, pos, r_, a_, hgvs, rsid, lid in rows: k = kidx[(chrom, pos, r_, a_)] ids[k] = (hgvs, rsid) loc_keys[lid].append(k) if allele in hidx: carries[hidx[allele], k] = 1 for allele, lid in con.execute("SELECT allele, location_id FROM allele_wildcards WHERE gene = ?", (gene,)): if allele in hidx: for k in loc_keys.get(lid, []): carries[hidx[allele], k] = -1 freq = np.full(len(names), np.nan) fr: dict[str, list[float]] = defaultdict(list) for allele, pop, f in con.execute("SELECT allele, population, freq FROM frequencies WHERE gene = ?", (gene,)): if population in ("global", pop): fr[allele].append(float(f)) for i, n in enumerate(names): if fr.get(n): freq[i] = float(np.mean(fr[n])) freq = np.nan_to_num(freq, nan=0.0) fa = {a: (f, v) for a, f, v in con.execute("SELECT allele, function, activity FROM alleles WHERE gene = ?", (gene,))} # fmt: skip act = {} for a, (_, v) in fa.items(): with contextlib.suppress(TypeError, ValueError): act[a] = float(v) out[gene] = GeneModel(gene, keys, ids, names, carries, freq, DELETION_ALLELE.get(gene), ref, basis, act, {a: f for a, (f, _) in fa.items() if f}) # fmt: skip con.close() return out def read_vcf_sites( path: Path, wanted: dict[str, set[int]] ) -> tuple[dict[tuple[str, int], list[dict[str, Any]]], dict[str, str]]: """Records at the wanted positions, and the ``##`` header meta (``##key=value``).""" recs: dict[tuple[str, int], list[dict[str, Any]]] = defaultdict(list) meta: dict[str, str] = {} with gzip.open(path, "rt") as fh: for ln in fh: if ln.startswith("##"): k, _, v = ln[2:].rstrip("\n").partition("=") meta.setdefault(k, v) continue if ln.startswith("#"): continue c, p, _ = ln.split("\t", 2) ps = wanted.get(c) if ps is None or int(p) not in ps: continue f = ln.rstrip("\n").split("\t") fmt = f[8].split(":") val = dict(zip(fmt, f[9].split(":"), strict=False)) recs[(c, int(p))].append({"id": f[2], "ref": f[3], "alts": f[4].split(","), "gt": val.get("GT", "./."), "gs": val.get("GS"), "baf": val.get("BAF")}) # fmt: skip return recs, meta def observations( m: GeneModel, recs: dict[tuple[str, int], list[dict[str, Any]]], masked: set[str] | None = None ) -> tuple[Observations, int]: """PGX-012 evidence per defining variant; returns the observations and the ploidy seen in the GTs. A record with any masked probe (PGX-010) makes the variant ``masked``.""" v = len(m.variants) state = np.full(v, NOT_COVERED, dtype=np.int8) dosage = np.full(v, -1, dtype=np.int8) gs = np.full(v, np.nan) baf = np.full(v, np.nan) ploidy = 2 for k, (c, p, r_, a_) in enumerate(m.variants): for rec in recs.get((c, p), []): if rec["ref"] != r_ or a_ not in rec["alts"]: continue ai = rec["alts"].index(a_) + 1 if masked and any(pn in masked for pn in rec["id"].replace(";", ",").split(",")): state[k] = MASKED break alleles = rec["gt"].replace("|", "/").split("/") ploidy = len(alleles) if any(a in (".", "") for a in alleles): state[k] = NO_CALL break state[k] = CALLED dosage[k] = sum(int(a) == ai for a in alleles) gs[k] = float(rec["gs"]) if rec["gs"] not in (None, ".") else math.nan baf[k] = float(rec["baf"]) if rec["baf"] not in (None, ".") else math.nan if len(rec["alts"]) > 1 and not math.isnan(baf[k]): baf[k] = math.nan # BAF of a merged multi-allelic record is not per-ALT break return Observations(state, dosage, gs, baf), ploidy def variant_records(m: GeneModel, recs: dict[tuple[str, int], list[dict[str, Any]]]) -> list[dict[str, Any] | None]: """The VCF record that carries each defining variant (same REF, ALT among the ALTs), or None.""" out: list[dict[str, Any] | None] = [] for c, p, r_, a_ in m.variants: out.append(next((rec for rec in recs.get((c, p), []) if rec["ref"] == r_ and a_ in rec["alts"]), None)) return out def _vcn(obs: Observations, k: int, n: int) -> int | None: """PGX-012 variant copy number: GT dosage at CN 2, else the ML k of BAF ~ N(k/n, σ²).""" if obs.state[k] != CALLED: return None if n == 2 or math.isnan(obs.baf[k]): return int(obs.dosage[k]) return int(min(n, max(0, round(obs.baf[k] * n)))) def solution_detail(m: GeneModel, obs: Observations, recs: list[dict[str, Any] | None], s: Solution, top: Solution, low_gs: float, max_edit: int) -> dict[str, Any]: # fmt: skip """PGX-018 … PGX-020, PGX-028: per-allele supporting / unresolved variants, no-call impact, change vs top.""" n = max(1, len(s.copies)) alleles = [] names = list(s.haplotypes) hap_of = {m.haplotypes[i]: i for i in s.copies} reason = {NO_CALL: "no_call", MASKED: "masked", NOT_COVERED: "not_on_array"} for name in names: core = name.split("x")[0] if "x" in name and name.split("x")[-1].isdigit() else name h = hap_of.get(core) sup, unres = [], [] if h is not None: for k in np.flatnonzero(m.carries[h] == 1).tolist(): c, p, r_, a_ = m.variants[k] hg, rs = m.variant_ids[k] rec = recs[k] if obs.state[k] == CALLED: sup.append({"chrom": c, "pos": p, "ref": r_, "alts": rec["alts"] if rec else [a_], "gt": rec["gt"] if rec else None, "gs": None if math.isnan(obs.gs[k]) else round(float(obs.gs[k]), 4), "baf": None if math.isnan(obs.baf[k]) else round(float(obs.baf[k]), 4), "variant_copy_number": _vcn(obs, k, n), "hgvs": hg, "rsid": rs, "probe_ids": rec["id"].split(",") if rec else [], "low_gs": bool(obs.gs[k] < low_gs)}) # fmt: skip else: unres.append({"hgvs": hg, "rsid": rs, "reason": reason[int(obs.state[k])], "probe_ids": rec["id"].split(",") if rec else []}) # fmt: skip background = None if "+" in name: background = name.split("+")[-1] alleles.append({"allele": name, "supporting_variants": sup, "unresolved_variant_sites": unres, "array_equivalent_alleles": sorted({mm[names.index(name)] for mm in s.members if len(mm) == len(names)}, key=_natural), "background_allele": background}) # fmt: skip exp = _expected(m, s.copies) exp_top = _expected(m, top.copies) called = obs.state == CALLED changed = [m.variant_ids[k][0] for k in np.flatnonzero(called & (exp != exp_top)).tolist()] flags = ["distant_alternative"] if len(changed) > max_edit else [] return {"alleles": alleles, "variant_change": changed if s is not top else None, "flags": flags} def _expected(m: GeneModel, copies: tuple[int, ...]) -> npt.NDArray[np.int16]: e = np.zeros(m.carries.shape[1], dtype=np.int16) for i in copies: e += np.maximum(m.carries[i], 0) return e def read_cn_vcf(path: Path) -> dict[str, int | None]: out: dict[str, int | None] = {} with gzip.open(path, "rt") as fh: for ln in fh: if ln.startswith("#"): continue f = ln.rstrip("\n").split("\t") region = f[7].split("REGION=")[1].split(";")[0] cn = f[9].split(":")[0] out[region] = None if cn == "." or "NoCall" in f[6] else int(cn) return out def _discover(dirs: list[Path]) -> tuple[dict[str, Path], dict[str, Path]]: snv, cn = {}, {} for d in dirs: for p in sorted(d.rglob("*.vcf.gz")): if p.name.endswith(".pgx_cn.vcf.gz"): cn[p.name[: -len(".pgx_cn.vcf.gz")]] = p elif p.name.endswith(".snv.vcf.gz"): snv[p.name[: -len(".snv.vcf.gz")]] = p return snv, cn def run_star_call(o: StarOptions, rep: Reporter) -> int: """``simurg pgx star-allele call`` (CLI-033).""" from simurg.core.atomic import check_outputs_absent from simurg.pipeline.pgx_pheno import db_meta, tsv_header, write_meta_tsv snv, cnv = _discover(o.vcfs) if not snv: raise SimurgError("E_NO_INPUT", f"no *.snv.vcf.gz under {', '.join(map(str, o.vcfs))}") missing_cn = sorted(set(snv) - set(cnv)) if missing_cn and not o.no_cn: raise SimurgError("E_PGX_CN_MISSING", f"{len(missing_cn)} sample(s) have no .pgx_cn.vcf.gz (e.g. " f"{missing_cn[0]})", hint="run pgx copy-number call, or give --no-cn") # fmt: skip for vp in snv.values(): if not Path(str(vp) + ".tbi").exists() and not Path(str(vp) + ".csi").exists(): raise SimurgError("E_PGX_VCF_UNINDEXED", f"{vp} is not indexed") check_outputs_absent([o.out / "diplotype_summary.tsv"], force=o.force) models = load_models(o.db, o.population, o.genes or None, ALL_BASES) mcon = sqlite3.connect(o.db) masks: dict[str, set[str]] = defaultdict(set) for man, probe in mcon.execute("SELECT manifest, probe_name FROM masks"): masks[man].add(probe) mcon.close() wanted: dict[str, set[int]] = defaultdict(set) for m in models.values(): for c, p, _, _ in m.variants: wanted[c].add(p) rows: list[dict[str, Any]] = [] sup_rows: list[dict[str, Any]] = [] unres_rows: list[dict[str, Any]] = [] not_cov_rows: dict[tuple[str, str], dict[str, Any]] = {} calls_json = [] for sid, path in sorted(snv.items()): recs, meta = read_vcf_sites(path, wanted) opts = meta.get("simurg_options", "") if not opts and not o.allow_foreign_vcf: raise SimurgError("E_PGX_VCF_UNSUITABLE", f"{path}: not a SimurgArray VCF (no ##simurg_options)", hint="give --allow-foreign-vcf to use it anyway") # fmt: skip kv = dict(x.partition("=")[::2] for x in opts.split(";") if x) if ( kv.get("unsquash_duplicates") == "true" or kv.get("exclude_loci", "none").lower() != "none" or kv.get("include_loci", "none").lower() != "none" ): # PGX-011 raise SimurgError("E_PGX_VCF_UNSUITABLE", f"{path}: made with unsquashed duplicates or a locus filter") cn = {} if o.no_cn or sid not in cnv else read_cn_vcf(cnv[sid]) masked = masks.get(meta.get("simurg_manifest", "").split(";")[0], set()) sample_name = sid genes_out = [] for gene, m in sorted(models.items()): gene_json, g_rows, g_sup, g_unres, g_ncov = call_one_gene(sid, sample_name, m, recs, masked, cn, o) if gene_json is None: continue genes_out.append(gene_json) rows += g_rows sup_rows += g_sup unres_rows += g_unres for x in g_ncov: not_cov_rows[(gene, str(x["hgvs"]))] = x if not o.no_cn: for gene, region in CN_ONLY_GENES.items(): if o.genes and gene not in o.genes: continue gj, row = cn_only_gene(sid, sample_name, gene, cn.get(region)) genes_out.append(gj) rows.append(row) calls_json.append({"sample_id": sid, "genes": genes_out}) cols = ["sample_id", "sample_name", "solution_id", "gene", "call_type", "solution", "array_equivalent_alleles_1", "array_equivalent_alleles_2", "no_call_variants", "alleles_impacted_by_no_call", "background_alleles", "variant_change", "score", "raw_score", "copy_number_solution", "flags"] # fmt: skip o.out.mkdir(parents=True, exist_ok=True) head = tsv_header(o.db, db_meta(o.db), "none")[:-1] # PGX-026 metadata (no guideline before phenotyping) write_meta_tsv(o.out / "diplotype_summary.tsv", head, cols, rows) write_meta_tsv(o.out / "supporting_variants.tsv", head, ["sample_id", "solution_id", "gene", "allele", "hgvs", "rsid", "chrom", "pos", "ref", "alts", "genotype", "variant_copy_number", "gs", "baf", "probe_ids", "low_gs"], sup_rows) # fmt: skip write_meta_tsv(o.out / "unresolved_variants.tsv", head, ["sample_id", "gene", "solution_id", "allele", "hgvs", "rsid", "reason", "probe_ids"], unres_rows) # fmt: skip write_meta_tsv(o.out / "variants_not_covered.tsv", head, ["gene", "hgvs", "rsid", "alleles_impacted"], [not_cov_rows[k] for k in sorted(not_cov_rows)]) # fmt: skip (o.out / "pgx_calls.json").write_text(json.dumps({"software": f"SimurgArray {__version__}", "db": o.db.name, "samples": calls_json}, indent=1)) # fmt: skip rep.info(f"star alleles: {len(calls_json)} sample(s), {len(models)} gene(s)") return ExitCode.OK def call_one_gene( sid: str, sample_name: str, m: GeneModel, recs: dict[tuple[str, int], list[dict[str, Any]]], masked: set[str], cn: dict[str, int | None], o: StarOptions, ) -> tuple[dict[str, Any] | None, list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: # fmt: skip """One gene of one sample: call, details, and the rows of the four TSVs (PGX-012 … PGX-031).""" if m.basis in ("variant_based", "tagging_variant"): return call_variant_gene(sid, sample_name, m, recs, masked, o) gene = m.gene obs, ploidy = observations(m, recs, masked) vrec = variant_records(m, recs) region = CN_REGION.get(gene) gcn = cn.get(region) if region else None structure = None if gene == "CYP2D6" and gcn is not None: structure = cyp2d6_structure(gcn, cn.get("CYP2D6.intron.2"), cn.get("CYP2D6.exon.9")) sols, feasible = call_gene(m, obs, cn=gcn if region else None, ploidy=ploidy, max_candidates=o.max_candidates, feasible_ratio=o.feasible_ratio, structure=structure) # fmt: skip if not sols: return None, [], [], [], [] top = sols[0] details = [solution_detail(m, obs, vrec, s_, top, o.low_gs, MAX_EDIT_DISTANCE) for s_ in sols] flags = [] if any(sv["low_gs"] for al in details[0]["alleles"] for sv in al["supporting_variants"]): flags.append("review_alternatives") # PGX-016 v_nc = [k for k in range(len(m.variants)) if obs.state[k] in (NO_CALL, MASKED)] v_ncov = [k for k in range(len(m.variants)) if obs.state[k] == NOT_COVERED] def impacted(k: int) -> list[str]: return sorted((m.haplotypes[h] for h in np.flatnonzero(m.carries[:, k] == 1).tolist()), key=_natural) def probes(k: int) -> list[str]: rec = vrec[k] return str(rec["id"]).split(",") if rec else [] exp_any = np.zeros(len(m.variants), dtype=np.int16) for i in top.copies: exp_any += m.carries[i] != 0 unexplained = [m.variant_ids[k][0] for k in range(len(m.variants)) if obs.state[k] == CALLED and obs.dosage[k] > 0 and exp_any[k] == 0] # fmt: skip if unexplained: flags.append("unexplained_variants") # PGX-031 no_call_list = [{"hgvs": m.variant_ids[k][0], "rsid": m.variant_ids[k][1], "probe_ids": probes(k), "alleles_impacted": impacted(k)} for k in v_nc] # fmt: skip not_cov = [{"hgvs": m.variant_ids[k][0], "rsid": m.variant_ids[k][1], "alleles_impacted": impacted(k)} for k in v_ncov] # fmt: skip covered = obs.state != NOT_COVERED defining = m.carries == 1 tested = sorted((m.haplotypes[h] for h in range(len(m.haplotypes)) if not defining[h].any() or (defining[h] & covered).any()), key=_natural) # fmt: skip imp_nc = sorted({a for k in v_nc for a in impacted(k)}, key=_natural) rows, sup_rows, unres_rows = [], [], [] for rank, (s_, d) in enumerate(zip(sols, details, strict=True), start=1): sflags = flags + d["flags"] + (["feasible"] if s_ in feasible else []) rows.append({"sample_id": sid, "sample_name": sample_name, "solution_id": rank, "gene": gene, "call_type": m.basis, "solution": s_.genotype, "array_equivalent_alleles_1": ";".join(d["alleles"][0]["array_equivalent_alleles"]), "array_equivalent_alleles_2": ";".join(d["alleles"][-1]["array_equivalent_alleles"]), "no_call_variants": ";".join(str(x["hgvs"]) for x in no_call_list), "alleles_impacted_by_no_call": ";".join(imp_nc), "background_alleles": ";".join(al["background_allele"] for al in d["alleles"] if al["background_allele"]), "variant_change": ";".join(d["variant_change"] or []), "score": f"{s_.score:.4f}", "raw_score": f"{s_.raw_score:.4f}", "copy_number_solution": ";".join(f"{r}={cn.get(r)}" for r in CN_SOLUTION_REGIONS.get(gene, ()) if cn.get(r) is not None), "flags": ";".join(sflags)}) # fmt: skip for al in d["alleles"]: for sv in al["supporting_variants"]: sup_rows.append({"sample_id": sid, "solution_id": rank, "gene": gene, "allele": al["allele"], "hgvs": sv["hgvs"], "rsid": sv["rsid"], "chrom": sv["chrom"], "pos": sv["pos"], "ref": sv["ref"], "alts": ";".join(sv["alts"]), "genotype": sv["gt"], "variant_copy_number": sv["variant_copy_number"], "gs": sv["gs"], "baf": sv["baf"], "probe_ids": ";".join(sv["probe_ids"]), "low_gs": str(sv["low_gs"]).lower()}) # fmt: skip for uv in al["unresolved_variant_sites"]: unres_rows.append({"sample_id": sid, "gene": gene, "solution_id": rank, "allele": al["allele"], "hgvs": uv["hgvs"], "rsid": uv["rsid"], "reason": uv["reason"], "probe_ids": ";".join(uv["probe_ids"])}) # fmt: skip ncov_rows = [{"gene": gene, "hgvs": m.variant_ids[k][0], "rsid": m.variant_ids[k][1], "alleles_impacted": ";".join(impacted(k))} for k in v_ncov] # fmt: skip cands = [] for s_, d in zip(sols, details, strict=True): cj = _sol_json(s_) cj.update({"alleles": d["alleles"], "variant_change": d["variant_change"], "flags": d["flags"], "variant_no_calls": no_call_list, "alleles_impacted_by_no_calls": imp_nc}) # fmt: skip cands.append(cj) gene_json = {"gene": gene, "call_type": m.basis, "genotype": ";".join(s_.genotype for s_ in feasible), "feasible_solutions": [s_.genotype for s_ in feasible], "candidates": cands, "copy_number": gcn, "flags": flags, "variant_no_calls": no_call_list, "variants_not_covered": not_cov, "unexplained_variants": unexplained, "alleles_tested": tested} # fmt: skip return gene_json, rows, sup_rows, unres_rows, ncov_rows def call_variant_gene( sid: str, sample_name: str, m: GeneModel, recs: dict[tuple[str, int], list[dict[str, Any]]], masked: set[str], o: StarOptions, ) -> tuple[dict[str, Any] | None, list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: # fmt: skip """PGX-007 / PGX-021: variant-based and tagging genes are reported per variant (``rs9923231 C/T`` or ``Reference``); the diplotype for the phenotype lookup is built from the carried variant alleles — with more than two, the two of lowest activity (CPIC DPYD rule), non-normal function before normal [DB: CPIC DPYD guideline].""" obs, ploidy = observations(m, recs, masked) vrec = variant_records(m, recs) carried: list[str] = [] variant_text: list[str] = [] sup_rows: list[dict[str, Any]] = [] for k, (c, p, r_, a_) in enumerate(m.variants): if obs.state[k] != CALLED or obs.dosage[k] <= 0: continue rec = vrec[k] hg, rs = m.variant_ids[k] gt = "/".join([r_] * (ploidy - int(obs.dosage[k])) + [a_] * int(obs.dosage[k])) variant_text.append(f"{rs or hg} {gt}") owners = [m.haplotypes[h] for h in np.flatnonzero(m.carries[:, k] == 1).tolist()] carried += owners[:1] * int(obs.dosage[k]) sup_rows.append({"sample_id": sid, "solution_id": 1, "gene": m.gene, "allele": owners[0] if owners else "", "hgvs": hg, "rsid": rs, "chrom": c, "pos": p, "ref": r_, "alts": ";".join(rec["alts"]) if rec else a_, "genotype": rec["gt"] if rec else "", "variant_copy_number": int(obs.dosage[k]), "gs": None if math.isnan(obs.gs[k]) else round(float(obs.gs[k]), 4), "baf": None if math.isnan(obs.baf[k]) else round(float(obs.baf[k]), 4), "probe_ids": rec["id"] if rec else "", "low_gs": str(bool(obs.gs[k] < o.low_gs)).lower()}) # fmt: skip def severity(a: str) -> tuple[float, int]: f = (m.function.get(a) or "").lower() return (m.activity.get(a, 1.0), 0 if f and "normal" not in f else 1) carried.sort(key=severity) ref = m.reference or "Reference" dip = (carried + [ref] * ploidy)[:ploidy] if m.basis == "tagging_variant": genotype = ";".join(f"{a} {'positive' if a in carried else 'negative'}" for a in m.haplotypes) else: genotype = "; ".join(variant_text) if variant_text else "Reference" v_nc = [k for k in range(len(m.variants)) if obs.state[k] in (NO_CALL, MASKED)] v_ncov = [k for k in range(len(m.variants)) if obs.state[k] == NOT_COVERED] flags = ["unresolved_due_to_no_call"] if v_nc else [] if m.basis == "tagging_variant": flags.append("tag_caveat") def probe_ids(k: int) -> list[str]: rec = vrec[k] return str(rec["id"]).split(",") if rec else [] no_call_list = [{"hgvs": m.variant_ids[k][0], "rsid": m.variant_ids[k][1], "probe_ids": probe_ids(k), "alleles_impacted": [m.haplotypes[h] for h in np.flatnonzero(m.carries[:, k] == 1).tolist()]} for k in v_nc] # fmt: skip not_cov = [{"hgvs": m.variant_ids[k][0], "rsid": m.variant_ids[k][1], "alleles_impacted": [m.haplotypes[h] for h in np.flatnonzero(m.carries[:, k] == 1).tolist()]} for k in v_ncov] # fmt: skip diplotype = "/".join(dip) imp_nc = sorted({m.haplotypes[h] for k in v_nc for h in np.flatnonzero(m.carries[:, k] == 1).tolist()}) cand = {"genotype": diplotype, "score": 1.0, "raw_score": 1.0, "equivalents": [diplotype], "copy_number": 2, "alleles": [{"allele": a, "supporting_variants": [], "unresolved_variant_sites": [], "array_equivalent_alleles": [a], "background_allele": None} for a in dip], "variant_change": None, "flags": [], "variant_no_calls": no_call_list, "alleles_impacted_by_no_calls": imp_nc} # fmt: skip row = {"sample_id": sid, "sample_name": sample_name, "solution_id": 1, "gene": m.gene, "call_type": "variant" if m.basis == "variant_based" else m.basis, "solution": genotype, "array_equivalent_alleles_1": "", "array_equivalent_alleles_2": "", "no_call_variants": ";".join(str(x["hgvs"]) for x in no_call_list), "alleles_impacted_by_no_call": ";".join(imp_nc), "background_alleles": "", "variant_change": "", "score": "1.0000", "raw_score": "1.0000", "copy_number_solution": "", "flags": ";".join([*flags, "feasible"])} # fmt: skip gene_json = {"gene": m.gene, "call_type": "variant" if m.basis == "variant_based" else m.basis, "genotype": genotype, "feasible_solutions": [diplotype], "candidates": [cand], "copy_number": None, "flags": flags, "variant_no_calls": no_call_list, "variants_not_covered": not_cov, "unexplained_variants": [], "alleles_tested": sorted(m.haplotypes, key=_natural), "caveat": TAG_CAVEAT if m.basis == "tagging_variant" else None} # fmt: skip ncov_rows = [{"gene": m.gene, "hgvs": m.variant_ids[k][0], "rsid": m.variant_ids[k][1], "alleles_impacted": ";".join(m.haplotypes[h] for h in np.flatnonzero(m.carries[:, k] == 1).tolist())} for k in v_ncov] # fmt: skip return gene_json, [row], sup_rows, [], ncov_rows def cn_only_gene(sid: str, sample_name: str, gene: str, cn: int | None) -> tuple[dict[str, Any], dict[str, Any]]: """PGX-006: genes with a whole-gene CN region and no named deletion allele — CN 0 → *0/*0, 1 → *1/*0, 2 → *1/*1, n ≥ 3 → *1x(n−1)/*1 (N when capped).""" if cn is None: geno = "" elif cn == 0: geno = "*0/*0" elif cn == 1: geno = "*1/*0" elif cn == 2: geno = "*1/*1" else: geno = f"*1x{'N' if cn >= 5 else cn - 1}/*1" flags = [] if geno else ["no_call"] cand = {"genotype": geno, "score": 1.0, "raw_score": 1.0, "equivalents": [geno] if geno else [], "copy_number": cn, "alleles": [], "variant_change": None, "flags": [], "variant_no_calls": [], "alleles_impacted_by_no_calls": []} # fmt: skip gj = {"gene": gene, "call_type": "copy_number", "genotype": geno, "feasible_solutions": [geno] if geno else [], "candidates": [cand] if geno else [], "copy_number": cn, "flags": flags, "variant_no_calls": [], "variants_not_covered": [], "unexplained_variants": [], "alleles_tested": ["*0", "*1"]} # fmt: skip row = {"sample_id": sid, "sample_name": sample_name, "solution_id": 1, "gene": gene, "call_type": "copy_number", "solution": geno, "array_equivalent_alleles_1": "", "array_equivalent_alleles_2": "", "no_call_variants": "", "alleles_impacted_by_no_call": "", "background_alleles": "", "variant_change": "", "score": "1.0000", "raw_score": "1.0000", "copy_number_solution": f"{gene}={'' if cn is None else cn}", "flags": ";".join([*flags, "feasible"])} # fmt: skip return gj, row def _sol_json(s: Solution) -> dict[str, Any]: return {"genotype": s.genotype, "score": round(s.score, 4), "raw_score": round(s.raw_score, 4), "equivalents": ["/".join(m) for m in s.members][:20], "copy_number": s.copy_number} # fmt: skip def _natural(a: str) -> tuple[Any, ...]: import re return tuple(int(x) if x.isdigit() else x for x in re.split(r"(\d+)", a))