rpx_benchmark.metrics

33class MetricCalculator(ABC):
34    """Abstract base for a family of metrics bound to a single task.
35
36    Subclasses are lightweight, stateless objects. The ``compute``
37    method accepts one prediction and its ground truth and returns a
38    dict mapping metric name → scalar float.
39
40    Subclasses must set ``name`` (human-readable identifier used for
41    registration) and implement :meth:`compute`. Subclasses should
42    take their own hyperparameters via ``__init__`` when needed.
43
44    Examples
45    --------
46    >>> from rpx_benchmark.metrics import MetricCalculator, register_metric
47    >>> from rpx_benchmark.api import TaskType
48    >>> @register_metric(TaskType.MONOCULAR_DEPTH)
49    ... class MeanPred(MetricCalculator):
50    ...     name = "mean_pred"
51    ...     def compute(self, prediction, ground_truth):
52    ...         return {"mean_pred": float(prediction.depth_map.mean())}
53    """
54
55    name: str = ""
56
57    @abstractmethod
58    def compute(self, prediction: Any, ground_truth: Any) -> Dict[str, float]:
59        """Return a dict of metric name → scalar for one sample.
60
61        Parameters
62        ----------
63        prediction : Any
64            Task-specific Prediction dataclass
65            (e.g. :class:`DepthPrediction`). The concrete type is
66            whatever the calculator is designed for.
67        ground_truth : Any
68            Task-specific GroundTruth dataclass
69            (e.g. :class:`DepthGroundTruth`).
70
71        Returns
72        -------
73        dict[str, float]
74            Metric values. Keys should be short snake_case names;
75            values must be numeric so the runner can average them.
76
77        Raises
78        ------
79        MetricError
80            When inputs have the wrong shape/type/content.
81        """

255class MetricSuite:
256    """Thin wrapper around the metric registry used by the runner.
257
258    Kept as a class rather than a function because historical API
259    expects ``MetricSuite.for_task(...).evaluate(pred, gt)``. New code
260    can call :func:`compute_metrics` directly.
261    """
262
263    def __init__(self, task: TaskType) -> None:
264        self.task = task
265
266    @classmethod
267    def for_task(cls, task: TaskType) -> "MetricSuite":
268        """Create a suite for the given task.
269
270        Raises
271        ------
272        MetricError
273            If no calculators are registered for ``task``.
274        """
275        if not get_calculators(task):
276            raise MetricError(
277                f"No metric calculator registered for task {task.value!r}.",
278                hint="Import rpx_benchmark.metrics to trigger built-ins.",
279            )
280        return cls(task=task)
281
282    def evaluate(self, prediction: Any, ground_truth: Any) -> Dict[str, float]:
283        """Run every registered calculator and return merged results.
284
285        Raises
286        ------
287        MetricError
288            Propagated from individual calculators when inputs are
289            shape-mismatched or wrong-typed.
290        """
291        return compute_metrics(self.task, prediction, ground_truth)
292
293    def aggregate(self, per_sample: List[Dict[str, Any]]) -> Dict[str, float]:
294        """Mean over numeric metric keys; non-numeric metadata is skipped.
295
296        Parameters
297        ----------
298        per_sample : list of dict
299            Per-sample rows. May contain metric floats and metadata
300            strings/enums in the same dict.
301
302        Returns
303        -------
304        dict[str, float]
305            One float per numeric key. Empty dict if ``per_sample`` is
306            empty or contains no numeric values.
307        """
308        if not per_sample:
309            return {}
310        numeric_keys = [
311            k
312            for k, v in per_sample[0].items()
313            if isinstance(v, (int, float)) and not isinstance(v, bool)
314        ]
315        out: Dict[str, float] = {}
316        for k in numeric_keys:
317            vals = [
318                m[k]
319                for m in per_sample
320                if isinstance(m.get(k), (int, float)) and not isinstance(m.get(k), bool)
321            ]
322            if vals:
323                out[k] = float(np.mean(vals))
324        return out
325
326    def build_result(self, per_sample: List[Dict[str, Any]]) -> "BenchmarkResult":
327        """Convenience: wrap ``per_sample`` and its aggregate in a
328        :class:`BenchmarkResult`.
329        """
330        return BenchmarkResult(
331            task=self.task,
332            per_sample=per_sample,
333            aggregated=self.aggregate(per_sample),
334            num_samples=len(per_sample),
335        )

