src/endopath/evaluation.py

"""Evaluation harness for the metrics in docs/prd.md section 9 (issue #15), minus
the OCR-vs-visual-retrieval benchmark (issue #16, which reuses this module's
scoring on its own held-out comparison).

Ground truth does not exist in this repo yet: docs/prd.md section 9 is explicit
that GDC's coarser structured fields are NOT a substitute for real
pathologist-reviewed labels (benchmark numbers can overstate real-report
performance by 11-32 points), and section 6 notes grade/histotype carry real
interobserver disagreement even among pathologists. So this harness is built
to run the moment a labeled file exists, and tested against synthetic fixtures
until then. `ground_truth_template()` emits the fillable format so a labeling
pass can start.

Ground-truth file format (JSON):

    {
      "cases": [
        {
          "case_barcode": "TCGA-XX-0001",
          "fields": {
            "histologic_type": {"references": ["endometrioid"], "applicable": true},
            "histologic_grade": {"references": ["2", "3"], "applicable": true},
            ...
          }
        }
      ]
    }

`references` is a list because section 6's interobserver point is a first-class
concern: more than one abstractor's value can be recorded per field, and a
prediction is scored correct if it agrees with ANY reference (crediting
agreement with a competent abstractor even where abstractors disagree with
each other), while `interobserver_agreement()` separately reports how often
the references disagree -- i.e. how noisy the reference itself is.

Calibration is kept a distinct check from raw accuracy (section 7 / open risk
#2): the whole review-workflow safety assumption is that a high-confidence
label predicts correctness better than a low-confidence one, which a single
accuracy number cannot show. It is reported as binned accuracy-by-confidence
(a reliability diagram), never folded into one number.
"""

from __future__ import annotations

import json
import math
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Iterable, Mapping, Optional

from endopath.schema import Case, ChecklistField, EndometrialChecklist, FieldStatus


# --- Inputs ----------------------------------------------------------------


@dataclass
class FieldPrediction:
    """One field's extracted value + the confidence attached to it."""

    value: Any
    confidence: Optional[float] = None


@dataclass
class GroundTruthField:
    references: list[Any]  # one value per reference abstractor; >1 enables interobserver scoring
    applicable: bool = True


@dataclass
class GroundTruthCase:
    case_barcode: str
    fields: dict[str, GroundTruthField]


CasePredictions = Mapping[str, FieldPrediction]  # field_name -> prediction


# --- Value comparison ------------------------------------------------------


def _normalize(value: Any) -> Any:
    return value.value if isinstance(value, Enum) else value


def values_match(pred: Any, ref: Any, *, numeric_tolerance: float = 0.0) -> bool:
    """True if a predicted value matches a reference value. Enums compare by
    their string value (predictions carry Histotype etc.; ground-truth JSON
    carries the plain string). Numeric fields (e.g. myometrial invasion
    percent) match within numeric_tolerance, since a reference stated as "50%"
    and a prediction of 48% shouldn't count as a hard miss. bool is excluded
    from numeric comparison so True/1.0 don't accidentally unify."""
    p, r = _normalize(pred), _normalize(ref)
    p_num = isinstance(p, (int, float)) and not isinstance(p, bool)
    r_num = isinstance(r, (int, float)) and not isinstance(r, bool)
    if p_num and r_num:
        return math.isclose(p, r, abs_tol=numeric_tolerance)
    return p == r


def prediction_is_correct(
    pred_value: Any, gt_field: GroundTruthField, *, numeric_tolerance: float = 0.0
) -> bool:
    """Interobserver-aware: correct if the prediction agrees with ANY recorded
    reference abstractor, not only a single designated one."""
    return any(
        values_match(pred_value, ref, numeric_tolerance=numeric_tolerance)
        for ref in gt_field.references
    )


# --- Field-level accuracy --------------------------------------------------


