src/endopath/dashboard.py

"""The cohort dashboard aggregate (#100, docs/prd.md sections 8.1, 4.3, 4.5, 4.7).

Section 8.1's Cohort dashboard reports, per checklist field, the distribution of confirmed values and of
coded absence reasons, and whether an absence reason tracks the other variables captured for the same
case. An absence that concentrates in high-grade tumours signals that the missingness follows the disease
rather than chance (section 4.5), which a single completion percentage cannot hold. It also lists, per
case, which absent variables would require ordering slides (section 4.7), grounded in the FIGO 2023
determinability verdict (`taxonomy.derive_stage` / `taxonomy.determinability_matrix`, #35).

This module is the read side. `storage.dashboard_summary` keeps the cohort completion rate and the
worst-field-first summary; this module reads each stored case back through `storage.get_case` (the same
path the export takes) and classifies every checklist field into the section 4.3 field-state vocabulary,
so a rule-suppressed field counts as a coded absence rather than a confirmed value (#43): the code reads
the field state, never a blank.

The two invariants this aggregate serves:

1. Numbers stay numbers. A confirmed value is counted as the primitive the report stated; the association
   strata are derived from a covariate (the FIGO grade reduced to low/high by the registry's `grade_binary`
   rule), never stored in its place.
2. No value without its evidence. A field is never simply present or blank: it is a confirmed value, one
   of the five coded absence reasons, or a pending draft a reviewer has not yet settled. Each of the three
   is counted on its own, so silence is never the response to a miss.
"""

from __future__ import annotations

from typing import Any, Optional

from sqlalchemy import Connection

from endopath import fields, storage, taxonomy
from endopath.schema import Case, ChecklistField, FieldStatus
from endopath.validation import FieldState

# The closed section-4.3 absence vocabulary, in the order the screen names it. Every field holds this
# whole set as its legend even where a code has a zero count, so the distribution reads against a fixed
# vocabulary rather than only the codes that happen to appear.
ABSENCE_VOCABULARY: tuple[str, ...] = (
    FieldState.NOT_APPLICABLE.value,
    FieldState.NOT_STATED.value,
    FieldState.INDETERMINATE.value,
    FieldState.DETERMINED_BY_JOIN.value,
    FieldState.CONSTRAINED.value,
)

# A confirmed value that names its own unassessability is the `indeterminate` state (the report says the
# feature cannot be assessed, which is a value), not a plain confirmed value. Held as a defensive set: a
# CAP value set that holds a "cannot be assessed" member reads as indeterminate, never as a plain value.
_INDETERMINATE_VALUES = frozenset({"cannot_be_assessed", "cannot_be_determined", "indeterminate"})

# How a missing variable is recovered, so the per-case list separates the absences that demand glass
# slides from the ones a data join fills (docs/prd.md sections 4.7, 8.3). A molecular class absent from a
# report of this era arrives from the clinical record, so it is a join, never a slide order; every other
# pathology variable is re-read from the slides. `pathologic_stage` is a derived summary, recovered by
# recovering its inputs rather than ordered on its own.
_RECOVERY_METHOD: dict[str, str] = {
    "histologic_type": "slides",
    "histologic_grade": "slides",
    "myometrial_invasion_percent": "slides",
    "cervical_stromal_invasion": "slides",
    "lymphovascular_space_invasion": "slides",
    "regional_lymph_node_status": "slides",
    "pathologic_stage": "derived",
    "molecular_classification": "join",
}

# The stratifier the absence-association is measured against: the FIGO grade reduced to low/high by the
# registry's `grade_binary` membership rule (#35), plus `unknown` for a case whose grade is not stated.
# Grade is the covariate docs/prd.md section 4.5 names ("concentrates in high-grade tumours").
_STRATIFIER_FIELD = "histologic_grade"
_STRATA: tuple[str, ...] = ("low", "high", "unknown")

# An association fires only with support on both known strata and a wide enough gap, so a one-case
# coincidence does not read as a signal.
_MIN_STRATUM_SUPPORT = 2
_SIGNAL_MARGIN = 0.20


def _primitive(value: object) -> object:
    """An enum to its value, everything else as-is, so a value read back as a raw string and one still
    with its enum both compare against the string vocabulary."""
    return value.value if hasattr(value, "value") else value


