PGx: full call detail (PGX-016..020, 027/028, 031), pgx coverage (PGX-009/035), full HGVS

- supporting_variants.tsv, unresolved_variants.tsv, variants_not_covered.tsv with PGX-026 metadata; per-allele
  supporting/unresolved sites for both alleles of every candidate; variant_change vs top, distant_alternative,
  unexplained_variants, alleles_impacted_by_no_call, alleles_tested.
- simurg pgx coverage: callable / not_covered / masked / equivalent_to per allele and manifest.
- HGVS on GRCh38 RefSeq accessions (NC_...:g.).
This commit is contained in:
kemal committed 2026-09-29 16:16:39 +01:00
1 parent 4b22006c04
commit 304c69cb94
10 files changed
+352 -40

No files matched your search

+26
View File
@@ -121,3 +121,29 @@ def test_db_build_deterministic(tmp_path: Path) -> None: # IN-053, CLI-035
meta = dict(con.execute("select key, value from meta").fetchall())
assert meta["content_sha256"][:12] in outs[0].name and json.loads(meta["allele_sources"])[0]["name"] == "CPIC"
assert con.execute("select phenotype from phenotypes").fetchone() == ("Poor Metabolizer",)
def test_coverage_statuses(tmp_path: Path) -> None: # PGX-009 / PGX-035
from simurg.pipeline.pgx_db import coverage_rows
src = tmp_path / "sources" / "cpic-20260101"
src.mkdir(parents=True)
for k, v in _tables().items():
(src / f"{k}.json").write_text(json.dumps(v))
(src / "source.json").write_text(json.dumps({"name": "CPIC", "version": "2026-01-01", "retrieved": "20260101"}))
fa = tmp_path / "g.fa"
fa.write_text(">chr1\n" + SEQ + "\n")
out = tmp_path / "db"
r = CliRunner().invoke(app, ["pgx", "db", "build", "--sources", str(tmp_path / "sources"), "--fasta", str(fa),
"--out", str(out), "--quiet"]) # fmt: skip
assert r.exit_code == 0, r.output
db = next(out.glob("pgx_db-1-*.sqlite"))
con = sqlite3.connect(db)
con.executemany("INSERT INTO probe_map VALUES (?, ?, ?, ?, ?, ?)",
[("M", "p_star2", "chr1", 5, "A", "G"), ("M", "p_star3", "chr1", 7, "GT", "G")]) # fmt: skip
con.execute("INSERT INTO masks VALUES ('M', 'p_star3', 'test')")
con.commit()
con.close()
st = {r["allele"]: r["status"] for r in coverage_rows(db, "M")}
assert st["*1"] == "callable" and st["*2"] == "callable"
assert st["*3"] == "masked" and st["*5"] == "not_covered"