rpx_benchmark.temporal_budget_sweep

  1"""Frame-budget sweep runner and degradation analysis for video tasks.
  2
  3Runs a video-task pipeline at multiple frame budgets (e.g. {50, 100, 150,
  4250}) on the same clips and produces a degradation analysis that answers:
  5
  6* **Temporal Cue Value (TCV):** per model × metric, how much does the full
  7  clip improve over the smallest budget?
  8* **Area Under Degradation Curve (AUDC):** integral of metric vs. budget,
  9  normalized — models with flat curves are budget-robust.
 10* **Critical Budget:** smallest budget where metric is within ``tol`` of
 11  full-budget performance.
 12* **Paired test:** Wilcoxon signed-rank across scenes at min vs. max budget.
 13
 14Design note
 15-----------
 16The frame-budget ablation is a **first-class diagnostic** unique to RPX
 17(no other benchmark offers it). It tests whether temporal cues actually
 18help each model and at what granularity. The sweep is task-agnostic:
 19any task that uses :class:`~rpx_benchmark.video_loader.VideoDepthDataset`
 20with ``frame_budget`` + ``sampling`` benefits.  Object tracking will
 21plug in through the same harness.
 22
 23The Φ MANOVA is computed **only** on the full-budget (canonical) run.
 24Budget ablations are reported as a separate degradation analysis.
 25"""
 26
 27from __future__ import annotations
 28
 29from dataclasses import dataclass, field
 30from pathlib import Path
 31from typing import Dict, List, Optional, Sequence, Tuple
 32
 33import numpy as np
 34
 35from .logging_utils import get_logger
 36
 37log = get_logger(__name__)
 38
 39__all__ = [
 40    "BudgetSweepConfig",
 41    "DegradationResult",
 42    "run_budget_sweep",
 43    "degradation_analysis",
 44]
 45
 46#: Default frame budgets for the temporal resolution ablation.
 47DEFAULT_BUDGETS: Tuple[int, ...] = (50, 100, 150, 250)
 48
 49
 50# --------------------------------------------------------------------------- #
 51# Config
 52# --------------------------------------------------------------------------- #
 53
 54
 55@dataclass
 56class BudgetSweepConfig:
 57    """Configuration for a frame-budget sweep.
 58
 59    Parameters
 60    ----------
 61    budgets : sequence of int
 62        Frame budgets to evaluate. The largest budget is treated as
 63        the canonical "full" run for degradation analysis. Example:
 64        ``(50, 100, 150, 250)``.
 65    sampling : str
 66        Sampling mode for frame selection. ``"stride"`` (default) gives
 67        uniform-with-endpoint-anchored selection — equivalent to
 68        "what if your camera ran at a lower FPS". ``"fps_se3"``
 69        selects frames that maximise spatial coverage of the camera path.
 70    metric_keys : list of str or None
 71        If provided, restrict the degradation analysis to these metric
 72        keys. If ``None``, analyze all numeric columns in the cell logs.
 73    tol : float
 74        Tolerance for "critical budget" — the smallest budget where
 75        metric is within ``tol`` (relative) of full-budget performance.
 76        Default: 0.05 (5%).
 77    """
 78
 79    budgets: Sequence[int] = DEFAULT_BUDGETS
 80    sampling: str = "stride"
 81    metric_keys: Optional[List[str]] = None
 82    tol: float = 0.05
 83
 84
 85# --------------------------------------------------------------------------- #
 86# Degradation result
 87# --------------------------------------------------------------------------- #
 88
 89
 90@dataclass
 91class DegradationResult:
 92    """Per-model × per-metric degradation analysis across frame budgets.
 93
 94    Attributes
 95    ----------
 96    model : str
 97    metric : str
 98    direction : str
 99        ``"lower"`` or ``"higher"`` — whether lower/higher is better.
100    budgets : list[int]
101        Frame budgets (ascending).
102    values : list[float]
103        Mean metric value at each budget (over scenes/phases).
104    per_scene_values : dict[int, list[float]]
105        Per-scene metric values at each budget (for paired tests).
106    tcv : float
107        Temporal Cue Value = metric@max_budget - metric@min_budget
108        (sign-adjusted so positive = temporal cues help).
109    audc : float
110        Area Under Degradation Curve, normalized by metric@max_budget.
111    critical_budget : int or None
112        Smallest budget where metric is within ``tol`` of full-budget.
113        ``None`` if no budget meets the tolerance.
114    wilcoxon_p : float or None
115        p-value of Wilcoxon signed-rank test (min vs max budget).
116        ``None`` if insufficient data.
117    """
118
119    model: str = ""
120    metric: str = ""
121    direction: str = "lower"
122    budgets: List[int] = field(default_factory=list)
123    values: List[float] = field(default_factory=list)
124    per_scene_values: Dict[int, List[float]] = field(default_factory=dict)
125    tcv: float = float("nan")
126    audc: float = float("nan")
127    critical_budget: Optional[int] = None
128    wilcoxon_p: Optional[float] = None
129
130
131# --------------------------------------------------------------------------- #
132# Degradation analysis
133# --------------------------------------------------------------------------- #
134
135
136def degradation_analysis(
137    cells_by_budget: Dict[int, List[dict]],
138    metric_keys: Optional[List[str]] = None,
139    tol: float = 0.05,
140    model_name: str = "model",
141) -> List[DegradationResult]:
142    """Compute degradation statistics from cell-log dicts at each budget.
143
144    Parameters
145    ----------
146    cells_by_budget : dict[int, list[dict]]
147        Mapping from frame_budget → list of cell-log row dicts. Each
148        dict must contain the metric columns + ``scene`` (or
149        ``scene_id``) + ``phase`` for scene-level pairing.
150    metric_keys : list[str] or None
151        Metrics to analyze. ``None`` means auto-detect all float columns.
152    tol : float
153        Critical-budget tolerance (default 5%).
154    model_name : str
155        Model name for the result objects.
156
157    Returns
158    -------
159    list[DegradationResult]
160        One result per metric, sorted by metric name.
161    """
162    if not cells_by_budget:
163        return []
164
165    budgets_sorted = sorted(cells_by_budget.keys())
166    max_budget = budgets_sorted[-1]
167    min_budget = budgets_sorted[0]
168
169    # Auto-detect metric keys from the max-budget cells.
170    sample_row = cells_by_budget[max_budget][0] if cells_by_budget[max_budget] else {}
171    if metric_keys is None:
172        skip = {"frame_budget", "latency_ms", "scene", "scene_id", "phase",
173                "difficulty", "id", "model", "task"}
174        metric_keys = sorted(
175            k for k, v in sample_row.items()
176            if isinstance(v, (int, float)) and not isinstance(v, bool)
177            and k not in skip
178        )
179
180    # Determine metric direction from specs registry (best-effort).
181    from .metrics.specs import get_spec, has_spec
182
183    results = []
184    for mkey in metric_keys:
185        direction = "lower"
186        if has_spec(mkey):
187            direction = get_spec(mkey).direction
188
189        # Collect per-budget mean and per-scene values.
190        means = []
191        per_scene: Dict[int, List[float]] = {}
192        for b in budgets_sorted:
193            vals = [
194                row[mkey] for row in cells_by_budget[b]
195                if mkey in row and isinstance(row[mkey], (int, float))
196                and np.isfinite(row[mkey])
197            ]
198            means.append(float(np.mean(vals)) if vals else float("nan"))
199            per_scene[b] = vals
200
201        # TCV: sign-adjusted difference (positive = temporal cues help).
202        val_max = means[-1]  # metric @ max_budget
203        val_min = means[0]   # metric @ min_budget
204        if direction == "lower":
205            # Lower is better → if max_budget is lower, temporal cues help.
206            tcv = val_min - val_max  # positive when full clip is better
207        else:
208            # Higher is better → if max_budget is higher, temporal cues help.
209            tcv = val_max - val_min
210
211        # AUDC: trapezoidal integral, normalized by full-budget metric.
212        areas = []
213        for i in range(len(budgets_sorted) - 1):
214            db = budgets_sorted[i + 1] - budgets_sorted[i]
215            avg_val = (means[i] + means[i + 1]) / 2.0
216            areas.append(db * avg_val)
217        total_area = sum(areas)
218        budget_range = budgets_sorted[-1] - budgets_sorted[0]
219        if abs(val_max) > 1e-12 and budget_range > 0:
220            audc = total_area / (budget_range * abs(val_max))
221        else:
222            audc = float("nan")
223
224        # Critical budget: smallest where metric is within tol of full.
225        critical = None
226        for i, b in enumerate(budgets_sorted):
227            if abs(val_max) < 1e-12:
228                break
229            rel_diff = abs(means[i] - val_max) / abs(val_max)
230            if rel_diff <= tol:
231                critical = b
232                break
233
234        # Wilcoxon signed-rank test (min vs max budget, scene-paired).
235        wilcoxon_p = None
236        if len(per_scene[min_budget]) >= 5 and len(per_scene[max_budget]) >= 5:
237            # Pair by position (assumes same scene ordering).
238            n = min(len(per_scene[min_budget]), len(per_scene[max_budget]))
239            diffs = np.array(per_scene[max_budget][:n]) - np.array(per_scene[min_budget][:n])
240            diffs = diffs[np.isfinite(diffs)]
241            if len(diffs) >= 5 and np.any(diffs != 0):
242                try:
243                    from scipy.stats import wilcoxon  # type: ignore[import-untyped]
244
245                    _, wilcoxon_p = wilcoxon(diffs, alternative="two-sided")
246                    wilcoxon_p = float(wilcoxon_p)
247                except ImportError:
248                    log.debug("scipy not available — Wilcoxon test skipped")
249
250        results.append(DegradationResult(
251            model=model_name,
252            metric=mkey,
253            direction=direction,
254            budgets=budgets_sorted,
255            values=means,
256            per_scene_values=per_scene,
257            tcv=tcv,
258            audc=audc,
259            critical_budget=critical,
260            wilcoxon_p=wilcoxon_p,
261        ))
262
263    return sorted(results, key=lambda r: r.metric)
264
265
266# --------------------------------------------------------------------------- #
267# Sweep runner (orchestrates multiple pipeline calls)
268# --------------------------------------------------------------------------- #
269
270
271def run_budget_sweep(
272    *,
273    run_fn,
274    make_config_fn,
275    budgets: Sequence[int] = DEFAULT_BUDGETS,
276    sampling: str = "stride",
277    model_name: str = "model",
278    metric_keys: Optional[List[str]] = None,
279    tol: float = 0.05,
280) -> Tuple[Dict[int, List[dict]], List[DegradationResult]]:
281    """Run a video task at multiple frame budgets and analyze degradation.
282
283    Parameters
284    ----------
285    run_fn : callable
286        ``run_fn(config) -> (result, dr_report, paths)`` — the task
287        pipeline function (e.g. ``run_video_depth``).
288    make_config_fn : callable
289        ``make_config_fn(budget, sampling) -> config`` — builds a
290        run config with the given ``frame_budget`` and ``sampling``.
291    budgets, sampling, model_name, metric_keys, tol :
292        Forwarded to :func:`degradation_analysis`.
293
294    Returns
295    -------
296    cells_by_budget : dict[int, list[dict]]
297        Raw cell-log dicts per budget.
298    degradation : list[DegradationResult]
299        One per metric.
300    """
301    cells_by_budget: Dict[int, List[dict]] = {}
302
303    for budget in sorted(budgets):
304        log.info("budget sweep: running at frame_budget=%d, sampling=%s", budget, sampling)
305        cfg = make_config_fn(budget, sampling)
306        result, _, paths = run_fn(cfg)
307
308        # Read back the cell-log from disk.
309        cells_path = paths.get("cells")
310        if cells_path and Path(cells_path).exists():
311            try:
312                import pyarrow.parquet as pq  # type: ignore[import-untyped]
313
314                table = pq.read_table(str(cells_path))
315                cells = table.to_pylist()
316            except ImportError:
317                log.warning("pyarrow not available — reading cells from result.per_sample")
318                cells = list(getattr(result, "per_sample", []))
319        else:
320            cells = list(getattr(result, "per_sample", []))
321
322        cells_by_budget[budget] = cells
323        log.info("budget sweep: budget=%d → %d cells collected", budget, len(cells))
324
325    analysis = degradation_analysis(
326        cells_by_budget,
327        metric_keys=metric_keys,
328        tol=tol,
329        model_name=model_name,
330    )
331
332    return cells_by_budget, analysis
333
334
335# --------------------------------------------------------------------------- #
336# Formatting helpers
337# --------------------------------------------------------------------------- #
338
339
340def format_degradation_markdown(results: List[DegradationResult]) -> str:
341    """Format degradation analysis as a Markdown summary table."""
342    if not results:
343        return "_No degradation results._\n"
344
345    lines = [
346        "## Frame-Budget Degradation Analysis\n",
347        "| Metric | Direction | Budgets | Values | TCV | AUDC | Critical Budget | Wilcoxon p |",
348        "|--------|-----------|---------|--------|-----|------|-----------------|------------|",
349    ]
350    for r in results:
351        vals_str = " → ".join(f"{v:.4f}" for v in r.values)
352        budgets_str = ", ".join(str(b) for b in r.budgets)
353        tcv_str = f"{r.tcv:+.4f}" if np.isfinite(r.tcv) else "—"
354        audc_str = f"{r.audc:.3f}" if np.isfinite(r.audc) else "—"
355        crit_str = str(r.critical_budget) if r.critical_budget is not None else "—"
356        p_str = f"{r.wilcoxon_p:.4f}" if r.wilcoxon_p is not None else "—"
357        lines.append(
358            f"| `{r.metric}` | {r.direction} | {budgets_str} "
359            f"| {vals_str} | {tcv_str} | {audc_str} | {crit_str} | {p_str} |"
360        )
361
362    lines.append("")
363    lines.append("**TCV** = Temporal Cue Value (positive = temporal cues help). "
364                 "**AUDC** = Area Under Degradation Curve (normalized). "
365                 "**Critical Budget** = smallest budget within 5% of full-budget.")
366    lines.append("")
367    return "\n".join(lines)