File size: 4,701 Bytes
ef393a2 11ea0f0 ef393a2 11ea0f0 6a78d3d ef393a2 6a78d3d ef393a2 6a78d3d ef393a2 6a78d3d ef393a2 6a78d3d ef393a2 | 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 100 101 102 103 104 105 106 107 108 | """Field coverage: a project's target dictionary and the record types that fulfil it (#237).
The coverage view (the Field coverage page) shows every field of a project's target dictionary, coloured by
the record type that sources it and metered by how full it is in the confirmed dataset. That mapping is
DB-backed here rather than hardcoded in the frontend, so a different project or dictionary is a different
format to review. The record types replace the old hardcoded source strings.
`seed_coverage` loads the TCGA-UCEC REDCap seed (`coverage_data`, generated and de-identified) into the DB
once. `build_coverage` reads it back as the API payload; the live fill is still computed in the browser
from `GET /api/dashboard` and `GET /api/longitudinal/cohort`, so this module stays a thin schema read.
"""
from __future__ import annotations
from sqlalchemy import Connection
from endopath import coverage_data, coverage_extractions, storage
def seed_coverage(conn: Connection, project: str = coverage_data.PROJECT_ID) -> None:
"""Seed the project's record types and target-dictionary coverage from the generated seed, once, then
load the populated record-type extractions. Each step is a no-op if already seeded."""
if not storage.has_coverage(conn, project):
for order, (record_type_id, name, blurb) in enumerate(coverage_data.RECORD_TYPES):
storage.upsert_record_type(
conn,
project=project,
record_type_id=record_type_id,
name=name,
source_class=record_type_id,
blurb=blurb,
sort_order=order,
)
for order, (name, label, section, record_type_id, fill_via, fill_key) in enumerate(
coverage_data.FIELDS
):
storage.upsert_coverage_field(
conn,
project=project,
dictionary_id=coverage_data.DICTIONARY_ID,
name=name,
label=label,
section=section,
record_type_id=record_type_id,
fill_via=fill_via,
fill_key=fill_key,
sort_order=order,
)
conn.commit()
seed_extractions(conn, project)
def seed_extractions(conn: Connection, project: str = coverage_extractions.PROJECT_ID) -> None:
"""Load each record type's committed, in-conversation sample into its extracted dictionary, once per
record type. A no-op for a record type that already carries extraction rows."""
for record_type_id, cases in coverage_extractions.RECORD_TYPE_EXTRACTIONS.items():
if storage.has_record_type_extractions(conn, project, record_type_id):
continue
for barcode, fields in cases.items():
for order, (field, (value, evidence)) in enumerate(fields.items()):
storage.upsert_record_type_extraction(
conn,
project=project,
record_type_id=record_type_id,
case_barcode=barcode,
field=field,
value=value,
evidence=evidence,
sort_order=order,
)
conn.commit()
def build_coverage(conn: Connection, project: str) -> dict:
"""The Field coverage payload for a project: its record types, every target-dictionary field with the
record type that fulfils it and its fill config, and each record type's populated sample extractions
(the extracted dictionary in action). Empty `record_types`/`fields` when unseeded."""
fields = [
{
"name": row["name"],
"label": row["label"],
"section": row["section"],
"source": row["record_type_id"],
"fill": {"via": row["fill_via"], "key": row["fill_key"]} if row["fill_via"] else None,
}
for row in storage.list_coverage_fields(conn, project)
]
record_types = storage.list_record_types(conn, project)
extractions: dict[str, list[dict]] = {}
for rt in record_types:
rows = storage.list_record_type_extractions(conn, project, rt["record_type_id"])
if rows:
extractions[rt["record_type_id"]] = [
{
"case": row["case_barcode"],
"field": row["field"],
"value": row["value"],
"evidence": row["evidence"],
}
for row in rows
]
return {
"dictionary_id": coverage_data.DICTIONARY_ID,
"record_types": record_types,
"fields": fields,
"extractions": extractions,
}
|