@dataclass
class AccuracyReport:
    n_applicable: int  # ground-truth fields that are applicable and carry a reference
    n_answered: int  # of those, the ones the system gave a non-null value for
    n_correct: int  # of the answered ones, those matching a reference
    per_field_correct: dict[str, int] = field(default_factory=dict)
    per_field_total: dict[str, int] = field(default_factory=dict)

    @property
    def accuracy(self) -> Optional[float]:
        """Accuracy over answered fields -- 'when the tool commits to a value,
        how often is it right'."""
        return self.n_correct / self.n_answered if self.n_answered else None

    @property
    def recall(self) -> Optional[float]:
        """Coverage-adjusted: correct over all applicable fields, so a tool
        that stays silent to protect its accuracy doesn't look better than one
        that answers."""
        return self.n_correct / self.n_applicable if self.n_applicable else None


def field_accuracy(
    predictions: Mapping[str, CasePredictions],
    ground_truth: Mapping[str, GroundTruthCase],
    *,
    numeric_tolerance: float = 0.0,
    numeric_tolerances: Optional[Mapping[str, float]] = None,
) -> AccuracyReport:
    """Score predictions against ground truth across the corpus. Only cases
    present in both maps, and only applicable ground-truth fields with at
    least one reference, are scored."""
    numeric_tolerances = numeric_tolerances or {}
    report = AccuracyReport(n_applicable=0, n_answered=0, n_correct=0)

    for barcode, gt_case in ground_truth.items():
        case_preds = predictions.get(barcode, {})
        for name, gt_field in gt_case.fields.items():
            if not gt_field.applicable or not gt_field.references:
                continue
            report.n_applicable += 1
            report.per_field_total[name] = report.per_field_total.get(name, 0) + 1

            pred = case_preds.get(name)
            if pred is None or pred.value is None:
                continue  # unanswered -- counts against recall, not accuracy
            report.n_answered += 1

            tol = numeric_tolerances.get(name, numeric_tolerance)
            if prediction_is_correct(pred.value, gt_field, numeric_tolerance=tol):
                report.n_correct += 1
                report.per_field_correct[name] = report.per_field_correct.get(name, 0) + 1

    return report


# --- Confidence calibration (distinct from accuracy) -----------------------


@dataclass
class CalibrationBin:
    lower: float
    upper: float
    count: int
    mean_confidence: Optional[float]
    accuracy: Optional[float]  # observed correctness within the bin


@dataclass
class CalibrationReport:
    bins: list[CalibrationBin]
    n_samples: int

    @property
    def expected_calibration_error(self) -> Optional[float]:
        """A supplementary summary only -- the bins are the primary output.
        ECE = sum over bins of (bin share) * |accuracy - mean confidence|."""
        if not self.n_samples:
            return None
        ece = 0.0
        for b in self.bins:
            if b.count and b.accuracy is not None and b.mean_confidence is not None:
                ece += (b.count / self.n_samples) * abs(b.accuracy - b.mean_confidence)
        return ece


def confidence_calibration(
    predictions: Mapping[str, CasePredictions],
    ground_truth: Mapping[str, GroundTruthCase],
    *,
    n_bins: int = 10,
    numeric_tolerance: float = 0.0,
    numeric_tolerances: Optional[Mapping[str, float]] = None,
) -> CalibrationReport:
    """Bin answered predictions by their confidence and report observed
    accuracy per bin -- a reliability diagram. This is the calibration check
    section 7 wants kept separate from raw accuracy: does a high-confidence
    label actually predict correctness on this corpus?"""
    numeric_tolerances = numeric_tolerances or {}
    samples: list[tuple[float, bool]] = []  # (confidence, correct)

    for barcode, gt_case in ground_truth.items():
        case_preds = predictions.get(barcode, {})
        for name, gt_field in gt_case.fields.items():
            if not gt_field.applicable or not gt_field.references:
                continue
            pred = case_preds.get(name)
            if pred is None or pred.value is None or pred.confidence is None:
                continue
            tol = numeric_tolerances.get(name, numeric_tolerance)
            correct = prediction_is_correct(pred.value, gt_field, numeric_tolerance=tol)
            samples.append((float(pred.confidence), correct))

    bins: list[CalibrationBin] = []
    for i in range(n_bins):
        lower = i / n_bins
        upper = (i + 1) / n_bins
        # Upper-inclusive only in the last bin so confidence == 1.0 lands somewhere.
        in_bin = [
            (c, ok)
            for c, ok in samples
            if (lower <= c < upper) or (i == n_bins - 1 and c == upper)
        ]
        if in_bin:
            mean_conf = sum(c for c, _ in in_bin) / len(in_bin)
            acc = sum(1 for _, ok in in_bin if ok) / len(in_bin)
        else:
            mean_conf = None
            acc = None
        bins.append(CalibrationBin(lower, upper, len(in_bin), mean_conf, acc))

    return CalibrationReport(bins=bins, n_samples=len(samples))


