def draft_gene(
spec_dir: Path,
gene: str,
*,
drugs: Sequence[str] = (),
alleles: Sequence[str] = (),
population: str | None = None,
declared_use: str = "unstated",
dry_run: bool = False,
offline: bool = False,
cpic_cache: Path | None = None,
client: CpicClient | CpicSnapshotClient | None = None,
) -> PgxDraftResult:
"""Draft one gene's `haplotypes.csv`, `allele_function.csv` and `diplotypes.csv` rows.
Re-runnable and additive: call it once per gene, in any order, as a module grows. Rows already in
the files are left exactly as they are.
`alleles` restricts the draft to a set of star alleles (RM34) and is applied to **all three** tables:
the defining variants of those alleles, their function rows, and only the diplotypes whose *both*
halves are in the set. Filtering one table and not the others would leave a module naming alleles it
never defines, which is precisely what `_cross_validate_haplotype_definitions` warns about. `*1` is
always kept; empty means everything CPIC publishes, exactly as before.
`offline` (0.5.1, RM38) is the flag this provider simply did not have — every sibling pass took one,
so a caller running the family under one switch had to know out of band that drafting ignored it,
and forgetting meant silent egress from a run documented as making none. A built CPIC snapshot
serves the draft; with none and `offline` set, nothing is drafted and the reason is returned.
"""
declared_use, declared_from = effective_declared_use(spec_dir, CPIC_TERMS, declared_use) # S105
skip_reason = check_declared_use(CPIC_TERMS, declared_use)
if skip_reason:
# Acquisition-time refusal: nothing is fetched, because the terms are accepted by taking it.
return PgxDraftResult(warnings=[skip_reason], skipped=True)
# **Asked before the client, not beside the rows it warns about** (R2-2). `source_build_mismatch`
# raises `EnrichmentError` on a present-but-unreadable `module_spec.yaml`, which is right — a
# module whose declaration cannot be read has no build to draft against — but asking it *after*
# the CPIC queries meant the failure landed once the work was already paid for. It reads a file
# beside the spec and costs nothing, so it belongs with the other precondition above. The warning
# itself is still appended in its old place, so the reported order does not move.
build_warning = source_build_mismatch(spec_dir, "CPIC", CPIC_GENOME_BUILD)
owned = client is None
if client is not None:
cpic = client
else:
reference = resolve_cpic_reference(cpic_cache)
if reference is not None:
cpic = CpicSnapshotClient(reference)
elif offline:
return PgxDraftResult(
warnings=[
"CPIC draft skipped: --offline and no built snapshot. Build one with "
"`just-dna-enricher cpic build --out <dir>`, or point at it with "
"$JUST_DNA_CPIC_CACHE."
],
skipped=True,
)
else:
cpic = CpicClient()
try:
# Read inside the `try` for the reason `knows_drug` is asked here: the client is closed in the
# `finally` below. `None` on the live client, which carries no release to name.
cpic_dataset = getattr(cpic, "dataset", None)
published = cpic.alleles_for_gene(gene)
diplotypes = cpic.diplotypes_for_gene(gene)
defining, defining_warnings = cpic.defining_variants(gene)
by_drug = {drug: cpic.recommendations(gene, drug) for drug in drugs}
# Asked here, inside the `try`, because the client is closed in the `finally` below — and
# asked only for the drugs that came back empty, since that is the only case whose message
# needs it. See `CpicClient.knows_drug`.
#
# **Its failure is caught, and that is R2-4.** By this line every substantive query has
# already returned: the alleles, the diplotypes, the defining variants and the
# recommendations are all in hand, and a complete draft exists. `knows_drug` exists only to
# sharpen the *sentence* explaining an empty result, so letting a transport failure here
# propagate would discard finished work to improve a message about it — the gnomAD rule
# (a per-item error must never sink a batch) arriving in a different tier.
#
# The tri-state it is typed with was designed and, from the live client, never delivered:
# `CpicClient.knows_drug` could only return `True`/`False` or raise, so the `known is None`
# branch below was reachable from the snapshot client alone. R2-13 is what makes this
# catchable at all — before it, the escape was a raw `httpx.HTTPStatusError`.
drug_known: dict[str, bool | None] = {}
partners: dict[str, list[str]] = {}
unaskable: dict[str, str] = {}
for drug, found in by_drug.items():
if found:
continue
try:
# Partners first (S102): a drug keyed on a gene *pair* has rows in the very table
# `recommendations` read, so "no row for it" was false and "no single-gene
# recommendation" was true but unexplained. A non-empty answer settles `knows_drug`
# too — the source evidently lists the drug — so that request is not paid.
partners[drug] = cpic.partner_genes(gene, drug)
drug_known[drug] = True if partners[drug] else cpic.knows_drug(drug)
except CpicError as exc:
drug_known[drug] = None
unaskable[drug] = str(exc)
finally:
if owned:
cpic.close()
warnings = list(defining_warnings)
if declared_from is not None:
warnings.append(
f"{gene}: CPIC use {declared_use!r} read from {declared_from}, recorded by an earlier run; "
f"pass --use to declare otherwise."
)
# CPIC's `sequence_location` is GRCh38 and `haplotypes.csv` gets a `chrom`/`start` from it, so a
# module declaring another build is about to record a position from the wrong assembly. Computed
# at the top of the function; reported here.
if build_warning:
warnings.append(f"{gene}: {build_warning}")
selected = _selected_alleles(alleles, [a.allele for a in published], gene)
if selected is not None:
# Filtered once, here, so the three tables and the drug rows all see the same set — the drug rows
# are joined onto these same diplotypes below.
kept = [d for d in diplotypes if _pair_in(d.diplotype, selected)]
# Counted over the **parsable** pairs only, which is what the filter actually decided. Counting
# every row instead read "567 of 16836 drafted" for a set of six alleles, because the 546
# copy-number rows `_pair_in` deliberately leaves alone were tallied as kept and then skipped by
# the notation rule below — two different findings, and the reader could not see either.
parsable = [d for d in diplotypes if _split_diplotype(d.diplotype) is not None]
kept_parsable = [d for d in kept if _split_diplotype(d.diplotype) is not None]
warnings.append(
f"{gene}: --allele kept {len(selected)} allele(s) {sorted(selected)} (`*1` is always kept, "
f"since it is defined by carrying no variants) — {len(kept_parsable)} of {len(parsable)} "
f"diplotype(s) drafted; the rest name an allele outside the set."
)
diplotypes = kept
published = [a for a in published if a.allele in selected]
defining = [v for v in defining if v.allele in selected]
haplotypes, haplotype_warnings = _haplotype_rows(defining)
warnings.extend(haplotype_warnings)
function_rows: list[AlleleFunctionRow] = []
for allele in published:
if not STAR_ALLELE_PATTERN.match(allele.allele or ""):
warnings.append(f"{gene}: allele {allele.allele!r} is not a star-allele string — skipped.")
continue
function_rows.append(
AlleleFunctionRow(
gene=allele.gene,
allele=allele.allele,
activity_value=allele.activity_value,
function_status=allele.function_status,
)
)
diplotype_rows: list[DiplotypeRow] = []
# Aggregated, not one line per row: CYP2C19 alone produced ~600 identical warnings, which buries
# every other finding in the run. A cap with the total stated beats a wall with nothing stated.
unscorable: dict[str, list[str]] = {}
unparsable: list[str] = []
for entry in diplotypes:
pair = _split_diplotype(entry.diplotype)
if pair is None:
# Aggregated for the same reason the activity scores four lines below are, which this
# loop had not learned: a real CYP2D6 draft skips 546 diplotypes whose copy-number
# notation carries `≥` (`*4x≥3/*95`), and 546 identical-shaped lines bury every other
# finding — including the three allele skips and the licence row. One line, examples,
# and the count, so nothing is silently dropped.
unparsable.append(entry.diplotype)
continue
if entry.phenotype is None:
continue # nothing to conclude; a row whose only content is the pair says nothing
if entry.activity_score and not _is_numeric(entry.activity_score):
# Two different things, and they were reported as one. `"≥3.0"` is a *bound*: real
# information the numeric bin columns cannot hold. `"n/a"` is CPIC saying it did not
# score this diplotype — an absence, which is an empty cell here and not a finding at
# all. Calling that "an inequality rather than a number" was simply wrong.
bucket = "unscored" if entry.activity_score.strip().lower() in _NOT_SCORED else "bounded"
unscorable.setdefault(bucket, []).append(f"{entry.diplotype}={entry.activity_score}")
diplotype_rows.append(
DiplotypeRow(
gene=entry.gene,
haplotype_a=pair[0],
haplotype_b=pair[1],
phenotype=entry.phenotype,
conclusion=f"{entry.gene} {entry.diplotype}: {entry.phenotype}",
)
)
# Two reasons a pair does not parse, and one sentence described both as copy number (S106):
# DPYD's 3,570 pairs are `c.1003G>T (*11)/Reference` — HGVS-named alleles, not star strings at all
# — and were reported as CYP2D6's `*4x≥3` shape. Bucketed by which it is, since the fix differs.
for bucket, entries in sorted(_unparsable_by_reason(unparsable).items()):
shown = ", ".join(entries[:3])
rest = f" (+{len(entries) - 3} more)" if len(entries) > 3 else ""
warnings.append(
f"{gene}: {len(entries)} diplotype(s) "
+ (
"carry copy-number notation and were skipped — CPIC writes `x≥3`, and `≥` is not a "
"star-string character"
if bucket == "copy_number"
else "name alleles that are not star strings and were skipped — CPIC names this gene's "
"alleles as HGVS variants (`c.1003G>T (*11)`), which the haplotype-name rule refuses "
"for their whitespace and this drafter does not yet translate (RM253)"
)
+ f". e.g. {shown}{rest}."
)
for bucket, entries in sorted(unscorable.items()):
shown = ", ".join(entries[:3])
rest = f" (+{len(entries) - 3} more)" if len(entries) > 3 else ""
warnings.append(
f"{gene}: {len(entries)} diplotype(s) carry no numeric activity score — "
+ (
"CPIC records them as not scored, so the cell is simply empty"
if bucket == "unscored"
else "CPIC gives a bound rather than a value; add a bin by hand if you need one"
)
+ f". e.g. {shown}{rest}."
)
# Drug rows are *added to* the phenotype table, not a replacement for it: the plain rows answer
# "what phenotype is this pair", the drug rows answer "and what does CPIC say to do about it for
# this drug". Different questions, and `_TABLE_DUPE_KEYS` keys on `drug`, so both coexist.
for drug, recommendations in by_drug.items():
if not recommendations:
# Three outcomes, not one sentence. "CPIC has no recommendations for X" was emitted
# identically for a typo and for a real drug CPIC scores in a shape this table does not
# hold — warfarin being exactly that: the guideline exists, it is a dosing algorithm over
# several genes rather than a per-phenotype recommendation, so nothing lands here and the
# author was told CPIC has nothing. Same distinction the rsID vocabulary makes between a
# mistyped id and a real one the source records differently.
known = drug_known.get(drug)
if partners.get(drug):
# The arm S102 found missing, and it was unreachable on the snapshot path: `known`
# is `None` there, so the "no row for it" sentence below was reached for a drug with
# 35 rows in that very table, all keyed on TPMT *and* NUDT15. Named here, before the
# other readings, because it is the one positive fact the source has for the drug.
detail = (
f"CPIC keys every recommendation for it that names {gene} on more than one gene "
f"({gene} together with {', '.join(partners[drug])}). A row about the pair is not "
f"a statement about {gene} alone, and no authored table is keyed on two genes yet, "
f"so nothing lands in diplotypes.csv; pairing across genes is the open RM28"
)
elif known is False:
detail = (
"CPIC does not list that drug at all — check the spelling (CPIC uses lowercase "
"generic names)"
)
elif drug in unaskable:
# A third reading of "could not establish", kept apart from the snapshot one below
# because the remedies differ: this one is re-runnable and that one is not. Folding
# them would put the snapshot's sentence in front of an author who has no snapshot.
detail = (
f"CPIC could not be asked whether it lists that drug ({unaskable[drug]}), so "
"whether the name is a typo or a real drug with no phenotype-keyed "
"recommendation is undetermined. The rest of the draft is unaffected and was "
"written; re-run to settle this line"
)
elif known is None:
# Deliberately does NOT say "re-run without --offline": the route is snapshot-first,
# so a provisioned cache is used whether or not --offline was passed, and advising a
# flag that changes nothing is the same defect as the joinability warning telling a
# GRCh37 author to re-run enrich.
detail = (
"the CPIC snapshot's recommendation table has no row for it. That table only "
"names drugs that already have a phenotype-keyed recommendation, so this does "
"not establish whether CPIC knows the drug at all. Only the live API can answer "
"that, and it is consulted only when no snapshot is present"
)
else:
# Reached with `partners` empty, so this is a drug CPIC lists and publishes no
# recommendation row for at all — warfarin, whose guideline is a dosing algorithm
# rather than a recommendation table (zero rows in the snapshot, measured for S102).
# A drug keyed on a gene pair is the arm above, not this one.
detail = (
"CPIC lists the drug but publishes no phenotype-keyed recommendation row for it "
"at all — a guideline shaped as a dosing algorithm (warfarin) has no row to draft"
)
warnings.append(f"{gene}: nothing drafted for {drug!r} — {detail}.")
continue
drug_rows, drug_warnings = _recommendation_rows(diplotypes, recommendations, population=population)
warnings.extend(drug_warnings)
diplotype_rows.extend(drug_rows)
# The licence row lands inside each table's own commit (RM232). Built before the first append,
# because the closure has to exist to be handed to it; `record_draft_provenance` below calls the
# same body for the run that covered something and appended nothing.
commit_licence = licence_commit(
sources=[CPIC_TERMS.source],
spec_dir=spec_dir,
dataset=cpic_dataset,
declared_use=declared_use,
error=CpicError,
)
reports = [
append_rows(spec_dir, csv_name, rows, dry_run=dry_run, before_commit=commit_licence)
for csv_name, rows in (
("haplotypes.csv", haplotypes),
("allele_function.csv", function_rows),
("diplotypes.csv", diplotype_rows),
)
if rows
]
if not dry_run and reports:
# A pass that consults a source must WRITE its `SourceRow`. This one did not, and of the three
# providers it is the one where that matters most: CPIC is CC BY-SA **with a no-sale clause**,
# so a module drafted from it and carrying no `sources.csv` leaves the compile gate nothing to
# refuse on — the restriction simply disappears. Building the row and not writing it is the
# exact `clingen.py` bug, and it was sitting in the newest provider.
#
# **`dataset` was the missing half and it is what makes the tautology visible here (RM73).**
# This provider copies `function_status` straight out of CPIC and `pgx._function_conflicts`
# then compares that very column against CPIC — the RM4 shape exactly, and it went unmarked
# for two releases because this was the one provider recording no release at all. `None` on
# the live client, deliberately: with no label there is nothing to establish a copy against,
# so the check simply runs, which is the conservative direction.
# **`withdraw_stale_dataset` is new here (RM228).** `dataset` was recorded and never
# withdrawn, so a module widened from a newer CPIC release kept a licence row naming the
# older one — `merge_sources_file` is never-clobber, which protects a curator's terms and
# turns the release label into a false claim. The digest restamp is driven by this provider's
# `kind` now rather than by remembering the call.
warnings.extend(
record_draft_provenance(
provider=_PROVIDER,
sources=[CPIC_TERMS.source],
spec_dir=spec_dir,
dataset=cpic_dataset,
covered=True,
drafted=any(report.added for report in reports),
declared_use=declared_use,
error=CpicError,
)
)
return PgxDraftResult(reports=reports, warnings=warnings)