feat: redesign around phenotype-driven triage, not variant filtering

A table with filters made the user do the work. Rare disease triage is a different task:
which few variants could explain *this* patient's phenotype, and why. The app now answers
that, and lets a reviewer act on the answer.

Domain
- a case is a proband: a VCF plus the HPO terms observed in the patient (samples -> cases)
- HPO's gene-to-phenotype annotations are loaded as reference data (scripts/load-hpo.py)
- each candidate can be shortlisted or dismissed with a reason and a note

Ranking (app/services/triage.py, 21 tests)
- weighted sum of phenotype match, rarity, consequence severity and the model's score,
  with every component shown next to the candidate
- rarity and consequence filter; phenotype only ranks, because a real diagnosis can sit in
  a gene nobody has annotated yet and filtering on it would hide exactly that case
- ClinVar is deliberately not an input: it appears beside the result as independent
  confirmation, so nothing ranks highly merely because ClinVar already said pathogenic

UI
- the funnel is the headline: variants called -> rare -> coding candidates -> phenotype-matched
- ranked candidates with evidence chips, not a grid of everything; filters are demoted
- a variant panel showing the score breakdown, the matched HPO terms, the raw VEP record and
  links out to Ensembl/gnomAD/ClinVar, with the decision controls
- a printable case report: phenotype, funnel, shortlisted variants with reasons, provenance

API: /cases with phenotypes, /cases/{id}/candidates (funnel + ranked + weights),
/variants/{id}, /variants/{id}/decision, /cases/{id}/report, /phenotypes for the picker.
Scoring moved under the case and now answers 503 with the reason when no model registry is
reachable, instead of a 500.

Verified end to end on a simulated proband (scripts/make-demo-case.sh: real GIAB HG002
background + one real ClinVar 2-star pathogenic NF2 variant). 13 variants called -> 1 coding
candidate, and the planted variant ranks first at 0.80 on phenotype 1.00, rarity 1.00 and
consequence 1.00, with ClinVar agreeing afterwards.

