base
fastvideo.eval.metrics.base.MetricResult
MetricResult(name: str, score: float | None, details: dict[str, Any] = dict())Standard result container returned by all metrics.
score is None when the metric was skipped (e.g. missing
required input). Check details["skipped"] for the reason.
fastvideo.eval.metrics.base.BaseMetric
BaseMetric()Abstract base class for all eval metrics.
Two execution shapes:
-
Per-sample (
is_set_metric=False, default) — implement :meth:compute. The Evaluator calls it once per input sample and returns one :class:MetricResultper sample. -
Set-vs-set (
is_set_metric=True) — implement :meth:accumulate(called once per sample to buffer features) and :meth:finalize(called once after all samples to compute the corpus-level result). Use :meth:resetto clear buffers and :meth:merge_fromto fold multi-GPU per-worker state together.
Optionally override :meth:setup to eagerly load models. Metrics
that chunk along the time dim for memory hardcode their own chunk
size in __init__ (see optical_flow for the canonical
example). Eval always processes one video per
:meth:Evaluator.evaluate call; compute / accumulate
receive a single sample, not a batch.
Source
class BaseMetric:
"""Abstract base class for all eval metrics.
Two execution shapes:
* **Per-sample** (``is_set_metric=False``, default) — implement
:meth:`compute`. The Evaluator calls it once per input sample and
returns one :class:`MetricResult` per sample.
* **Set-vs-set** (``is_set_metric=True``) — implement
:meth:`accumulate` (called once per sample to buffer features)
and :meth:`finalize` (called once after all samples to compute
the corpus-level result). Use :meth:`reset` to clear buffers and
:meth:`merge_from` to fold multi-GPU per-worker state together.
Optionally override :meth:`setup` to eagerly load models. Metrics
that chunk along the time dim for memory hardcode their own chunk
size in ``__init__`` (see ``optical_flow`` for the canonical
example). Eval always processes one video per
:meth:`Evaluator.evaluate` call; ``compute`` / ``accumulate``
receive a single sample, not a batch.
"""
name: str = ""
requires_reference: bool = True
higher_is_better: bool = True
dependencies: list[str] = []
needs_gpu: bool = False
backbone: str | None = None
is_set_metric: bool = False
# Default time-dim chunk size for metrics that batch internally over
# frames or frame-pairs. Override in subclass __init__ if needed
# (see ``optical_flow``, ``motion_smoothness``, ``dynamic_degree``).
_chunk_size: int | None = None
def __init__(self) -> None:
self._device: torch.device = torch.device("cpu")
@property
def device(self) -> torch.device:
return self._device
def to(self, device: str | torch.device) -> BaseMetric:
"""Move metric (and its internal models) to *device*."""
self._device = torch.device(device)
return self
def setup(self) -> None: # noqa: B027 - intentionally optional override
"""Eagerly load models. Called once by :class:`EvalWorker`.
Default is a no-op; metrics with no eager state (pixel math,
closed-form ops) inherit this. Override only if your metric
needs to load weights.
"""
def _skip(self, sample: dict, reason: str) -> MetricResult:
"""Return a skipped result (``score=None`` + reason in details)."""
return MetricResult(name=self.name, score=None, details={"skipped": reason})
def compute(self, sample: dict) -> MetricResult:
"""Per-sample metrics: compute the score for one sample.
``sample["video"]`` is ``(T, C, H, W)`` float in ``[0, 1]``.
``sample["reference"]`` (if used) has the same shape. Return
``self._skip(sample, reason)`` for missing inputs.
"""
raise NotImplementedError(f"{type(self).__name__}.compute is not implemented")
# --- set-vs-set protocol (only invoked when is_set_metric=True) ---
def reset(self) -> None: # noqa: B027 - intentionally optional override
"""Clear accumulator state at the start of each evaluate() call."""
def accumulate(self, sample: dict) -> None:
"""Buffer per-sample features for a corpus-level metric."""
raise NotImplementedError(f"{type(self).__name__}.accumulate is not implemented")
def finalize(self) -> MetricResult:
"""Compute the corpus-level result from buffered state."""
raise NotImplementedError(f"{type(self).__name__}.finalize is not implemented")
def merge_from(self, other: BaseMetric) -> None: # noqa: B027 - intentionally optional override
"""Multi-GPU: fold another worker's accumulator state into this one."""</code></pre></details>
fastvideo.eval.metrics.base.BaseMetric.to
to(device: str | torch.device) -> BaseMetric
Move metric (and its internal models) to device.
Source
def to(self, device: str | torch.device) -> BaseMetric:
"""Move metric (and its internal models) to *device*."""
self._device = torch.device(device)
return self
fastvideo.eval.metrics.base.BaseMetric.setup
setup() -> None
Eagerly load models. Called once by :class:EvalWorker.
Default is a no-op; metrics with no eager state (pixel math,
closed-form ops) inherit this. Override only if your metric
needs to load weights.
Source
def setup(self) -> None: # noqa: B027 - intentionally optional override
"""Eagerly load models. Called once by :class:`EvalWorker`.
Default is a no-op; metrics with no eager state (pixel math,
closed-form ops) inherit this. Override only if your metric
needs to load weights.
"""</code></pre></details>
fastvideo.eval.metrics.base.BaseMetric.compute
compute(sample: dict) -> MetricResult
Per-sample metrics: compute the score for one sample.
sample["video"] is (T, C, H, W) float in [0, 1].
sample["reference"] (if used) has the same shape. Return
self._skip(sample, reason) for missing inputs.
Source
def compute(self, sample: dict) -> MetricResult:
"""Per-sample metrics: compute the score for one sample.
``sample["video"]`` is ``(T, C, H, W)`` float in ``[0, 1]``.
``sample["reference"]`` (if used) has the same shape. Return
``self._skip(sample, reason)`` for missing inputs.
"""
raise NotImplementedError(f"{type(self).__name__}.compute is not implemented")</code></pre></details>
fastvideo.eval.metrics.base.BaseMetric.reset
reset() -> None
Clear accumulator state at the start of each evaluate() call.
Source
def reset(self) -> None: # noqa: B027 - intentionally optional override
"""Clear accumulator state at the start of each evaluate() call."""
fastvideo.eval.metrics.base.BaseMetric.accumulate
accumulate(sample: dict) -> None
Buffer per-sample features for a corpus-level metric.
Source
def accumulate(self, sample: dict) -> None:
"""Buffer per-sample features for a corpus-level metric."""
raise NotImplementedError(f"{type(self).__name__}.accumulate is not implemented")
fastvideo.eval.metrics.base.BaseMetric.finalize
finalize() -> MetricResult
Compute the corpus-level result from buffered state.
Source
def finalize(self) -> MetricResult:
"""Compute the corpus-level result from buffered state."""
raise NotImplementedError(f"{type(self).__name__}.finalize is not implemented")
fastvideo.eval.metrics.base.BaseMetric.merge_from
merge_from(other: BaseMetric) -> None
Multi-GPU: fold another worker's accumulator state into this one.
Source
def merge_from(self, other: BaseMetric) -> None: # noqa: B027 - intentionally optional override
"""Multi-GPU: fold another worker's accumulator state into this one."""