From 82d6153a6362168d0be9104ae0ac6e5fb1b33a6e Mon Sep 17 00:00:00 2001 From: Kemal Yaylali Date: Tue, 29 Sep 2026 13:24:54 +0100 Subject: [PATCH] PGx DB: pgx db build from CPIC (PGX-001..005, IN-053, CLI-035) - scripts/pgx/fetch_sources.sh: CPIC API tables (CC0) into a dated folder with version and hashes. - simurg/pgx/db.py: CPIC HGVS locations and values -> left-normalised VCF variants (SNV, del, ins, dup, repeats, delins, chrM); IUPAC values are core-allele wildcards; alleles without definitions are flagged. - pipeline/pgx_db.py: deterministic SQLite (content-hash name), probe_map via the SNV VCF site mapper; GSA v4 PGx: 22 genes, 1354 alleles, 5002 probe links; two builds byte-identical. --- THIRD_PARTY_NOTICES.md | 1 + docs/messages.md | 1 + scripts/pgx/fetch_sources.sh | 27 ++++ src/simurg/cli/main.py | 28 +++++ src/simurg/core/errors.py | 1 + src/simurg/pgx/db.py | 223 ++++++++++++++++++++++++++++++++++ src/simurg/pipeline/pgx_db.py | 223 ++++++++++++++++++++++++++++++++++ tests/unit/test_pgx_db.py | 123 +++++++++++++++++++ 8 files changed, 627 insertions(+) create mode 100644 scripts/pgx/fetch_sources.sh create mode 100644 src/simurg/pgx/db.py create mode 100644 src/simurg/pipeline/pgx_db.py create mode 100644 tests/unit/test_pgx_db.py diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index b9647c8..4f6f11a 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -15,6 +15,7 @@ used only as external subprocess oracles and are not copied or linked. | [PyO3](https://github.com/PyO3/pyo3), [rust-numpy](https://github.com/PyO3/rust-numpy), [ndarray](https://github.com/rust-ndarray/ndarray) and their Rust dependencies | MIT / Apache-2.0 / BSD-2-Clause | Compiled into the optional `simurg_core` extension (`rust/simurg-core`); list with `cargo tree` | | [UCSC Genome Browser](https://genome.ucsc.edu/) `cytoBand` tables (hg38, hg19) and `ncbiRefSeq` (RefSeq transcripts, NCBI) | Free for all uses (UCSC data terms; RefSeq is public domain) | `simurg/cn/data/cytoBand_GRCh3{7,8}.txt.gz` and the derived centromere/acrocentric mask BEDs are shipped; `scripts/build_annotation_bundle.sh` downloads cytoBand + ncbiRefSeq for gene lists; `simurg/pgx/data/pgx_cn_regions.GRCh38.v1.bed` is derived from ncbiRefSeq exon coordinates (`scripts/pgx/build_cn_regions.py`) | | [illuminaio](https://github.com/HenrikBengtsson/illuminaio) | GPL-2 | Documentation of IDAT field codes only; oracle via subprocess | +| [CPIC](https://cpicpgx.org) database via the CPIC API (allele definitions, allele function, diplotype→phenotype tables) | CC0 1.0 | downloaded by `scripts/pgx/fetch_sources.sh` into `data/` (never committed); `simurg pgx db build` derives the local PGx DB from it; source version and file hashes are recorded in the DB `meta` table | ## BSD-2-Clause (BeadArrayFiles, verbatim from its README) diff --git a/docs/messages.md b/docs/messages.md index fcd114c..f010884 100644 --- a/docs/messages.md +++ b/docs/messages.md @@ -44,6 +44,7 @@ Generated from `src/simurg/core/errors.py` — do not edit by hand. | `E_OUTPUT_EXISTS` | error | Outputs already exist; use --force or --resume | CLI-010 | | `E_PGX_BATCH_TOO_SMALL` | error | Too few (passing) samples in the batch for PGx CN calling | PCN-004 | | `E_PGX_CN_FAILED` | error | PGx CN calling failed for the sample | PCN-002 | +| `E_PGX_SOURCE` | error | PGx source tables missing; run scripts/pgx/fetch_sources.sh | PGX-001 | | `E_PGX_TRAIN_TOO_SMALL` | error | Too few (passing) samples to train a PGx CN model | PCN-005 | | `E_QC_CONFIG_EXT` | error | QC config file must be .yaml or .yml | QC-021 | | `E_QC_CONFIG_MISSING` | error | QC config file does not exist | QC-021 | diff --git a/scripts/pgx/fetch_sources.sh b/scripts/pgx/fetch_sources.sh new file mode 100644 index 0000000..4196c2b --- /dev/null +++ b/scripts/pgx/fetch_sources.sh @@ -0,0 +1,27 @@ +#!/usr/bin/env bash +# Download the public PGx sources for `simurg pgx db build` (spec PGX-001) into a dated folder. +# bash scripts/pgx/fetch_sources.sh [OUT_DIR] (default data/pgx/sources) +# CPIC (CC0): tables from the CPIC API (https://api.cpicpgx.org, PostgREST) as JSON; the newest change-log date is +# recorded as the data version. Never committed (data/ is ignored). +set -euo pipefail +OUT=${1:-data/pgx/sources} +DATE=$(date -u +%Y%m%d) +D="$OUT/cpic-$DATE" +mkdir -p "$D" +API=https://api.cpicpgx.org/v1 +for t in gene allele allele_definition allele_location_value sequence_location gene_result gene_result_lookup \ + gene_result_diplotype population change_log; do + curl -sSf --retry 3 -H "Accept: application/json" "$API/$t" -o "$D/$t.json" + python3 -c "import json,sys; print(sys.argv[1], len(json.load(open(sys.argv[2]))))" "$t" "$D/$t.json" +done +python3 - "$D" <<'PY' +import hashlib, json, pathlib, sys +d = pathlib.Path(sys.argv[1]) +log = json.loads((d / "change_log.json").read_text()) +ver = max(e["date"] for e in log if e.get("date")) +files = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(d.glob("*.json")) if p.name != "source.json"} +(d / "source.json").write_text(json.dumps({"name": "CPIC", "version": ver, "retrieved": d.name.split("-")[1], + "url": "https://api.cpicpgx.org/v1", "license": "CC0-1.0", + "files": files}, indent=1)) +print("CPIC data version", ver) +PY diff --git a/src/simurg/cli/main.py b/src/simurg/cli/main.py index 2051591..e8bf0f3 100644 --- a/src/simurg/cli/main.py +++ b/src/simurg/cli/main.py @@ -810,6 +810,34 @@ pgx_cn_app = typer.Typer(help="Targeted PGx copy number (PCN-001).") pgx_app.add_typer(pgx_cn_app, name="copy-number") +pgx_db_app = typer.Typer(help="PGx allele-definition database (PGX-001).") +pgx_app.add_typer(pgx_db_app, name="db") + + +@pgx_db_app.command("build") +def pgx_db_build_cmd( + sources: Annotated[Path, typer.Option("--sources", help="Folder from scripts/pgx/fetch_sources.sh (required).")], + fasta: Annotated[Path, typer.Option("--fasta", help="GRCh38 FASTA for variant normalisation (required).")], + out: Annotated[Path, typer.Option("--out", help="Output folder for pgx_db--.sqlite.")] = Path("."), + manifests: Annotated[ + list[Path] | None, typer.Option("--manifests", help="BPM manifest(s) for the probe map (repeatable).") + ] = None, + csv: Annotated[list[Path] | None, typer.Option("--csv", help="CSV manifest(s), same order (indels).")] = None, + mask: Annotated[list[Path] | None, typer.Option("--mask", help="Probe mask TSV(s) (PGX-010).")] = None, + force: ForceOpt = False, + log_json: LogJson = False, + debug: Debug = False, + quiet: Quiet = False, +) -> None: + """Build the PGx DB from public sources (CLI-035, PGX-002).""" + from simurg.pipeline.pgx_db import PgxDbOptions, run_db_build + + rep = _reporter("pgx db build", out, log_json, debug, quiet) + o = PgxDbOptions(sources=sources, fasta=fasta, out=out, manifests=manifests or [], csvs=csv or [], + masks=mask or [], force=force) # fmt: skip + run_guarded(rep, lambda: run_db_build(o, rep)) + + @pgx_cn_app.command("train") def pgx_cn_train_cmd( gtcs: Annotated[Path, typer.Option("--gtcs", help="Folder searched recursively for training GTCs (required).")], diff --git a/src/simurg/core/errors.py b/src/simurg/core/errors.py index 9869a54..41ad17b 100644 --- a/src/simurg/core/errors.py +++ b/src/simurg/core/errors.py @@ -171,6 +171,7 @@ CATALOGUE: dict[str, MessageDef] = dict( _m("W_PGX_NO_ALT_CONTIG", "warning", "FASTA lacks a contig carrying a PGx region; REF written as N", "IN-062"), _m("W_PGX_BATCH_COMPOSITION", "warning", "Batch spans several lots/batches or > 14 days of scans", "PCN-006"), _m("E_PGX_CN_FAILED", "error", "PGx CN calling failed for the sample", "PCN-002"), + _m("E_PGX_SOURCE", "error", "PGx source tables missing; run scripts/pgx/fetch_sources.sh", "PGX-001"), # --- methylation control QC (MET-001 .. MET-018) ------------------------------------------------- _m("E_METH_NO_ASSAY", "error", "Probe section not found", "MET-002"), _m("E_METH_NO_CONTROLS", "error", "Missing control probes in manifest", "MET-002"), diff --git a/src/simurg/pgx/db.py b/src/simurg/pgx/db.py new file mode 100644 index 0000000..4ea4f73 --- /dev/null +++ b/src/simurg/pgx/db.py @@ -0,0 +1,223 @@ +"""PGx allele definitions from public sources → normalised VCF-style variants — spec PGX-001 … PGX-005. +Pure functions: the reference sequence comes in as a ``fetch(contig, start0, end0)`` callable. + +CPIC tables (CC0, https://api.cpicpgx.org) [DB: CPIC allele definition, allele function, diplotype→phenotype]: +``sequence_location`` gives each defining position as HGVS on the chromosome (``g.``/``m.``); ``allele_location_value`` +gives each allele's value there. Values are literal sequences (``A``, ``AA`` for a dup, ``TA(7)`` for a repeat) or +``del…``/``ins…`` forms; the reference allele definition carries the reference value. Every value becomes a +replacement of a reference window, then a left-normalised, anchored VCF (POS, REF, ALT) — the representation the +SNV VCFs use (IN-046), so definitions and array records compare directly. +""" + +from __future__ import annotations + +import re +from collections.abc import Callable +from dataclasses import dataclass, field +from typing import Any + +Fetch = Callable[[str, int, int], str] +HGVS_RE = re.compile(r"^([gm])\.(\d+)(?:_(\d+))?(.*)$") +REPEAT_RE = re.compile(r"^([ACGT]+)\((\d+)\)$") +IUPAC_WILDCARD = set("RYSWKMBDHVN") # CPIC core-allele definitions: the position does not constrain the allele + + +@dataclass(frozen=True) +class Window: + """Reference interval [start, end] (1-based, inclusive) replaced by allele values; ``insertion`` windows are the + single anchor base S of ``g.S_EinsX`` (values are anchor + inserted sequence).""" + + start: int + end: int + kind: str # snv | del | ins | dup | repeat | delins + unit: str = "" # repeat unit + + +def parse_location(hgvs: str) -> Window | None: + """First HGVS expression of a CPIC ``chromosomelocation`` (several are ``;``-joined for one site).""" + first = hgvs.split(";")[0].strip() + m = HGVS_RE.match(first) + if not m: + return None + s = int(m.group(2)) + e = int(m.group(3)) if m.group(3) else s + rest = m.group(4) + if re.fullmatch(r"[ACGT]>[ACGT]", rest): + return Window(s, s, "snv") + if rest.startswith("delins"): + return Window(s, e, "delins") + if rest.startswith("del"): + return Window(s, e, "del") + if rest.startswith("ins"): + return Window(s, s, "ins") + if rest.startswith("dup"): + return Window(s, e, "dup") + rm = re.fullmatch(r"([ACGT]+)\[(\d+)\]", rest) + if rm: + return Window(s, s, "repeat", rm.group(1)) + return None + + +def value_sequence(value: str, w: Window, fetch: Fetch, contig: str) -> tuple[int, int, str] | None: + """(start, end, sequence) that the value puts in place of the reference window; None = unusable/ambiguous.""" + v = value.strip() + if " or " in v or ";" in v or not v: + return None + if w.kind == "repeat": + m = REPEAT_RE.match(v) + unit = w.unit + n_ref = 0 + while fetch(contig, w.start - 1 + n_ref * len(unit), w.start - 1 + (n_ref + 1) * len(unit)) == unit: + n_ref += 1 + end = w.start - 1 + n_ref * len(unit) + if m: + return w.start, end, m.group(1) * int(m.group(2)) + if re.fullmatch(r"[ACGT]+", v): + return w.start, end, v + return None + if w.kind == "ins": + anchor = fetch(contig, w.start - 1, w.start) + if v == "del": + return w.start, w.start, anchor + if v.startswith("ins"): + return w.start, w.start, anchor + v[3:] + return None + if w.kind == "delins": + if v.startswith("del"): + return w.start, w.end, fetch(contig, w.start - 1, w.end) + if v.startswith("ins"): + return w.start, w.end, v[3:] + return None + if v.startswith("del"): + return w.start, w.end, "" + if re.fullmatch(r"[ACGT]+", v): + return w.start, w.end, v + return None + + +def normalize(fetch: Fetch, contig: str, start: int, end: int, alt: str) -> tuple[int, str, str]: + """Replacement of reference [start, end] (1-based; end = start − 1 for a pure insertion before start) by ``alt`` + → left-normalised, anchored (POS, REF, ALT) [LIT: Tan et al. 2015 Bioinformatics 31:2202 (vt normalize)].""" + ref = fetch(contig, start - 1, end) + pos = start + # trim the common suffix, then the common prefix + while ref and alt and ref[-1] == alt[-1]: + ref, alt = ref[:-1], alt[:-1] + while len(ref) > 0 and len(alt) > 0 and ref[0] == alt[0]: + ref, alt = ref[1:], alt[1:] + pos += 1 + if ref and alt and len(ref) == len(alt): + return pos, ref, alt # SNV / MNV + # left shift the indel while the base before it equals its last base + while pos > 1 and (ref or alt): + prev = fetch(contig, pos - 2, pos - 1) + body = ref or alt + if prev != body[-1] or (ref and alt): + break + if ref: + ref = prev + ref[:-1] + else: + alt = prev + alt[:-1] + pos -= 1 + anchor = fetch(contig, pos - 2, pos - 1) + return pos - 1, anchor + ref, anchor + alt + + +@dataclass +class VariantSite: + location_id: int + gene: str + chrom: str # contig name as in the FASTA + hgvs: str + rsid: str | None + ref_value: str | None + alleles: dict[str, tuple[int, str, str]] = field(default_factory=dict) # value → normalised (pos, ref, alt) + + +@dataclass +class AlleleDef: + gene: str + name: str + function: str | None + activity: str | None + reference: bool + structural: bool + variants: list[tuple[int, str]] = field(default_factory=list) # (location_id, value) where ≠ reference + frequencies: dict[str, float] = field(default_factory=dict) + wildcards: list[int] = field(default_factory=list) # location ids with an IUPAC value (either base) + + @property + def undefined(self) -> bool: + """A non-reference allele without any defining variant cannot be called (e.g. CPIC NUDT15 *2).""" + return not self.reference and not self.variants + + +def cpic_definitions( + t: dict[str, list[dict[str, Any]]], fetch: Fetch, contig_of: Callable[[str], str | None] +) -> tuple[dict[str, list[AlleleDef]], dict[int, VariantSite], list[str]]: + """CPIC tables → (gene → allele definitions, location id → site, problems).""" + problems: list[str] = [] + gene_chr = {g["symbol"]: g.get("chr") for g in t["gene"]} + locs = {loc["id"]: loc for loc in t["sequence_location"]} + values: dict[int, list[tuple[int, str]]] = {} + for v in t["allele_location_value"]: + values.setdefault(v["alleledefinitionid"], []).append((v["locationid"], v["variantallele"])) + func = {(a["genesymbol"], a["name"]): a for a in t["allele"]} + defs: dict[str, list[AlleleDef]] = {} + sites: dict[int, VariantSite] = {} + ref_value: dict[int, str] = {} + for d in sorted(t["allele_definition"], key=lambda d: (d["genesymbol"], d["name"])): + if d.get("matchesreferencesequence"): + for lid, val in values.get(d["id"], []): + ref_value[lid] = val + for d in sorted(t["allele_definition"], key=lambda d: (d["genesymbol"], d["name"])): + g = d["genesymbol"] + a = func.get((g, d["name"]), {}) + freq = {k: float(v) for k, v in (a.get("frequency") or {}).items() if isinstance(v, int | float)} + ref_def, sv = bool(d.get("matchesreferencesequence")), bool(d.get("structuralvariation")) + ad = AlleleDef(g, d["name"], a.get("clinicalfunctionalstatus"), a.get("activityvalue"), ref_def, sv, [], freq) + for lid, val in sorted(values.get(d["id"], [])): + loc = locs.get(lid) + if loc is None: + continue + if lid not in sites: + hg = loc.get("chromosomelocation") or "" + chrom = "chrM" if hg.startswith("m.") else (gene_chr.get(g) or "") + contig = contig_of(chrom) if chrom else None + if contig is None: + problems.append(f"{g} {loc['name']}: no contig for {chrom!r}") + continue + sites[lid] = VariantSite(lid, g, contig, hg, loc.get("dbsnpid"), ref_value.get(lid)) + site = sites[lid] + if len(val) == 1 and val in IUPAC_WILDCARD: + ad.wildcards.append(lid) + elif val != site.ref_value: + ad.variants.append((lid, val)) + defs.setdefault(g, []).append(ad) + for lid, site in sites.items(): + w = parse_location(site.hgvs) + if w is None: + problems.append(f"{site.gene} {site.hgvs}: unparsed location") + continue + vals = {val for d in defs.get(site.gene, []) for (l2, val) in d.variants if l2 == lid} + for val in sorted(vals): + seq = value_sequence(val, w, fetch, site.chrom) + if seq is None: + problems.append(f"{site.gene} {site.hgvs}: value {val!r} not representable") + continue + s, e, alt = seq + if w.kind == "ins": + s, e, alt = s + 1, s, alt[1:] + site.alleles[val] = normalize(fetch, site.chrom, s, e, alt) + return defs, sites, problems + + +def result_basis(gene: str, names: list[str]) -> str: + """PGX-002 ``genes.result_basis``.""" + if gene in ("HLA-A", "HLA-B"): + return "tagging_variant" + if gene in ("G6PD", "APOE"): + return "haplotype_named" + if names and all(n.startswith("*") for n in names): + return "star_allele" + return "variant_based" diff --git a/src/simurg/pipeline/pgx_db.py b/src/simurg/pipeline/pgx_db.py new file mode 100644 index 0000000..f79b332 --- /dev/null +++ b/src/simurg/pipeline/pgx_db.py @@ -0,0 +1,223 @@ +"""``simurg pgx db build`` — the PGx allele-definition database (spec PGX-001 … PGX-005, PGX-010, IN-053, CLI-035). + +I/O and orchestration; the parsing and normalisation live in ``simurg.pgx.db``. Sources are the dated folders written +by ``scripts/pgx/fetch_sources.sh`` (``cpic-YYYYMMDD/`` with ``source.json``). The build is deterministic: rows are +inserted in sorted order into a fresh SQLite file with fixed pragmas, and the file is named by its content hash. +""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from simurg import __version__ +from simurg.core.errors import ExitCode, SimurgError +from simurg.core.log import Reporter + +DB_SCHEMA = 1 +CPIC_TABLES = ("gene", "allele", "allele_definition", "allele_location_value", "sequence_location", "gene_result", + "gene_result_lookup") # fmt: skip +GENE_PAD = 50_000 # probes within this distance of a gene's defining positions are resolved for probe_map + +SCHEMA_SQL = """ +CREATE TABLE meta(key TEXT PRIMARY KEY, value TEXT NOT NULL); +CREATE TABLE genes(gene TEXT PRIMARY KEY, result_basis TEXT NOT NULL, chrom TEXT); +CREATE TABLE alleles(gene TEXT, allele TEXT, kind TEXT, function TEXT, activity TEXT, structural INTEGER, + defined INTEGER, PRIMARY KEY(gene, allele)); +CREATE TABLE allele_variants(gene TEXT, allele TEXT, build TEXT, chrom TEXT, pos INTEGER, ref TEXT, alt TEXT, + hgvs TEXT, rsid TEXT, location_id INTEGER, value TEXT); +CREATE TABLE allele_wildcards(gene TEXT, allele TEXT, location_id INTEGER); +CREATE TABLE sites(gene TEXT, location_id INTEGER, chrom TEXT, hgvs TEXT, rsid TEXT, ref_value TEXT, + PRIMARY KEY(gene, location_id)); +CREATE TABLE site_values(gene TEXT, location_id INTEGER, value TEXT, pos INTEGER, ref TEXT, alt TEXT); +CREATE TABLE probe_map(manifest TEXT, probe_name TEXT, chrom TEXT, pos INTEGER, ref TEXT, alt TEXT); +CREATE TABLE masks(manifest TEXT, probe_name TEXT, reason TEXT); +CREATE TABLE limitations(manifest TEXT, gene TEXT, allele TEXT, status TEXT, reported_as TEXT); +CREATE TABLE frequencies(gene TEXT, allele TEXT, population TEXT, freq REAL); +CREATE TABLE phenotypes(guideline TEXT, gene TEXT, lookup_key TEXT, activity_score TEXT, phenotype TEXT); +CREATE INDEX av_pos ON allele_variants(chrom, pos); +CREATE INDEX pm_pos ON probe_map(manifest, chrom, pos); +""" + + +@dataclass +class PgxDbOptions: + sources: Path + fasta: Path + out: Path + manifests: list[Path] = field(default_factory=list) + csvs: list[Path] = field(default_factory=list) + masks: list[Path] = field(default_factory=list) + force: bool = False + + +def _latest(sources: Path, prefix: str) -> Path | None: + cands = sorted(p for p in sources.glob(f"{prefix}-*") if (p / "source.json").exists()) + return cands[-1] if cands else None + + +def _load_cpic(d: Path) -> dict[str, list[dict[str, Any]]]: + t = {} + for name in CPIC_TABLES: + p = d / f"{name}.json" + if not p.exists(): + raise SimurgError("E_PGX_SOURCE", f"{p} missing", hint="run scripts/pgx/fetch_sources.sh") + t[name] = json.loads(p.read_text(encoding="utf-8")) + return t + + +def _probe_sites( + bpm_path: Path, + csv_path: Path | None, + genome: Any, + cmap: Any, + windows: dict[str, list[tuple[int, int]]], + rep: Reporter, +) -> list[Any]: + """Plus-strand, normalised sites of the manifest probes inside the gene windows (the SNV VCF representation).""" + from simurg.formats.bpm import read_bpm + from simurg.harmonize.sites import ManifestView, map_locus + + bpm = read_bpm(bpm_path) + csvm = None + if csv_path is not None: + from simurg.formats.csv_manifest import read_csv_manifest + + csvm = read_csv_manifest(csv_path) + elif any("D" in s for s in bpm.snps): + rep.warn("W_SKIP_INDEL_CONTEXT", f"{bpm_path.name}: no CSV manifest, indel probes are not mapped") + src = csvm or bpm + mv = ManifestView(bpm.names, bpm.snps, bpm.chroms, bpm.map_infos.tolist(), src.ilmn_strands, src.source_strands, + src.ref_strands if any(src.ref_strands) else bpm.ref_strands, + csvm.source_seqs if csvm else [""] * bpm.num_loci, bpm.assay_types.tolist(), + bpm.intensity_only.tolist()) # fmt: skip + out = [] + for i in range(bpm.num_loci): + c = cmap.get(bpm.chroms[i]) + if c is None or c not in windows: + continue + p = int(bpm.map_infos[i]) + if not any(a - GENE_PAD <= p <= b + GENE_PAD for a, b in windows[c]): + continue + site, _ = map_locus(i, mv, genome.fetch, cmap.get, genome.build or "GRCh38") + if site is not None: + out.append(site) + return out + + +def run_db_build(o: PgxDbOptions, rep: Reporter) -> int: + """``simurg pgx db build`` (CLI-035).""" + from simurg.formats.bpm import read_bpm + from simurg.formats.fasta import ContigMap, Genome + from simurg.pgx.db import cpic_definitions, result_basis + + cdir = _latest(o.sources, "cpic") + if cdir is None: + raise SimurgError("E_PGX_SOURCE", f"no cpic-YYYYMMDD source folder in {o.sources}", + hint="run scripts/pgx/fetch_sources.sh") # fmt: skip + csrc = json.loads((cdir / "source.json").read_text()) + genome = Genome(o.fasta) + if genome.build not in ("GRCh38", "unknown", ""): + raise SimurgError("E_BUILD_MISMATCH", f"the PGx DB is GRCh38 only (FASTA is {genome.build})") # PCN-015 + cmap = ContigMap([e.name for e in genome.entries], "auto") + t = _load_cpic(cdir) + defs, sites, problems = cpic_definitions(t, genome.fetch, cmap.get) + for p in problems: + rep.info(f"definition note: {p}") + rows: dict[str, list[tuple[Any, ...]]] = {k: [] for k in ("genes", "alleles", "allele_variants", "allele_wildcards", + "sites", "site_values", "frequencies", "phenotypes", + "probe_map", "masks")} # fmt: skip + gene_chr = {g["symbol"]: g.get("chr") for g in t["gene"]} + for gene, ds in sorted(defs.items()): + rows["genes"].append((gene, result_basis(gene, [d.name for d in ds]), gene_chr.get(gene))) + for d in ds: + kind = "reference" if d.reference else ("star_allele" if d.name.startswith("*") else "named_haplotype") + rows["alleles"].append( + (gene, d.name, kind, d.function, d.activity, int(d.structural), int(not d.undefined)) + ) + for lid, val in d.variants: + s = sites.get(lid) + if s is None or val not in s.alleles: + continue + pos, ref, alt = s.alleles[val] + rows["allele_variants"].append( + (gene, d.name, "GRCh38", s.chrom, pos, ref, alt, s.hgvs, s.rsid, lid, val) + ) + for lid in d.wildcards: + rows["allele_wildcards"].append((gene, d.name, lid)) + for pop, f in sorted(d.frequencies.items()): + rows["frequencies"].append((gene, d.name, pop, f)) + for lid, s in sorted(sites.items()): + rows["sites"].append((s.gene, lid, s.chrom, s.hgvs, s.rsid, s.ref_value)) + for val, (pos, ref, alt) in sorted(s.alleles.items()): + rows["site_values"].append((s.gene, lid, val, pos, ref, alt)) + results = {r["id"]: r for r in t["gene_result"]} + for lk in t["gene_result_lookup"]: + res = results.get(lk["phenotypeid"]) + if res is None: + continue + key = json.dumps(lk.get("lookupkey"), sort_keys=True, ensure_ascii=False) + rows["phenotypes"].append(("cpic", res["genesymbol"], key, lk.get("totalactivityscore"), res["result"])) + # probe map (PGX-003): manifest probes whose normalised site carries a DB variant + windows: dict[str, list[tuple[int, int]]] = {} + for av in rows["allele_variants"]: + windows.setdefault(av[3], []).append((av[4], av[4] + len(av[5]))) + var_keys = {(av[3], av[4], av[5], av[6]) for av in rows["allele_variants"]} + manifests = [] + csvs = dict(zip(o.manifests, o.csvs, strict=False)) + for bpm_path in o.manifests: + name = read_bpm(bpm_path).manifest_name + manifests.append(name) + n = 0 + for ps in _probe_sites(bpm_path, csvs.get(bpm_path), genome, cmap, windows, rep): + for alt in ps.alleles: + if alt != ps.ref and (ps.contig, ps.pos, ps.ref, alt) in var_keys: + rows["probe_map"].append((name, ps.name, ps.contig, ps.pos, ps.ref, alt)) + n += 1 + rep.info(f"{name}: {n} probe-to-variant link(s)") + for mp in o.masks: + for ln in mp.read_text(encoding="utf-8").splitlines(): + if ln.strip() and not ln.startswith("#"): + cols = ln.split("\t") + rows["masks"].append((cols[2] if len(cols) > 2 else mp.stem, cols[0], cols[1] if len(cols) > 1 else "")) + meta = {"schema": "simurg.pgx_db", "schema_version": str(DB_SCHEMA), "builder": f"SimurgArray {__version__}", + "genome_build": "GRCh38", "fasta_sha256": genome.sha256, "sqlite_version": sqlite3.sqlite_version, + "allele_sources": json.dumps([{"name": "CPIC", "version": csrc["version"], "retrieved": csrc["retrieved"], + "license": csrc.get("license")}]), + "phenotype_sources": json.dumps([{"name": "CPIC", "version": csrc["version"]}]), + "cpic_files_sha256": json.dumps(csrc.get("files", {}), sort_keys=True), + "manifests": json.dumps(sorted(manifests)), "definition_notes": json.dumps(problems)} # fmt: skip + # content hash over the sorted rows and meta (IN-053: the name and meta carry it; a file cannot hold its own hash) + for rs in rows.values(): + rs.sort(key=lambda r: tuple(str(x) for x in r)) + sha = hashlib.sha256(json.dumps({"rows": rows, "meta": sorted(meta.items())}, sort_keys=True, + default=str).encode()).hexdigest() # fmt: skip + meta["content_sha256"] = sha + o.out.mkdir(parents=True, exist_ok=True) + tmp = o.out / ".pgx_db.part.sqlite" + tmp.unlink(missing_ok=True) + con = sqlite3.connect(tmp) + con.execute("PRAGMA page_size=4096") + con.execute("PRAGMA journal_mode=OFF") + con.executescript(SCHEMA_SQL) + for table, rs in rows.items(): + if rs: + ph = ",".join("?" * len(rs[0])) + con.executemany(f"INSERT INTO {table} VALUES ({ph})", rs) + con.executemany("INSERT INTO meta VALUES (?, ?)", sorted(meta.items())) + con.commit() + con.execute("VACUUM") + con.close() + final = o.out / f"pgx_db-{DB_SCHEMA}-{sha[:12]}.sqlite" + if final.exists() and not o.force: + tmp.unlink() + rep.info(f"identical PGx DB already present: {final}") + return ExitCode.OK + tmp.replace(final) + rep.info(f"PGx DB: {len(defs)} genes, {len(rows['alleles'])} alleles, {len(rows['allele_variants'])} variant " + f"rows, {len(rows['probe_map'])} probe links → {final} (content sha256 {sha})") # fmt: skip + return ExitCode.OK diff --git a/tests/unit/test_pgx_db.py b/tests/unit/test_pgx_db.py new file mode 100644 index 0000000..e72e533 --- /dev/null +++ b/tests/unit/test_pgx_db.py @@ -0,0 +1,123 @@ +"""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",)