Tests: api 75, ml 18, loader 16, web 27; ruff, mypy, svelte-check, terraform validate, both
kustomize overlays and the Nextflow stub run all clean.
This commit is contained in:
Kemal Yaylali
2026-09-12 08:30:44 +01:00
parent abde5ec6e4
commit 07a01715fd
47 changed files with 2159 additions and 539 deletions
+225
View File
@@ -0,0 +1,225 @@
import logging
import uuid
from dataclasses import asdict
from datetime import UTC, datetime
from fastapi import APIRouter, HTTPException, Query, status
from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import selectinload
from app.db import SessionDep
from app.models import (
Case,
CasePhenotype,
DecisionState,
Job,
JobStatus,
Variant,
VariantDecision,
)
from app.schemas import (
CandidateOut,
CandidatePage,
CaseCreate,
CaseOut,
FunnelOut,
JobOut,
ProvenanceOut,
ReportOut,
ScoreOut,
)
from app.services import candidates as case_view
from app.services import events, triage
from app.services.scoring import score_job
logger = logging.getLogger(__name__)
router = APIRouter()
async def _case_or_404(session: SessionDep, case_id: uuid.UUID) -> Case:
case = await case_view.get_case(session, case_id)
if case is None:
raise HTTPException(404, "case not found")
return case
async def _shortlisted_counts(session: SessionDep, case_ids: list[uuid.UUID]) -> dict[uuid.UUID, int]:
if not case_ids:
return {}
rows = await session.execute(
select(Job.case_id, func.count(VariantDecision.id))
.join(Variant, Variant.job_id == Job.id)
.join(VariantDecision, VariantDecision.variant_id == Variant.id)
.where(Job.case_id.in_(case_ids), VariantDecision.state == DecisionState.shortlisted)
.group_by(Job.case_id)
)
return dict(rows.all()) # type: ignore[arg-type]
async def _latest_jobs(session: SessionDep, case_ids: list[uuid.UUID]) -> dict[uuid.UUID, Job]:
if not case_ids:
return {}
jobs = await session.scalars(
select(Job).where(Job.case_id.in_(case_ids)).order_by(Job.created_at.desc())
)
latest: dict[uuid.UUID, Job] = {}
for job in jobs: # ordered newest first, so the first one wins
latest.setdefault(job.case_id, job)
return latest
def _as_case_out(case: Case, job: Job | None, shortlisted: int) -> CaseOut:
out = CaseOut.model_validate(case)
out.latest_job = JobOut.model_validate(job) if job is not None else None
out.shortlisted = shortlisted
return out
@router.get("", response_model=list[CaseOut])
async def list_cases(session: SessionDep):
cases = (
await session.scalars(
select(Case).order_by(Case.created_at.desc()).options(selectinload(Case.phenotypes))
)
).all()
ids = [c.id for c in cases]
jobs, counts = await _latest_jobs(session, ids), await _shortlisted_counts(session, ids)
return [_as_case_out(c, jobs.get(c.id), counts.get(c.id, 0)) for c in cases]
@router.post("", response_model=CaseOut, status_code=status.HTTP_201_CREATED)
async def create_case(payload: CaseCreate, session: SessionDep):
case = Case(
name=payload.name,
vcf_uri=payload.vcf_uri,
assembly=payload.assembly,
phenotypes=[CasePhenotype(hpo_id=p.hpo_id, label=p.label) for p in payload.phenotypes],
)
session.add(case)
try:
await session.commit()
except IntegrityError: # cases.name is unique
await session.rollback()
raise HTTPException(409, f"a case named {payload.name!r} already exists") from None
await session.refresh(case, attribute_names=["phenotypes"])
return _as_case_out(case, None, 0)
@router.get("/{case_id}", response_model=CaseOut)
async def get_case(case_id: uuid.UUID, session: SessionDep):
case = await _case_or_404(session, case_id)
counts = await _shortlisted_counts(session, [case_id])
return _as_case_out(case, await case_view.latest_job(session, case_id), counts.get(case_id, 0))
@router.get("/{case_id}/jobs", response_model=list[JobOut])
async def list_jobs(case_id: uuid.UUID, session: SessionDep):
result = await session.scalars(
select(Job).where(Job.case_id == case_id).order_by(Job.created_at.desc())
)
return result.all()
@router.post("/{case_id}/annotate", response_model=JobOut, status_code=status.HTTP_202_ACCEPTED)
async def annotate(case_id: uuid.UUID, session: SessionDep):
case = await _case_or_404(session, case_id)
# Commit `running` before launching: a local run that dies instantly is marked failed by its
# watcher, and a later status write here would overwrite that.
job = Job(case_id=case.id, status=JobStatus.running)
session.add(job)
await session.commit()
try:
job.workflow_ref = await events.launch(job.id, case.vcf_uri, case.assembly)
except events.LaunchError as e:
job.status = JobStatus.failed
job.log = str(e)
job.finished_at = datetime.now(UTC)
await session.commit()
await session.refresh(job)
return job
@router.post("/{case_id}/score", response_model=ScoreOut)
async def score(case_id: uuid.UUID, session: SessionDep):
await _case_or_404(session, case_id)
job = await case_view.latest_job(session, case_id, status=JobStatus.succeeded)
if job is None:
raise HTTPException(409, "no finished annotation to score")
try:
scored, version = await score_job(job.id, session)
except Exception as e:
# No registry, no model behind the alias, a model that will not load: all of these are
# the environment being unready, not a bug in the request. Say so rather than throwing 500.
logger.exception("scoring case %s failed", case_id)
raise HTTPException(503, f"could not score with the model: {e}") from e
return ScoreOut(case_id=case_id, scored=scored, model_version=version)
@router.get("/{case_id}/candidates", response_model=CandidatePage)
async def list_candidates(
case_id: uuid.UUID,
session: SessionDep,
gene: str | None = None,
impact: str | None = Query(None, pattern="^(HIGH|MODERATE|LOW|MODIFIER)$"),
max_af: float | None = Query(None, ge=0, le=1),
state: str | None = Query(None, pattern="^(shortlisted|dismissed|undecided)$"),
limit: int = Query(50, ge=1, le=500),
offset: int = Query(0, ge=0),
) -> CandidatePage:
case = await _case_or_404(session, case_id)
view = await case_view.build(session, case)
items = view.candidates
if gene:
items = [c for c in items if (c.variant.gene or "").upper() == gene.upper()]
if impact:
items = [c for c in items if c.variant.impact == impact]
if max_af is not None:
items = [c for c in items if c.variant.gnomad_af is None or c.variant.gnomad_af <= max_af]
if state == "undecided":
items = [c for c in items if c.variant.decision is None]
elif state is not None:
items = [c for c in items if c.variant.decision and c.variant.decision.state.value == state]
labels = view.labels
return CandidatePage(
# The funnel describes the whole case, not the filtered view.
funnel=FunnelOut(**asdict(view.funnel)),
weights=triage.WEIGHTS,
items=[CandidateOut.from_candidate(c, labels) for c in items[offset : offset + limit]],
total=len(items),
limit=limit,
offset=offset,
)
@router.get("/{case_id}/report", response_model=ReportOut)
async def report(case_id: uuid.UUID, session: SessionDep) -> ReportOut:
case = await _case_or_404(session, case_id)
view = await case_view.build(session, case)
labels = view.labels
def decided(state: DecisionState) -> list[CandidateOut]:
return [
CandidateOut.from_candidate(c, labels)
for c in view.candidates
if c.variant.decision is not None and c.variant.decision.state == state
]
shortlisted, dismissed = decided(DecisionState.shortlisted), decided(DecisionState.dismissed)
prediction = next((c.variant.prediction for c in view.candidates if c.variant.prediction), None)
return ReportOut(
case=_as_case_out(case, view.job, len(shortlisted)),
funnel=FunnelOut(**asdict(view.funnel)),
generated_at=datetime.now(UTC),
provenance=ProvenanceOut(
job_id=view.job.id if view.job else None,
vep_version=view.job.vep_version if view.job else None,
finished_at=view.job.finished_at if view.job else None,
model_name=prediction.model_name if prediction else None,
model_version=prediction.model_version if prediction else None,
),
shortlisted=shortlisted,
dismissed=dismissed,
)