# --- Checklist completeness ------------------------------------------------


@dataclass
class CompletenessReport:
    n_fields: int
    n_determined: int  # value present, legitimately not-applicable, or actively flagged not-found

    @property
    def completeness(self) -> Optional[float]:
        return self.n_determined / self.n_fields if self.n_fields else None


def checklist_completeness(fields: Mapping[str, ChecklistField]) -> CompletenessReport:
    """Coverage of the full nine-to-fourteen-field protocol: a field counts as
    determined if it has a value, is legitimately not-applicable (with a
    reason), or was actively flagged not-found -- docs/prd.md section 8.3's "silence
    is never the response to a miss". A field still sitting empty at
    NEEDS_REVIEW with no value is an undetermined gap."""
    n_determined = 0
    for f in fields.values():
        if f.value is not None or not f.applicable or f.status == FieldStatus.FLAGGED:
            n_determined += 1
    return CompletenessReport(n_fields=len(fields), n_determined=n_determined)


# --- Interobserver agreement (how noisy the reference itself is) ------------


@dataclass
class InterobserverReport:
    per_field_disagreement_rate: dict[str, float]  # fraction of multi-reference fields that disagree
    n_multi_reference_fields: int

    @property
    def overall_disagreement_rate(self) -> Optional[float]:
        rates = self.per_field_disagreement_rate
        if not rates:
            return None
        return sum(rates.values()) / len(rates)


def interobserver_agreement(
    ground_truth: Mapping[str, GroundTruthCase],
    *,
    numeric_tolerance: float = 0.0,
    numeric_tolerances: Optional[Mapping[str, float]] = None,
) -> InterobserverReport:
    """For every field that has more than one reference abstractor, how often
    do the abstractors disagree among themselves? Section 5: grade/histotype
    have documented interobserver disagreement, so a single reference should
    not be treated as absolute ground truth."""
    numeric_tolerances = numeric_tolerances or {}
    disagree: dict[str, int] = {}
    total: dict[str, int] = {}

    for gt_case in ground_truth.values():
        for name, gt_field in gt_case.fields.items():
            refs = gt_field.references
            if len(refs) < 2:
                continue
            total[name] = total.get(name, 0) + 1
            tol = numeric_tolerances.get(name, numeric_tolerance)
            all_agree = all(
                values_match(refs[0], other, numeric_tolerance=tol) for other in refs[1:]
            )
            if not all_agree:
                disagree[name] = disagree.get(name, 0) + 1

    rates = {name: disagree.get(name, 0) / n for name, n in total.items()}
    return InterobserverReport(per_field_disagreement_rate=rates, n_multi_reference_fields=len(total))


# --- Time saved vs a manual-abstraction baseline ---------------------------


@dataclass
class TimeSavedReport:
    manual_seconds_per_case: float
    assisted_seconds_per_case: float

    @property
    def seconds_saved_per_case(self) -> float:
        return self.manual_seconds_per_case - self.assisted_seconds_per_case

    @property
    def percent_saved(self) -> Optional[float]:
        if not self.manual_seconds_per_case:
            return None
        return self.seconds_saved_per_case / self.manual_seconds_per_case


