picarones/evaluation/metrics/robustness_projection.py

"""Projection de robustesse synthétique sur le corpus réel —
Sprint 81 (A.I.8).

Sprint 81 — A.I.8 du plan d'évolution 2026.

Pourquoi ce module
------------------
Le module ``picarones/core/robustness.py`` (Sprint 8) génère des
courbes CER vs niveau de dégradation **synthétique** (bruit, flou,
rotation, résolution).  ``picarones/core/image_quality.py`` mesure
le bruit/flou/contraste **réels** des images du corpus.  Ce
sprint **projette** les caractéristiques réelles sur les courbes
synthétiques pour estimer le **déficit attendu de CER** sur le
corpus dans son état actuel.

Lecture concrète
----------------
*« 30 % de vos documents ont un bruit équivalent à σ=15 où
Tesseract perd 8 points de CER — soit un déficit attendu global
de 2,4 points (30 % × 8 points). »*

Méthode
-------
1. Pour chaque document, on extrait la valeur de qualité réelle
   (``noise_level``, ``blur_score``, ``contrast_score``…) depuis
   ``ImageQualityResult``.
2. Pour chaque type de dégradation, on interpole linéairement la
   ``DegradationCurve`` synthétique : CER attendu à ce niveau.
3. On agrège : CER moyen attendu, % docs au-dessus du seuil
   critique de la courbe, déficit projeté = CER_attendu -
   CER_baseline (niveau nul).

Sortie
------
``project_robustness_on_corpus(curves, image_qualities)`` retourne
``{engine_name: {degradation_type: {expected_cer_mean,
deficit_vs_baseline, n_docs_above_critical, n_docs}}}``.

Limites
-------
- Mapping ``image_quality → degradation level`` : on suppose que
  ``noise_level`` (ImageQualityResult) correspond à σ
  (DegradationCurve), et idem pour ``blur_score`` ↔ rayon de
  flou.  Si un corpus expose ces valeurs avec une échelle
  différente, le mapping est documenté et l'utilisateur peut
  passer ``quality_to_level`` custom.
- Interpolation **linéaire** entre les points de la courbe.  Au-
  delà des bornes, on **clip** au point extrême (pas
  d'extrapolation hasardeuse).
"""

from __future__ import annotations

import logging
import statistics
from typing import Callable, Iterable, Optional

logger = logging.getLogger(__name__)


# Mapping par défaut entre attributs ImageQualityResult et types
# de dégradation synthétique.  L'utilisateur peut passer un dict
# custom pour modifier ce mapping.
_DEFAULT_QUALITY_FIELD: dict[str, str] = {
    "noise":      "noise_level",       # σ
    "blur":       "blur_score",        # Variance laplacienne (inverse)
    "contrast":   "contrast_score",
    "rotation":   "rotation_angle",
    "resolution": "resolution_score",  # peut être absent
}


def _interpolate_cer(
    levels: list[float],
    cer_values: list[Optional[float]],
    target_level: float,
) -> Optional[float]:
    """Interpolation linéaire : retourne CER attendu à
    ``target_level``.

    - Si ``target_level`` est en-dessous du minimum de levels,
      retourne le CER au minimum (clip).
    - Si au-dessus du maximum, retourne le CER au maximum.
    - Sinon, interpolation linéaire entre les deux points
      encadrants.
    - Retourne ``None`` si aucun ``cer_value`` valide.
    """
    if not levels:
        return None
    # Filtrer les paires (level, cer) où cer est None
    pairs = [
        (lvl, cer) for lvl, cer in zip(levels, cer_values)
        if cer is not None
    ]
    if not pairs:
        return None
    pairs.sort(key=lambda p: p[0])
    # Clip
    if target_level <= pairs[0][0]:
        return pairs[0][1]
    if target_level >= pairs[-1][0]:
        return pairs[-1][1]
    # Interpolation
    for i in range(len(pairs) - 1):
        lo_lvl, lo_cer = pairs[i]
        hi_lvl, hi_cer = pairs[i + 1]
        if lo_lvl <= target_level <= hi_lvl:
            if hi_lvl == lo_lvl:
                return lo_cer
            ratio = (target_level - lo_lvl) / (hi_lvl - lo_lvl)
            return lo_cer + (hi_cer - lo_cer) * ratio
    return None  # ne devrait pas arriver