174def available_metrics() -> Dict[TaskType, List[str]]:
175    """List the registered metric names grouped by task.
176
177    Returns
178    -------
179    dict
180        ``{TaskType: [calculator_name, ...]}``. Only tasks with at
181        least one registered calculator appear.
182    """
183    return {task: [c.name for c in calcs] for task, calcs in _CALCULATORS.items() if calcs}

164def clear_registry() -> None:
165    """Remove every registered calculator. Primarily for tests."""
166    _CALCULATORS.clear()

186def compute_metrics(
187    task: TaskType,
188    prediction: Any,
189    ground_truth: Any,
190) -> Dict[str, float]:
191    """Run every registered calculator for ``task`` and merge outputs.
192
193    Parameters
194    ----------
195    task : TaskType
196    prediction : Any
197    ground_truth : Any
198
199    Returns
200    -------
201    dict[str, float]
202        Union of all calculators' output dicts. Later calculators may
203        overwrite earlier ones if they emit the same key.
204
205    Raises
206    ------
207    MetricError
208        If no calculators are registered for ``task``.
209    """
210    calcs = _CALCULATORS.get(task)
211    if not calcs:
212        raise MetricError(
213            f"No metric calculator registered for task {task.value!r}.",
214            hint=(
215                "Either import the built-in calculators (e.g. "
216                "`import rpx_benchmark.metrics`) or register a custom "
217                "one via @register_metric."
218            ),
219        )
220    merged: Dict[str, float] = {}
221    for calc in calcs:
222        merged.update(calc.compute(prediction, ground_truth))
223    return merged

169def get_calculators(task: TaskType) -> List[MetricCalculator]:
170    """Return the calculators registered for ``task`` (empty list if none)."""
171    return list(_CALCULATORS.get(task, []))

 91def register_metric(
 92    task: TaskType,
 93) -> Callable[[Type[MetricCalculator]], Type[MetricCalculator]]:
 94    """Class decorator registering a calculator under a task.
 95
 96    Parameters
 97    ----------
 98    task : TaskType
 99        Task the calculator belongs to. Multiple calculators can be
100        registered against the same task and will be composed at
101        evaluation time.
102
103    Returns
104    -------
105    Callable
106        The decorator; when applied to a class it instantiates it
107        once with no arguments and appends to the registry.
108
109    Raises
110    ------
111    MetricError
112        If ``task`` is not a :class:`TaskType`, or if the decorated
113        object is not a :class:`MetricCalculator` subclass.
114
115    Examples
116    --------
117    >>> from rpx_benchmark.api import TaskType
118    >>> from rpx_benchmark.metrics import MetricCalculator, register_metric
119    >>> @register_metric(TaskType.MONOCULAR_DEPTH)
120    ... class MyMetric(MetricCalculator):
121    ...     name = "my_metric"
122    ...     def compute(self, prediction, ground_truth):
123    ...         return {"my_metric": 0.0}
124    """
125    if not isinstance(task, TaskType):
126        raise MetricError(
127            f"register_metric expected a TaskType, got {type(task).__name__}",
128            hint="Use one of the members of rpx_benchmark.api.TaskType.",
129        )
130
131    def decorator(cls: Type[MetricCalculator]) -> Type[MetricCalculator]:
132        if not (isinstance(cls, type) and issubclass(cls, MetricCalculator)):
133            raise MetricError(
134                f"@register_metric can only decorate MetricCalculator subclasses; got {cls!r}",
135            )
136        instance = cls()
137        _CALCULATORS[task].append(instance)
138        log.debug("registered metric %s for task %s", instance.name or cls.__name__, task.value)
139        return cls
140
141    return decorator

144def unregister_metric(task: TaskType, name: str) -> bool:
145    """Remove a previously registered calculator by its ``name`` field.
146
147    Parameters
148    ----------
149    task : TaskType
150    name : str
151        Name attribute of the calculator to remove.
152
153    Returns
154    -------
155    bool
156        True if a calculator was removed; False if no match was found.
157    """
158    before = len(_CALCULATORS.get(task, []))
159    _CALCULATORS[task] = [c for c in _CALCULATORS.get(task, []) if c.name != name]
160    after = len(_CALCULATORS[task])
161    return after < before