Buckets:
| """AIDE / Driver-WM protocol metrics. | |
| Shapes: pred, gt are [N, T_PRED, J, 2] pixel coordinates (gt may carry a 3rd channel = conf, | |
| which is ignored except for the secondary conf-masked metric). All metrics use ALL joints | |
| regardless of confidence (paper protocol). | |
| MPJPE mean over clips x horizons x joints of the Euclidean error (px) | |
| d-nMPJPE 100 * MPJPE / DIAG | |
| PCK@t fraction of (clip, horizon, joint) with error <= t * max(W,H) (paper normaliser, | |
| locked by Gate 0; t * DIAG variant reported as pck*_diag) | |
| HM subset top int(0.1 N) clips by mean over the HM_N_TRANSITIONS future frame-to-frame | |
| transitions and joints of the GT displacement | |
| Clip-level bootstrap CIs (default 10k resamples) and paired differences are provided. | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[2])) | |
| from aide.data import protocol as P # noqa: E402 | |
| # --------------------------------------------------------------------------- core | |
| def joint_errors(pred, gt): | |
| """[N,H,J] Euclidean errors in px.""" | |
| pred = np.asarray(pred, dtype=np.float64)[..., :2] | |
| gt = np.asarray(gt, dtype=np.float64)[..., :2] | |
| assert pred.shape == gt.shape, (pred.shape, gt.shape) | |
| return np.sqrt(((pred - gt) ** 2).sum(-1)) | |
| def mpjpe(err): | |
| return float(err.mean()) | |
| def dn_mpjpe(err): | |
| return 100.0 * mpjpe(err) / P.DIAG | |
| def pck(err, thr_frac, norm="max_side"): | |
| return 100.0 * float((err <= P.pck_threshold_px(thr_frac, norm)).mean()) | |
| def per_horizon(err): | |
| return err.mean(axis=(0, 2)) # [H] | |
| def per_part(err): | |
| return {k: float(err[:, :, idx].mean()) for k, idx in P.PARTS.items()} | |
| def motion_score(gt_future, n_transitions=P.HM_N_TRANSITIONS): | |
| """Per-clip mean GT displacement (px) over the last `n_transitions` frame pairs of the | |
| future block and all joints. gt_future: [N, T_PRED, J, >=2]; n_transitions=4 uses the | |
| 4 transitions inside the 5 future frames. For the sensitivity variant that also counts | |
| the last-observed -> first-future transition, pass the 6-frame block and n_transitions=5. | |
| """ | |
| g = np.asarray(gt_future, dtype=np.float64)[..., :2] | |
| d = np.sqrt(((g[:, 1:] - g[:, :-1]) ** 2).sum(-1)) # [N, T-1, J] | |
| d = d[:, -n_transitions:] | |
| return d.mean(axis=(1, 2)) | |
| def hm_mask(gt_future, n_top=None, n_transitions=P.HM_N_TRANSITIONS): | |
| """Boolean mask of the High-Motion subset and the implied threshold (min score in subset).""" | |
| s = motion_score(gt_future, n_transitions) | |
| n_top = P.hm_top(len(s)) if n_top is None else n_top | |
| order = np.argsort(-s, kind="stable") | |
| m = np.zeros(len(s), bool) | |
| m[order[:n_top]] = True | |
| thr = float(s[order[n_top - 1]]) if n_top > 0 else float("nan") | |
| return m, thr, s | |
| # --------------------------------------------------------------------------- classification | |
| def macro_f1(y_true, y_pred, n_classes=None): | |
| """Macro-F1 (%) over the classes present in y_true or y_pred (sklearn default).""" | |
| from sklearn.metrics import f1_score | |
| return 100.0 * float(f1_score(y_true, y_pred, average="macro", zero_division=0)) | |
| def accuracy(y_true, y_pred): | |
| return 100.0 * float((np.asarray(y_true) == np.asarray(y_pred)).mean()) | |
| # --------------------------------------------------------------------------- bootstrap | |
| def bootstrap_ci(per_clip, n_boot=10000, seed=0, alpha=0.05, stat="mean"): | |
| """Percentile CI of the mean (or of a statistic) of per-clip values by clip resampling. | |
| per_clip: [N] or [N, ...] (mean is taken over all trailing axes after resampling).""" | |
| v = np.asarray(per_clip, dtype=np.float64) | |
| if v.ndim > 1: | |
| v = v.reshape(len(v), -1).mean(1) | |
| n = len(v) | |
| rng = np.random.default_rng(seed) | |
| idx = rng.integers(0, n, size=(n_boot, n)) | |
| if stat == "mean": | |
| b = v[idx].mean(1) | |
| else: | |
| b = np.array([stat(v[i]) for i in idx]) | |
| return float(np.quantile(b, alpha / 2)), float(np.quantile(b, 1 - alpha / 2)) | |
| def paired_diff(per_clip_a, per_clip_b, n_boot=10000, seed=0): | |
| """mean(a - b) with bootstrap CI and the fraction of resamples with diff < 0.""" | |
| a = np.asarray(per_clip_a, dtype=np.float64).reshape(len(per_clip_a), -1).mean(1) | |
| b = np.asarray(per_clip_b, dtype=np.float64).reshape(len(per_clip_b), -1).mean(1) | |
| d = a - b | |
| lo, hi = bootstrap_ci(d, n_boot, seed) | |
| rng = np.random.default_rng(seed + 1) | |
| idx = rng.integers(0, len(d), size=(n_boot, len(d))) | |
| return {"mean": float(d.mean()), "ci_lo": lo, "ci_hi": hi, | |
| "p_lt0": float((d[idx].mean(1) < 0).mean())} | |
| # --------------------------------------------------------------------------- full evaluation | |
| def evaluate(pred, gt, n_boot=10000, seed=0, ref_pred=None, conf_min=0.1, hm_top=None, ref_preds=None): | |
| """Everything the harness reports. gt: [N,T_PRED,J,3] (x,y,conf). Returns a flat dict of | |
| {metric: {"value", "ci_lo", "ci_hi", "n"}} plus meta. | |
| ref_pred: zero-velocity reference -> diff_vs_ref_{all,hm}; ref_preds: {name: pred} -> diff_vs_<name>_{all,hm} | |
| (paired per-clip differences, negative = better than the reference). | |
| """ | |
| pred = np.asarray(pred, dtype=np.float64) | |
| gt = np.asarray(gt, dtype=np.float64) | |
| N = len(gt) | |
| err = joint_errors(pred, gt) # [N,H,J] | |
| clip_err = err.mean(axis=(1, 2)) # [N] | |
| out, meta = {}, {} | |
| def put(name, value, per_clip=None, n=N): | |
| lo, hi = (float("nan"), float("nan")) if per_clip is None else bootstrap_ci(per_clip, n_boot, seed) | |
| out[name] = {"value": float(value), "ci_lo": lo, "ci_hi": hi, "n": int(n)} | |
| put("mpjpe_all", mpjpe(err), clip_err) | |
| put("dn_mpjpe", dn_mpjpe(err), 100.0 * clip_err / P.DIAG) | |
| for t in P.PCK_THRESHOLDS: | |
| for norm, suf in (("max_side", ""), ("diag", "_diag")): | |
| thr = P.pck_threshold_px(t, norm) | |
| put(f"pck{int(round(t * 100)):02d}{suf}", pck(err, t, norm), 100.0 * (err <= thr).mean(axis=(1, 2))) | |
| ph = per_horizon(err) | |
| for h in range(err.shape[1]): | |
| put(f"mpjpe_h{h + 1}", ph[h], err[:, h].mean(1)) | |
| for k, v in per_part(err).items(): | |
| put(f"mpjpe_{k}", v, err[:, :, P.PARTS[k]].mean(axis=(1, 2))) | |
| # secondary: only GT joints with conf >= conf_min | |
| cm = gt[..., 2] >= conf_min | |
| if cm.any(): | |
| w = cm.sum(axis=(1, 2)) | |
| pc = np.where(w > 0, (err * cm).sum(axis=(1, 2)) / np.maximum(w, 1), np.nan) | |
| put(f"mpjpe_conf{conf_min:g}", float(err[cm].mean()), pc[~np.isnan(pc)], n=int((~np.isnan(pc)).sum())) | |
| meta["conf_mask_fraction"] = float(cm.mean()) | |
| # High-Motion subset | |
| m, thr, score = hm_mask(gt, n_top=hm_top) | |
| nh = int(m.sum()) | |
| meta["hm_threshold"] = thr | |
| meta["hm_n"] = nh | |
| meta["hm_clip_ids_idx"] = np.flatnonzero(m).tolist() | |
| if nh > 0: | |
| put("mpjpe_hm", mpjpe(err[m]), clip_err[m], n=nh) | |
| put("dn_mpjpe_hm", dn_mpjpe(err[m]), 100.0 * clip_err[m] / P.DIAG, n=nh) | |
| for t in P.PCK_THRESHOLDS: | |
| put(f"pck{int(round(t * 100)):02d}_hm", pck(err[m], t), | |
| 100.0 * (err[m] <= P.pck_threshold_px(t)).mean(axis=(1, 2)), n=nh) | |
| phm = per_horizon(err[m]) | |
| for h in range(err.shape[1]): | |
| put(f"hm_h{h + 1}", phm[h], err[m, h].mean(1), n=nh) | |
| for k, v in per_part(err[m]).items(): | |
| put(f"hm_{k}", v, err[m][:, :, P.PARTS[k]].mean(axis=(1, 2)), n=nh) | |
| # paired differences against reference predictors ("ref" = zero-velocity, plus named refs) | |
| refs = dict(ref_preds or {}) | |
| if ref_pred is not None: | |
| refs = {"ref": ref_pred, **refs} | |
| for name, rp in refs.items(): | |
| ref_err = joint_errors(rp, gt) | |
| d = paired_diff(err.mean(axis=(1, 2)), ref_err.mean(axis=(1, 2)), n_boot, seed) | |
| out[f"diff_vs_{name}_all"] = {"value": d["mean"], "ci_lo": d["ci_lo"], "ci_hi": d["ci_hi"], "n": N} | |
| meta[f"diff_vs_{name}_all_p_lt0"] = d["p_lt0"] | |
| if nh > 0: | |
| d = paired_diff(err[m].mean(axis=(1, 2)), ref_err[m].mean(axis=(1, 2)), n_boot, seed) | |
| out[f"diff_vs_{name}_hm"] = {"value": d["mean"], "ci_lo": d["ci_lo"], "ci_hi": d["ci_hi"], "n": nh} | |
| meta[f"diff_vs_{name}_hm_p_lt0"] = d["p_lt0"] | |
| for h in range(err.shape[1]): # per-horizon paired differences (point estimates only) | |
| meta[f"diff_vs_{name}_h{h + 1}"] = float((err[:, h] - ref_err[:, h]).mean()) | |
| return out, meta | |
| def evaluate_semantic(logits: dict, labels: dict, n_boot=10000, seed=0): | |
| """logits: {task: [N,C]}, labels: {task: [N]} -> {f1_task, acc_task} with bootstrap CIs | |
| (F1 CI by clip resampling of the (y, yhat) pairs).""" | |
| out = {} | |
| for t, lg in logits.items(): | |
| y = np.asarray(labels[t]) | |
| yh = np.asarray(lg).argmax(-1) | |
| N = len(y) | |
| rng = np.random.default_rng(seed) | |
| idx = rng.integers(0, N, size=(min(n_boot, 2000), N)) # F1 bootstrap is O(n_boot*N) | |
| f1b = np.array([macro_f1(y[i], yh[i]) for i in idx]) | |
| out[f"f1_{t}"] = {"value": macro_f1(y, yh), "ci_lo": float(np.quantile(f1b, 0.025)), | |
| "ci_hi": float(np.quantile(f1b, 0.975)), "n": N} | |
| lo, hi = bootstrap_ci(100.0 * (y == yh), n_boot, seed) | |
| out[f"acc_{t}"] = {"value": accuracy(y, yh), "ci_lo": lo, "ci_hi": hi, "n": N} | |
| return out | |
Xet Storage Details
- Size:
- 9.37 kB
- Xet hash:
- 3e9634481a84009422d3a9ccbeacb252c2d88ca4de019fc80a913b574aee6463
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.