rpx_benchmark.metrics.depth_paper

  1"""Canonical RPX D1-F paper metrics and fixed camera calibration.
  2
  3The RPX paper declares one RGB-camera calibration for the released 640x480
  4benchmark.  The release does not include per-device calibration files, so this
  5module deliberately fails on any other resolution rather than inventing a
  6rescaling rule.
  7"""
  8
  9from __future__ import annotations
 10
 11from dataclasses import dataclass
 12
 13import numpy as np
 14
 15from ..api import DepthGroundTruth, DepthPrediction, TaskType
 16from ..exceptions import MetricError
 17from .registry import MetricSuite
 18
 19DEPTH_MIN_M = 0.3
 20DEPTH_MAX_M = 5.0
 21FSCORE_THRESHOLD_M = 0.05
 22EXPECTED_HW = (480, 640)
 23
 24
 25@dataclass(frozen=True)
 26class D1Calibration:
 27    fx: float = 615.0
 28    fy: float = 615.0
 29    cx: float = 320.0
 30    cy: float = 240.0
 31    width: int = 640
 32    height: int = 480
 33    version: str = "rpx-paper-d1-rgb-640x480-v1"
 34    provenance: str = "RPX-4.pdf Appendix I.1; fixed paper-declared RGB intrinsics"
 35
 36    def to_dict(self) -> dict[str, object]:
 37        return {
 38            "fx": self.fx,
 39            "fy": self.fy,
 40            "cx": self.cx,
 41            "cy": self.cy,
 42            "width": self.width,
 43            "height": self.height,
 44            "version": self.version,
 45            "provenance": self.provenance,
 46            "intrinsics_policy": "fixed paper-declared calibration; never rescaled",
 47            "coordinate_system": "RGB camera",
 48            "pixel_origin": "top-left pixel centre",
 49            "released_depth_alignment": "depth aligned to RGB color stream",
 50            "per_device_calibration": "not present in released HF manifests",
 51            "extrinsics_used": False,
 52        }
 53
 54
 55D1_CALIBRATION = D1Calibration()
 56PAPER_METRIC_KEYS = (
 57    "absrel",
 58    "rmse",
 59    "silog",
 60    "delta1",
 61    "irmse",
 62    "fscore_5cm",
 63)
 64FAST_PAPER_METRIC_KEYS = PAPER_METRIC_KEYS[:-1]
 65PREDICTION_EVALUATION_POLICY = {
 66    "version": "rpx-d1f-prediction-clip-v1",
 67    "operation": "clip",
 68    "minimum_m": DEPTH_MIN_M,
 69    "maximum_m": DEPTH_MAX_M,
 70    "applies_to": "evaluation copy before every D1-F metric; raw NPZ prediction is preserved",
 71    "reason": "enforce the declared finite positive D1-F evaluation domain uniformly",
 72    "trigger_observation": {
 73        "sample_id": "scene026__0__00155",
 74        "coordinates_yx": [[5, 323], [6, 323]],
 75        "raw_prediction_m": [-0.044158936, -0.061798096],
 76        "ground_truth_m": [1.134, 1.138],
 77        "paper_valid": [True, True],
 78    },
 79}
 80
 81
 82def valid_depth_mask(gt_m: np.ndarray) -> np.ndarray:
 83    """Paper validity mask: finite GT strictly inside the D435 range."""
 84    return np.isfinite(gt_m) & (gt_m > DEPTH_MIN_M) & (gt_m < DEPTH_MAX_M)
 85
 86
 87def clip_prediction_for_evaluation(prediction: np.ndarray) -> np.ndarray:
 88    """Return the uniformly range-clipped D1-F evaluation copy."""
 89    pred = np.asarray(prediction, dtype=np.float32)
 90    if not np.isfinite(pred).all():
 91        raise MetricError("Depth prediction contains non-finite values.")
 92    return np.clip(pred, DEPTH_MIN_M, DEPTH_MAX_M)
 93
 94
 95def _point_cloud(depth: np.ndarray, valid: np.ndarray) -> np.ndarray:
 96    y, x = np.nonzero(valid)
 97    z = depth[valid].astype(np.float64, copy=False)
 98    return np.column_stack(
 99        (
100            (x.astype(np.float64) - D1_CALIBRATION.cx) * z / D1_CALIBRATION.fx,
101            (y.astype(np.float64) - D1_CALIBRATION.cy) * z / D1_CALIBRATION.fy,
102            z,
103        )
104    )
105
106
107def _matched_fraction(tree, points: np.ndarray, *, chunk_size: int = 65_536) -> float:
108    matched = 0
109    for start in range(0, len(points), chunk_size):
110        distances, _ = tree.query(
111            points[start : start + chunk_size],
112            k=1,
113            workers=1,
114        )
115        matched += int(np.count_nonzero(distances < FSCORE_THRESHOLD_M))
116    return float(matched / len(points))
117
118
119def point_cloud_fscore_5cm(
120    pred_m: np.ndarray,
121    gt_m: np.ndarray,
122    valid: np.ndarray,
123) -> float:
124    """Bidirectional nearest-neighbour F-score at a strict 5 cm threshold."""
125    try:
126        from scipy.spatial import cKDTree
127    except ImportError as exc:  # pragma: no cover - Docker and test env include scipy
128        raise MetricError(
129            "F-Score@5cm requires scipy.spatial.cKDTree.",
130            hint="Install scipy>=1.10 or use the RPX depth-paper Docker image.",
131        ) from exc
132
133    gt_points = _point_cloud(gt_m, valid)
134    pred_points = _point_cloud(pred_m, valid)
135    if len(gt_points) == 0 or len(pred_points) == 0:
136        raise MetricError("F-Score@5cm received an empty valid point cloud.")
137
138    precision = _matched_fraction(cKDTree(gt_points), pred_points)
139    recall = _matched_fraction(cKDTree(pred_points), gt_points)
140    denom = precision + recall
141    return 0.0 if denom == 0.0 else float(2.0 * precision * recall / denom)
142
143
144def compute_d1_paper_metrics(
145    prediction: np.ndarray,
146    ground_truth: np.ndarray,
147    *,
148    include_fscore: bool = True,
149) -> dict[str, float]:
150    """Compute the D1-F headline metrics for one 640x480 frame.
151
152    ``include_fscore=False`` retains the exact paper mask and formulas for
153    the five image-space metrics while deferring the expensive point-cloud
154    calculation.  This is used by the staged production workflow; the saved
155    predictions can be evaluated for F-Score later without another model
156    forward.
157    """
158    pred_raw = np.asarray(prediction, dtype=np.float32)
159    gt = np.asarray(ground_truth, dtype=np.float32)
160    if pred_raw.shape != gt.shape:
161        raise MetricError(
162            f"Depth prediction shape {pred_raw.shape} does not match ground-truth shape {gt.shape}."
163        )
164    if gt.shape != EXPECTED_HW:
165        raise MetricError(
166            f"RPX D1-F paper calibration requires depth shape {EXPECTED_HW}; got {gt.shape}.",
167            hint="Do not rescale the fixed paper intrinsics; use the released 640x480 frames.",
168        )
169
170    pred = clip_prediction_for_evaluation(pred_raw)
171    valid = valid_depth_mask(gt)
172    if not valid.any():
173        raise MetricError("Depth frame contains no finite GT pixels in the 0.3-5.0 m range.")
174    pred_valid = pred[valid]
175    gt_valid = gt[valid]
176    diff = pred_valid - gt_valid
177    log_error = np.log(pred_valid) - np.log(gt_valid)
178    ratio = np.maximum(pred_valid / gt_valid, gt_valid / pred_valid)
179    silog_variance = max(
180        float(np.mean(log_error**2) - np.mean(log_error) ** 2),
181        0.0,
182    )
183    metrics = {
184        "absrel": float(np.mean(np.abs(diff) / gt_valid)),
185        "rmse": float(np.sqrt(np.mean(diff**2))),
186        "silog": float(100.0 * np.sqrt(silog_variance)),
187        "delta1": float(np.mean(ratio < 1.25)),
188        "irmse": float(np.sqrt(np.mean((1.0 / pred_valid - 1.0 / gt_valid) ** 2))),
189    }
190    if include_fscore:
191        metrics["fscore_5cm"] = point_cloud_fscore_5cm(pred, gt, valid)
192    if not all(np.isfinite(value) for value in metrics.values()):
193        raise MetricError("D1-F paper metric computation produced a non-finite value.")
194    return metrics
195
196
197class PaperDepthMetricSuite(MetricSuite):
198    """Metric suite dedicated to the strict RPX D1-F paper protocol."""
199
200    def __init__(self) -> None:
201        super().__init__(TaskType.MONOCULAR_DEPTH)
202
203    def evaluate(self, prediction: object, ground_truth: object) -> dict[str, float]:
204        if not isinstance(prediction, DepthPrediction):
205            raise MetricError(
206                f"PaperDepthMetricSuite expected DepthPrediction, got {type(prediction).__name__}."
207            )
208        if not isinstance(ground_truth, DepthGroundTruth):
209            raise MetricError(
210                "PaperDepthMetricSuite expected DepthGroundTruth, "
211                f"got {type(ground_truth).__name__}."
212            )
213        return compute_d1_paper_metrics(prediction.depth_map, ground_truth.depth_map)
214
215
216class FastPaperDepthMetricSuite(PaperDepthMetricSuite):
217    """Strict D1-F suite with only F-Score deferred to post-processing."""
218
219    def evaluate(self, prediction: object, ground_truth: object) -> dict[str, float]:
220        if not isinstance(prediction, DepthPrediction):
221            raise MetricError(
222                f"FastPaperDepthMetricSuite expected DepthPrediction, got "
223                f"{type(prediction).__name__}."
224            )
225        if not isinstance(ground_truth, DepthGroundTruth):
226            raise MetricError(
227                "FastPaperDepthMetricSuite expected DepthGroundTruth, "
228                f"got {type(ground_truth).__name__}."
229            )
230        return compute_d1_paper_metrics(
231            prediction.depth_map,
232            ground_truth.depth_map,
233            include_fscore=False,
234        )