rpx_benchmark.paper_tracking_analysis

  1"""Paper-level statistical analysis for RPX D3 tracking cells."""
  2
  3from __future__ import annotations
  4
  5import hashlib
  6import json
  7from pathlib import Path
  8from typing import Any, Mapping, Sequence
  9
 10import numpy as np
 11
 12from .exceptions import ConfigError, DatasetError
 13from .metrics.tracking_paper import PAPER_TRACKING_METRICS
 14from .paper_depth_analysis import (
 15    DIFFICULTIES,
 16    PHASES,
 17    _holm_adjust,
 18    _hotelling_transition,
 19    _interaction_test,
 20    _nested_effect,
 21    _phase_contrasts,
 22)
 23
 24TRACKING_DIRECTIONS = {
 25    "hota": "higher",
 26    "deta": "higher",
 27    "assa": "higher",
 28    "idf1": "higher",
 29    "mota": "higher",
 30    "idsw": "lower",
 31}
 32EGO_SPLIT_CELLS = {"easy": 33, "medium": 33, "hard": 34}
 33_PHASE_NAMES = {0: "clutter", 1: "interaction", 2: "clean"}
 34
 35
 36def _phase_test(cube: np.ndarray) -> dict[str, float]:
 37    n_scenes, n_phases, n_metrics = cube.shape
 38    if n_phases != 3 or n_metrics != len(PAPER_TRACKING_METRICS):
 39        raise ConfigError(f"Unexpected D3 MANOVA cube shape: {cube.shape}")
 40    responses = cube.reshape(n_scenes * n_phases, n_metrics)
 41    subjects = np.repeat(np.eye(n_scenes), n_phases, axis=0)
 42    phase = np.tile(_phase_contrasts(), (n_scenes, 1))
 43    return _nested_effect(
 44        responses,
 45        subjects,
 46        np.column_stack((subjects, phase)),
 47        df_h=2,
 48    )
 49
 50
 51def _load_bounds(path: Path | None) -> tuple[dict[str, Any] | None, str | None]:
 52    if path is None:
 53        return None, None
 54    raw = path.read_bytes()
 55    value = json.loads(raw)
 56    if not value.get("version") or not value.get("provenance"):
 57        raise ConfigError("D3 JEDI bounds require version and provenance.")
 58    epsilon = value.get("epsilon")
 59    if not isinstance(epsilon, (int, float)) or not 0 < float(epsilon) <= 1:
 60        raise ConfigError("D3 JEDI epsilon must be in (0, 1].")
 61    metrics = value.get("metrics")
 62    if not isinstance(metrics, dict) or set(metrics) != set(PAPER_TRACKING_METRICS):
 63        raise ConfigError(
 64            f"D3 JEDI bounds must define exactly {list(PAPER_TRACKING_METRICS)}."
 65        )
 66    for metric, direction in TRACKING_DIRECTIONS.items():
 67        spec = metrics[metric]
 68        if spec.get("direction") != direction:
 69            raise ConfigError(f"D3 JEDI direction mismatch for {metric}.")
 70        best, worst = spec.get("best"), spec.get("worst")
 71        if not isinstance(best, (int, float)) or not isinstance(worst, (int, float)):
 72            raise ConfigError(f"D3 JEDI bounds for {metric} must be numeric.")
 73        if not np.isfinite([best, worst]).all():
 74            raise ConfigError(f"D3 JEDI bounds for {metric} must be finite.")
 75        if (direction == "higher" and best <= worst) or (
 76            direction == "lower" and best >= worst
 77        ):
 78            raise ConfigError(f"D3 JEDI bounds contradict direction for {metric}.")
 79    return value, hashlib.sha256(raw).hexdigest()
 80
 81
 82def _jedi(values: Mapping[str, float], bounds: Mapping[str, Any]) -> dict[str, Any]:
 83    desirabilities: dict[str, float] = {}
 84    for metric in PAPER_TRACKING_METRICS:
 85        spec = bounds["metrics"][metric]
 86        desirability = (values[metric] - spec["worst"]) / (spec["best"] - spec["worst"])
 87        desirabilities[metric] = float(np.clip(desirability, 0, 1))
 88    epsilon = float(bounds["epsilon"])
 89    score = float(
 90        np.exp(np.mean(np.log([max(value, epsilon) for value in desirabilities.values()])))
 91    )
 92    return {"score": score, "desirabilities": desirabilities}
 93
 94
 95def analyze_tracking_cells(
 96    paths: Sequence[str | Path],
 97    *,
 98    jedi_bounds_path: str | Path | None = None,
 99) -> tuple[dict[str, Any], "Any"]:
