metric
fastvideo.eval.metrics.common.lpips.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: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.
fastvideo.eval.metrics.common.lpips.metric.BaseMetric.to
to(device: str | torch.device) -> BaseMetricMove metric (and its internal models) to device.
fastvideo.eval.metrics.common.lpips.metric.BaseMetric.setup
setup() -> NoneEagerly 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) -> MetricResultPer-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() -> NoneClear accumulator state at the start of each evaluate() call.
fastvideo.eval.metrics.common.lpips.metric.BaseMetric.accumulate
accumulate(sample: dict) -> NoneBuffer per-sample features for a corpus-level metric.
fastvideo.eval.metrics.common.lpips.metric.BaseMetric.finalize
finalize() -> MetricResultCompute the corpus-level result from buffered state.
fastvideo.eval.metrics.common.lpips.metric.BaseMetric.merge_from
merge_from(other: BaseMetric) -> NoneMulti-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()},
)</code></pre></details>
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()},
)</code></pre></details>