Skip to content
FastVideo
Esc
↑↓navigate↵open⌘Jpreview
On this page

metric

VBench Motion Smoothness — AMT-S frame interpolation quality.

Takes every-other frame, uses AMT-S to interpolate the missing middle frames, then compares interpolated vs actual frames. Score = (255 - mean_pixel_diff) / 255. Higher = smoother motion.

fastvideo.eval.metrics.vbench.motion_smoothness.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.vbench.motion_smoothness.metric.BaseMetric.to

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

Move metric (and its internal models) to device.

fastvideo.eval.metrics.vbench.motion_smoothness.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.vbench.motion_smoothness.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.vbench.motion_smoothness.metric.BaseMetric.reset

reset() -> None

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

fastvideo.eval.metrics.vbench.motion_smoothness.metric.BaseMetric.accumulate

accumulate(sample: dict) -> None

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

fastvideo.eval.metrics.vbench.motion_smoothness.metric.BaseMetric.finalize

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

fastvideo.eval.metrics.vbench.motion_smoothness.metric.BaseMetric.merge_from

merge_from(other: BaseMetric) -> None

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

fastvideo.eval.metrics.vbench.motion_smoothness.metric.register

register(name: str)

Decorator to register a metric class.

Usage::

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

fastvideo.eval.metrics.vbench.motion_smoothness.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.vbench.motion_smoothness.metric.MotionSmoothnessMetric

MotionSmoothnessMetric()
Source
@register("vbench.motion_smoothness")
class MotionSmoothnessMetric(BaseMetric):

name = "vbench.motion_smoothness"
requires_reference = False
higher_is_better = True
needs_gpu = True
dependencies = ["omegaconf"]

def __init__(self) -> None:
    super().__init__()
    self._model: Any = None
    self._embt: Any = None
    self._chunk_size = 8

def to(self, device):
    super().to(device)
    if self._model is not None:
        self._model = self._model.to(self.device)
    if self._embt is not None:
        self._embt = self._embt.to(self.device)
    return self

def setup(self) -> None:
    if self._model is not None:
        return
    from omegaconf import OmegaConf
    import vbench.third_party.amt as _amt_pkg
    from vbench.third_party.amt.utils.build_utils import build_from_cfg
    from fastvideo.eval.models import ensure_checkpoint

    amt_dir = os.path.dirname(_amt_pkg.__file__)
    cfg_path = os.path.join(amt_dir, "cfgs", "AMT-S.yaml")

    ckpt_path = ensure_checkpoint(
        "amt-s.pth",
        source="https://huggingface.co/lalala125/AMT/resolve/main/amt-s.pth",
    )

    network_cfg = OmegaConf.load(cfg_path).network
    self._model = build_from_cfg(network_cfg)
    ckpt = torch.load(ckpt_path, map_location="cpu", weights_only=False)
    self._model.load_state_dict(ckpt["state_dict"])
    self._model.to(self.device)
    self._model.eval()

    self._embt = torch.tensor(1 / 2).float().view(1, 1, 1, 1).to(self.device)

def _get_scale(self, h: int, w: int) -> float:
    """Pick a downscale factor that keeps AMT's correlation volume
    within free GPU memory.

    Re-queries free memory on every call (rather than caching at setup
    time) so the scale adapts to whatever's actually available — other
    metric replicas already loaded, residual generator allocations,
    another process sharing the GPU, etc. The upstream version cached
    ``total_memory`` at setup, which on a shared/loaded GPU lets AMT
    attempt a 30+ GB correlation volume reshape and OOM.
    """
    if self.device.type != "cuda":
        return 1.0
    # Free memory that won't be claimed by other allocations during this
    # forward pass. min(free, total) is conservative against transient
    # spikes; mem_get_info returns (free, total) in bytes.
    free_bytes, _ = torch.cuda.mem_get_info(self.device)
    anchor_resolution = 1024 * 512
    anchor_memory = 1500 * 1024**2
    anchor_memory_bias = 2500 * 1024**2
    if free_bytes <= anchor_memory_bias:
        # Less than the model + scratch overhead is free; force the
        # most aggressive downscale we support.
        return 1 / 16
    scale = anchor_resolution / (h * w) * np.sqrt((free_bytes - anchor_memory_bias) / anchor_memory)
    if scale >= 1.0:
        return 1.0
    scale = 1 / np.floor(1 / np.sqrt(scale) * 16) * 16
    return float(scale)

_MIN_AMT_SCALE = 1 / 16

