blindspot / shared /contracts.py
yashwrites's picture
Signs on the film, draw tool, My search off by default, example films and Radiopaedia links (part 2)
2e6a666 verified
Raw History Blame Contribute Delete
18.8 kB
"""Pydantic mirrors of shared/schemas/*.json. FROZEN CONTRACT — owned by the orchestrator.
Subagents import from here; they never edit it. Changes go through a CONTRACT CHANGE REQUEST
in docs/PROGRESS.md. shared/tests/test_contracts.py checks these models against the JSON Schemas.
"""
from __future__ import annotations
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
# --------------------------------------------------------------------------- vocabularies
Label = Literal[
"pneumothorax",
"effusion",
"consolidation",
"atelectasis",
"nodule",
"mass",
"calcification",
"fracture",
"pleural_thickening",
"cardiomegaly",
"emphysema",
"fibrosis",
"diffuse_nodule",
"pancreatic_tumour",
"liver_tumour",
"brain_tumour",
"lung_tumour",
"colon_tumour",
]
FocalLabel = Literal[
"pneumothorax",
"effusion",
"consolidation",
"atelectasis",
"nodule",
"mass",
"calcification",
"fracture",
"pleural_thickening",
"pancreatic_tumour",
"liver_tumour",
"brain_tumour",
"lung_tumour",
"colon_tumour",
]
Modality = Literal["cxr", "ct", "mr"]
BodyRegion = Literal["chest", "abdomen", "brain"]
Plane = Literal["axial", "coronal", "sagittal"]
VOLUME_LABELS: tuple[str, ...] = ("pancreatic_tumour", "liver_tumour", "brain_tumour", "lung_tumour", "colon_tumour")
PatternLabel = Literal["cardiomegaly", "emphysema", "fibrosis", "diffuse_nodule"]
LearnerFocalLabel = Literal[
"pneumothorax",
"effusion",
"consolidation",
"atelectasis",
"nodule",
"mass",
"calcification",
"fracture",
"pleural_thickening",
"not_sure",
"pancreatic_tumour",
"liver_tumour",
"brain_tumour",
"lung_tumour",
"colon_tumour",
]
Side = Literal["right", "left", "bilateral", "midline"]
Kind = Literal["focal", "pattern"]
Split = Literal["practice", "assess_A", "assess_B", "bench", "holdout"]
Mode = Literal["practice", "drill", "assess_A", "assess_B", "review"]
Level = Literal["MS1", "MS2", "MS3", "MS4", "intern", "resident", "PA/NP student", "other"]
Confidence = Literal[1, 2, 3, 4, 5]
OutcomeResult = Literal[
"found",
"mislabeled",
"missed_search",
"missed_recognition",
"missed_decision",
"pattern_found",
"pattern_missed",
"pattern_false",
"true_positive",
"duplicate",
"false_positive",
"true_negative",
"unmatched",
]
DebriefFindingResult = Literal[
"found",
"mislabeled",
"missed_search",
"missed_recognition",
"missed_decision",
"pattern_found",
"pattern_missed",
]
Verdict = Literal["all_found", "partly_found", "missed", "overcall", "correct_normal", "missed_normal_call"]
ReviewStatus = Literal["ai_draft", "student_reviewed", "radiologist_reviewed"]
DebriefSource = Literal["live", "cache", "template"]
XY = tuple[float, float]
BBox = tuple[float, float, float, float]
class _Strict(BaseModel):
model_config = ConfigDict(extra="forbid")
# --------------------------------------------------------------------------- case.json / finding.json
class Geometry(_Strict):
kind: Literal["polygon", "bbox"]
bbox: BBox
polygon: list[XY] | None = None
mask_path: str | None = None
class Component(_Strict):
name: str
label_value: int
class Measure(_Strict):
long_mm: float
slice: int
plane: Literal["axial"] = "axial"
class Finding(_Strict):
finding_id: str
label: Label
source_label: str
kind: Kind
geometry: Geometry
centroid: XY
area_frac: float = Field(ge=0, le=1)
label_value: int | None = None
label_values: list[int] | None = None
centroid3: tuple[float, float, float] | None = None
slice_range: tuple[int, int] | None = None
measure: Measure | None = None
components: list[Component] | None = None
volume_mm3: float | None = None
side: Side | None = None
zones: list[str] = []
primary_zone: str | None = None
relative_location: str | None = None
contrast: float | None = None
model_prob: float | None = None
edge_dist: float | None = None
difficulty: float | None = None
readers: int | None = None
agreement: float | None = None
@property
def short_id(self) -> str:
return self.finding_id.split("#", 1)[1]
class Window(_Strict):
wc: float
ww: float
class VolumeInfo(_Strict):
shape: tuple[int, int, int]
spacing: tuple[float, float, float]
window: Window
data_path: str
mask_path: str
labels: dict[str, str]
sequence: str | None = None
original_shape: list[int] | None = None
crop_origin: list[int] | None = None
class Provenance(_Strict):
dataset: str
segmented_by: str
readers: int | None = None
institution: str | None = None
license: str | None = None
citation: str | None = None
url: str | None = None
grade: Literal["radiologist", "clinician", "model", "unknown"] = "unknown"
@property
def badge(self) -> str:
return f"Segmented by {self.segmented_by} ({self.dataset})"
class Case(_Strict):
case_id: str
source: Literal["chestx-det", "vindr-cxr", "nih-bbox", "synthetic", "msd"]
modality: Modality = "cxr"
body_region: BodyRegion | None = None
volume: VolumeInfo | None = None
provenance: Provenance | None = None
source_split: str
split: Split
image_path: str
width: int
height: int
pixel_spacing_mm: float | None = None
is_normal: bool
findings: list[Finding]
anatomy_path: str | None = None
zones_path: str | None = None
zones_approximate: bool = False
cardiothoracic_ratio: float | None = None
difficulty_prior: float = 0.0
features: dict[str, float] = {}
license_tag: str
attribution: str
qa_flags: list[str] = []
# --------------------------------------------------------------------------- telemetry_event.json
TelemetryKind = Literal[
"move", "down", "up", "wheel", "enter", "leave", "loupe", "wl", "pan", "slice", "plane", "window"
]
class TelemetryEvent(_Strict):
t: float = Field(ge=0)
kind: TelemetryKind
x: float | None = None
y: float | None = None
zoom: float = Field(ge=0)
vp: BBox
loupe: bool
plane: Plane | None = None
slice: int | None = None
# --------------------------------------------------------------------------- attempt_submit.json
class Mark(_Strict):
mark_id: str
x: float
y: float
label: LearnerFocalLabel
confidence: Confidence
plane: Plane | None = None
slice: int | None = None
voxel: tuple[float, float, float] | None = None
polygon: list[XY] | None = None
tool: Literal["point", "draw"] | None = None
class Measurement(_Strict):
mark_id: str
long_mm: float = Field(ge=0)
plane: Plane
slice: int
p0: list[float] | None = None
p1: list[float] | None = None
class PatternSelection(_Strict):
label: PatternLabel
confidence: Confidence
class ClientTiming(_Strict):
shown_at: str
submitted_at: str
class AttemptSubmit(_Strict):
marks: list[Mark]
patterns: list[PatternSelection]
declared_normal: bool
normal_confidence: Confidence | None = None
telemetry: list[TelemetryEvent] = Field(max_length=20000)
hints_used: int = Field(ge=0, le=3)
client_timing: ClientTiming
measurements: list[Measurement] = []
# --------------------------------------------------------------------------- submit_result.json
class Outcome(_Strict):
target: str
result: OutcomeResult
dwell_ms: float | None = None
zone: str | None = None
matched: str | None = None
learner_label: str | None = None
size_verdict: dict[str, Any] | None = None
slices_viewed: bool | None = None
class SignGeometry(_Strict):
kind: Literal["polyline", "polygon", "circle", "arrow", "segment", "band"]
points: list[XY]
radius: float | None = None
plane: str | None = None
slice: int | None = None
class Sign(_Strict):
id: str
name: str
text: str
geometry: SignGeometry
schematic: str | None = None
class SizeVerdict(_Strict):
your_mm: float
reference_mm: float
diff_mm: float
diff_pct: float
ok: bool
plane: str | None = None
class SliceDwell(_Strict):
plane: str
slice: int
ms: float
has_finding: bool
finding_ids: list[str] | None = None
class RevealFinding(_Strict):
finding_id: str
label: str
display: str
kind: Kind
polygon: list[XY] | None = None
bbox: BBox
centroid: XY | None = None
side: str | None = None
zones: list[str]
primary_zone: str | None = None
relative_location: str | None = None
result: OutcomeResult | None = None
dwell_ms: float | None = None
slice_range: list[int] | None = None
centroid3: list[float] | None = None
label_values: list[int] | None = None
components: list[Component] | None = None
measure: Measure | None = None
size_verdict: SizeVerdict | None = None
signs: list[Sign] | None = None
class RevealMark(_Strict):
mark_id: str
result: Literal["true_positive", "duplicate", "false_positive", "unmatched"]
matched_finding: str | None = None
zone: str | None = None
voxel: list[float] | None = None
plane: str | None = None
slice: int | None = None
polygon: list[XY] | None = None
outline_verdict: Literal["on_target", "partly", "too_broad", "off"] | None = None
class Arrow(_Strict):
from_mark: str | None
to_finding: str
text: str
label: str | None = None
from_xy: XY | None = None
to_xy: XY | None = None
class SearchSummary(_Strict):
lung_coverage_pct: float
unvisited_review_areas: list[str]
visited_review_areas: list[str] = []
first_visits: list[str] = []
time_to_first_mark_s: float | None = None
total_time_s: float | None = None
zoom_used: bool = False
loupe_used: bool = False
heatmap_png_b64: str | None = None
slice_dwell: list[SliceDwell] | None = None
slices_viewed_pct: float | None = None
finding_slices_viewed: dict[str, bool] | None = None
class Reveal(_Strict):
findings: list[RevealFinding]
marks: list[RevealMark]
arrows: list[Arrow]
search: SearchSummary
ctr: float | None = None
is_normal: bool = False
zones_approximate: bool = False
maskvol_url: str | None = None
modality: Modality = "cxr"
provenance: dict[str, Any] | None = None
class FactsCard(_Strict):
headline: str
lines: list[str]
class SubmitResult(_Strict):
score: float = Field(ge=0, le=100)
success: bool
outcomes: list[Outcome]
reveal: Reveal
facts_card: FactsCard
debrief_status: Literal["pending", "disabled"]
class AssessmentRecorded(_Strict):
recorded: Literal[True] = True
index: int | None = None
total: int | None = None
# --------------------------------------------------------------------------- debrief_facts.json
class FactsFinding(_Strict):
id: str
label: str
display: str
kind: Kind
side: str | None = None
primary_zone: str | None = None
zones: list[str] = []
relative_location: str | None = None
size: str | None = None
difficulty: Literal["easy", "moderate", "hard"] | None = None
zones_approximate: bool = False
ctr: float | None = None
slice_range: list[int] | None = None
size_mm: float | None = None
components: list[str] | None = None
signs_drawn: list[str] | None = None
class FactsCase(_Strict):
case_id: str
is_normal: bool
projection: str
pixel_spacing_mm: float | None = None
findings: list[FactsFinding]
modality: Modality = "cxr"
body_region: str | None = None
provenance: str | None = None
class FactsMark(_Strict):
id: str
label: str
confidence: int
zone: str | None
plane: str | None = None
slice: int | None = None
tool: str | None = None
class FactsMeasurement(_Strict):
mark_id: str
long_mm: float
plane: str
class FactsPattern(_Strict):
label: str
confidence: int
class FactsLearner(_Strict):
level: str
declared_normal: bool
normal_confidence: int | None = None
hints_used: int
time_to_submit_s: float
marks: list[FactsMark]
pattern_selections: list[FactsPattern]
measurements: list[FactsMeasurement] = []
class SpatialRelation(_Strict):
from_: str = Field(alias="from")
to: str
text: str
model_config = ConfigDict(extra="forbid", populate_by_name=True, serialize_by_alias=True)
class FactsSearch(_Strict):
lung_coverage_pct: float
unvisited_review_areas: list[str]
first_visits: list[str]
zoom_used: bool = False
loupe_used: bool = False
slices_viewed_pct: float | None = None
finding_slices_viewed: dict[str, bool] | None = None
class DebriefFacts(BaseModel):
model_config = ConfigDict(extra="forbid")
schema_: Literal["debrief_facts.v1"] = Field(default="debrief_facts.v1", alias="schema")
case: FactsCase
learner: FactsLearner
outcomes: list[Outcome]
spatial_relations: list[SpatialRelation]
search: FactsSearch
history: dict[str, Any]
teaching_cards: list[str]
model_config = ConfigDict(extra="forbid", populate_by_name=True, serialize_by_alias=True)
# --------------------------------------------------------------------------- debrief_output.json
class DebriefFindingOut(_Strict):
finding_id: str
result: DebriefFindingResult
where_to_look: str
what_it_looks_like: list[str]
why: str
class DebriefOvercall(_Strict):
mark_id: str
explanation: str
possible_mimics: list[str]
class DebriefOutput(_Strict):
headline: str
verdict: Verdict
findings: list[DebriefFindingOut]
overcalls: list[DebriefOvercall]
search_coaching: str
calibration_note: str
next_step: str
fact_ids: list[str]
class DebriefResponse(_Strict):
status: Literal["pending", "ready", "failed", "disabled"]
debrief: DebriefOutput | None = None
source: DebriefSource | None = None
provenance: ReviewStatus | None = None
validator: dict[str, Any] | None = None
latency_ms: float | None = None
error: str | None = None
# --------------------------------------------------------------------------- teaching_card.json
class CardReview(_Strict):
status: ReviewStatus
reviewer: str | None
date: str | None
notes: str | None
class TeachingCard(_Strict):
label: str
display_name: str
kind: Kind
one_liner: str
key_signs: list[str] = Field(min_length=1)
where_it_hides: list[str]
mimics: list[str]
commonly_confused_with: list[str]
search_tip: str
radiopaedia_url: str | None
review: CardReview
modality: Modality | None = None
# --------------------------------------------------------------------------- api.json
TutorMode = Literal["live", "offline", "paused_credits", "paused_budget", "paused_rate", "paused_error"]
class TutorStatus(_Strict):
mode: TutorMode
reason: str | None = None
resume_at: str | None = None
spend_usd: dict[str, float] | None = None
budget_usd: dict[str, float] | None = None
class Health(_Strict):
ok: bool
offline: bool
cases: int
version: str | None = None
tutor: TutorStatus | None = None
modalities: list[Modality] | None = None
cases_by_modality: dict[str, int] | None = None
class SessionCreate(_Strict):
display_name: str = Field(min_length=1, max_length=80)
level: Level
participant_code: str | None = None
mode: Mode
settings: dict[str, Any] = {}
class SessionCreated(_Strict):
session_id: str
learner_id: str
mode: Mode
class VolumePreset(_Strict):
name: str
wc: float
ww: float
class NextCaseVolume(_Strict):
"""Voxels only. The label volume is never sent before submit."""
shape: list[int]
spacing: list[float]
window: Window
data_url: str
sequence: str | None = None
presets: list[VolumePreset] | None = None
class NextCaseCase(_Strict):
case_id: str
image_url: str
width: int
height: int
modality: Modality = "cxr"
body_region: str | None = None
volume: NextCaseVolume | None = None
provenance: dict[str, Any] | None = None
class NextCase(_Strict):
"""INVARIANT: never contains ground truth."""
attempt_id: str
case: NextCaseCase
index: int
total: int | None = None
hints_enabled: bool = True
done: bool = False
class HintRequest(_Strict):
marks: list[Mark]
telemetry: list[TelemetryEvent]
class HintResponse(_Strict):
level: int = Field(ge=1, le=3)
text: str
remaining: int = Field(ge=0)
class AskRequest(_Strict):
question: str = Field(min_length=1, max_length=500)
class AskResponse(_Strict):
answer: str
remaining: int
source: Literal["live", "template"] = "live"
class AssessmentSummary(BaseModel):
model_config = ConfigDict(extra="allow")
session_id: str
mode: Mode
n_cases: int
sensitivity: float | None
specificity: float | None
localization_fraction: float | None
false_positives_per_image: float
miss_type_mix: dict[str, int]
score_mean: float
cases: list[dict[str, Any]] = []
class ReviewRating(_Strict):
reviewer: str
role: str
item_type: Literal["debrief", "card"]
item_id: str
accuracy: int = Field(ge=1, le=5)
teaching: int = Field(ge=1, le=5)
safety_flag: bool
comment: str | None = None
card_edits: dict[str, Any] | None = None
class SignSchematic(_Strict):
id: str
name: str
description: str
labels: list[str]
modality: str | None = None
svg: str
radiopaedia_url: str | None = None
review_status: str = "ai_draft"
class SusSubmit(_Strict):
learner_id: str
answers: list[int] = Field(min_length=10, max_length=10)
class SusResult(_Strict):
score: float
# Ground-truth field names that must never appear in pre-submit responses (used by leak tests).
GROUND_TRUTH_KEYS: frozenset[str] = frozenset(
{
"findings",
"is_normal",
"label",
"polygon",
"mask_path",
"zones",
"primary_zone",
"side",
"relative_location",
"difficulty_prior",
"cardiothoracic_ratio",
"anatomy_path",
"zones_path",
"centroid",
"area_frac",
"source_label",
"qa_flags",
"outcomes",
"reveal",
"facts_card",
"maskvol_url",
"label_values",
"slice_range",
"measure",
"centroid3",
"components",
"mask_path",
}
)