def field_disposition(field: ChecklistField) -> tuple[str, Optional[Any]]:
    """Reduce one stored checklist field to a dashboard disposition, reading the field state rather than a
    blank (#43, CLAUDE.md field-state corollary):

    - `("present", value)` a confirmed, applicable field with a value: it counts in the value
      distribution.
    - `("<absence_code>", None)` a coded absence from the section 4.3 vocabulary.
    - `("pending", value_or_None)` an applicable field a reviewer has not yet settled (still needs review).

    A gate-suppressed field is `not_applicable`; a flagged field was searched for and confirmed absent, so
    it is `not_stated`; a confirmed value naming its own unassessability is `indeterminate`.
    """
    if not field.applicable:
        return FieldState.NOT_APPLICABLE.value, None
    if field.status is FieldStatus.FLAGGED:
        return FieldState.NOT_STATED.value, None

    value = _primitive(field.value)
    if field.status is FieldStatus.CONFIRMED:
        if value is None:
            # A reviewer confirmed the field has no value: searched and absent.
            return FieldState.NOT_STATED.value, None
        if isinstance(value, str) and value in _INDETERMINATE_VALUES:
            return FieldState.INDETERMINATE.value, None
        return FieldState.PRESENT.value, value
    # Needs review: a drafted value a reviewer has not confirmed, or an empty draft. Neither a confirmed
    # value nor a coded absence yet.
    return "pending", value


def _stratum_for_case(case: Case) -> str:
    """The FIGO-grade stratum a case falls in: `low` or `high` by the registry `grade_binary` rule, or
    `unknown` when the grade is not stated."""
    grade = _primitive(case.checklist.histologic_grade.value)
    band = taxonomy.grade_binary(grade)
    return band if band in ("low", "high") else "unknown"


def _association(per_case: list[tuple[str, tuple[str, Optional[Any]]]]) -> dict:
    """Whether a field's absence tracks the FIGO-grade stratum (docs/prd.md section 4.5). Builds a
    present-vs-absent contingency across the strata and fires a signal when the absent rate on one known
    stratum exceeds the other by a wide margin with support on both, the shape of missingness that follows
    the disease rather than chance.

    `per_case` is `(stratum, disposition)` for every case with the field, where a disposition of
    `pending` is neither present nor a coded absence and so sits outside the contingency.
    """
    counts: dict[str, dict[str, int]] = {s: {"present": 0, "absent": 0} for s in _STRATA}
    for stratum, (kind, _value) in per_case:
        bucket = counts[stratum]
        if kind == FieldState.PRESENT.value:
            bucket["present"] += 1
        elif kind in ABSENCE_VOCABULARY:
            bucket["absent"] += 1
        # a pending field counts toward neither series

    rows: list[dict] = []
    for stratum in _STRATA:
        present = counts[stratum]["present"]
        absent = counts[stratum]["absent"]
        total = present + absent
        rows.append(
            {
                "stratum": stratum,
                "present": present,
                "absent": absent,
                "total": total,
                "absent_rate": (absent / total) if total else 0.0,
            }
        )

    known = [r for r in rows if r["stratum"] != "unknown" and r["total"] >= _MIN_STRATUM_SUPPORT]
    signal = False
    detail: Optional[str] = None
    if len(known) >= 2:
        high = max(known, key=lambda r: r["absent_rate"])
        low = min(known, key=lambda r: r["absent_rate"])
        if high["absent"] > 0 and high["absent_rate"] - low["absent_rate"] >= _SIGNAL_MARGIN:
            signal = True
            detail = (
                f"absence concentrates in {high['stratum']}-grade cases "
                f"({high['absent_rate'] * 100:.0f}% absent, against "
                f"{low['absent_rate'] * 100:.0f}% in {low['stratum']}-grade)"
            )
    return {"stratifier": _STRATIFIER_FIELD, "rows": rows, "signal": signal, "detail": detail}


def _case_assertions(case: Case) -> dict[str, Any]:
    """Canonical assertions for the FIGO projection, from the stored checklist. Shared with the export
    projection (#40), so the dashboard verdict and the exported stage read the same assertions."""
    from endopath import projection

    return projection.assertions_from_checklist(case.checklist)


def _case_verdict(case: Case) -> tuple[Optional[str], Optional[str]]:
    """The FIGO 2023 determinability verdict for one case (`report_determinability.case_verdict`'s logic,
    #35): `determined`, `determined_by_join`, `constrained`, or `unmeasurable`, with the stage or candidate
    set. Best-effort: a projection error leaves the verdict unnamed rather than failing the dashboard."""
    try:
        result = taxonomy.derive_stage(_case_assertions(case), "figo_endo/2023")
    except Exception:
        return None, None
    if isinstance(result, taxonomy.Constrained):
        return "constrained", "|".join(result.candidates)
    if isinstance(result, taxonomy.Indeterminate):
        return "unmeasurable", None
    if getattr(result, "modifier", None):
        return "determined_by_join", f"{result.code} ({result.modifier})"
    return "determined", getattr(result, "code", None)


# The checklist fields whose values feed the FIGO 2023 projection (`projection.assertions_from_checklist`).
# A verdict rests on an unreviewed draft when any of these that holds a value has not been confirmed.
_FIGO_INPUT_FIELDS: tuple[str, ...] = (
    "histologic_type",
    "histologic_grade",
    "myometrial_invasion_percent",
    "cervical_stromal_invasion",
    "lymphovascular_space_invasion",
    "molecular_classification",
)