100    import pandas as pd
101
102    frames = [pd.read_parquet(path) for path in paths]
103    cells = pd.concat(frames, ignore_index=True)
104    required = {
105        "model",
106        "task",
107        "split",
108        "scene",
109        "phase",
110        *(f"metric:{metric}" for metric in PAPER_TRACKING_METRICS),
111    }
112    missing = required - set(cells.columns)
113    if missing:
114        raise DatasetError(f"D3 cells are missing columns: {sorted(missing)}")
115    cells = cells[cells["task"] == "object_tracking"].copy()
116    if cells.empty:
117        raise DatasetError("No object_tracking cells found.")
118    models = set(cells["model"].astype(str))
119    if len(models) != 1:
120        raise DatasetError(f"D3 analysis requires one model, got {sorted(models)}.")
121    if cells.duplicated(["scene", "phase"]).any():
122        raise DatasetError("D3 cells contain duplicate scene-phase rows.")
123    if len(cells) != 300 or cells["scene"].nunique() != 100:
124        raise DatasetError(
125            f"D3 analysis requires 300 cells/100 scenes; got {len(cells)}/"
126            f"{cells['scene'].nunique()}."
127        )
128
129    scenes = sorted(cells["scene"].astype(str).unique())
130    raw = np.empty((100, 3, len(PAPER_TRACKING_METRICS)), dtype=np.float64)
131    difficulties: list[str] = []
132    for scene_index, scene in enumerate(scenes):
133        scene_rows = cells[cells["scene"].astype(str) == scene]
134        tiers = set(scene_rows["split"].astype(str))
135        if len(tiers) != 1 or next(iter(tiers)) not in DIFFICULTIES:
136            raise DatasetError(f"Invalid/changing difficulty for {scene}: {tiers}")
137        difficulties.append(next(iter(tiers)))
138        for phase in range(3):
139            phase_rows = scene_rows[scene_rows["phase"].astype(int) == phase]
140            if len(phase_rows) != 1:
141                raise DatasetError(f"Missing D3 cell for {scene}/phase {phase}.")
142            row = phase_rows.iloc[0]
143            for metric_index, metric in enumerate(PAPER_TRACKING_METRICS):
144                value = float(row[f"metric:{metric}"])
145                if not np.isfinite(value):
146                    raise DatasetError(f"Non-finite {metric} for {scene}/phase {phase}.")
147                raw[scene_index, phase, metric_index] = value
148
149    oriented = raw.copy()
150    for index, metric in enumerate(PAPER_TRACKING_METRICS):
151        if TRACKING_DIRECTIONS[metric] == "lower":
152            oriented[:, :, index] *= -1
153    flat = oriented.reshape(-1, len(PAPER_TRACKING_METRICS))
154    standard_deviation = flat.std(axis=0, ddof=0)
155    if np.any(standard_deviation <= 0):
156        raise DatasetError(
157            f"D3 analysis requires {len(PAPER_TRACKING_METRICS)} non-constant metrics."
158        )
159    standardized = ((flat - flat.mean(axis=0)) / standard_deviation).reshape(oriented.shape)
160
161    transitions: list[dict[str, Any]] = []
162    for label, first, second in (
163        ("clutter_to_interaction", 0, 1),
164        ("interaction_to_clean", 1, 2),
165        ("clutter_to_clean", 0, 2),
166    ):
167        transitions.append(
168            {
169                "transition": label,
170                **_hotelling_transition(standardized[:, first], standardized[:, second]),
171            }
172        )
173    adjusted = _holm_adjust([value["p_value_raw"] for value in transitions])
174    for transition, adjusted_p in zip(transitions, adjusted, strict=True):
175        transition["p_value_holm"] = adjusted_p
176
177    tier_array = np.asarray(difficulties)
178    per_tier = {
179        tier: {
180            "n_scenes": int(np.sum(tier_array == tier)),
181            **_phase_test(standardized[tier_array == tier]),
182        }
183        for tier in DIFFICULTIES
184    }
185    phase_means = {
186        _PHASE_NAMES[phase]: {
187            metric: float(raw[:, phase, index].mean())
188            for index, metric in enumerate(PAPER_TRACKING_METRICS)
189        }
190        for phase in range(3)
191    }
192    overall_means = {
193        metric: float(raw[:, :, index].mean())
194        for index, metric in enumerate(PAPER_TRACKING_METRICS)
195    }
196    bounds, bounds_sha = _load_bounds(Path(jedi_bounds_path) if jedi_bounds_path else None)
197    if bounds is None:
198        jedi: dict[str, Any] = {"status": "bounds_missing"}
199    else:
200        jedi = {
201            "status": "computed",
202            "bounds_version": bounds["version"],
203            "bounds_sha256": bounds_sha,
204            "overall": _jedi(overall_means, bounds),
205            "per_phase": {
206                phase: _jedi(values, bounds) for phase, values in phase_means.items()
207            },
208        }
209
210    analysis = {
211        "schema_version": "rpx-d3-paper-analysis-v1",
212        "model": next(iter(models)),
213        "metrics": list(PAPER_TRACKING_METRICS),
214        "directions": TRACKING_DIRECTIONS,
215        "n_scenes": 100,
216        "n_cells": 300,
217        "overall_phase_manova": _phase_test(standardized),
218        "transitions": transitions,
219        "per_tier": per_tier,
220        "phase_by_difficulty": _interaction_test(standardized, difficulties),
221        "metric_means": {"overall": overall_means, "per_phase": phase_means},
222        "jedi": jedi,
223        "jedi_status": jedi["status"],
224    }
225    return analysis, cells
226
227
228def analyze_ego_tracking_cells(
229    paths: Sequence[str | Path],
230) -> tuple[dict[str, Any], "Any"]:
231    """Aggregate the one-clip-per-scene RPX ego tracking protocol."""
232
233    import pandas as pd
234
235    cells = pd.concat([pd.read_parquet(path) for path in paths], ignore_index=True)
236    required = {
237        "model",
238        "task",
239        "split",
240        "scene",
241        "phase",
242        *(f"metric:{metric}" for metric in PAPER_TRACKING_METRICS),
243    }
244    missing = required - set(cells.columns)
245    if missing:
246        raise DatasetError(f"Ego tracking cells are missing columns: {sorted(missing)}")
247    cells = cells[cells["task"] == "object_tracking"].copy()
248    if "dataset_protocol" in cells:
249        cells = cells[cells["dataset_protocol"] == "ego"].copy()
250    if cells.empty:
251        raise DatasetError("No ego object_tracking cells found.")
252    models = set(cells["model"].astype(str))
253    if len(models) != 1:
254        raise DatasetError(
255            f"Ego tracking analysis requires one model, got {sorted(models)}."
256        )
257    if cells.duplicated(["scene", "phase"]).any():
258        raise DatasetError("Ego tracking cells contain duplicate scene-phase rows.")
259    split_counts = cells.groupby("split").size().to_dict()
260    if split_counts != EGO_SPLIT_CELLS or cells["scene"].nunique() != 100:
261        raise DatasetError(
262            "Ego analysis requires 100 scenes split 33/33/34; got "
263            f"{split_counts} across {cells['scene'].nunique()} scenes."
264        )
265
266    metric_columns = {metric: f"metric:{metric}" for metric in PAPER_TRACKING_METRICS}
267    for metric, column in metric_columns.items():
268        values = cells[column].astype(float).to_numpy()
269        if not np.isfinite(values).all():
270            raise DatasetError(f"Non-finite ego tracking metric: {metric}.")
271    metric_means = {
272        "overall": {
273            metric: float(cells[column].astype(float).mean())
274            for metric, column in metric_columns.items()
275        },
276        "per_difficulty": {
277            split: {
278                metric: float(rows[column].astype(float).mean())
279                for metric, column in metric_columns.items()
280            }
281            for split, rows in cells.groupby("split", sort=True)
282        },
283    }
284    hardware_columns = (
285        "latency_ms",
286        "throughput_fps",
287        "peak_gpu_memory_allocated_mb",
288        "peak_gpu_memory_reserved_mb",
289        "clip_wall_time_s",
290        "parameter_count",
291    )
292    hardware_means = {
293        column: float(cells[column].astype(float).mean())
294        for column in hardware_columns
295        if column in cells and np.isfinite(cells[column].astype(float)).all()
296    }
297    analysis = {
298        "schema_version": "rpx-ego-tracking-analysis-v1",
299        "dataset_protocol": "ego",
300        "model": next(iter(models)),
301        "metrics": list(PAPER_TRACKING_METRICS),
302        "directions": TRACKING_DIRECTIONS,
303        "n_scenes": 100,
304        "n_cells": len(cells),
305        "split_cells": split_counts,
306        "metric_means": metric_means,
307        "hardware_means": hardware_means,
308    }
309    return analysis, cells
310
311
312def write_ego_tracking_analysis(
313    analysis: Mapping[str, Any],
314    cells: Any,
315    output_dir: str | Path,
316) -> dict[str, Path]:
317    import pandas as pd
318
319    output = Path(output_dir)
320    output.mkdir(parents=True, exist_ok=True)
321    paths = {
322        "combined_cells": output / "combined_cells.parquet",
323        "json": output / "paper_analysis.json",
324        "markdown": output / "paper_analysis.md",
325        "table": output / "paper_table.csv",
326    }
327    cells.to_parquet(paths["combined_cells"], index=False)
328    paths["json"].write_text(
329        json.dumps(dict(analysis), indent=2, sort_keys=True) + "\n",
330        encoding="utf-8",
331    )
332    rows = []
333    for split, values in (
334        [("overall", analysis["metric_means"]["overall"])]
335        + list(analysis["metric_means"]["per_difficulty"].items())
336    ):
337        rows.append({"model": analysis["model"], "split": split, **values})
338    pd.DataFrame(rows).to_csv(paths["table"], index=False)
339    lines = [
340        "# RPX ego tracking analysis",
341        "",
342        f"- Model: `{analysis['model']}`",
343        f"- Clips/scenes: {analysis['n_cells']}",
344        "- Protocol: one ego video per scene",
345        "",
346        "| Split | HOTA | DetA | AssA | IDF1 | MOTA | ID switches |",
347        "|---|---:|---:|---:|---:|---:|---:|",
348    ]
349    for row in rows:
350        lines.append(
351            f"| {row['split']} | {row['hota']:.6g} | {row['deta']:.6g} | "
352            f"{row['assa']:.6g} | {row['idf1']:.6g} | "
353            f"{row['mota']:.6g} | {row['idsw']:.6g} |"
354        )
355    paths["markdown"].write_text("\n".join(lines) + "\n", encoding="utf-8")
356    return paths
357
358
359def write_tracking_analysis(
360    analysis: Mapping[str, Any],
361    cells: Any,
362    output_dir: str | Path,
363) -> dict[str, Path]:
364    import pandas as pd
365
366    output = Path(output_dir)
367    output.mkdir(parents=True, exist_ok=True)
368    paths = {
369        "combined_cells": output / "combined_cells.parquet",
370        "json": output / "paper_analysis.json",
371        "markdown": output / "paper_analysis.md",
372        "table": output / "paper_table.csv",
373    }
374    cells.to_parquet(paths["combined_cells"], index=False)
375    paths["json"].write_text(
376        json.dumps(dict(analysis), indent=2, sort_keys=True) + "\n",
377        encoding="utf-8",
378    )
379    overall = analysis["overall_phase_manova"]
380    table = {
381        "model": analysis["model"],
382        "phi": overall["phi_wilks"],
383        "wilks_lambda": overall["wilks_lambda"],
384        "p_value": overall["p_value"],
385        "partial_eta_squared": overall["partial_eta_squared"],
386        "jedi_status": analysis["jedi_status"],
387    }
388    pd.DataFrame([table]).to_csv(paths["table"], index=False)
389    lines = [
390        "# RPX D3 paper analysis",
391        "",
392        f"- Model: `{analysis['model']}`",
393        f"- Cells: {analysis['n_cells']} (100 scenes × 3 phases)",
394        f"- Φ: {overall['phi_wilks']:.6g}",
395        f"- Wilks’ Λ: {overall['wilks_lambda']:.6g}",
396        f"- Rao p: {overall['p_value']:.6g}",
397        f"- Partial η²: {overall['partial_eta_squared']:.6g}",
398        f"- JEDI: `{analysis['jedi_status']}`",
399        "",
400        "## Per-phase means",
401        "",
402        "| Phase | HOTA | DetA | AssA | IDF1 | MOTA | ID switches |",
403        "|---|---:|---:|---:|---:|---:|---:|",
404    ]
405    for phase in PHASES:
406        values = analysis["metric_means"]["per_phase"][phase]
407        lines.append(
408            f"| {phase} | {values['hota']:.6g} | {values['deta']:.6g} | "
409            f"{values['assa']:.6g} | {values['idf1']:.6g} | "
410            f"{values['mota']:.6g} | {values['idsw']:.6g} |"
411        )
412    paths["markdown"].write_text("\n".join(lines) + "\n", encoding="utf-8")
413    return paths