metric

fastvideo.eval.metrics.common.psnr.metric.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: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.

fastvideo.eval.metrics.common.psnr.metric.BaseMetric.to

to(device: str | torch.device) -> BaseMetric

Move metric (and its internal models) to device.

fastvideo.eval.metrics.common.psnr.metric.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.

fastvideo.eval.metrics.common.psnr.metric.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.

fastvideo.eval.metrics.common.psnr.metric.BaseMetric.reset

reset() -> None

Clear accumulator state at the start of each evaluate() call.

fastvideo.eval.metrics.common.psnr.metric.BaseMetric.accumulate

accumulate(sample: dict) -> None

Buffer per-sample features for a corpus-level metric.

fastvideo.eval.metrics.common.psnr.metric.BaseMetric.finalize

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

fastvideo.eval.metrics.common.psnr.metric.BaseMetric.merge_from

merge_from(other: BaseMetric) -> None

Multi-GPU: fold another worker's accumulator state into this one.

fastvideo.eval.metrics.common.psnr.metric.register

register(name: str)

Decorator to register a metric class.

Usage::

@register("ssim")
class SSIMMetric(BaseMetric):
    ...

fastvideo.eval.metrics.common.psnr.metric.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.common.psnr.metric.PSNRMetric

PSNRMetric(max_val: float = 1.0, chunk_size: int = 32)

@register("common.psnr")
class PSNRMetric(BaseMetric):
    name = "common.psnr"
    requires_reference = True
    higher_is_better = True
    # On CPU the squared-diff is memory-bandwidth-bound at 1080p and
    # fights the loader for the host bus; trivial on GPU.
    needs_gpu = True
def __init__(self, max_val: float = 1.0, chunk_size: int = 32) -> None:
    super().__init__()
    self.max_val = max_val
    # Keeps the (gen - ref)**2 intermediate under ~800 MB at 1080p.
    self._chunk_size = chunk_size

def compute(self, sample: dict) -> MetricResult:
    gen = sample["video"].float().to(self.device)  # (T, C, H, W)
    ref = sample["reference"].float().to(self.device)
    n = min(gen.shape[0], ref.shape[0])
    gen, ref = gen[:n], ref[:n]

    chunk = self._chunk_size or n
    mse_parts = []
    for i in range(0, n, chunk):
        g = gen[i:i + chunk]
        r = ref[i:i + chunk]
        mse_parts.append(((g - r)**2).mean(dim=(1, 2, 3)))
    mse = torch.cat(mse_parts)  # (T,)
    psnr = 10.0 * torch.log10(self.max_val**2 / mse.clamp(min=1e-10))

    return MetricResult(
        name=self.name,
        score=psnr.mean().item(),
        details={"per_frame": psnr.tolist()},
    )</code></pre></details>
fastvideo.eval.metrics.common.psnr.metric.PSNRMetric.compute
compute(sample: dict) -> MetricResult
Source
def compute(self, sample: dict) -> MetricResult:
    gen = sample["video"].float().to(self.device)  # (T, C, H, W)
    ref = sample["reference"].float().to(self.device)
    n = min(gen.shape[0], ref.shape[0])
    gen, ref = gen[:n], ref[:n]
chunk = self._chunk_size or n
mse_parts = &#91;&#93;
for i in range(0, n, chunk):
    g = gen&#91;i:i + chunk&#93;
    r = ref&#91;i:i + chunk&#93;
    mse_parts.append(((g - r)**2).mean(dim=(1, 2, 3)))
mse = torch.cat(mse_parts)  # (T,)
psnr = 10.0 * torch.log10(self.max_val**2 / mse.clamp(min=1e-10))

return MetricResult(
    name=self.name,
    score=psnr.mean().item(),
    details=&#123;&quot;per_frame&quot;: psnr.tolist()&#125;,
)</code></pre></details>