def _verdict_basis(case: Case) -> str:
    """Whether the FIGO verdict rests only on confirmed values (`confirmed`) or draws on at least one
    unreviewed model draft (`draft`). The projection reads a field's value regardless of review state, so a
    determined stage can rest on drafts a reviewer has not settled; the dashboard marks those provisional
    rather than presenting them as established (#224)."""
    for name in _FIGO_INPUT_FIELDS:
        field = getattr(case.checklist, name)
        if _primitive(field.value) is None:
            continue
        if field.status is not FieldStatus.CONFIRMED:
            return "draft"
    return "confirmed"


def _requires_slides_for_case(
    case: Case, dispositions: dict[str, tuple[str, Optional[Any]]]
) -> Optional[dict]:
    """The absent variables one case would order slides to recover (docs/prd.md section 4.7). A variable
    counts when it is `not_stated` and its recovery method is the slides (`_RECOVERY_METHOD`): a molecular
    class arrives from a join and a gated field is genuinely inapplicable, so neither orders a slide.
    Returns None when the case needs no slides.
    """
    variables: list[dict] = []
    join_variables: list[dict] = []
    for field_name, (kind, _value) in dispositions.items():
        if kind != FieldState.NOT_STATED.value:
            continue
        method = _RECOVERY_METHOD.get(field_name, "slides")
        label = fields.BY_NAME[field_name].label
        if method == "slides":
            variables.append({"field_name": field_name, "label": label})
        elif method == "join":
            join_variables.append({"field_name": field_name, "label": label})

    if not variables:
        return None
    verdict, stage = _case_verdict(case)
    return {
        "case_barcode": case.case_barcode,
        "verdict": verdict,
        # Whether the verdict rests only on confirmed values or on at least one unreviewed draft, so the
        # dashboard can mark a draft-based projection provisional rather than stating it as fact (#224).
        "verdict_basis": _verdict_basis(case) if verdict is not None else None,
        "stage": stage,
        "variables": variables,
        "count": len(variables),
        "join_variables": join_variables,
    }


def _distribution_key(field_name: str, value: Any) -> str:
    """The category a confirmed value counts under in the distribution. Two free-text fields store raw
    strings that normalize onto a bounded set (#165): a reported `pathologic_stage` folds to its FIGO
    stage, a reported `regional_lymph_node_status` to its pn_category, so formatting variants and combined
    TNM tokens stop reading as distinct categories. A value the parser cannot resolve keeps its verbatim
    string, so it stays visible in the distribution rather than vanishing into a wrong bucket."""
    if field_name == "pathologic_stage":
        return taxonomy.parse_reported_stage(value).figo_stage or str(value)
    if field_name == "regional_lymph_node_status":
        return taxonomy.parse_reported_nodal(value) or str(value)
    return str(value)


def _stage_component_distributions(components: dict[str, dict[str, int]]) -> list[dict]:
    """The reported stage's parsed T/N/M components as separate distributions (#165 follow-on), so the
    dashboard shows the separated tabs beside the folded FIGO stage. Each is sorted by count; a dimension
    no case stated is dropped."""
    out: list[dict] = []
    for dimension, counts in components.items():
        if not counts:
            continue
        values = sorted(
            ({"value": v, "count": c} for v, c in counts.items()),
            key=lambda item: (-item["count"], item["value"]),
        )
        out.append({"dimension": dimension, "values": values})
    return out


