rpx_benchmark.deployment
Deployment-readiness metrics for RPX: TS, STR, SGC, ESD, weighted scoring.
These are the core novel contributions of the RPX benchmark paper.
Four metrics characterise how well a model generalises under deployment conditions:
TS — Temporal Stability: pose-compensated frame-to-frame consistency STR — State-Transition Robustness: performance drop / recovery across scene phases SGC — Stack-Level Geometric Coherence: mask–depth boundary alignment ESD — Effort-Stratified Difficulty: per-difficulty breakdown (Easy/Medium/Hard)
Plus the weighted phase scoring scheme: S_p = 0.25·M(p,Easy) + 0.35·M(p,Med) + 0.40·M(p,Hard) S_overall = (S_clutter + S_interaction + S_clean) / 3 Δ_int = S_interaction − S_clutter (interaction drop) Δ_rec = S_clean − S_interaction (recovery)
1"""Deployment-readiness metrics for RPX: TS, STR, SGC, ESD, weighted scoring. 2 3These are the core novel contributions of the RPX benchmark paper. 4 5Four metrics characterise how well a model generalises under deployment conditions: 6 7TS — Temporal Stability: pose-compensated frame-to-frame consistency 8STR — State-Transition Robustness: performance drop / recovery across scene phases 9SGC — Stack-Level Geometric Coherence: mask–depth boundary alignment 10ESD — Effort-Stratified Difficulty: per-difficulty breakdown (Easy/Medium/Hard) 11 12Plus the weighted phase scoring scheme: 13 S_p = 0.25·M(p,Easy) + 0.35·M(p,Med) + 0.40·M(p,Hard) 14 S_overall = (S_clutter + S_interaction + S_clean) / 3 15 Δ_int = S_interaction − S_clutter (interaction drop) 16 Δ_rec = S_clean − S_interaction (recovery) 17""" 18 19from __future__ import annotations 20 21from dataclasses import dataclass, field 22from typing import Any, Dict, List, Sequence, Tuple 23 24import numpy as np 25 26from .api import ESD_WEIGHTS, Difficulty, Phase 27 28# ------------------------------------------------------------------ # 29# D435 intrinsics (paper-declared, 640×480) 30# ------------------------------------------------------------------ # 31 32_D435_FX: float = 615.0 33_D435_FY: float = 615.0 34_D435_CX: float = 320.0 35_D435_CY: float = 240.0 36 37 38# ------------------------------------------------------------------ # 39# SE(3) depth reprojection (true geometric, replaces 2D-rotation proxy) 40# ------------------------------------------------------------------ # 41 42 43def se3_reproject_depth( 44 depth: np.ndarray, 45 pose_src: np.ndarray, 46 pose_tgt: np.ndarray, 47 fx: float = _D435_FX, 48 fy: float = _D435_FY, 49 cx: float = _D435_CX, 50 cy: float = _D435_CY, 51) -> tuple: 52 """Reproject a depth map from *src* camera to *tgt* camera via SE(3). 53 54 1. Back-project ``depth`` pixels to 3D in src camera frame. 55 2. Transform 3D points to world using ``pose_src`` (world-from-camera). 56 3. Transform from world to tgt camera using ``inv(pose_tgt)``. 57 4. Project to tgt image plane. 58 5. Scatter the projected depth into a (H, W) buffer. 59 60 Parameters 61 ---------- 62 depth : (H, W) float, metres. 63 pose_src, pose_tgt : (4, 4) world-from-camera SE(3). 64 fx, fy, cx, cy : pinhole intrinsics. 65 66 Returns 67 ------- 68 reproj_depth : (H, W) float — reprojected depth in the *tgt* frame. 69 Pixels that no source point projects to are zero. 70 valid_mask : (H, W) bool — True where a source point landed. 71 """ 72 H, W = depth.shape 73 depth64 = depth.astype(np.float64) 74 75 # Pixel grid. 76 v, u = np.mgrid[0:H, 0:W].astype(np.float64) 77 z = depth64 78 valid_src = (z > 0) & np.isfinite(z) 79 80 # Back-project to 3D in src camera frame. 81 x_cam = (u - cx) * z / fx 82 y_cam = (v - cy) * z / fy 83 ones = np.ones_like(z) 84 pts_cam = np.stack([x_cam, y_cam, z, ones], axis=-1) # (H, W, 4) 85 86 # src camera → world. 87 pose_src_64 = np.asarray(pose_src, dtype=np.float64) 88 pts_world = np.einsum("ij,hwj->hwi", pose_src_64, pts_cam) # (H, W, 4) 89 90 # world → tgt camera. 91 pose_tgt_64 = np.asarray(pose_tgt, dtype=np.float64) 92 pose_tgt_inv = np.linalg.inv(pose_tgt_64) 93 pts_tgt = np.einsum("ij,hwj->hwi", pose_tgt_inv, pts_world) # (H, W, 4) 94 95 z_tgt = pts_tgt[..., 2] 96 x_tgt = pts_tgt[..., 0] 97 y_tgt = pts_tgt[..., 1] 98 99 # Project to tgt image plane. 100 eps = 1e-8 101 u_tgt = (fx * x_tgt / (z_tgt + eps) + cx) 102 v_tgt = (fy * y_tgt / (z_tgt + eps) + cy) 103 104 # Round to nearest integer pixel. 105 u_int = np.round(u_tgt).astype(np.int64) 106 v_int = np.round(v_tgt).astype(np.int64) 107 108 # Validity: source depth positive, target depth positive, lands in image. 109 in_bounds = (u_int >= 0) & (u_int < W) & (v_int >= 0) & (v_int < H) 110 ok = valid_src & in_bounds & (z_tgt > 0) 111 112 # Scatter into output buffer (nearest, z-buffer style). 113 reproj = np.zeros((H, W), dtype=np.float64) 114 valid_out = np.zeros((H, W), dtype=bool) 115 u_ok = u_int[ok] 116 v_ok = v_int[ok] 117 z_ok = z_tgt[ok] 118 119 # Simple scatter — last write wins. For typical RPX clips with small 120 # baseline this is adequate (very few collisions). 121 reproj[v_ok, u_ok] = z_ok 122 valid_out[v_ok, u_ok] = True 123 124 return reproj, valid_out 125 126 127# ------------------------------------------------------------------ # 128# Result dataclasses 129# ------------------------------------------------------------------ # 130 131 132@dataclass 133class TemporalStabilityResult: 134 """TS score per task type. 135 136 TS_seg = E[IoU(P_t, warp(P_{t+1}, ΔT))] (segmentation) 137 TS_depth = E[||D_t − warp(D_{t+1}, ΔT)||_1 / N_valid] (depth) 138 139 Warping uses the relative SE(3) pose from the T265 ground-truth track. 140 When exact warping is not feasible, a consistency proxy (unchanged-pixel 141 fraction) is used as a lower bound. 142 """ 143 144 ts_score: float # primary TS value (higher = more stable) 145 num_pairs: int # number of consecutive frame pairs evaluated 146 per_pair: List[float] = field(default_factory=list) 147 148 149@dataclass 150class StateTransitionRobustnessResult: 151 """STR captures performance change across phase boundaries. 152 153 STR_{C→I} = M(interaction) − M(clutter) ← interaction drop (negative = worse) 154 STR_{I→L} = M(clean) − M(interaction) ← recovery (positive = better) 155 """ 156 157 str_c_to_i: float # interaction drop Δ = M_I − M_C 158 str_i_to_l: float # recovery Δ = M_L − M_I 159 metric_clutter: float 160 metric_interaction: float 161 metric_clean: float 162 163 164@dataclass 165class StackGeometricCoherenceResult: 166 """SGC measures mask–depth boundary alignment. 167 168 SGC = F-score(boundary(mask), boundary(depth_gradient > τ)) 169 Boundary pixels are extracted via Sobel gradient magnitude thresholding. 170 """ 171 172 sgc_score: float # F-score of mask/depth boundary overlap 173 precision: float 174 recall: float 175 num_samples: int 176 177 178@dataclass 179class ESDResult: 180 """Per-difficulty metric breakdown (Effort-Stratified Difficulty).""" 181 182 easy: float | None 183 medium: float | None 184 hard: float | None 185 metric_key: str # which metric was stratified (e.g. "absrel", "miou") 186 187 def weighted_score(self) -> float: 188 """S_p = 0.25·Easy + 0.35·Medium + 0.40·Hard.""" 189 total, weight = 0.0, 0.0 190 for diff, w in ESD_WEIGHTS.items(): 191 val = getattr(self, diff.value) 192 if val is not None: 193 total += w * val 194 weight += w 195 return total / weight if weight > 0 else 0.0 196 197 198@dataclass 199class WeightedPhaseScore: 200 """Full deployment-readiness scoring table. 201 202 Per phase: S_p = 0.25·M(p,Easy) + 0.35·M(p,Medium) + 0.40·M(p,Hard) 203 Overall: S_overall = (S_C + S_I + S_L) / 3 204 Delta int: Δ_int = S_I − S_C 205 Delta rec: Δ_rec = S_L − S_I 206 """ 207 208 clutter: ESDResult 209 interaction: ESDResult 210 clean: ESDResult 211 212 @property 213 def s_clutter(self) -> float: 214 return self.clutter.weighted_score() 215 216 @property 217 def s_interaction(self) -> float: 218 return self.interaction.weighted_score() 219 220 @property 221 def s_clean(self) -> float: 222 return self.clean.weighted_score() 223 224 @property 225 def s_overall(self) -> float: 226 return (self.s_clutter + self.s_interaction + self.s_clean) / 3.0 227 228 @property 229 def delta_int(self) -> float: 230 """Interaction drop: S_I − S_C (negative = model degrades on interaction).""" 231 return self.s_interaction - self.s_clutter 232 233 @property 234 def delta_rec(self) -> float: 235 """Recovery: S_L − S_I (positive = model recovers after interaction).""" 236 return self.s_clean - self.s_interaction 237 238 def to_dict(self) -> Dict[str, float]: 239 return { 240 "s_clutter": self.s_clutter, 241 "s_interaction": self.s_interaction, 242 "s_clean": self.s_clean, 243 "s_overall": self.s_overall, 244 "delta_int": self.delta_int, 245 "delta_rec": self.delta_rec, 246 } 247 248 249@dataclass 250class DeploymentReadinessReport: 251 """Aggregated deployment-readiness report for a model on a task.""" 252 253 task: str 254 model_name: str 255 weighted_phase_score: WeightedPhaseScore | None = None 256 temporal_stability: TemporalStabilityResult | None = None 257 state_transition: StateTransitionRobustnessResult | None = None 258 geometric_coherence: StackGeometricCoherenceResult | None = None 259 260 # --- Tier 1: hardware-agnostic model properties ----------------------- 261 params_m: float | None = None # parameter count in millions 262 flops_g: float | None = None # FLOPs in giga-ops (batch=1) 263 macs_g: float | None = None # MACs (= FLOPs / 2) 264 actmem_gb_fp16: float | None = None # optional activation memory at FP16 265 memory_traffic_gb: float | None = None # estimated DRAM traffic (GB) 266 arithmetic_intensity: float | None = None # FLOPs / Bytes (FLOP/Byte) 267 268 # --- Tier 2: roofline bounds ------------------------------------------ 269 roofline: Dict | None = None # {gpu_name: RooflineBound.to_dict()} 270 271 # --- Tier 3: measured (hardware-specific) ------------------------------ 272 latency_ms_per_sample: float | None = None # wall-clock inference latency (p50) 273 peak_memory_mb: float | None = None # peak resident memory (device-agnostic) 274 system_card: Dict | None = None # SystemCard.to_dict() 275 276 # --- Operating point (precision × accuracy × cost triple) ------------- 277 operating_point: "OperatingPoint | None" = None 278 279 def summary(self) -> Dict[str, Any]: 280 out: Dict[str, Any] = {"task": self.task, "model": self.model_name} 281 if self.weighted_phase_score: 282 out.update(self.weighted_phase_score.to_dict()) 283 if self.temporal_stability: 284 out["ts_score"] = self.temporal_stability.ts_score 285 if self.state_transition: 286 out["str_c_to_i"] = self.state_transition.str_c_to_i 287 out["str_i_to_l"] = self.state_transition.str_i_to_l 288 if self.geometric_coherence: 289 out["sgc_score"] = self.geometric_coherence.sgc_score 290 # Tier 1 291 if self.params_m is not None: 292 out["params_m"] = self.params_m 293 if self.flops_g is not None: 294 out["flops_g"] = self.flops_g 295 if self.macs_g is not None: 296 out["macs_g"] = self.macs_g 297 if self.memory_traffic_gb is not None: 298 out["memory_traffic_gb"] = self.memory_traffic_gb 299 if self.arithmetic_intensity is not None: 300 out["arithmetic_intensity"] = self.arithmetic_intensity 301 # Tier 2 302 if self.roofline is not None: 303 out["roofline"] = self.roofline 304 # Tier 3 305 if self.latency_ms_per_sample is not None: 306 out["latency_ms_per_sample"] = self.latency_ms_per_sample 307 if self.peak_memory_mb is not None: 308 out["peak_memory_mb"] = self.peak_memory_mb 309 if self.system_card is not None: 310 out["system_card"] = self.system_card 311 return out 312 313 314# ------------------------------------------------------------------ # 315# Metric computation functions 316# ------------------------------------------------------------------ # 317 318 319def compute_esd( 320 per_sample_metrics: List[Dict[str, float]], 321 per_sample_difficulties: List[Difficulty | None], 322 metric_key: str, 323) -> ESDResult: 324 """Compute per-difficulty metric averages from per-sample results. 325 326 Args: 327 per_sample_metrics: list of metric dicts, one per sample. 328 per_sample_difficulties: difficulty label per sample (may be None). 329 metric_key: which metric key to stratify (e.g. "absrel", "miou"). 330 331 Returns: 332 ESDResult with easy/medium/hard averages. 333 """ 334 buckets: Dict[Difficulty, List[float]] = {d: [] for d in Difficulty} 335 336 for metrics, diff in zip(per_sample_metrics, per_sample_difficulties, strict=False): 337 if diff is None or metric_key not in metrics: 338 continue 339 buckets[diff].append(metrics[metric_key]) 340 341 def mean_or_none(vals: List[float]) -> float | None: 342 return float(np.mean(vals)) if vals else None 343 344 return ESDResult( 345 easy=mean_or_none(buckets[Difficulty.EASY]), 346 medium=mean_or_none(buckets[Difficulty.MEDIUM]), 347 hard=mean_or_none(buckets[Difficulty.HARD]), 348 metric_key=metric_key, 349 ) 350 351 352def compute_weighted_phase_score( 353 per_sample_metrics: List[Dict[str, float]], 354 per_sample_phases: List[Phase | None], 355 per_sample_difficulties: List[Difficulty | None], 356 metric_key: str, 357) -> WeightedPhaseScore: 358 """Compute the full weighted phase scoring table. 359 360 Groups samples by (phase, difficulty) and computes: 361 S_p = 0.25·M(p,Easy) + 0.35·M(p,Medium) + 0.40·M(p,Hard) 362 for each phase, then overall score and transition deltas. 363 """ 364 phase_sample_metrics: Dict[Phase, Tuple[List, List]] = {p: ([], []) for p in Phase} 365 366 for m, ph, diff in zip( 367 per_sample_metrics, per_sample_phases, per_sample_difficulties, strict=False 368 ): 369 if ph is None: 370 continue 371 phase_sample_metrics[ph][0].append(m) 372 phase_sample_metrics[ph][1].append(diff) 373 374 def esd_for_phase(ph: Phase) -> ESDResult: 375 metrics_list, diff_list = phase_sample_metrics[ph] 376 return compute_esd(metrics_list, diff_list, metric_key) 377 378 return WeightedPhaseScore( 379 clutter=esd_for_phase(Phase.CLUTTER), 380 interaction=esd_for_phase(Phase.INTERACTION), 381 clean=esd_for_phase(Phase.CLEAN), 382 ) 383 384 385def compute_str( 386 phase_scores: Dict[Phase, float], 387) -> StateTransitionRobustnessResult: 388 """Compute STR from per-phase aggregated scores. 389 390 Args: 391 phase_scores: dict mapping Phase → scalar metric value. 392 """ 393 m_c = phase_scores.get(Phase.CLUTTER, 0.0) 394 m_i = phase_scores.get(Phase.INTERACTION, 0.0) 395 m_l = phase_scores.get(Phase.CLEAN, 0.0) 396 return StateTransitionRobustnessResult( 397 str_c_to_i=m_i - m_c, 398 str_i_to_l=m_l - m_i, 399 metric_clutter=m_c, 400 metric_interaction=m_i, 401 metric_clean=m_l, 402 ) 403 404 405def compute_temporal_stability_seg( 406 pred_masks: Sequence[np.ndarray], 407 camera_poses: Sequence[np.ndarray | None], 408) -> TemporalStabilityResult: 409 """TS_seg = E[IoU(P_t, warp(P_{t+1}, ΔT))]. 410 411 When T265 pose data is available, we use the relative rotation to compensate 412 for camera motion before computing IoU between adjacent frames. Without 413 pixel-accurate warping (which requires depth for backprojection), we apply 414 a simplified affine proxy using the in-plane rotation component only. 415 416 This gives a conservative lower-bound TS_seg that is still a meaningful 417 stability signal when scenes have modest depth variation. 418 419 Args: 420 pred_masks: sequence of predicted segmentation masks (H×W int). 421 camera_poses: per-frame 4×4 SE(3) matrices (camera-to-world), or None. 422 423 Returns: 424 TemporalStabilityResult. 425 """ 426 if len(pred_masks) < 2: 427 return TemporalStabilityResult(ts_score=1.0, num_pairs=0) 428 429 per_pair = [] 430 for t in range(len(pred_masks) - 1): 431 m_t = np.asarray(pred_masks[t], dtype=np.int32) 432 m_t1 = np.asarray(pred_masks[t + 1], dtype=np.int32) 433 434 # Attempt pose-compensated warp if poses are available 435 if camera_poses[t] is not None and camera_poses[t + 1] is not None: 436 m_t1_warped = _warp_mask_approx(m_t1, camera_poses[t], camera_poses[t + 1]) 437 else: 438 m_t1_warped = m_t1 439 440 # Per-class IoU then mean 441 classes = np.unique(np.concatenate([m_t.flatten(), m_t1_warped.flatten()])) 442 classes = classes[classes >= 0] 443 if len(classes) == 0: 444 per_pair.append(1.0) 445 continue 446 ious = [] 447 for c in classes: 448 inter = float(((m_t == c) & (m_t1_warped == c)).sum()) 449 union = float(((m_t == c) | (m_t1_warped == c)).sum()) 450 ious.append(inter / union if union > 0 else 1.0) 451 per_pair.append(float(np.mean(ious))) 452 453 return TemporalStabilityResult( 454 ts_score=float(np.mean(per_pair)), 455 num_pairs=len(per_pair), 456 per_pair=per_pair, 457 ) 458 459 460def compute_temporal_stability_depth( 461 pred_depths: Sequence[np.ndarray], 462 camera_poses: Sequence[np.ndarray | None], 463) -> TemporalStabilityResult: 464 """TS_depth = E[||D_t − warp(D_{t+1}, ΔT)||_1 / N_valid]. 465 466 Normalised to [0,1] by dividing by the max depth range to give a 467 higher-is-better stability score (TS = 1 − normalised_L1). 468 """ 469 if len(pred_depths) < 2: 470 return TemporalStabilityResult(ts_score=1.0, num_pairs=0) 471 472 per_pair = [] 473 for t in range(len(pred_depths) - 1): 474 d_t = np.asarray(pred_depths[t], dtype=np.float32) 475 d_t1 = np.asarray(pred_depths[t + 1], dtype=np.float32) 476 477 if camera_poses[t] is not None and camera_poses[t + 1] is not None: 478 d_t1_warped = _warp_depth_approx(d_t1, camera_poses[t], camera_poses[t + 1]) 479 else: 480 d_t1_warped = d_t1 481 482 valid = (d_t > 0) & (d_t1_warped > 0) 483 if valid.sum() == 0: 484 per_pair.append(1.0) 485 continue 486 l1 = float(np.abs(d_t[valid] - d_t1_warped[valid]).mean()) 487 depth_range = max(float(d_t[valid].max() - d_t[valid].min()), 1e-3) 488 ts = max(0.0, 1.0 - l1 / depth_range) 489 per_pair.append(ts) 490 491 return TemporalStabilityResult( 492 ts_score=float(np.mean(per_pair)), 493 num_pairs=len(per_pair), 494 per_pair=per_pair, 495 ) 496 497 498def compute_sgc( 499 pred_masks: Sequence[np.ndarray], 500 pred_depths: Sequence[np.ndarray], 501 depth_gradient_threshold: float = 0.1, 502 boundary_dilation: int = 2, 503) -> StackGeometricCoherenceResult: 504 """Stack-Level Geometric Coherence: boundary F-score between mask and depth edges. 505 506 SGC = F-score(boundary(mask), boundary(depth_gradient > τ)) 507 508 A high SGC means segmentation boundaries are geometrically consistent with 509 the depth discontinuities — indicating the model perceives coherent surfaces. 510 511 Args: 512 pred_masks: sequence of predicted segmentation masks (H×W int). 513 pred_depths: sequence of predicted depth maps (H×W float32, metres). 514 depth_gradient_threshold: τ for depth gradient thresholding. 515 boundary_dilation: pixel tolerance for boundary matching. 516 """ 517 if len(pred_masks) == 0: 518 return StackGeometricCoherenceResult( 519 sgc_score=0.0, precision=0.0, recall=0.0, num_samples=0 520 ) 521 522 precisions, recalls = [], [] 523 for mask, depth in zip(pred_masks, pred_depths, strict=False): 524 mask = np.asarray(mask, dtype=np.int32) 525 depth = np.asarray(depth, dtype=np.float32) 526 527 mask_boundary = _extract_boundary(mask, dilation=boundary_dilation) 528 depth_boundary = _extract_depth_boundary( 529 depth, threshold=depth_gradient_threshold, dilation=boundary_dilation 530 ) 531 532 tp = float((mask_boundary & depth_boundary).sum()) 533 fp = float((mask_boundary & ~depth_boundary).sum()) 534 fn = float((~mask_boundary & depth_boundary).sum()) 535 536 p = tp / (tp + fp) if (tp + fp) > 0 else 1.0 537 r = tp / (tp + fn) if (tp + fn) > 0 else 1.0 538 precisions.append(p) 539 recalls.append(r) 540 541 mean_p = float(np.mean(precisions)) 542 mean_r = float(np.mean(recalls)) 543 f1 = (2 * mean_p * mean_r) / (mean_p + mean_r) if (mean_p + mean_r) > 0 else 0.0 544 545 return StackGeometricCoherenceResult( 546 sgc_score=f1, 547 precision=mean_p, 548 recall=mean_r, 549 num_samples=len(pred_masks), 550 ) 551 552 553# ------------------------------------------------------------------ # 554# Internal helpers 555# ------------------------------------------------------------------ # 556 557 558def _extract_boundary(mask: np.ndarray, dilation: int = 2) -> np.ndarray: 559 """Boolean boundary map from a semantic mask using finite differences.""" 560 boundary = np.zeros_like(mask, dtype=bool) 561 boundary[:-1, :] |= mask[:-1, :] != mask[1:, :] 562 boundary[:, :-1] |= mask[:, :-1] != mask[:, 1:] 563 if dilation > 0: 564 boundary = _binary_dilate(boundary, dilation) 565 return boundary 566 567 568def _extract_depth_boundary(depth: np.ndarray, threshold: float, dilation: int = 2) -> np.ndarray: 569 """Boolean boundary map from depth via Sobel gradient magnitude.""" 570 gy = np.zeros_like(depth) 571 gx = np.zeros_like(depth) 572 gy[1:-1, :] = (depth[2:, :] - depth[:-2, :]) / 2.0 573 gx[:, 1:-1] = (depth[:, 2:] - depth[:, :-2]) / 2.0 574 grad_mag = np.sqrt(gx**2 + gy**2) 575 boundary = grad_mag > threshold 576 if dilation > 0: 577 boundary = _binary_dilate(boundary, dilation) 578 return boundary 579 580 581def _binary_dilate(mask: np.ndarray, radius: int) -> np.ndarray: 582 """Simple box-dilation of a boolean mask (avoids scipy dependency).""" 583 result = mask.copy() 584 for _ in range(radius): 585 tmp = result.copy() 586 tmp[1:, :] |= result[:-1, :] 587 tmp[:-1, :] |= result[1:, :] 588 tmp[:, 1:] |= result[:, :-1] 589 tmp[:, :-1] |= result[:, 1:] 590 result = tmp 591 return result 592 593 594def _relative_pose(pose_t: np.ndarray, pose_t1: np.ndarray) -> np.ndarray: 595 """Compute relative SE(3): T_{t→t+1} = pose_t^{-1} @ pose_{t+1}.""" 596 return np.linalg.inv(pose_t) @ pose_t1 597 598 599def _warp_mask_approx( 600 mask: np.ndarray, 601 pose_t: np.ndarray, 602 pose_t1: np.ndarray, 603) -> np.ndarray: 604 """Approximate mask warp using in-plane rotation from relative pose. 605 606 Full pixel-accurate warping requires dense depth for backprojection. 607 This approximation applies the in-plane rotation (yaw component) from 608 the relative SE(3) transform as a 2D affine warp, which is a reasonable 609 proxy for planar/forward-facing camera motion. 610 """ 611 try: 612 import cv2 613 614 T_rel = _relative_pose(pose_t, pose_t1) 615 R_rel = T_rel[:3, :3] 616 angle_rad = float(np.arctan2(R_rel[1, 0], R_rel[0, 0])) 617 h, w = mask.shape 618 cx, cy = w / 2.0, h / 2.0 619 M = cv2.getRotationMatrix2D((cx, cy), float(np.degrees(angle_rad)), 1.0) 620 warped = cv2.warpAffine( 621 mask.astype(np.float32), 622 M, 623 (w, h), 624 flags=cv2.INTER_NEAREST, 625 borderValue=-1, 626 ).astype(np.int32) 627 return warped 628 except ImportError: 629 return mask # fall back to unwarped if cv2 not available 630 631 632# ================================================================== # 633# Operating point (precision × accuracy × cost triple, no composite) 634# ================================================================== # 635 636 637@dataclass 638class OperatingPoint: 639 """One (precision, accuracy, cost) triple for a model. 640 641 A model can publish multiple operating points — e.g. fp32, fp16, 642 int8 — each reported separately on the three RPX axes (task 643 accuracy, scene-change robustness, compute cost). RPX does not 644 combine these into a single composite score: the reader picks the 645 operating point and the axis their deployment cares about. 646 """ 647 648 precision: str # "fp32", "fp16", "bf16" 649 # Task performance (primary metric — δ1, mIoU, accuracy, etc.) 650 task_metric: float 651 task_metric_name: str # e.g. "delta1", "miou" 652 higher_is_better: bool 653 # Robustness 654 str_score: float # STR value (near 0 = robust) 655 # Hardware-agnostic cost (Tier 1) 656 flops_g: float 657 params_m: float 658 memory_traffic_gb: float | None = None 659 660 661def _warp_depth_approx( 662 depth: np.ndarray, 663 pose_t: np.ndarray, 664 pose_t1: np.ndarray, 665) -> np.ndarray: 666 """Approximate depth warp (same in-plane rotation proxy as mask warp).""" 667 if not np.isfinite(pose_t).all() or not np.isfinite(pose_t1).all(): 668 return depth.copy() 669 try: 670 import cv2 671 672 T_rel = _relative_pose(pose_t, pose_t1) 673 R_rel = T_rel[:3, :3] 674 angle_rad = float(np.arctan2(R_rel[1, 0], R_rel[0, 0])) 675 h, w = depth.shape 676 cx, cy = w / 2.0, h / 2.0 677 M = cv2.getRotationMatrix2D((cx, cy), float(np.degrees(angle_rad)), 1.0) 678 warped = cv2.warpAffine( 679 depth, 680 M, 681 (w, h), 682 flags=cv2.INTER_LINEAR, 683 borderValue=0.0, 684 ) 685 return warped 686 except ImportError: 687 return depth
def
se3_reproject_depth( depth: numpy.ndarray, pose_src: numpy.ndarray, pose_tgt: numpy.ndarray, fx: float = 615.0, fy: float = 615.0, cx: float = 320.0, cy: float = 240.0) -> tuple:
44def se3_reproject_depth( 45 depth: np.ndarray, 46 pose_src: np.ndarray, 47 pose_tgt: np.ndarray, 48 fx: float = _D435_FX, 49 fy: float = _D435_FY, 50 cx: float = _D435_CX, 51 cy: float = _D435_CY, 52) -> tuple: 53 """Reproject a depth map from *src* camera to *tgt* camera via SE(3). 54 55 1. Back-project ``depth`` pixels to 3D in src camera frame. 56 2. Transform 3D points to world using ``pose_src`` (world-from-camera). 57 3. Transform from world to tgt camera using ``inv(pose_tgt)``. 58 4. Project to tgt image plane. 59 5. Scatter the projected depth into a (H, W) buffer. 60 61 Parameters 62 ---------- 63 depth : (H, W) float, metres. 64 pose_src, pose_tgt : (4, 4) world-from-camera SE(3). 65 fx, fy, cx, cy : pinhole intrinsics. 66 67 Returns 68 ------- 69 reproj_depth : (H, W) float — reprojected depth in the *tgt* frame. 70 Pixels that no source point projects to are zero. 71 valid_mask : (H, W) bool — True where a source point landed. 72 """ 73 H, W = depth.shape 74 depth64 = depth.astype(np.float64) 75 76 # Pixel grid. 77 v, u = np.mgrid[0:H, 0:W].astype(np.float64) 78 z = depth64 79 valid_src = (z > 0) & np.isfinite(z) 80 81 # Back-project to 3D in src camera frame. 82 x_cam = (u - cx) * z / fx 83 y_cam = (v - cy) * z / fy 84 ones = np.ones_like(z) 85 pts_cam = np.stack([x_cam, y_cam, z, ones], axis=-1) # (H, W, 4) 86 87 # src camera → world. 88 pose_src_64 = np.asarray(pose_src, dtype=np.float64) 89 pts_world = np.einsum("ij,hwj->hwi", pose_src_64, pts_cam) # (H, W, 4) 90 91 # world → tgt camera. 92 pose_tgt_64 = np.asarray(pose_tgt, dtype=np.float64) 93 pose_tgt_inv = np.linalg.inv(pose_tgt_64) 94 pts_tgt = np.einsum("ij,hwj->hwi", pose_tgt_inv, pts_world) # (H, W, 4) 95 96 z_tgt = pts_tgt[..., 2] 97 x_tgt = pts_tgt[..., 0] 98 y_tgt = pts_tgt[..., 1] 99 100 # Project to tgt image plane. 101 eps = 1e-8 102 u_tgt = (fx * x_tgt / (z_tgt + eps) + cx) 103 v_tgt = (fy * y_tgt / (z_tgt + eps) + cy) 104 105 # Round to nearest integer pixel. 106 u_int = np.round(u_tgt).astype(np.int64) 107 v_int = np.round(v_tgt).astype(np.int64) 108 109 # Validity: source depth positive, target depth positive, lands in image. 110 in_bounds = (u_int >= 0) & (u_int < W) & (v_int >= 0) & (v_int < H) 111 ok = valid_src & in_bounds & (z_tgt > 0) 112 113 # Scatter into output buffer (nearest, z-buffer style). 114 reproj = np.zeros((H, W), dtype=np.float64) 115 valid_out = np.zeros((H, W), dtype=bool) 116 u_ok = u_int[ok] 117 v_ok = v_int[ok] 118 z_ok = z_tgt[ok] 119 120 # Simple scatter — last write wins. For typical RPX clips with small 121 # baseline this is adequate (very few collisions). 122 reproj[v_ok, u_ok] = z_ok 123 valid_out[v_ok, u_ok] = True 124 125 return reproj, valid_out
Reproject a depth map from src camera to tgt camera via SE(3).
- Back-project
depthpixels to 3D in src camera frame. - Transform 3D points to world using
pose_src(world-from-camera). - Transform from world to tgt camera using
inv(pose_tgt). - Project to tgt image plane.
- Scatter the projected depth into a (H, W) buffer.
Parameters
depth : (H, W) float, metres. pose_src, pose_tgt : (4, 4) world-from-camera SE(3). fx, fy, cx, cy : pinhole intrinsics.
Returns
reproj_depth : (H, W) float — reprojected depth in the *tgt* frame. Pixels that no source point projects to are zero. valid_mask : (H, W) bool — True where a source point landed.
@dataclass
class
TemporalStabilityResult:
133@dataclass 134class TemporalStabilityResult: 135 """TS score per task type. 136 137 TS_seg = E[IoU(P_t, warp(P_{t+1}, ΔT))] (segmentation) 138 TS_depth = E[||D_t − warp(D_{t+1}, ΔT)||_1 / N_valid] (depth) 139 140 Warping uses the relative SE(3) pose from the T265 ground-truth track. 141 When exact warping is not feasible, a consistency proxy (unchanged-pixel 142 fraction) is used as a lower bound. 143 """ 144 145 ts_score: float # primary TS value (higher = more stable) 146 num_pairs: int # number of consecutive frame pairs evaluated 147 per_pair: List[float] = field(default_factory=list)
TS score per task type.
TS_seg = E[IoU(P_t, warp(P_{t+1}, ΔT))] (segmentation) TS_depth = E[||D_t − warp(D_{t+1}, ΔT)||_1 / N_valid] (depth)
Warping uses the relative SE(3) pose from the T265 ground-truth track. When exact warping is not feasible, a consistency proxy (unchanged-pixel fraction) is used as a lower bound.
@dataclass
class
StateTransitionRobustnessResult:
150@dataclass 151class StateTransitionRobustnessResult: 152 """STR captures performance change across phase boundaries. 153 154 STR_{C→I} = M(interaction) − M(clutter) ← interaction drop (negative = worse) 155 STR_{I→L} = M(clean) − M(interaction) ← recovery (positive = better) 156 """ 157 158 str_c_to_i: float # interaction drop Δ = M_I − M_C 159 str_i_to_l: float # recovery Δ = M_L − M_I 160 metric_clutter: float 161 metric_interaction: float 162 metric_clean: float
STR captures performance change across phase boundaries.
STR_{C→I} = M(interaction) − M(clutter) ← interaction drop (negative = worse) STR_{I→L} = M(clean) − M(interaction) ← recovery (positive = better)
@dataclass
class
StackGeometricCoherenceResult:
165@dataclass 166class StackGeometricCoherenceResult: 167 """SGC measures mask–depth boundary alignment. 168 169 SGC = F-score(boundary(mask), boundary(depth_gradient > τ)) 170 Boundary pixels are extracted via Sobel gradient magnitude thresholding. 171 """ 172 173 sgc_score: float # F-score of mask/depth boundary overlap 174 precision: float 175 recall: float 176 num_samples: int
SGC measures mask–depth boundary alignment.
SGC = F-score(boundary(mask), boundary(depth_gradient > τ)) Boundary pixels are extracted via Sobel gradient magnitude thresholding.
@dataclass
class
ESDResult:
179@dataclass 180class ESDResult: 181 """Per-difficulty metric breakdown (Effort-Stratified Difficulty).""" 182 183 easy: float | None 184 medium: float | None 185 hard: float | None 186 metric_key: str # which metric was stratified (e.g. "absrel", "miou") 187 188 def weighted_score(self) -> float: 189 """S_p = 0.25·Easy + 0.35·Medium + 0.40·Hard.""" 190 total, weight = 0.0, 0.0 191 for diff, w in ESD_WEIGHTS.items(): 192 val = getattr(self, diff.value) 193 if val is not None: 194 total += w * val 195 weight += w 196 return total / weight if weight > 0 else 0.0
Per-difficulty metric breakdown (Effort-Stratified Difficulty).
def
weighted_score(self) -> float:
188 def weighted_score(self) -> float: 189 """S_p = 0.25·Easy + 0.35·Medium + 0.40·Hard.""" 190 total, weight = 0.0, 0.0 191 for diff, w in ESD_WEIGHTS.items(): 192 val = getattr(self, diff.value) 193 if val is not None: 194 total += w * val 195 weight += w 196 return total / weight if weight > 0 else 0.0
S_p = 0.25·Easy + 0.35·Medium + 0.40·Hard.
@dataclass
class
WeightedPhaseScore:
199@dataclass 200class WeightedPhaseScore: 201 """Full deployment-readiness scoring table. 202 203 Per phase: S_p = 0.25·M(p,Easy) + 0.35·M(p,Medium) + 0.40·M(p,Hard) 204 Overall: S_overall = (S_C + S_I + S_L) / 3 205 Delta int: Δ_int = S_I − S_C 206 Delta rec: Δ_rec = S_L − S_I 207 """ 208 209 clutter: ESDResult 210 interaction: ESDResult 211 clean: ESDResult 212 213 @property 214 def s_clutter(self) -> float: 215 return self.clutter.weighted_score() 216 217 @property 218 def s_interaction(self) -> float: 219 return self.interaction.weighted_score() 220 221 @property 222 def s_clean(self) -> float: 223 return self.clean.weighted_score() 224 225 @property 226 def s_overall(self) -> float: 227 return (self.s_clutter + self.s_interaction + self.s_clean) / 3.0 228 229 @property 230 def delta_int(self) -> float: 231 """Interaction drop: S_I − S_C (negative = model degrades on interaction).""" 232 return self.s_interaction - self.s_clutter 233 234 @property 235 def delta_rec(self) -> float: 236 """Recovery: S_L − S_I (positive = model recovers after interaction).""" 237 return self.s_clean - self.s_interaction 238 239 def to_dict(self) -> Dict[str, float]: 240 return { 241 "s_clutter": self.s_clutter, 242 "s_interaction": self.s_interaction, 243 "s_clean": self.s_clean, 244 "s_overall": self.s_overall, 245 "delta_int": self.delta_int, 246 "delta_rec": self.delta_rec, 247 }
Full deployment-readiness scoring table.
Per phase: S_p = 0.25·M(p,Easy) + 0.35·M(p,Medium) + 0.40·M(p,Hard) Overall: S_overall = (S_C + S_I + S_L) / 3 Delta int: Δ_int = S_I − S_C Delta rec: Δ_rec = S_L − S_I
clutter: ESDResult
interaction: ESDResult
clean: ESDResult
delta_int: float
229 @property 230 def delta_int(self) -> float: 231 """Interaction drop: S_I − S_C (negative = model degrades on interaction).""" 232 return self.s_interaction - self.s_clutter
Interaction drop: S_I − S_C (negative = model degrades on interaction).
delta_rec: float
234 @property 235 def delta_rec(self) -> float: 236 """Recovery: S_L − S_I (positive = model recovers after interaction).""" 237 return self.s_clean - self.s_interaction
Recovery: S_L − S_I (positive = model recovers after interaction).
@dataclass
class
DeploymentReadinessReport:
250@dataclass 251class DeploymentReadinessReport: 252 """Aggregated deployment-readiness report for a model on a task.""" 253 254 task: str 255 model_name: str 256 weighted_phase_score: WeightedPhaseScore | None = None 257 temporal_stability: TemporalStabilityResult | None = None 258 state_transition: StateTransitionRobustnessResult | None = None 259 geometric_coherence: StackGeometricCoherenceResult | None = None 260 261 # --- Tier 1: hardware-agnostic model properties ----------------------- 262 params_m: float | None = None # parameter count in millions 263 flops_g: float | None = None # FLOPs in giga-ops (batch=1) 264 macs_g: float | None = None # MACs (= FLOPs / 2) 265 actmem_gb_fp16: float | None = None # optional activation memory at FP16 266 memory_traffic_gb: float | None = None # estimated DRAM traffic (GB) 267 arithmetic_intensity: float | None = None # FLOPs / Bytes (FLOP/Byte) 268 269 # --- Tier 2: roofline bounds ------------------------------------------ 270 roofline: Dict | None = None # {gpu_name: RooflineBound.to_dict()} 271 272 # --- Tier 3: measured (hardware-specific) ------------------------------ 273 latency_ms_per_sample: float | None = None # wall-clock inference latency (p50) 274 peak_memory_mb: float | None = None # peak resident memory (device-agnostic) 275 system_card: Dict | None = None # SystemCard.to_dict() 276 277 # --- Operating point (precision × accuracy × cost triple) ------------- 278 operating_point: "OperatingPoint | None" = None 279 280 def summary(self) -> Dict[str, Any]: 281 out: Dict[str, Any] = {"task": self.task, "model": self.model_name} 282 if self.weighted_phase_score: 283 out.update(self.weighted_phase_score.to_dict()) 284 if self.temporal_stability: 285 out["ts_score"] = self.temporal_stability.ts_score 286 if self.state_transition: 287 out["str_c_to_i"] = self.state_transition.str_c_to_i 288 out["str_i_to_l"] = self.state_transition.str_i_to_l 289 if self.geometric_coherence: 290 out["sgc_score"] = self.geometric_coherence.sgc_score 291 # Tier 1 292 if self.params_m is not None: 293 out["params_m"] = self.params_m 294 if self.flops_g is not None: 295 out["flops_g"] = self.flops_g 296 if self.macs_g is not None: 297 out["macs_g"] = self.macs_g 298 if self.memory_traffic_gb is not None: 299 out["memory_traffic_gb"] = self.memory_traffic_gb 300 if self.arithmetic_intensity is not None: 301 out["arithmetic_intensity"] = self.arithmetic_intensity 302 # Tier 2 303 if self.roofline is not None: 304 out["roofline"] = self.roofline 305 # Tier 3 306 if self.latency_ms_per_sample is not None: 307 out["latency_ms_per_sample"] = self.latency_ms_per_sample 308 if self.peak_memory_mb is not None: 309 out["peak_memory_mb"] = self.peak_memory_mb 310 if self.system_card is not None: 311 out["system_card"] = self.system_card 312 return out
Aggregated deployment-readiness report for a model on a task.
DeploymentReadinessReport( task: str, model_name: str, weighted_phase_score: WeightedPhaseScore | None = None, temporal_stability: TemporalStabilityResult | None = None, state_transition: StateTransitionRobustnessResult | None = None, geometric_coherence: StackGeometricCoherenceResult | None = None, params_m: float | None = None, flops_g: float | None = None, macs_g: float | None = None, actmem_gb_fp16: float | None = None, memory_traffic_gb: float | None = None, arithmetic_intensity: float | None = None, roofline: Optional[Dict] = None, latency_ms_per_sample: float | None = None, peak_memory_mb: float | None = None, system_card: Optional[Dict] = None, operating_point: OperatingPoint | None = None)
def
summary(self) -> Dict[str, Any]:
280 def summary(self) -> Dict[str, Any]: 281 out: Dict[str, Any] = {"task": self.task, "model": self.model_name} 282 if self.weighted_phase_score: 283 out.update(self.weighted_phase_score.to_dict()) 284 if self.temporal_stability: 285 out["ts_score"] = self.temporal_stability.ts_score 286 if self.state_transition: 287 out["str_c_to_i"] = self.state_transition.str_c_to_i 288 out["str_i_to_l"] = self.state_transition.str_i_to_l 289 if self.geometric_coherence: 290 out["sgc_score"] = self.geometric_coherence.sgc_score 291 # Tier 1 292 if self.params_m is not None: 293 out["params_m"] = self.params_m 294 if self.flops_g is not None: 295 out["flops_g"] = self.flops_g 296 if self.macs_g is not None: 297 out["macs_g"] = self.macs_g 298 if self.memory_traffic_gb is not None: 299 out["memory_traffic_gb"] = self.memory_traffic_gb 300 if self.arithmetic_intensity is not None: 301 out["arithmetic_intensity"] = self.arithmetic_intensity 302 # Tier 2 303 if self.roofline is not None: 304 out["roofline"] = self.roofline 305 # Tier 3 306 if self.latency_ms_per_sample is not None: 307 out["latency_ms_per_sample"] = self.latency_ms_per_sample 308 if self.peak_memory_mb is not None: 309 out["peak_memory_mb"] = self.peak_memory_mb 310 if self.system_card is not None: 311 out["system_card"] = self.system_card 312 return out
def
compute_esd( per_sample_metrics: List[Dict[str, float]], per_sample_difficulties: List[rpx_benchmark.api.Difficulty | None], metric_key: str) -> ESDResult:
320def compute_esd( 321 per_sample_metrics: List[Dict[str, float]], 322 per_sample_difficulties: List[Difficulty | None], 323 metric_key: str, 324) -> ESDResult: 325 """Compute per-difficulty metric averages from per-sample results. 326 327 Args: 328 per_sample_metrics: list of metric dicts, one per sample. 329 per_sample_difficulties: difficulty label per sample (may be None). 330 metric_key: which metric key to stratify (e.g. "absrel", "miou"). 331 332 Returns: 333 ESDResult with easy/medium/hard averages. 334 """ 335 buckets: Dict[Difficulty, List[float]] = {d: [] for d in Difficulty} 336 337 for metrics, diff in zip(per_sample_metrics, per_sample_difficulties, strict=False): 338 if diff is None or metric_key not in metrics: 339 continue 340 buckets[diff].append(metrics[metric_key]) 341 342 def mean_or_none(vals: List[float]) -> float | None: 343 return float(np.mean(vals)) if vals else None 344 345 return ESDResult( 346 easy=mean_or_none(buckets[Difficulty.EASY]), 347 medium=mean_or_none(buckets[Difficulty.MEDIUM]), 348 hard=mean_or_none(buckets[Difficulty.HARD]), 349 metric_key=metric_key, 350 )
Compute per-difficulty metric averages from per-sample results.
Args: per_sample_metrics: list of metric dicts, one per sample. per_sample_difficulties: difficulty label per sample (may be None). metric_key: which metric key to stratify (e.g. "absrel", "miou").
Returns: ESDResult with easy/medium/hard averages.
def
compute_weighted_phase_score( per_sample_metrics: List[Dict[str, float]], per_sample_phases: List[rpx_benchmark.api.Phase | None], per_sample_difficulties: List[rpx_benchmark.api.Difficulty | None], metric_key: str) -> WeightedPhaseScore:
353def compute_weighted_phase_score( 354 per_sample_metrics: List[Dict[str, float]], 355 per_sample_phases: List[Phase | None], 356 per_sample_difficulties: List[Difficulty | None], 357 metric_key: str, 358) -> WeightedPhaseScore: 359 """Compute the full weighted phase scoring table. 360 361 Groups samples by (phase, difficulty) and computes: 362 S_p = 0.25·M(p,Easy) + 0.35·M(p,Medium) + 0.40·M(p,Hard) 363 for each phase, then overall score and transition deltas. 364 """ 365 phase_sample_metrics: Dict[Phase, Tuple[List, List]] = {p: ([], []) for p in Phase} 366 367 for m, ph, diff in zip( 368 per_sample_metrics, per_sample_phases, per_sample_difficulties, strict=False 369 ): 370 if ph is None: 371 continue 372 phase_sample_metrics[ph][0].append(m) 373 phase_sample_metrics[ph][1].append(diff) 374 375 def esd_for_phase(ph: Phase) -> ESDResult: 376 metrics_list, diff_list = phase_sample_metrics[ph] 377 return compute_esd(metrics_list, diff_list, metric_key) 378 379 return WeightedPhaseScore( 380 clutter=esd_for_phase(Phase.CLUTTER), 381 interaction=esd_for_phase(Phase.INTERACTION), 382 clean=esd_for_phase(Phase.CLEAN), 383 )
Compute the full weighted phase scoring table.
Groups samples by (phase, difficulty) and computes: S_p = 0.25·M(p,Easy) + 0.35·M(p,Medium) + 0.40·M(p,Hard) for each phase, then overall score and transition deltas.
def
compute_str( phase_scores: Dict[rpx_benchmark.api.Phase, float]) -> StateTransitionRobustnessResult:
386def compute_str( 387 phase_scores: Dict[Phase, float], 388) -> StateTransitionRobustnessResult: 389 """Compute STR from per-phase aggregated scores. 390 391 Args: 392 phase_scores: dict mapping Phase → scalar metric value. 393 """ 394 m_c = phase_scores.get(Phase.CLUTTER, 0.0) 395 m_i = phase_scores.get(Phase.INTERACTION, 0.0) 396 m_l = phase_scores.get(Phase.CLEAN, 0.0) 397 return StateTransitionRobustnessResult( 398 str_c_to_i=m_i - m_c, 399 str_i_to_l=m_l - m_i, 400 metric_clutter=m_c, 401 metric_interaction=m_i, 402 metric_clean=m_l, 403 )
Compute STR from per-phase aggregated scores.
Args: phase_scores: dict mapping Phase → scalar metric value.
def
compute_temporal_stability_seg( pred_masks: Sequence[numpy.ndarray], camera_poses: Sequence[numpy.ndarray | None]) -> TemporalStabilityResult:
406def compute_temporal_stability_seg( 407 pred_masks: Sequence[np.ndarray], 408 camera_poses: Sequence[np.ndarray | None], 409) -> TemporalStabilityResult: 410 """TS_seg = E[IoU(P_t, warp(P_{t+1}, ΔT))]. 411 412 When T265 pose data is available, we use the relative rotation to compensate 413 for camera motion before computing IoU between adjacent frames. Without 414 pixel-accurate warping (which requires depth for backprojection), we apply 415 a simplified affine proxy using the in-plane rotation component only. 416 417 This gives a conservative lower-bound TS_seg that is still a meaningful 418 stability signal when scenes have modest depth variation. 419 420 Args: 421 pred_masks: sequence of predicted segmentation masks (H×W int). 422 camera_poses: per-frame 4×4 SE(3) matrices (camera-to-world), or None. 423 424 Returns: 425 TemporalStabilityResult. 426 """ 427 if len(pred_masks) < 2: 428 return TemporalStabilityResult(ts_score=1.0, num_pairs=0) 429 430 per_pair = [] 431 for t in range(len(pred_masks) - 1): 432 m_t = np.asarray(pred_masks[t], dtype=np.int32) 433 m_t1 = np.asarray(pred_masks[t + 1], dtype=np.int32) 434 435 # Attempt pose-compensated warp if poses are available 436 if camera_poses[t] is not None and camera_poses[t + 1] is not None: 437 m_t1_warped = _warp_mask_approx(m_t1, camera_poses[t], camera_poses[t + 1]) 438 else: 439 m_t1_warped = m_t1 440 441 # Per-class IoU then mean 442 classes = np.unique(np.concatenate([m_t.flatten(), m_t1_warped.flatten()])) 443 classes = classes[classes >= 0] 444 if len(classes) == 0: 445 per_pair.append(1.0) 446 continue 447 ious = [] 448 for c in classes: 449 inter = float(((m_t == c) & (m_t1_warped == c)).sum()) 450 union = float(((m_t == c) | (m_t1_warped == c)).sum()) 451 ious.append(inter / union if union > 0 else 1.0) 452 per_pair.append(float(np.mean(ious))) 453 454 return TemporalStabilityResult( 455 ts_score=float(np.mean(per_pair)), 456 num_pairs=len(per_pair), 457 per_pair=per_pair, 458 )
TS_seg = E[IoU(P_t, warp(P_{t+1}, ΔT))].
When T265 pose data is available, we use the relative rotation to compensate for camera motion before computing IoU between adjacent frames. Without pixel-accurate warping (which requires depth for backprojection), we apply a simplified affine proxy using the in-plane rotation component only.
This gives a conservative lower-bound TS_seg that is still a meaningful stability signal when scenes have modest depth variation.
Args: pred_masks: sequence of predicted segmentation masks (H×W int). camera_poses: per-frame 4×4 SE(3) matrices (camera-to-world), or None.
Returns: TemporalStabilityResult.
def
compute_temporal_stability_depth( pred_depths: Sequence[numpy.ndarray], camera_poses: Sequence[numpy.ndarray | None]) -> TemporalStabilityResult:
461def compute_temporal_stability_depth( 462 pred_depths: Sequence[np.ndarray], 463 camera_poses: Sequence[np.ndarray | None], 464) -> TemporalStabilityResult: 465 """TS_depth = E[||D_t − warp(D_{t+1}, ΔT)||_1 / N_valid]. 466 467 Normalised to [0,1] by dividing by the max depth range to give a 468 higher-is-better stability score (TS = 1 − normalised_L1). 469 """ 470 if len(pred_depths) < 2: 471 return TemporalStabilityResult(ts_score=1.0, num_pairs=0) 472 473 per_pair = [] 474 for t in range(len(pred_depths) - 1): 475 d_t = np.asarray(pred_depths[t], dtype=np.float32) 476 d_t1 = np.asarray(pred_depths[t + 1], dtype=np.float32) 477 478 if camera_poses[t] is not None and camera_poses[t + 1] is not None: 479 d_t1_warped = _warp_depth_approx(d_t1, camera_poses[t], camera_poses[t + 1]) 480 else: 481 d_t1_warped = d_t1 482 483 valid = (d_t > 0) & (d_t1_warped > 0) 484 if valid.sum() == 0: 485 per_pair.append(1.0) 486 continue 487 l1 = float(np.abs(d_t[valid] - d_t1_warped[valid]).mean()) 488 depth_range = max(float(d_t[valid].max() - d_t[valid].min()), 1e-3) 489 ts = max(0.0, 1.0 - l1 / depth_range) 490 per_pair.append(ts) 491 492 return TemporalStabilityResult( 493 ts_score=float(np.mean(per_pair)), 494 num_pairs=len(per_pair), 495 per_pair=per_pair, 496 )
TS_depth = E[||D_t − warp(D_{t+1}, ΔT)||_1 / N_valid].
Normalised to [0,1] by dividing by the max depth range to give a higher-is-better stability score (TS = 1 − normalised_L1).
def
compute_sgc( pred_masks: Sequence[numpy.ndarray], pred_depths: Sequence[numpy.ndarray], depth_gradient_threshold: float = 0.1, boundary_dilation: int = 2) -> StackGeometricCoherenceResult:
499def compute_sgc( 500 pred_masks: Sequence[np.ndarray], 501 pred_depths: Sequence[np.ndarray], 502 depth_gradient_threshold: float = 0.1, 503 boundary_dilation: int = 2, 504) -> StackGeometricCoherenceResult: 505 """Stack-Level Geometric Coherence: boundary F-score between mask and depth edges. 506 507 SGC = F-score(boundary(mask), boundary(depth_gradient > τ)) 508 509 A high SGC means segmentation boundaries are geometrically consistent with 510 the depth discontinuities — indicating the model perceives coherent surfaces. 511 512 Args: 513 pred_masks: sequence of predicted segmentation masks (H×W int). 514 pred_depths: sequence of predicted depth maps (H×W float32, metres). 515 depth_gradient_threshold: τ for depth gradient thresholding. 516 boundary_dilation: pixel tolerance for boundary matching. 517 """ 518 if len(pred_masks) == 0: 519 return StackGeometricCoherenceResult( 520 sgc_score=0.0, precision=0.0, recall=0.0, num_samples=0 521 ) 522 523 precisions, recalls = [], [] 524 for mask, depth in zip(pred_masks, pred_depths, strict=False): 525 mask = np.asarray(mask, dtype=np.int32) 526 depth = np.asarray(depth, dtype=np.float32) 527 528 mask_boundary = _extract_boundary(mask, dilation=boundary_dilation) 529 depth_boundary = _extract_depth_boundary( 530 depth, threshold=depth_gradient_threshold, dilation=boundary_dilation 531 ) 532 533 tp = float((mask_boundary & depth_boundary).sum()) 534 fp = float((mask_boundary & ~depth_boundary).sum()) 535 fn = float((~mask_boundary & depth_boundary).sum()) 536 537 p = tp / (tp + fp) if (tp + fp) > 0 else 1.0 538 r = tp / (tp + fn) if (tp + fn) > 0 else 1.0 539 precisions.append(p) 540 recalls.append(r) 541 542 mean_p = float(np.mean(precisions)) 543 mean_r = float(np.mean(recalls)) 544 f1 = (2 * mean_p * mean_r) / (mean_p + mean_r) if (mean_p + mean_r) > 0 else 0.0 545 546 return StackGeometricCoherenceResult( 547 sgc_score=f1, 548 precision=mean_p, 549 recall=mean_r, 550 num_samples=len(pred_masks), 551 )
Stack-Level Geometric Coherence: boundary F-score between mask and depth edges.
SGC = F-score(boundary(mask), boundary(depth_gradient > τ))
A high SGC means segmentation boundaries are geometrically consistent with the depth discontinuities — indicating the model perceives coherent surfaces.
Args: pred_masks: sequence of predicted segmentation masks (H×W int). pred_depths: sequence of predicted depth maps (H×W float32, metres). depth_gradient_threshold: τ for depth gradient thresholding. boundary_dilation: pixel tolerance for boundary matching.
@dataclass
class
OperatingPoint:
638@dataclass 639class OperatingPoint: 640 """One (precision, accuracy, cost) triple for a model. 641 642 A model can publish multiple operating points — e.g. fp32, fp16, 643 int8 — each reported separately on the three RPX axes (task 644 accuracy, scene-change robustness, compute cost). RPX does not 645 combine these into a single composite score: the reader picks the 646 operating point and the axis their deployment cares about. 647 """ 648 649 precision: str # "fp32", "fp16", "bf16" 650 # Task performance (primary metric — δ1, mIoU, accuracy, etc.) 651 task_metric: float 652 task_metric_name: str # e.g. "delta1", "miou" 653 higher_is_better: bool 654 # Robustness 655 str_score: float # STR value (near 0 = robust) 656 # Hardware-agnostic cost (Tier 1) 657 flops_g: float 658 params_m: float 659 memory_traffic_gb: float | None = None
One (precision, accuracy, cost) triple for a model.
A model can publish multiple operating points — e.g. fp32, fp16, int8 — each reported separately on the three RPX axes (task accuracy, scene-change robustness, compute cost). RPX does not combine these into a single composite score: the reader picks the operating point and the axis their deployment cares about.