--- title: metric ---

fastvideo.eval.metrics.common.lpips.metric.BaseMetric

BaseMetric()

Abstract base class for all eval metrics.

Two execution shapes:

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.lpips.metric.BaseMetric.to

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

Move metric (and its internal models) to device.

fastvideo.eval.metrics.common.lpips.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.lpips.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.lpips.metric.BaseMetric.reset

reset() -> None

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

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

accumulate(sample: dict) -> None

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

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

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

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

merge_from(other: BaseMetric) -> None

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

fastvideo.eval.metrics.common.lpips.metric.register

register(name: str)

Decorator to register a metric class.

Usage::

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

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

LPIPSMetric(net: str = 'alex', chunk_size: int = 8)
Source
@register("common.lpips")
class LPIPSMetric(BaseMetric):
    name = "common.lpips"
    requires_reference = True
    higher_is_better = False
    needs_gpu = True
    dependencies = ["lpips"]

    def __init__(self, net: str = "alex", chunk_size: int = 8) -> None:
        super().__init__()
        self.net = net
        # Chunk the per-frame forward: AlexNet feature maps at 1080p
        # peak ~60 GB on a 121-frame clip; chunk=8 caps it at ~5 GB.
        self._chunk_size = chunk_size
        self._model: Any = None

    def to(self, device: str | torch.device) -> LPIPSMetric:
        super().to(device)
        if self._model is not None:
            self._model = self._model.to(self.device)
        return self

    def setup(self) -> None:
        if self._model is not None:
            return
        import lpips as lpips_lib
        self._model = lpips_lib.LPIPS(net=self.net).to(self.device)
        self._model.eval()

    def compute(self, sample: dict) -> MetricResult:
        if self._model is None:
            self.setup()

        gen = sample["video"].float().to(self.device, non_blocking=True)
        ref = sample["reference"].float().to(self.device, non_blocking=True)

        n = min(gen.shape[0], ref.shape[0])
        gen, ref = gen[:n] * 2.0 - 1.0, ref[:n] * 2.0 - 1.0

        chunk = self._chunk_size or n
        all_scores = []
        with torch.no_grad():
            for i in range(0, n, chunk):
                s = self._model(gen[i:i + chunk], ref[i:i + chunk]).squeeze()
                if s.dim() == 0:
                    s = s.unsqueeze(0)
                all_scores.append(s)
        scores = torch.cat(all_scores)  # (n,)

        return MetricResult(
            name=self.name,
            score=float(scores.mean()),
            details={"per_frame": scores.tolist()},
        )

fastvideo.eval.metrics.common.lpips.metric.LPIPSMetric.to

to(device: str | torch.device) -> LPIPSMetric
Source
def to(self, device: str | torch.device) -> LPIPSMetric:
    super().to(device)
    if self._model is not None:
        self._model = self._model.to(self.device)
    return self

fastvideo.eval.metrics.common.lpips.metric.LPIPSMetric.setup

setup() -> None
Source
def setup(self) -> None:
    if self._model is not None:
        return
    import lpips as lpips_lib
    self._model = lpips_lib.LPIPS(net=self.net).to(self.device)
    self._model.eval()

fastvideo.eval.metrics.common.lpips.metric.LPIPSMetric.compute

compute(sample: dict) -> MetricResult
Source
def compute(self, sample: dict) -> MetricResult:
    if self._model is None:
        self.setup()

    gen = sample["video"].float().to(self.device, non_blocking=True)
    ref = sample["reference"].float().to(self.device, non_blocking=True)

    n = min(gen.shape[0], ref.shape[0])
    gen, ref = gen[:n] * 2.0 - 1.0, ref[:n] * 2.0 - 1.0

    chunk = self._chunk_size or n
    all_scores = []
    with torch.no_grad():
        for i in range(0, n, chunk):
            s = self._model(gen[i:i + chunk], ref[i:i + chunk]).squeeze()
            if s.dim() == 0:
                s = s.unsqueeze(0)
            all_scores.append(s)
    scores = torch.cat(all_scores)  # (n,)

    return MetricResult(
        name=self.name,
        score=float(scores.mean()),
        details={"per_frame": scores.tolist()},
    )