rpx_benchmark.metrics.depth

  1"""Monocular absolute depth metric calculators.
  2
  3Registers :class:`DepthErrorMetrics` which computes AbsRel, RMSE, and
  4three thresholded accuracy ratios (δ<1.25, δ<1.25², δ<1.25³) against
  5the subset of pixels where the ground-truth depth map is valid
  6(``gt > 0``). This matches the paper's eval protocol: we mask out
  7D435-invalid pixels so models are not penalised where the sensor
  8itself produced no measurement.
  9"""
 10
 11from __future__ import annotations
 12
 13from typing import Dict
 14
 15import numpy as np
 16
 17from ..api import DepthGroundTruth, DepthPrediction, TaskType
 18from ..exceptions import MetricError
 19from .registry import MetricCalculator, register_metric
 20
 21
 22@register_metric(TaskType.MONOCULAR_DEPTH)
 23class DepthErrorMetrics(MetricCalculator):
 24    """AbsRel / RMSE / δ-accuracy for monocular metric depth.
 25
 26    Notes
 27    -----
 28    All metrics are computed in metres. Neither raw logits nor any
 29    kind of scale alignment is applied — we deliberately measure the
 30    model's *absolute* metric accuracy, which is the property that
 31    matters for robot policies that consume depth directly.
 32
 33    For empty validity masks (fully invalid GT frames) we return the
 34    safe no-op sentinel ``{"absrel": 0, "rmse": 0, "delta1": 1, ...}``
 35    so a single bad frame does not break aggregate means. In practice
 36    the dataset pipeline should filter these out upstream.
 37
 38    Metric definitions
 39    ------------------
 40
 41    .. math::
 42
 43        \\text{AbsRel} &= \\frac{1}{N} \\sum_i \\frac{|\\hat d_i - d_i|}{d_i}
 44
 45        \\text{RMSE}   &= \\sqrt{\\frac{1}{N}\\sum_i(\\hat d_i - d_i)^2}
 46
 47        \\delta_k      &= \\frac{1}{N}\\,\\#\\Big\\{i : \\max\\!\\big(\\tfrac{\\hat d_i}{d_i},
 48                          \\tfrac{d_i}{\\hat d_i}\\big) < 1.25^{\\,k}\\Big\\}
 49    """
 50
 51    name = "depth_error_metrics"
 52
 53    def compute(
 54        self,
 55        prediction: DepthPrediction,
 56        ground_truth: DepthGroundTruth,
 57    ) -> Dict[str, float]:
 58        if not isinstance(prediction, DepthPrediction):
 59            raise MetricError(
 60                f"DepthErrorMetrics expected DepthPrediction, got {type(prediction).__name__}",
 61            )
 62        if not isinstance(ground_truth, DepthGroundTruth):
 63            raise MetricError(
 64                f"DepthErrorMetrics expected DepthGroundTruth, got {type(ground_truth).__name__}",
 65            )
 66
 67        pred = np.asarray(prediction.depth_map, dtype=np.float32)
 68        gt = np.asarray(ground_truth.depth_map, dtype=np.float32)
 69
 70        if pred.shape != gt.shape:
 71            raise MetricError(
 72                f"Depth prediction shape {pred.shape} does not match ground-truth shape {gt.shape}",
 73                hint=(
 74                    "The adapter's OutputAdapter should resize predictions "
 75                    "back to the sample RGB resolution before returning."
 76                ),
 77            )
 78
 79        valid = gt > 0
 80        if not valid.any():
 81            return {"rmse": 0.0, "absrel": 0.0, "delta1": 1.0, "delta2": 1.0, "delta3": 1.0}
 82
 83        pred_v = pred[valid]
 84        gt_v = gt[valid]
 85
 86        # Guard against pathological predictions (0 or negative) that
 87        # would blow up the ratio-based thresholds.
 88        pred_v = np.clip(pred_v, a_min=1e-6, a_max=None)
 89
 90        rmse = float(np.sqrt(np.mean((pred_v - gt_v) ** 2)))
 91        absrel = float(np.mean(np.abs(pred_v - gt_v) / gt_v))
 92
 93        thresh = np.maximum(pred_v / gt_v, gt_v / pred_v)
 94        delta1 = float(np.mean(thresh < 1.25))
 95        delta2 = float(np.mean(thresh < 1.25**2))
 96        delta3 = float(np.mean(thresh < 1.25**3))
 97
 98        return {
 99            "rmse": rmse,
100            "absrel": absrel,
101            "delta1": delta1,
102            "delta2": delta2,
103            "delta3": delta3,
104        }
105
106
107def _log_depth_pair(prediction: DepthPrediction, ground_truth: DepthGroundTruth):
108    from .depth_alignment import default_valid_mask
109
110    if not isinstance(prediction, DepthPrediction) or not isinstance(ground_truth, DepthGroundTruth):
111        raise MetricError("Expected DepthPrediction and DepthGroundTruth")
112    pred = np.asarray(prediction.depth_map, dtype=np.float64)
113    gt = np.asarray(ground_truth.depth_map, dtype=np.float64)
114    if pred.shape != gt.shape:
115        raise MetricError(f"Depth shape mismatch: {pred.shape} != {gt.shape}")
116    return pred, gt, default_valid_mask(pred, gt)
117
118
119def _log_scores(pred: np.ndarray, gt: np.ndarray) -> Dict[str, float]:
120    if not pred.size:
121        return {"rmselog": 0.0, "silog": 0.0}
122    error = np.log(pred) - np.log(gt)
123    return {"rmselog": float(np.sqrt(np.mean(error ** 2))),
124            "silog": float(100.0 * np.std(error))}
125
126
127@register_metric(TaskType.MONOCULAR_DEPTH)
128class DepthLogMetrics(MetricCalculator):
129    """Natural-log RMSE and SILog (100 times the log-error standard deviation).
130
131    Uses the shared 0.3 < GT < 5 m validity mask. Empty frames retain the
132    historical zero sentinel; callers should filter empty GT frames upstream.
133    """
134
135    name = "depth_log_metrics"
136
137    def compute(self, prediction: DepthPrediction, ground_truth: DepthGroundTruth) -> Dict[str, float]:
138        pred, gt, valid = _log_depth_pair(prediction, ground_truth)
139        return _log_scores(pred[valid], gt[valid])
140
141
142@register_metric(TaskType.MONOCULAR_DEPTH)
143class DepthRangeStratified(MetricCalculator):
144    """Depth quality in near/mid/far bins; fewer than 100 pixels yields NaN."""
145
146    name = "depth_range_stratified"
147
148    def compute(self, prediction: DepthPrediction, ground_truth: DepthGroundTruth) -> Dict[str, float]:
149        from .depth_temporal import DEPTH_RANGE_BINS
150
151        pred, gt, valid = _log_depth_pair(prediction, ground_truth)
152        result: Dict[str, float] = {}
153        for name, (lo, hi) in DEPTH_RANGE_BINS.items():
154            selected = valid & (gt >= lo) & (gt < hi)
155            values = dict.fromkeys(("absrel", "rmse", "rmselog", "silog", "delta1"), float("nan"))
156            if selected.sum() >= 100:
157                p, g = pred[selected], gt[selected]
158                values.update(_log_scores(p, g))
159                values.update(absrel=float(np.mean(np.abs(p-g)/g)),
160                              rmse=float(np.sqrt(np.mean((p-g)**2))),
161                              delta1=float(np.mean(np.maximum(p/g, g/p) < 1.25)))
162            result.update({f"{metric}_{name}": value for metric, value in values.items()})
163        return result