def enrich_clinical_assertions(
spec_dir: Path,
*,
mode: str = "best_effort",
offline: bool = False,
clinvar_cache: Path | None = None,
download: bool = True,
write: bool = True,
) -> ClinicalAssertionResult:
"""Fill `clinical_assertions.csv` from the coordinates already in `resolution.csv`.
Existing rows are authoritative and merged, never clobbered — the standing rule, with its standing
consequence: to regenerate after a machinery change, delete the file first. The merge key is
`(variant_key, variation_id)`, which is the archive's grain: ClinVar genuinely holds several
records for one allele under different conditions, and collapsing them would pick a condition on
the author's behalf.
A resolved allele the archive has no record for gets a `not_found` row, unlike the gene-validity
pass's silence. The two differ because the sources differ: ClinVar covers the genome, so "asked
and absent" is a fact about the allele, while a gene-curation body's silence only means nobody has
assessed the gene yet.
`download` provisions the published snapshot when no local one is found, exactly as `enrich()`
does — `--offline` is the switch, and there is deliberately no second flag. With no snapshot
reachable the pass is a no-op with a warning (`skipped_no_snapshot`), never a failure.
"""
spec_dir = Path(spec_dir)
resolution_path = sidecar_path(spec_dir, "resolution.csv", error=ClinicalAssertionError)
assertions_path = sidecar_path(spec_dir, "clinical_assertions.csv", error=ClinicalAssertionError)
if write:
# Fail on a placeholder or half-edited licence table now, before the fetch (S98, RM231).
require_sources_file(spec_dir, error=ClinicalAssertionError)
if not resolution_path.exists():
raise ClinicalAssertionError(
f"no resolution.csv in {spec_dir} — the clinical-assertion pass reads resolved "
f"coordinates, so run `just-dna-enricher enrich` first."
)
resolution_rows, errors, _ = load_csv_rows(resolution_path, ResolutionRow, resolution_path.name)
if errors:
raise ClinicalAssertionError(f"{resolution_path.name} is invalid: {errors[0]}")
# **The overlay, before anything reads a coordinate** (RM136). An INPUT read: this pass consumes
# `resolution.csv` and writes a different file, so it must see what the module asserts rather than
# what the last enrichment wrote. The merge baseline below stays RAW — feeding post-overlay rows
# to a pass that writes that same file would bake the correction in (RM83).
resolution_rows = overlaid_input_rows(
spec_dir, "resolution.csv", resolution_rows, error=ClinicalAssertionError
)
existing_rows: list[ClinicalAssertionRow] = []
if assertions_path.exists():
parsed, errors, _ = load_csv_rows(assertions_path, ClinicalAssertionRow, assertions_path.name)
if errors:
raise ClinicalAssertionError(f"existing {assertions_path.name} is invalid: {errors[0]}")
existing_rows = parsed
alleles, off_build = _alleles_from_resolution(resolution_rows)
if off_build:
# One counted line naming the build, not one per row: an author on GRCh37 needs to know the
# pass declined rather than that the archive found nothing.
logger.warning(
"Clinical-assertion enrichment skipped %d row(s) whose genome_build is not %s: the "
"ClinVar snapshot is %s-only and its lookup key carries no assembly, so a coordinate "
"from another build would return a different variant's record under this module's key. "
"Examples: %s",
len(off_build),
ASSERTION_GENOME_BUILD,
ASSERTION_GENOME_BUILD,
off_build[:3],
)
reference = resolve_clinvar_reference(clinvar_cache)
if reference is None and not offline and download:
# Provisioning wired the way `enrich()` wires it: best-effort, so a failure degrades to a
# skip rather than sinking the pass.
try:
ensure_clinvar_snapshot(clinvar_cache)
reference = resolve_clinvar_reference(clinvar_cache)
except Exception as exc:
logger.warning("ClinVar snapshot provisioning failed (%s); the pass is skipped.", exc)
if reference is None:
logger.warning(
"Clinical-assertion pass skipped: no ClinVar snapshot is reachable%s. Any existing "
"clinical_assertions.csv is kept as the pin; compiles stay reproducible from it. "
"Provision one with `just-dna-enricher clinvar pull`.",
" and --offline was given" if offline else "",
)
out = sorted(existing_rows, key=_sort_key)
if write and existing_rows:
_write_assertions_csv(out, assertions_path)
return ClinicalAssertionResult(
rows=out, mode=mode, skipped_no_snapshot=True, off_build=sorted(set(off_build))
)
dataset = snapshot_dataset(reference)
# **Every in-scope allele is asked about on every run, and that is a deliberate departure from
# `enrich_frequencies`.** That pass skips an allele any existing row covers because gnomAD costs a
# slot of a 10-per-minute budget; this one reads a local parquet, where the same skip protects
# nothing and costs correctness. ClinVar *grows*: skipping on `variant_key` meant a newly-published
# record could never reach an allele the table already mentioned, and a `not_found` row — an
# absence recorded before the first submission — became a permanent pin. Dedup happens per record
# instead, on the `(variant_key, variation_id)` grain this table is documented to have.
wanted = list(alleles)
def _unusable(reason: object) -> ClinicalAssertionResult:
"""Degrade rather than sink the run, keeping any existing table as the pin.
The rule the resolver link and the cross-check already follow for a located-but-unusable
snapshot. **Two causes reach it, and they are the same cause**: a query that fails, and a
record whose values this reader cannot load. Treating the second as a crash and the first as a
skip would be an arbitrary split — both mean *this snapshot is not one this reader can use* —
and the second is not hypothetical: the published ClinVar repo still carries a pre-split
`clinvar.parquet` whose columns are raw VCF INFO fields, so a reader that ever globbed it
would see ClinVar's own `Pathogenic/Likely_pathogenic` where the vocabulary is expected.
"""
logger.warning(
"ClinVar reference at %s is present but not usable (%s); the clinical-assertion pass "
"is skipped this run. Rebuild it with `just-dna-enricher clinvar build`.",
reference,
reason,
)
kept = sorted(existing_rows, key=_sort_key)
if write and existing_rows:
_write_assertions_csv(kept, assertions_path)
return ClinicalAssertionResult(
rows=kept, mode=mode, skipped_no_snapshot=True, off_build=sorted(set(off_build))
)
try:
records = lookup_clin_sig(reference, [(c, s, r, a) for _k, _rs, c, s, r, a in wanted])
except Exception as exc:
return _unusable(exc)
fetched_at = now_utc_iso()
# `ClinicalAssertionRow._KEY_FIELDS`, not a restated tuple: this pass and the published
# `hints.key_fields("clinical_assertions.csv")` then cannot disagree (S51). The two `seen.add`
# calls below build the same shape for a row that does not exist yet.
seen = {merge_key(row) for row in existing_rows}
out: list[ClinicalAssertionRow] = list(existing_rows)
covered: list[str] = []
missing: list[str] = []
#: Alleles whose recorded absence this run has answered — see the withdrawal below.
answered: set[str] = set()
for key, rsid, chrom, start, ref, alt in wanted:
found = records.get((chrom, start, ref, alt), [])
if not found:
missing.append(key)
# A `not_found` row is a FACT — the archive was consulted and has no record for this
# allele — and is materially different from an allele that was never queried, which has
# no row at all. Carried under `(key, None)` so re-running against the same snapshot
# re-states the absence rather than appending a second copy of it.
if (key, None) in seen:
continue
seen.add((key, None))
out.append(
ClinicalAssertionRow(
variant_key=key,
rsid=rsid,
chrom=chrom,
start=start,
ref=ref,
alt=alt,
genome_build=ASSERTION_GENOME_BUILD,
dataset=dataset,
source=CLINVAR_SOURCE,
status="not_found",
fetched_at=fetched_at,
)
)
continue
covered.append(key)
answered.add(key)
for record in found:
variation_id = _text(record.get("variation_id"))
if (key, variation_id) in seen:
continue
seen.add((key, variation_id))
# A record this reader cannot load is the snapshot being unusable, not this row being
# interesting — same verdict as a failed query, reached above. Caught narrowly
# (`ValidationError`, not `Exception`) so a genuine bug in the lines below still surfaces
# as a bug instead of being reported as somebody else's bad parquet.
try:
out.append(
ClinicalAssertionRow(
variant_key=key,
rsid=rsid,
chrom=chrom,
start=start,
ref=ref,
alt=alt,
genome_build=ASSERTION_GENOME_BUILD,
clin_sig=_text(record.get("clin_sig")),
clin_sig_raw=_text(record.get("clin_sig_raw")),
review_status=_text(record.get("review_status")),
review_stars=record.get("review_stars"),
condition=_text(record.get("condition")),
variation_id=variation_id,
dataset=dataset,
source=CLINVAR_SOURCE,
status="resolved",
fetched_at=fetched_at,
)
)
except ValidationError as exc:
return _unusable(f"record {variation_id} at {chrom}:{start} — {exc}")
# **A recorded absence the archive has since answered is withdrawn, not left beside the answer.**
# This is the one place the pass removes a row it wrote earlier, and the narrowness is the point:
# `not_found` is this pass's own bookkeeping — "the archive was consulted and has none" — and it
# stops being true the moment a record exists. Keeping both would make the table assert an absence
# and a presence for one allele, which is worse than either. Nothing else is touched: a `resolved`
# row, and any row a curator wrote about an allele the archive still lacks, both survive untouched.
if answered:
out = [r for r in out if not (r.status == "not_found" and r.variant_key in answered)]
out.sort(key=_sort_key)
result = ClinicalAssertionResult(
rows=out,
covered=sorted(set(covered)),
missing=sorted(set(missing)),
off_build=sorted(set(off_build)),
dataset=dataset,
mode=mode,
)
# `off_build` is deliberately NOT part of this gate, for the reason `not_covered` is outside the
# frequency pass's: `strict` means "a reproducible artifact", and a coordinate on another assembly
# is reproducibly out of this snapshot's reach. Refusing would make a GRCh37 module uncompilable
# for a reason no authored edit could fix.
if mode == "strict" and result.missing:
raise ClinicalAssertionError(
f"strict clinical-assertion enrichment: {len(result.missing)} resolved allele(s) have no "
f"ClinVar record: {result.missing}. Most variants are not in ClinVar at all, so this is "
f"usually correct data rather than a read failure — use mode='best_effort'."
)
if write:
# Same rule as every other pass that consults a source: record its terms, or the module cannot
# account for it. ClinVar is public-domain and asks to be cited, which is what this row carries.
# Inside the table's commit, so neither file exists without the other (S98, RM231).
_write_assertions_csv(
out,
assertions_path,
before_commit=lambda: record_source_terms(
{row.source for row in out if row.source},
"clinical_assertion",
spec_dir,
error=ClinicalAssertionError,
),
)
return result