File size: 3,820 Bytes
eea689d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """The target data dictionary an Intake submission points the tool at.
Intake points at a report source and at a target dictionary (docs/prd.md section
8.1). A dictionary names the variables to capture, each variable's permitted
values, and whether confirming a variable requires a licensed reference source.
The built-in dictionary is the CAP endometrium checklist, derived from
`endopath.fields` so it never drifts from `GET /api/fields`. An uploaded
dictionary arrives on the ingestion request in the same shape.
"""
from __future__ import annotations
import json
from typing import Optional
from pydantic import BaseModel, Field
from sqlalchemy import Connection
from endopath import fields, storage
# The CAP protocol the built-in checklist transcribes, recorded in
# data/taxonomy/sources.csv. The dictionary carries the source_id so a reviewer
# can trace every variable to the standard it came from.
BUILTIN_DICTIONARY_ID = "cap_uterus/5.1.0.0"
BUILTIN_DICTIONARY_TITLE = "CAP endometrium checklist"
class DictionaryVariable(BaseModel):
"""One variable to capture. `permitted_values` is None for an open
vocabulary (a stage code, a lymph node string) and a closed list for an
enumerated field. `licence_required` marks a variable whose confirmation
needs a licensed reference source, so a deployment without that licence can
gate it."""
name: str
label: Optional[str] = None
permitted_values: Optional[list[str]] = None
licence_required: bool = False
class DataDictionary(BaseModel):
"""A target dictionary: an id, a title, and the variables to capture. Both
the built-in CAP checklist and an uploaded dictionary use this shape."""
id: str
title: str
variables: list[DictionaryVariable] = Field(default_factory=list)
@property
def licence_required_variables(self) -> list[str]:
return [v.name for v in self.variables if v.licence_required]
def register(conn: Connection, dictionary: DataDictionary) -> DataDictionary:
"""Store a submitted dictionary so the compilation job (#91) can read it back by id after the
request that submitted it has returned. Returns the stored dictionary."""
storage.upsert_dictionary(
conn,
dictionary_id=dictionary.id,
title=dictionary.title,
variables_json=json.dumps([v.model_dump() for v in dictionary.variables]),
)
return dictionary
def get_registered(conn: Connection, dictionary_id: str) -> Optional[DataDictionary]:
"""The stored dictionary, reconstructed from its row, or None if it was never submitted."""
row = storage.get_dictionary(conn, dictionary_id)
if row is None:
return None
variables = [DictionaryVariable(**item) for item in json.loads(row["variables_json"])]
return DataDictionary(id=row["id"], title=row["title"], variables=variables)
def reset(conn: Connection) -> None:
"""Clear the stored dictionaries, the queue, and their artifacts. For tests and a fresh process."""
storage.clear_intake_state(conn)
def builtin_cap_dictionary() -> DataDictionary:
"""The CAP endometrium checklist as a dictionary, one variable per checklist
field, permitted values from each field's closed vocabulary (None for the
open ones). No variable is licence-gated: every value is extracted from the
report and stored with its citation, not drawn from a licensed source."""
variables = [
DictionaryVariable(
name=spec.name,
label=spec.label,
permitted_values=list(spec.enum_values) if spec.enum_values else None,
licence_required=False,
)
for spec in fields.FIELDS
]
return DataDictionary(
id=BUILTIN_DICTIONARY_ID,
title=BUILTIN_DICTIONARY_TITLE,
variables=variables,
)
|