rpx_benchmark.phi_jedi_summary

  1"""Glue between the runner's per-sample metrics and Φ / JEDI.
  2
  3The runner produces a flat list of per-sample metric dicts annotated with
  4``scene``, ``phase``, and ``difficulty``. This module groups those rows into
  5a ``(N, P, K)`` tensor and calls :mod:`rpx_benchmark.phi` and
  6:mod:`rpx_benchmark.jedi` to produce a single :class:`PhiJediSummary`
  7payload that the runner attaches to the deployment-readiness report.
  8
  9Design notes
 10------------
 11
 12- **Complete-case analysis** (paper §3.2). Scenes that lack predictions for
 13  any phase are dropped from the MANOVA; ``n_eff`` records how many
 14  survived.
 15- **JEDI uses the same registered ``MetricSpec`` set as Φ**, but its
 16  direction/bounds handling is independent of the MANOVA sign-flipping.
 17  The convention paper §3.2 specifies is: lower-is-better metrics are
 18  negated *before standardisation* for Φ, separately from the
 19  direction-aware bounds used by JEDI.
 20- The runner already aggregates per-sample metrics into a ``BenchmarkResult``;
 21  this helper consumes the same list before aggregation so we can group
 22  cells without losing the (scene, phase) keying.
 23"""
 24
 25from __future__ import annotations
 26
 27from dataclasses import dataclass, field
 28from typing import Any, Dict, List, Mapping, Optional, Sequence
 29
 30import numpy as np
 31
 32from .exceptions import MetricError
 33from .jedi import JEDIModelResult, aggregate_jedi, compute_jedi
 34from .logging_utils import get_logger
 35from .metrics.specs import MetricSpec, get_spec, has_spec
 36from .phi import (
 37    PhiMixedDesign,
 38    PhiOneway,
 39    PhiTransition,
 40    compute_phi_mixed_design,
 41    compute_phi_oneway,
 42    compute_phi_per_transition,
 43    phi_interpretation,
 44    standardise,
 45)
 46
 47log = get_logger(__name__)
 48
 49
 50# Canonical phase order matches paper §3.1: Phase 0 = Clutter,
 51# Phase 1 = Interaction, Phase 2 = Clean. Alphabetical sort of the
 52# string labels would put Clean first, which silently mislabels the
 53# per-transition tests. We keep an explicit map so the runner's phase
 54# strings (rpx_benchmark.api.Phase values) always land on the right
 55# tensor axis.
 56CANONICAL_PHASE_ORDER: tuple[str, ...] = ("clutter", "interaction", "clean")
 57CANONICAL_PHASE_SHORT: dict[str, str] = {
 58    "clutter": "clu",
 59    "interaction": "int",
 60    "clean": "cln",
 61}
 62
 63
 64def _phase_sort_key(phase: str) -> tuple[int, str]:
 65    """Return a sort key that places canonical phases in paper order
 66    (Clutter → Interaction → Clean), with any unknown phase strings
 67    appended afterwards in alphabetical order (and clearly flagged in
 68    the summary's `notes`)."""
 69    try:
 70        return (CANONICAL_PHASE_ORDER.index(phase), phase)
 71    except ValueError:
 72        return (len(CANONICAL_PHASE_ORDER), phase)
 73
 74
 75@dataclass
 76class PhiJediSummary:
 77    """Bundled Φ + JEDI output for one (model, task) pair.
 78
 79    Attributes
 80    ----------
 81    phi : PhiOneway | None
 82        One-way Φ from repeated-measures MANOVA. ``None`` when the
 83        data shape rules out the test (e.g. K > N).
 84    phi_per_transition : Dict[str, PhiTransition]
 85        Pairwise Hotelling-T² Φ per transition. Empty if Φ could not
 86        be computed.
 87    phi_interpretation : str | None
 88        Cohen anchor label for the conservative Φ.
 89    jedi : JEDIModelResult | None
 90        Aggregated JEDI score across all (scene, phase) cells.
 91    metrics_used : List[str]
 92        The metric keys that fed the analysis (in the order they appear
 93        on each row of the ``(N, P, K)`` tensor).
 94    n_eff : int
 95        Number of scenes that survived complete-case filtering.
 96    n_dropped : int
 97        Number of scenes dropped because they lacked a complete
 98        ``(P, K)`` block.
 99    notes : List[str]
100        Free-text caveats surfaced during computation (e.g. "Pillai used
101        because Wilks Λ was non-finite").
102    """
103
104    phi: Optional[PhiOneway] = None
105    phi_per_transition: Dict[str, PhiTransition] = field(default_factory=dict)
106    phi_interpretation: Optional[str] = None
107    phi_mixed: Optional[PhiMixedDesign] = None
108    """Two-way mixed-design MANOVA: per-tier Φ + interaction effect.
109    ``None`` when scenes don't carry a ``difficulty`` tag."""
110    jedi: Optional[JEDIModelResult] = None
111    metrics_used: List[str] = field(default_factory=list)
112    n_eff: int = 0
113    n_dropped: int = 0
114    notes: List[str] = field(default_factory=list)
115    # Per-metric per-phase raw means, ``{metric: {phase_label: float}}``.
116    # Populated alongside the MANOVA tensor so the paper-table filler
117    # (``scripts/fill_paper_table.py``) can populate per-phase columns.
118    per_phase_metric_means: Dict[str, Dict[str, float]] = field(default_factory=dict)
119
120    def to_dict(self) -> Dict[str, Any]:
121        return {
122            "phi": self.phi.to_dict() if self.phi else None,
123            "phi_per_transition": {
124                k: v.to_dict() for k, v in self.phi_per_transition.items()
125            },
126            "phi_interpretation": self.phi_interpretation,
127            "phi_mixed": self.phi_mixed.to_dict() if self.phi_mixed else None,
128            "jedi": self.jedi.to_dict() if self.jedi else None,
129            "metrics_used": list(self.metrics_used),
130            "n_eff": self.n_eff,
131            "n_dropped": self.n_dropped,
132            "notes": list(self.notes),
133            "per_phase_metric_means": {
134                m: dict(by_phase) for m, by_phase in self.per_phase_metric_means.items()
135            },
136        }
137
138
139# --------------------------------------------------------------------------- #
140# Internals
141# --------------------------------------------------------------------------- #
142
143
144def _select_metric_keys(
145    per_sample_metrics: Sequence[Mapping[str, Any]],
146    explicit_keys: Optional[Sequence[str]],
147) -> List[str]:
148    """Pick which metric keys to feed into Φ / JEDI.
149
150    If ``explicit_keys`` is provided, validate that each has a registered
151    spec and that at least one row has a numeric value for it. Otherwise,
152    auto-detect: take the intersection of "numeric keys in row 0" and
153    "keys with registered specs", excluding metadata (phase, scene, etc.).
154    """
155    metadata = {"id", "phase", "difficulty", "scene", "latency_ms"}
156
157    if explicit_keys:
158        keys = list(explicit_keys)
159        for k in keys:
160            if not has_spec(k):
161                raise MetricError(
162                    f"Metric {k!r} has no registered MetricSpec",
163                    hint="Register a spec via rpx_benchmark.metrics.register_spec.",
164                )
165        return keys
166
167    if not per_sample_metrics:
168        return []
169    row0 = per_sample_metrics[0]
170    keys = []
171    for k, v in row0.items():
172        if k in metadata:
173            continue
174        if isinstance(v, bool):
175            continue
176        if not isinstance(v, (int, float)):
177            continue
178        if has_spec(k):
179            keys.append(k)
180    return keys
181
182
183def _group_by_scene_phase(
184    per_sample_metrics: Sequence[Mapping[str, Any]],
185    metric_keys: Sequence[str],
186) -> tuple[np.ndarray, List[str], List[Any], int, int, Dict[Any, str]]:
187    """Group per-sample rows into a (N, P, K) tensor.
188
189    Returns
190    -------
191    tensor : ndarray, shape (N_eff, P, K)
192        Per (scene, phase) means of each metric.
193    phase_order : List[str]
194        The phase labels in tensor-axis order. Sorted alphabetically by
195        the string value to keep results stable across runs.
196    scene_ids : List
197        The N_eff scene identifiers retained (complete-case).
198    n_dropped : int
199        Scenes dropped because they lacked at least one (scene, phase)
200        observation in every phase.
201    K : int
202        Number of metric columns kept.
203    """
204    K = len(metric_keys)
205    # Bucket rows by (scene, phase), keeping a list of per-row vectors so
206    # we can average across multiple samples of the same (scene, phase).
207    buckets: Dict[tuple, Dict[Any, List[List[float]]]] = {}
208    phases_seen: set = set()
209    scenes_seen: List[Any] = []
210    scene_difficulty: Dict[Any, str] = {}
211
212    for row in per_sample_metrics:
213        scene = row.get("scene")
214        phase = row.get("phase")
215        if scene is None or phase is None:
216            continue
217        vec: List[float] = []
218        complete = True
219        for k in metric_keys:
220            v = row.get(k)
221            if v is None or not isinstance(v, (int, float)) or isinstance(v, bool):
222                complete = False
223                break
224            vec.append(float(v))
225        if not complete:
226            continue
227        phases_seen.add(phase)
228        if scene not in buckets:
229            buckets[scene] = {}
230            scenes_seen.append(scene)
231        buckets[scene].setdefault(phase, []).append(vec)
232        # Difficulty is assumed constant across phases of a scene (ESD
233        # design). First non-null wins.
234        if scene not in scene_difficulty:
235            diff = row.get("difficulty")
236            if diff is not None:
237                scene_difficulty[scene] = str(
238                    diff.value if hasattr(diff, "value") else diff
239                )
240
241    if not buckets or not phases_seen:
242        return np.zeros((0, 0, K)), [], [], 0, K, {}
243
244    # Canonical paper order: Clutter → Interaction → Clean.
245    # Unknown phase strings (if any) trail afterwards alphabetically.
246    phase_order = sorted((str(p) for p in phases_seen), key=_phase_sort_key)
247    P = len(phase_order)
248    if P < 2:
249        return np.zeros((0, P, K)), phase_order, [], 0, K, scene_difficulty
250
251    # Drop scenes that don't have at least one observation for every phase.
252    complete_scenes: List[Any] = []
253    cells: List[np.ndarray] = []
254    for scene in scenes_seen:
255        cell = np.empty((P, K), dtype=np.float64)
256        ok = True
257        for i, p in enumerate(phase_order):
258            rows = buckets[scene].get(p)
259            if not rows:
260                ok = False
261                break
262            cell[i] = np.mean(rows, axis=0)
263        if ok:
264            cells.append(cell)
265            complete_scenes.append(scene)
266
267    n_eff = len(complete_scenes)
268    n_dropped = len(scenes_seen) - n_eff
269    if n_eff == 0:
270        return np.zeros((0, P, K)), phase_order, [], n_dropped, K, scene_difficulty
271
272    tensor = np.stack(cells, axis=0)  # (N_eff, P, K)
273    return tensor, phase_order, complete_scenes, n_dropped, K, scene_difficulty
274
275
276def _orient_for_phi(
277    tensor: np.ndarray,
278    metric_keys: Sequence[str],
279    specs: Mapping[str, MetricSpec],
280) -> np.ndarray:
281    """Flip lower-is-better columns so higher consistently means better.
282
283    Paper §3.2: *"Lower-is-better metrics are negated before
284    standardisation so that higher values consistently indicate better
285    performance."* The negation is for Φ only; JEDI handles direction
286    via its own bounds policy.
287    """
288    out = tensor.copy()
289    for j, key in enumerate(metric_keys):
290        spec = specs[key]
291        if spec.direction == "lower":
292            out[:, :, j] = -out[:, :, j]
293    return out
294
295
296# --------------------------------------------------------------------------- #
297# Public entry point
298# --------------------------------------------------------------------------- #
299
300
301def summarize_phi_jedi(
302    per_sample_metrics: Sequence[Mapping[str, Any]],
303    *,
304    metric_keys: Optional[Sequence[str]] = None,
305    jedi_epsilon: float = 0.0,
306    apply_holm: bool = True,
307) -> PhiJediSummary:
308    """Compute Φ + JEDI for one (model, task) run.
309
310    Parameters
311    ----------
312    per_sample_metrics : sequence of dict
313        The runner's per-sample metric rows. Each row must carry
314        numeric metric keys plus ``scene`` and ``phase`` metadata.
315    metric_keys : sequence of str, optional
316        Explicit metric keys to feed into the analysis (in canonical
317        order). When ``None``, auto-detected as the numeric keys in
318        row 0 that have registered :class:`MetricSpec`s.
319    jedi_epsilon : float
320        Lower floor on individual desirabilities (paper §3.3). Default
321        ``0.0`` (strict).
322    apply_holm : bool
323        Pass through to :func:`compute_phi_per_transition` for the
324        Holm-Bonferroni adjustment of the three pairwise tests.
325
326    Returns
327    -------
328    PhiJediSummary
329        Populated as fully as the data allows; ``phi`` may be ``None``
330        when N_eff is too small or K is too large.
331    """
332    metric_keys = _select_metric_keys(per_sample_metrics, metric_keys)
333    if not metric_keys:
334        return PhiJediSummary(notes=["No registered metric keys found in inputs."])
335
336    specs = {k: get_spec(k) for k in metric_keys}
337
338    tensor, phase_order, scene_ids, n_dropped, K, scene_difficulty = _group_by_scene_phase(
339        per_sample_metrics, metric_keys
340    )
341    n_eff = tensor.shape[0]
342
343    summary = PhiJediSummary(
344        metrics_used=list(metric_keys),
345        n_eff=n_eff,
346        n_dropped=n_dropped,
347    )
348
349    # Per-phase per-metric raw means: average each metric column across the
350    # N_eff scenes within a phase. Drives the per-phase columns of the
351    # paper's per-task tables.
352    if n_eff > 0 and tensor.shape[1] >= 1:
353        for j, key in enumerate(metric_keys):
354            by_phase: Dict[str, float] = {}
355            for i, phase_label in enumerate(phase_order):
356                by_phase[str(phase_label)] = float(tensor[:, i, j].mean())
357            summary.per_phase_metric_means[key] = by_phase
358
359    if n_eff < 2 or tensor.shape[1] < 2:
360        summary.notes.append(
361            f"Insufficient data for Φ: N_eff={n_eff}, P={tensor.shape[1]} (need ≥2 of each)."
362        )
363        # JEDI can still run per-cell even if Φ can't.
364    elif n_eff <= K:
365        summary.notes.append(
366            f"K={K} ≥ N_eff={n_eff}; MANOVA singular. Φ skipped."
367        )
368    else:
369        z = standardise(_orient_for_phi(tensor, metric_keys, specs))
370        try:
371            # Passing metric_keys lets phi.py's collinearity guard name the
372            # offending pair (e.g. "absrel vs rmse |r|=0.98") in warnings
373            # and error hints, instead of falling back to column indices.
374            phi = compute_phi_oneway(z, metric_names=list(metric_keys))
375            summary.phi = phi
376            summary.phi_interpretation = phi_interpretation(phi.phi_conservative)
377        except MetricError as e:
378            summary.notes.append(f"compute_phi_oneway failed: {e}")
379        try:
380            raw_transitions = compute_phi_per_transition(z, apply_holm=apply_holm)
381            # Translate "0->1" / "1->2" / "0->2" into phase-name labels
382            # ("clu->int" etc.) using the canonical phase_order so the
383            # JSON output is unambiguous and matches paper §3.2.
384            relabelled: Dict[str, PhiTransition] = {}
385            for key, t in raw_transitions.items():
386                try:
387                    p1_idx, p2_idx = (int(x) for x in key.split("->"))
388                    p1_name = phase_order[p1_idx]
389                    p2_name = phase_order[p2_idx]
390                    short_p1 = CANONICAL_PHASE_SHORT.get(p1_name, p1_name)
391                    short_p2 = CANONICAL_PHASE_SHORT.get(p2_name, p2_name)
392                    new_key = f"{short_p1}->{short_p2}"
393                except (ValueError, IndexError):
394                    new_key = key
395                relabelled[new_key] = PhiTransition(
396                    label=new_key,
397                    phi=t.phi,
398                    eta2=t.eta2,
399                    T2=t.T2,
400                    p_value=t.p_value,
401                    p_holm=t.p_holm,
402                    n_eff=t.n_eff,
403                )
404            summary.phi_per_transition = relabelled
405        except MetricError as e:
406            summary.notes.append(f"compute_phi_per_transition failed: {e}")
407        # Flag unknown phase strings if any leaked through.
408        unknown = [p for p in phase_order if p not in CANONICAL_PHASE_ORDER]
409        if unknown:
410            summary.notes.append(
411                f"non-canonical phase labels present: {unknown}; "
412                "transition labels include them with their raw names."
413            )
414
415        # Two-way mixed-design MANOVA: per-tier Φ + interaction effect.
416        # Only runs when every retained scene has a `difficulty` label and
417        # at least two tiers are present.
418        tier_by_scene = [scene_difficulty.get(s) for s in scene_ids]
419        if all(t is not None for t in tier_by_scene) and len(set(tier_by_scene)) >= 2:
420            try:
421                summary.phi_mixed = compute_phi_mixed_design(z, [t for t in tier_by_scene if t is not None])
422            except MetricError as e:
423                summary.notes.append(f"compute_phi_mixed_design failed: {e}")
424        else:
425            missing = sum(1 for t in tier_by_scene if t is None)
426            n_tiers = len(set(t for t in tier_by_scene if t is not None))
427            if missing > 0:
428                summary.notes.append(
429                    f"mixed-design MANOVA skipped: {missing}/{n_eff} scenes "
430                    "lack a `difficulty` tier; ESD tier file probably absent."
431                )
432            elif n_tiers < 2:
433                summary.notes.append(
434                    f"mixed-design MANOVA skipped: only {n_tiers} tier(s) present."
435                )
436
437    # JEDI per cell (doesn't need standardisation; uses raw values via specs).
438    if n_eff > 0 and tensor.shape[1] >= 1:
439        per_cell: List[tuple[str, Any]] = []
440        for i in range(n_eff):
441            for j, phase_label in enumerate(phase_order):
442                raw = {k: float(tensor[i, j, idx]) for idx, k in enumerate(metric_keys)}
443                try:
444                    r = compute_jedi(raw, epsilon=jedi_epsilon, specs=specs)
445                except MetricError as e:
446                    summary.notes.append(
447                        f"compute_jedi failed on scene={scene_ids[i]} phase={phase_label}: {e}"
448                    )
449                    continue
450                per_cell.append((str(phase_label), r))
451        if per_cell:
452            summary.jedi = aggregate_jedi(per_cell)
453
454    return summary
455
456
457__all__ = [
458    "PhiJediSummary",
459    "summarize_phi_jedi",
460]