src/conformal.py

"""
Conformal gradability.

Point verdicts (PASS/ACCEPTABLE/FAIL) give no statistical guarantee. Split
conformal prediction turns the continuous quality score into a *calibrated*
decision: at a target error rate alpha, it emits a prediction set - {gradable},
{ungradable}, or {uncertain} (abstain / route to human) - with a finite-sample
coverage guarantee, provided a labelled calibration set.

Calibration can come from expert-labelled images, or, for demonstration, from a
controlled-degradation surrogate (pristine == gradable, heavily degraded ==
ungradable). The surrogate is clearly flagged; swap in real labels for a
defensible guarantee.
"""

from __future__ import annotations
import numpy as np


class ConformalGradability:
    def __init__(self, alpha=0.1):
        self.alpha = alpha
        self.qhat_g = None      # threshold on gradable nonconformity
        self.qhat_u = None
        self.fitted = False
        self.source = None

    def fit(self, scores, labels, source="labelled"):
        """scores in [0,1] (higher=better quality). labels: 1=gradable, 0=ungradable."""
        s = np.asarray(scores, float); y = np.asarray(labels, int)
        g, u = s[y == 1], s[y == 0]
        if len(g) < 5 or len(u) < 5:
            raise ValueError("need >=5 gradable and >=5 ungradable calibration points")
        # nonconformity: gradable class scored by (1 - quality); ungradable by quality
        ncf_g = 1.0 - g
        ncf_u = u
        n = len(g)
        k = int(np.ceil((n + 1) * (1 - self.alpha)))
        self.qhat_g = float(np.sort(ncf_g)[min(k - 1, n - 1)])
        n2 = len(u); k2 = int(np.ceil((n2 + 1) * (1 - self.alpha)))
        self.qhat_u = float(np.sort(ncf_u)[min(k2 - 1, n2 - 1)])
        self.fitted = True
        self.source = source
        return self

    def predict(self, score):
        """Return calibrated prediction set for one quality score in [0,1]."""
        if not self.fitted:
            return dict(set=["uncalibrated"], label="uncalibrated", score=score)
        s = float(score)
        in_g = (1.0 - s) <= self.qhat_g      # plausibly gradable
        in_u = s <= self.qhat_u              # plausibly ungradable
        if in_g and not in_u:
            lab, pset = "gradable", ["gradable"]
        elif in_u and not in_g:
            lab, pset = "ungradable", ["ungradable"]
        elif in_g and in_u:
            lab, pset = "uncertain", ["gradable", "ungradable"]
        else:
            lab, pset = "uncertain", ["gradable", "ungradable"]
        return dict(set=pset, label=lab, score=s,
                    coverage=1 - self.alpha, source=self.source)


def synthetic_calibration(reference_results, degrade_fn, alpha=0.1):
    """Build a demonstration calibrator from analysed reference images.

    reference_results: list of pipeline result bundles (need summary.composite).
    degrade_fn(rgb)->rgb applies a heavy degradation to manufacture ungradables.
    """
    scores, labels = [], []
    for r in reference_results:
        if "summary" not in r:
            continue
        scores.append(r["summary"]["composite"] / 100.0); labels.append(1)
    # manufacture ungradable examples by heavy degradation
    from .fov import detect_fov
    from .qc_metrics import compute_all_metrics
    from .quality_score import composite_score
    for r in reference_results:
        if "rgb" not in r:
            continue
        bad = degrade_fn(r["rgb"])
        f = detect_fov(bad)
        c = composite_score(compute_all_metrics(bad, f))["composite"] / 100.0
        scores.append(c); labels.append(0)
    cg = ConformalGradability(alpha=alpha)
    cg.fit(scores, labels, source="degradation surrogate (demo)")
    return cg