def _safe_amt_forward(self, in0: Any, in1: Any, embt: Any, scale: Any) -> Any:
    """Run one chunk through AMT with OOM-retry on two axes.

    Recovery strategy on ``CUDA out of memory``:

    1. **Halve the batch** until batch=1. AMT's per-pair memory
       dominates; splitting helps until each pair is on its own.
    2. **Halve the scale_factor** passed to AMT (which controls its
       internal feature-map resolution and therefore the correlation
       volume size). Bottoms out at ``_MIN_AMT_SCALE`` — beyond that
       the feature maps are too coarse to produce meaningful
       interpolation and we re-raise.

    Self-tunes under memory pressure: the upstream autoscale formula
    in :meth:`_get_scale` mis-extrapolates at large resolutions
    (treats memory as linear in pixel count, but AMT's correlation
    volume grows quadratically). This retry path makes the metric
    robust to that without rewriting the formula.
    """
    try:
        return self._model(in0, in1, embt, scale_factor=scale, eval=True)["imgt_pred"]
    except torch.cuda.OutOfMemoryError:
        torch.cuda.empty_cache()
        bs = in0.shape[0]
        if bs > 1:
            half = bs // 2
            a = self._safe_amt_forward(in0[:half], in1[:half], embt[:half], scale)
            b = self._safe_amt_forward(in0[half:], in1[half:], embt[half:], scale)
            return torch.cat([a, b], dim=0)
        if scale > self._MIN_AMT_SCALE:
            return self._safe_amt_forward(in0, in1, embt, scale / 2)
        raise

@torch.no_grad()
def compute(self, sample: dict) -> MetricResult:
    from vbench.third_party.amt.utils.utils import (
        img2tensor,
        tensor2img,
        check_dim_and_resize,
        InputPadder,
    )

    video = sample["video"]  # (T, C, H, W) [0, 1]
    chunk = self._chunk_size or 8

    frames_np = (video * 255).to(torch.uint8).cpu().numpy()
    frames_np = [f.transpose(1, 2, 0) for f in frames_np]  # list of (H,W,C)

    even_indices = list(range(0, len(frames_np), 2))
    if len(even_indices) <= 1:
        return MetricResult(name=self.name, score=1.0, details={})

    even_frames = [frames_np[i] for i in even_indices]
    inputs = [img2tensor(f).to(self.device) for f in even_frames]
    inputs = check_dim_and_resize(inputs)

    h, w = inputs[0].shape[-2:]
    scale = self._get_scale(h, w)
    padding = int(16 / scale)
    padder = InputPadder(inputs[0].shape, padding)
    inputs = padder.pad(*inputs)

    n_pairs = len(inputs) - 1
    all_in0 = [inputs[i] for i in range(n_pairs)]
    all_in1 = [inputs[i + 1] for i in range(n_pairs)]
    all_gt = [
        frames_np[even_indices[i] + 1] if even_indices[i] + 1 < len(frames_np) else frames_np[-1]
        for i in range(n_pairs)
    ]

    all_preds = []
    for start in range(0, len(all_in0), chunk):
        end = min(start + chunk, len(all_in0))
        in0_batch = torch.cat(all_in0[start:end], dim=0).to(self.device)
        in1_batch = torch.cat(all_in1[start:end], dim=0).to(self.device)
        embt = self._embt.expand(in0_batch.shape[0], -1, -1, -1)
        pred = self._safe_amt_forward(in0_batch, in1_batch, embt, scale)
        all_preds.append(pred.cpu())
    all_preds = torch.cat(all_preds, dim=0)

    diffs: list[float] = []
    for i in range(n_pairs):
        pred = all_preds[i:i + 1]
        pred_unpadded = padder.unpad(pred)[0]
        pred_np = tensor2img(pred_unpadded)
        gt_np = all_gt[i]
        # gt comes from frames_np; pred goes through check_dim_and_resize +
        # AMT pad/unpad, which can reshape. Match shapes before absdiff.
        if gt_np.shape[:2] != pred_np.shape[:2]:
            gt_np = cv2.resize(gt_np, (pred_np.shape[1], pred_np.shape[0]), interpolation=cv2.INTER_AREA)
        diffs.append(float(np.mean(cv2.absdiff(gt_np, pred_np))))

    vfi_score = float(np.mean(diffs)) if diffs else 0.0
    return MetricResult(
        name=self.name,
        score=(255.0 - vfi_score) / 255.0,
        details={"vfi_score": vfi_score},
    )</code></pre></details>


fastvideo.eval.metrics.vbench.motion_smoothness.metric.MotionSmoothnessMetric.to


to(device)


Source
def to(self, device):
    super().to(device)
    if self._model is not None:
        self._model = self._model.to(self.device)
    if self._embt is not None:
        self._embt = self._embt.to(self.device)
    return self


fastvideo.eval.metrics.vbench.motion_smoothness.metric.MotionSmoothnessMetric.setup


setup() -> None


