src/endopath/coverage.py

"""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,
    }