def _extract_quality_value(
    quality: dict, degradation_type: str,
    custom_mapping: Optional[dict[str, str]] = None,
) -> Optional[float]:
    """Extrait la valeur de qualité pertinente pour un type de
    dégradation depuis un ``ImageQualityResult.as_dict()``."""
    mapping = custom_mapping or _DEFAULT_QUALITY_FIELD
    field = mapping.get(degradation_type)
    if field is None:
        return None
    value = quality.get(field)
    if value is None:
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def project_robustness_on_corpus(
    curves: Iterable,
    image_qualities: list[dict],
    *,
    quality_to_level: Optional[Callable[[dict, str], Optional[float]]] = None,
    critical_threshold: Optional[float] = None,
) -> dict:
    """Projette les courbes de robustesse sur les qualités réelles.

    Parameters
    ----------
    curves:
        Itérable de ``DegradationCurve`` (ou dicts compatibles
        avec ``engine_name``, ``degradation_type``, ``levels``,
        ``cer_values``, ``critical_threshold_level``).
    image_qualities:
        Liste de dicts ``ImageQualityResult.as_dict()`` (un par
        document).  Si vide, retourne une projection vide.
    quality_to_level:
        Fonction custom ``(quality_dict, degradation_type) →
        Optional[float]`` pour adapter le mapping qualité→niveau.
        Par défaut, utilise ``_DEFAULT_QUALITY_FIELD``.
    critical_threshold:
        Override pour le seuil critique de CER (défaut : utilise
        ``DegradationCurve.cer_threshold``).

    Returns
    -------
    dict
        ``{
            engine_name: {
                degradation_type: {
                    "n_docs": int,
                    "n_docs_with_data": int,    # qualité disponible
                    "expected_cer_mean": float, # moyenne CER attendu
                    "expected_cer_median": float,
                    "baseline_cer": float,      # CER à niveau min
                    "deficit_vs_baseline": float,
                    "n_docs_above_critical": int,
                    "critical_threshold_level": float | None,
                    "critical_threshold_cer": float,
                },
            },
        }``
    """
    extractor = quality_to_level or (
        lambda q, dt: _extract_quality_value(q, dt)
    )
    out: dict[str, dict] = {}

    for curve in curves:
        # Accepter dict ou DegradationCurve
        if hasattr(curve, "as_dict"):
            data = curve.as_dict()
        else:
            data = curve
        engine = data.get("engine_name")
        deg_type = data.get("degradation_type")
        levels = data.get("levels") or []
        cer_values = data.get("cer_values") or []
        crit_lvl = data.get("critical_threshold_level")
        crit_cer = (
            critical_threshold
            if critical_threshold is not None
            else data.get("cer_threshold", 0.20)
        )
        if not engine or not deg_type:
            continue

        per_doc_cer: list[float] = []
        n_docs_with_data = 0
        n_above_critical = 0
        for quality in image_qualities:
            level = extractor(quality, deg_type)
            if level is None:
                continue
            n_docs_with_data += 1
            cer = _interpolate_cer(levels, cer_values, level)
            if cer is None:
                continue
            per_doc_cer.append(cer)
            if cer > crit_cer:
                n_above_critical += 1

        if not per_doc_cer:
            continue

        # Baseline = CER au niveau minimum (sans dégradation)
        baseline = _interpolate_cer(
            levels, cer_values,
            min(levels) if levels else 0.0,
        )
        expected_mean = statistics.fmean(per_doc_cer)
        expected_median = statistics.median(per_doc_cer)
        deficit = (
            expected_mean - baseline
            if baseline is not None else None
        )

        out.setdefault(engine, {})[deg_type] = {
            "n_docs": len(image_qualities),
            "n_docs_with_data": n_docs_with_data,
            "expected_cer_mean": expected_mean,
            "expected_cer_median": expected_median,
            "baseline_cer": baseline,
            "deficit_vs_baseline": deficit,
            "n_docs_above_critical": n_above_critical,
            "critical_threshold_level": crit_lvl,
            "critical_threshold_cer": crit_cer,
        }
    return out


def aggregate_projection_per_engine(projection: dict) -> dict:
    """Pour chaque moteur, agrège le déficit projeté en sommant
    sur tous les types de dégradation.

    Lecture : *« déficit total attendu pour Tesseract = 5,2 points
    de CER si on considère les 4 dégradations indépendamment »*.

    Note : la sommation **suppose l'indépendance** des
    dégradations, ce qui n'est pas strictement vrai mais reste
    une approximation utile pour le diagnostic.
    """
    out: dict[str, dict] = {}
    for engine, per_type in projection.items():
        total_deficit = 0.0
        n_types_with_data = 0
        max_deficit_type: Optional[tuple[str, float]] = None
        for deg_type, stats in per_type.items():
            deficit = stats.get("deficit_vs_baseline")
            if deficit is None:
                continue
            total_deficit += deficit
            n_types_with_data += 1
            if max_deficit_type is None or deficit > max_deficit_type[1]:
                max_deficit_type = (deg_type, deficit)
        out[engine] = {
            "total_expected_deficit": total_deficit,
            "n_degradation_types": n_types_with_data,
            "worst_degradation_type": (
                max_deficit_type[0] if max_deficit_type else None
            ),
            "worst_degradation_deficit": (
                max_deficit_type[1] if max_deficit_type else None
            ),
        }
    return out


__all__ = [
    "project_robustness_on_corpus",
    "aggregate_projection_per_engine",
]