Source
def setup(self) -> None:
    if self._model is not None:
        return
    from omegaconf import OmegaConf
    import vbench.third_party.amt as _amt_pkg
    from vbench.third_party.amt.utils.build_utils import build_from_cfg
    from fastvideo.eval.models import ensure_checkpoint

amt_dir = os.path.dirname(_amt_pkg.__file__)
cfg_path = os.path.join(amt_dir, &quot;cfgs&quot;, &quot;AMT-S.yaml&quot;)

ckpt_path = ensure_checkpoint(
    &quot;amt-s.pth&quot;,
    source=&quot;https://huggingface.co/lalala125/AMT/resolve/main/amt-s.pth&quot;,
)

network_cfg = OmegaConf.load(cfg_path).network
self._model = build_from_cfg(network_cfg)
ckpt = torch.load(ckpt_path, map_location=&quot;cpu&quot;, weights_only=False)
self._model.load_state_dict(ckpt&#91;&quot;state_dict&quot;&#93;)
self._model.to(self.device)
self._model.eval()

self._embt = torch.tensor(1 / 2).float().view(1, 1, 1, 1).to(self.device)</code></pre></details>


fastvideo.eval.metrics.vbench.motion_smoothness.metric.MotionSmoothnessMetric.compute


compute(sample: dict) -> MetricResult


Source
@torch.no_grad()
def compute(self, sample: dict) -> MetricResult:
    from vbench.third_party.amt.utils.utils import (
        img2tensor,
        tensor2img,
        check_dim_and_resize,
        InputPadder,
    )

video = sample&#91;&quot;video&quot;&#93;  # (T, C, H, W) &#91;0, 1&#93;
chunk = self._chunk_size or 8

frames_np = (video * 255).to(torch.uint8).cpu().numpy()
frames_np = &#91;f.transpose(1, 2, 0) for f in frames_np&#93;  # list of (H,W,C)

even_indices = list(range(0, len(frames_np), 2))
if len(even_indices) &lt;= 1:
    return MetricResult(name=self.name, score=1.0, details=&#123;&#125;)

even_frames = &#91;frames_np&#91;i&#93; for i in even_indices&#93;
inputs = &#91;img2tensor(f).to(self.device) for f in even_frames&#93;
inputs = check_dim_and_resize(inputs)

h, w = inputs&#91;0&#93;.shape&#91;-2:&#93;
scale = self._get_scale(h, w)
padding = int(16 / scale)
padder = InputPadder(inputs&#91;0&#93;.shape, padding)
inputs = padder.pad(*inputs)

n_pairs = len(inputs) - 1
all_in0 = &#91;inputs&#91;i&#93; for i in range(n_pairs)&#93;
all_in1 = &#91;inputs&#91;i + 1&#93; for i in range(n_pairs)&#93;
all_gt = &#91;
    frames_np&#91;even_indices&#91;i&#93; + 1&#93; if even_indices&#91;i&#93; + 1 &lt; len(frames_np) else frames_np&#91;-1&#93;
    for i in range(n_pairs)
&#93;

all_preds = &#91;&#93;
for start in range(0, len(all_in0), chunk):
    end = min(start + chunk, len(all_in0))
    in0_batch = torch.cat(all_in0&#91;start:end&#93;, dim=0).to(self.device)
    in1_batch = torch.cat(all_in1&#91;start:end&#93;, dim=0).to(self.device)
    embt = self._embt.expand(in0_batch.shape&#91;0&#93;, -1, -1, -1)
    pred = self._safe_amt_forward(in0_batch, in1_batch, embt, scale)
    all_preds.append(pred.cpu())
all_preds = torch.cat(all_preds, dim=0)

diffs: list&#91;float&#93; = &#91;&#93;
for i in range(n_pairs):
    pred = all_preds&#91;i:i + 1&#93;
    pred_unpadded = padder.unpad(pred)&#91;0&#93;
    pred_np = tensor2img(pred_unpadded)
    gt_np = all_gt&#91;i&#93;
    # gt comes from frames_np; pred goes through check_dim_and_resize +
    # AMT pad/unpad, which can reshape. Match shapes before absdiff.
    if gt_np.shape&#91;:2&#93; != pred_np.shape&#91;:2&#93;:
        gt_np = cv2.resize(gt_np, (pred_np.shape&#91;1&#93;, pred_np.shape&#91;0&#93;), interpolation=cv2.INTER_AREA)
    diffs.append(float(np.mean(cv2.absdiff(gt_np, pred_np))))

vfi_score = float(np.mean(diffs)) if diffs else 0.0
return MetricResult(
    name=self.name,
    score=(255.0 - vfi_score) / 255.0,
    details=&#123;&quot;vfi_score&quot;: vfi_score&#125;,
)</code></pre></details>

Was this page helpful?