kemal
cc89e9b334
- fetch_sources.sh downloads ClinPGx clinicalVariants (CC BY-SA 4.0); pgx db build adds genotype-only variant-based genes (levels 1A-2B, level 3 for the parity list), rsIDs resolved from the manifest's probes: 70 genes in the DB. - pgx phenotype --guideline dpwg: CPIC mappings for CPIC genes (as ClinPGx curates DPWG), no phenotype for DPWG-only genes (flags dpwg_via_cpic_mapping / dpwg_mapping_not_available). K-029 updated.
202 lines
10 KiB
Python
202 lines
10 KiB
Python
"""PGx DB build (spec PGX-001 … PGX-005, IN-053, CLI-035)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from simurg.cli.main import app
|
|
from simurg.pgx.db import Window, cpic_definitions, normalize, parse_location, value_sequence
|
|
|
|
SEQ = "ACGTACGTTTTTAGAGAGAGCCCGGGTATATATAGGCCAATTGGCCAA" * 3
|
|
|
|
|
|
def fetch(contig: str, s0: int, e0: int) -> str:
|
|
return SEQ[s0:e0]
|
|
|
|
|
|
def test_parse_location() -> None:
|
|
assert parse_location("g.5C>T") == Window(5, 5, "snv")
|
|
assert parse_location("g.9_12del") == Window(9, 12, "del")
|
|
assert parse_location("g.13_14insGA") == Window(13, 13, "ins")
|
|
assert parse_location("g.20dup") == Window(20, 20, "dup")
|
|
assert parse_location("g.13AG[4]") == Window(13, 13, "repeat", "AG")
|
|
assert parse_location("g.21_22delinsTT") == Window(21, 22, "delins")
|
|
assert parse_location("m.961delT; m.961delTinsC(n)") == Window(961, 961, "del")
|
|
|
|
|
|
def test_normalize_left_aligns() -> None: # the SNV VCF representation (IN-046)
|
|
# deleting one T of TTTTT (positions 8-12) → anchored at the base before the run
|
|
assert normalize(fetch, "1", 11, 11, "") == (7, "GT", "G")
|
|
# inserting AG into AGAGAGAG (13-20) → left-most position
|
|
assert normalize(fetch, "1", 21, 20, "AG") == (12, "T", "TAG")
|
|
assert normalize(fetch, "1", 5, 5, "G") == (5, "A", "G")
|
|
w = parse_location("g.13AG[4]")
|
|
assert w is not None
|
|
s, e, alt = value_sequence("AG(5)", w, fetch, "1") or (0, 0, "")
|
|
assert (s, e) == (13, 20) and normalize(fetch, "1", s, e, alt) == (12, "T", "TAG")
|
|
assert value_sequence("AG(3) or AG(4)", w, fetch, "1") is None # ambiguous CPIC value
|
|
|
|
|
|
def _tables() -> dict[str, list[dict]]:
|
|
loc = [
|
|
{"id": 1, "name": "5A>G", "chromosomelocation": "g.5A>G", "genesymbol": "GENE1", "dbsnpid": "rs1"},
|
|
{"id": 2, "name": "11delT", "chromosomelocation": "g.11del", "genesymbol": "GENE1", "dbsnpid": "rs2"},
|
|
{"id": 3, "name": "30C>T", "chromosomelocation": "g.30G>T", "genesymbol": "GENE1", "dbsnpid": "rs3"},
|
|
]
|
|
defs = [
|
|
{"id": 10, "genesymbol": "GENE1", "name": "*1", "matchesreferencesequence": True, "structuralvariation": False},
|
|
{
|
|
"id": 11,
|
|
"genesymbol": "GENE1",
|
|
"name": "*2",
|
|
"matchesreferencesequence": False,
|
|
"structuralvariation": False,
|
|
},
|
|
{
|
|
"id": 12,
|
|
"genesymbol": "GENE1",
|
|
"name": "*3",
|
|
"matchesreferencesequence": False,
|
|
"structuralvariation": False,
|
|
},
|
|
{"id": 13, "genesymbol": "GENE1", "name": "*5", "matchesreferencesequence": False, "structuralvariation": True},
|
|
]
|
|
alv = [
|
|
{"alleledefinitionid": 10, "locationid": 1, "variantallele": "A"},
|
|
{"alleledefinitionid": 10, "locationid": 2, "variantallele": "T"},
|
|
{"alleledefinitionid": 10, "locationid": 3, "variantallele": "G"},
|
|
{"alleledefinitionid": 11, "locationid": 1, "variantallele": "G"},
|
|
{"alleledefinitionid": 11, "locationid": 3, "variantallele": "K"}, # IUPAC: not constraining
|
|
{"alleledefinitionid": 12, "locationid": 2, "variantallele": "delT"},
|
|
]
|
|
allele = [
|
|
{"genesymbol": "GENE1", "name": "*1", "clinicalfunctionalstatus": "Normal function", "activityvalue": "1.0",
|
|
"frequency": {"European": 0.7}},
|
|
{"genesymbol": "GENE1", "name": "*2", "clinicalfunctionalstatus": "No function", "activityvalue": "0.0",
|
|
"frequency": {"European": 0.2}},
|
|
{"genesymbol": "GENE1", "name": "*3", "clinicalfunctionalstatus": "Decreased function", "activityvalue": "0.5",
|
|
"frequency": None},
|
|
] # fmt: skip
|
|
return {"gene": [{"symbol": "GENE1", "chr": "chr1"}], "sequence_location": loc, "allele_definition": defs,
|
|
"allele_location_value": alv, "allele": allele,
|
|
"gene_result": [{"id": 50, "genesymbol": "GENE1", "result": "Poor Metabolizer"}],
|
|
"gene_result_lookup": [{"phenotypeid": 50, "lookupkey": {"No function": 2}, "totalactivityscore": "0.0"}]} # fmt: skip
|
|
|
|
|
|
def test_cpic_definitions() -> None: # PGX-002: variants, wildcards, undefined structural alleles
|
|
defs, sites, problems = cpic_definitions(_tables(), fetch, lambda c: "1")
|
|
assert not problems
|
|
by = {d.name: d for d in defs["GENE1"]}
|
|
assert by["*1"].reference and not by["*1"].variants
|
|
assert [sites[lid].alleles[v] for lid, v in by["*2"].variants] == [(5, "A", "G")]
|
|
assert by["*2"].wildcards == [3]
|
|
assert [sites[lid].alleles[v] for lid, v in by["*3"].variants] == [(7, "GT", "G")]
|
|
assert by["*5"].undefined and by["*5"].structural
|
|
|
|
|
|
def test_db_build_deterministic(tmp_path: Path) -> None: # IN-053, CLI-035
|
|
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")
|
|
outs = []
|
|
for k in range(2):
|
|
out = tmp_path / f"db{k}"
|
|
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
|
|
outs.append(next(out.glob("pgx_db-1-*.sqlite")))
|
|
assert outs[0].name == outs[1].name and outs[0].read_bytes() == outs[1].read_bytes()
|
|
con = sqlite3.connect(outs[0])
|
|
assert con.execute("select result_basis from genes").fetchall() == [("star_allele",)]
|
|
assert con.execute("select allele, pos, ref, alt from allele_variants order by allele").fetchall() == [
|
|
("*2", 5, "A", "G"), ("*3", 7, "GT", "G")] # fmt: skip
|
|
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"
|
|
|
|
|
|
def test_pharmvar_core_alleles() -> None: # PGX-001 PharmVar source
|
|
from simurg.pgx.db import pharmvar_core_alleles
|
|
|
|
vcf = "##fileformat=VCFv4.1\n#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\nchr19\t100\trs1\tA\tC\t.\t.\t.\n"
|
|
core = pharmvar_core_alleles("CYP2A6", {"CYP2A6-6.2/GRCh38/CYP2A6_9.vcf": vcf,
|
|
"CYP2A6-6.2/GRCh38/CYP2A6_9.001.vcf": vcf}) # fmt: skip
|
|
assert core == {"*9": [("chr19", 100, "A", "C", "rs1")]} # sub-alleles (*9.001) are not core definitions
|
|
|
|
|
|
def test_clinpgx_variant_genes() -> None: # PGX-008 variant-based genes from ClinPGx
|
|
from simurg.pgx.db import clinpgx_variant_genes
|
|
|
|
tsv = ("variant\tgene\ttype\tlevel of evidence\tchemicals\tphenotypes\n"
|
|
"rs1\tOPRM1\tEfficacy\t3\tx\t\nrs2\tFOO\tEfficacy\t3\tx\t\nrs3\tFOO\tEfficacy\t2A\tx\t\n"
|
|
"CYP2C9*3\tCYP2C9\tPK\t1A\tx\t\nrs4\tCYP2C9\tPK\t1A\tx\t\nrs5\tA,B\tPK\t1A\tx\t\n") # fmt: skip
|
|
assert clinpgx_variant_genes(tsv, {"CYP2C9"}) == {"OPRM1": {"rs1"}, "FOO": {"rs3"}}
|
|
|
|
|
|
def test_dpwg_mode(tmp_path: Path) -> None: # PGX-025
|
|
from simurg.core.log import Reporter
|
|
from simurg.pipeline.pgx_pheno import PhenoOptions, run_phenotype
|
|
|
|
src = tmp_path / "sources" / "cpic-20260101"
|
|
src.mkdir(parents=True)
|
|
t = _tables()
|
|
t["gene"][0]["symbol"] = "CYP2D6"
|
|
for tbl in ("sequence_location", "allele_definition", "allele", "gene_result"):
|
|
for r in t[tbl]:
|
|
r["genesymbol"] = "CYP2D6"
|
|
for k, v in t.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")
|
|
r = CliRunner().invoke(app, ["pgx", "db", "build", "--sources", str(tmp_path / "sources"), "--fasta", str(fa),
|
|
"--out", str(tmp_path / "db"), "--quiet"]) # fmt: skip
|
|
assert r.exit_code == 0, r.output
|
|
db = next((tmp_path / "db").glob("pgx_db-1-*.sqlite"))
|
|
calls = tmp_path / "calls"
|
|
calls.mkdir()
|
|
(calls / "pgx_calls.json").write_text(json.dumps({"samples": [{"sample_id": "s1", "genes": [
|
|
{"gene": "CYP2D6", "genotype": "*2/*2", "feasible_solutions": ["*2/*2"], "candidates": [
|
|
{"genotype": "*2/*2", "score": 1.0, "raw_score": 1.0}]},
|
|
{"gene": "CYP3A4", "genotype": "*1/*22", "feasible_solutions": ["*1/*22"], "candidates": []}]}]})) # fmt: skip
|
|
rep = Reporter(command="t", quiet=True)
|
|
assert run_phenotype(PhenoOptions(calls, db, guideline="dpwg"), rep) == 0
|
|
doc = json.loads((calls / "s1.pgx.json").read_text())
|
|
by = {g["gene"]: g for g in doc["genes"]}
|
|
assert by["CYP2D6"]["phenotype"] == "Poor Metabolizer" and "dpwg_via_cpic_mapping" in by["CYP2D6"]["flags"]
|
|
assert by["CYP3A4"]["phenotype"] is None and "dpwg_mapping_not_available" in by["CYP3A4"]["flags"]
|