def time_saved(
    manual_seconds: Iterable[float], assisted_seconds: Iterable[float]
) -> TimeSavedReport:
    """Mean manual-abstraction time vs mean tool-assisted review time per case.
    Inputs are measured elsewhere (a timing study); this only reduces them."""
    manual = list(manual_seconds)
    assisted = list(assisted_seconds)
    if not manual or not assisted:
        raise ValueError("time_saved needs at least one manual and one assisted timing")
    return TimeSavedReport(
        manual_seconds_per_case=sum(manual) / len(manual),
        assisted_seconds_per_case=sum(assisted) / len(assisted),
    )


# --- Aggregate + ground-truth I/O ------------------------------------------


@dataclass
class EvaluationReport:
    accuracy: AccuracyReport
    calibration: CalibrationReport
    interobserver: InterobserverReport


def evaluate(
    predictions: Mapping[str, CasePredictions],
    ground_truth: Mapping[str, GroundTruthCase],
    *,
    n_calibration_bins: int = 10,
    numeric_tolerance: float = 0.0,
    numeric_tolerances: Optional[Mapping[str, float]] = None,
) -> EvaluationReport:
    """Run accuracy, calibration, and interobserver scoring together over a
    corpus. Completeness and time-saved are reported per-case / from a timing
    study respectively, so they're separate calls rather than folded in here."""
    return EvaluationReport(
        accuracy=field_accuracy(
            predictions,
            ground_truth,
            numeric_tolerance=numeric_tolerance,
            numeric_tolerances=numeric_tolerances,
        ),
        calibration=confidence_calibration(
            predictions,
            ground_truth,
            n_bins=n_calibration_bins,
            numeric_tolerance=numeric_tolerance,
            numeric_tolerances=numeric_tolerances,
        ),
        interobserver=interobserver_agreement(
            ground_truth,
            numeric_tolerance=numeric_tolerance,
            numeric_tolerances=numeric_tolerances,
        ),
    )


def predictions_from_case(case: Case) -> dict[str, FieldPrediction]:
    """Adapter: turn an extracted Case's checklist into the prediction mapping
    this harness scores, so callers don't hand-build FieldPredictions."""
    return {
        name: FieldPrediction(value=f.value, confidence=f.confidence)
        for name, f in case.checklist.all_fields().items()
    }


def predictions_from_cases(cases: Iterable[Case]) -> dict[str, dict[str, FieldPrediction]]:
    return {case.case_barcode: predictions_from_case(case) for case in cases}


def load_ground_truth(path: Path | str) -> dict[str, GroundTruthCase]:
    """Parse the ground-truth JSON format documented in this module's
    docstring. Raises if the file is malformed, so a labeling mistake surfaces
    loudly rather than silently scoring against nothing."""
    data = json.loads(Path(path).read_text())
    out: dict[str, GroundTruthCase] = {}
    for case in data["cases"]:
        barcode = case["case_barcode"]
        fields = {
            name: GroundTruthField(
                references=list(spec["references"]),
                applicable=spec.get("applicable", True),
            )
            for name, spec in case["fields"].items()
        }
        out[barcode] = GroundTruthCase(case_barcode=barcode, fields=fields)
    return out


def ground_truth_template(case_barcodes: Iterable[str]) -> dict:
    """Emit the fillable ground-truth structure for a set of cases: every
    protocol field, references left empty for a labeler to fill. Lets the
    labeling pass the issue flags as a dependency start against a concrete
    format before any code consumes it."""
    field_names = list(EndometrialChecklist.model_fields)
    return {
        "cases": [
            {
                "case_barcode": barcode,
                "fields": {
                    name: {"references": [], "applicable": True} for name in field_names
                },
            }
            for barcode in case_barcodes
        ]
    }