def cohort_dashboard(conn: Connection, project: Optional[str] = None) -> dict:
    """The cohort dashboard payload (#100, docs/prd.md sections 8.1, 4.3, 4.5, 4.7).

    Holds the completion rate and worst-field-first summary from `storage.dashboard_summary`, then, per
    checklist field, the confirmed-value distribution, the coded-absence-reason distribution over the
    section-4.3 vocabulary, and the absence-versus-grade association; and, per case, the absent variables
    that would require ordering slides.

    `project` scopes the whole payload to one project's cases (issue #149), passed through to the summary
    and the batched cohort read. Omitted, the whole cohort is summarized.
    """
    summary = storage.dashboard_summary(conn, project)

    # value_counts[field] maps a confirmed value to its count; absence_counts[field] maps an absence code
    # to its count; per_case_strata[field] holds (stratum, disposition) for the association.
    value_counts: dict[str, dict[str, int]] = {name: {} for name in fields.FIELD_NAMES}
    absence_counts: dict[str, dict[str, int]] = {name: {} for name in fields.FIELD_NAMES}
    pending_counts: dict[str, int] = {name: 0 for name in fields.FIELD_NAMES}
    totals: dict[str, int] = {name: 0 for name in fields.FIELD_NAMES}
    per_case_strata: dict[str, list[tuple[str, tuple[str, Optional[Any]]]]] = {
        name: [] for name in fields.FIELD_NAMES
    }
    requires_slides: list[dict] = []
    # The reported pathologic stage's parsed T/N/M components, kept as their own distributions so the
    # dashboard can separate the tabs beside the folded FIGO stage (#165 follow-on).
    stage_components: dict[str, dict[str, int]] = {"T category": {}, "N category": {}, "M category": {}}

    # One batched read of the whole cohort rather than a get_case per barcode (issue #131), scoped to the
    # project when one is named (issue #149).
    for case, _ in storage.load_all_cases(conn, project):
        stratum = _stratum_for_case(case)
        dispositions: dict[str, tuple[str, Optional[Any]]] = {}

        for field_name, field in case.checklist.all_fields().items():
            if field_name not in value_counts:
                continue
            disposition = field_disposition(field)
            dispositions[field_name] = disposition
            totals[field_name] += 1
            per_case_strata[field_name].append((stratum, disposition))

            kind, value = disposition
            if kind == FieldState.PRESENT.value:
                key = _distribution_key(field_name, value)
                value_counts[field_name][key] = value_counts[field_name].get(key, 0) + 1
                if field_name == "pathologic_stage":
                    parsed = taxonomy.parse_reported_stage(value)
                    for dimension, component in (
                        ("T category", parsed.t_category),
                        ("N category", parsed.n_category),
                        ("M category", parsed.m_category),
                    ):
                        if component:
                            stage_components[dimension][component] = (
                                stage_components[dimension].get(component, 0) + 1
                            )
            elif kind == "pending":
                pending_counts[field_name] += 1
            else:  # a coded absence
                absence_counts[field_name][kind] = absence_counts[field_name].get(kind, 0) + 1

        slides = _requires_slides_for_case(case, dispositions)
        if slides is not None:
            requires_slides.append(slides)

    field_distributions: list[dict] = []
    for field_name in fields.FIELD_NAMES:
        values = sorted(
            ({"value": v, "count": c} for v, c in value_counts[field_name].items()),
            key=lambda item: (-item["count"], item["value"]),
        )
        absence = [
            {"reason": reason, "count": absence_counts[field_name][reason]}
            for reason in ABSENCE_VOCABULARY
            if absence_counts[field_name].get(reason)
        ]
        confirmed = sum(value_counts[field_name].values())
        absent = sum(absence_counts[field_name].values())
        not_stated = absence_counts[field_name].get(FieldState.NOT_STATED.value, 0)
        not_applicable = absence_counts[field_name].get(FieldState.NOT_APPLICABLE.value, 0)
        # A field a gate removed has no meaning for that case, so it leaves the denominator rather than
        # counting against completion.
        applicable = totals[field_name] - not_applicable
        # Resolved: a confirmed value, or an absence code that records an answer. `indeterminate` is the
        # report reporting that the feature cannot be assessed, `determined_by_join` is a value the
        # clinical record supplies, and `constrained` is a value a source narrowed. Only `not_stated`
        # means a source was searched and is silent, so only it counts as missing.
        resolved = confirmed + (absent - not_stated - not_applicable)
        field_distributions.append(
            {
                "field_name": field_name,
                "label": fields.BY_NAME[field_name].label,
                # The JSON-schema type of the field's value (number/integer/string), so the frontend picks
                # a histogram for a numeric distribution and a categorical bar chart otherwise (#100).
                "value_type": fields.BY_NAME[field_name].value_type,
                "confirmed": confirmed,
                "pending": pending_counts[field_name],
                "absent": absent,
                # The absence codes split by what they mean, so a reader never sums five different
                # statements into one missing-data figure (#108). `resolved + pending + not_stated`
                # equals `applicable`.
                "not_stated": not_stated,
                "not_applicable": not_applicable,
                "resolved": resolved,
                "applicable": applicable,
                "total": totals[field_name],
                "values": values,
                "absence": absence,
                "association": _association(per_case_strata[field_name]),
                # The reported stage's parsed T/N/M sub-distributions, so the tile can separate the tabs
                # (#165 follow-on). Empty for every other field.
                "components": _stage_component_distributions(stage_components)
                if field_name == "pathologic_stage"
                else [],
            }
        )

    requires_slides.sort(key=lambda entry: (-entry["count"], entry["case_barcode"]))

    return {
        **summary,
        "absence_vocabulary": list(ABSENCE_VOCABULARY),
        "stratifier": {
            "field_name": _STRATIFIER_FIELD,
            "label": fields.BY_NAME[_STRATIFIER_FIELD].label,
            "strata": list(_STRATA),
        },
        "field_distributions": field_distributions,
        "requires_slides": requires_slides,
        "requires_slides_case_count": len(requires_slides),
    }