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.
This commit is contained in:
kemal committed 2026-09-29 13:24:54 +01:00
1 parent c3e0681b6e
commit 0c5dd03ca8
8 files changed
+627

No files matched your search

+28
View File
@@ -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-<schema>-<sha12>.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).")],