VBench Dynamic Degree — RAFT optical flow motion detection.

For each consecutive frame pair, computes optical flow via RAFT and takes the mean of the top 5% flow magnitudes. If enough pairs exceed an adaptive threshold, the video is classified as dynamic (1.0) vs static (0.0).

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

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

Move metric (and its internal models) to device.

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

reset() -> None

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

`fastvideo.eval.metrics.vbench.dynamic_degree.metric.BaseMetric.accumulate`

accumulate(sample: dict) -> None

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

`fastvideo.eval.metrics.vbench.dynamic_degree.metric.BaseMetric.finalize`

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

`fastvideo.eval.metrics.vbench.dynamic_degree.metric.BaseMetric.merge_from`

merge_from(other: BaseMetric) -> None

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

`fastvideo.eval.metrics.vbench.dynamic_degree.metric.register`

register(name: str)

Decorator to register a metric class.

Usage::

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

`fastvideo.eval.metrics.vbench.dynamic_degree.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.dynamic_degree.metric.DynamicDegreeMetric`

DynamicDegreeMetric()

Source

@register("vbench.dynamic_degree")
class DynamicDegreeMetric(BaseMetric):

    name = "vbench.dynamic_degree"
    requires_reference = False
    higher_is_better = True
    needs_gpu = True
    dependencies = ["easydict"]

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

    def to(self, device):
        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
        from easydict import EasyDict
        from vbench.third_party.RAFT.core.raft import RAFT

        args = EasyDict(small=False, mixed_precision=False, alternate_corr=False, dropout=0.0)
        model = torch.nn.DataParallel(RAFT(args))

        from fastvideo.eval.models import ensure_checkpoint
        ckpt_path = ensure_checkpoint(
            "raft-things.pth",
            source="sbalani/raft-things",
            filename="raft-things.pth",
        )
        model.load_state_dict(torch.load(ckpt_path, map_location="cpu"))
        model = model.module
        model.to(self.device)
        model.eval()
        self._model = model

    def _get_score(self, flow: torch.Tensor) -> float:
        """Top-5% mean flow magnitude (matching VBench dynamic_degree.get_score)."""
        flo = flow.permute(1, 2, 0).cpu().numpy()
        rad = np.sqrt(flo[..., 0]**2 + flo[..., 1]**2)
        h, w = rad.shape
        cut = max(1, int(h * w * 0.05))
        rad_flat = rad.flatten()
        return float(np.mean(np.sort(rad_flat)[-cut:]))

    @torch.no_grad()
    def compute(self, sample: dict) -> MetricResult:
        from vbench.third_party.RAFT.core.utils_core.utils import InputPadder

        video = sample["video"]  # (T, C, H, W) [0, 1]
        T, _, H, W = video.shape

        # fps controls the temporal sampling stride for optical flow.
        # vbench computes flow at 8fps (interval = round(fps/8)). The metric
        # cannot auto-derive fps from a tensor, so a missing fps would silently
        # use a wrong stride and produce a wrong score. Skip explicitly.
        if "fps" not in sample:
            return self._skip(sample, "missing 'fps' (required to set the "
                              "8fps optical-flow sampling stride)")
        fps = float(sample["fps"])
        interval = max(1, round(fps / 8.0))

        video_255 = video * 255.0
        chunk = self._chunk_size or 16

        # Cap chunk so the RAFT correlation volume doesn't overflow int32.
        # RAFT downsamples 8x in the feature encoder; CorrBlock's tensor is
        # shape (B*H1*W1, 1, H2, W2) with H1=H2=H/8, W1=W2=W/8. Its element
        # count is B*(H/8)^2*(W/8)^2 — F.avg_pool2d's index space starts to
        # overflow int32 around 2^31. Safety factor 2x.
        h_red = max(1, H // 8)
        w_red = max(1, W // 8)
        max_chunk = max(1, (1 << 30) // (h_red * h_red * w_red * w_red))
        chunk = min(chunk, max_chunk)

        indices = list(range(0, T, interval))
        n = len(indices)
        all_img1 = [video_255[indices[i]] for i in range(n - 1)]
        all_img2 = [video_255[indices[i + 1]] for i in range(n - 1)]

        scores: list[float] = []
        for start in range(0, len(all_img1), chunk):
            end = min(start + chunk, len(all_img1))
            img1_batch = torch.stack(all_img1[start:end]).to(self.device)
            img2_batch = torch.stack(all_img2[start:end]).to(self.device)
            padder = InputPadder(img1_batch.shape)
            img1p, img2p = padder.pad(img1_batch, img2_batch)
            _, flow = self._model(img1p, img2p, iters=20, test_mode=True)
            for i in range(flow.shape[0]):
                scores.append(self._get_score(flow[i]))

        scale = min(H, W)
        thres = 6.0 * (scale / 256.0)
        count_needed = round(4 * (n / 16.0))
        count_above = sum(1 for s in scores if s > thres)
        is_dynamic = 1.0 if count_above >= count_needed else 0.0

        return MetricResult(
            name=self.name,
            score=is_dynamic,
            details={
                "per_pair_magnitude": scores,
                "threshold": thres,
                "count_above": count_above,
                "count_needed": count_needed,
                "fps": fps,
                "interval": interval,
                "n_frames_used": n
            },
        )

`fastvideo.eval.metrics.vbench.dynamic_degree.metric.DynamicDegreeMetric.to`

to(device)

Source

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

`fastvideo.eval.metrics.vbench.dynamic_degree.metric.DynamicDegreeMetric.setup`

setup() -> None

Source

def setup(self) -> None:
    if self._model is not None:
        return
    from easydict import EasyDict
    from vbench.third_party.RAFT.core.raft import RAFT

    args = EasyDict(small=False, mixed_precision=False, alternate_corr=False, dropout=0.0)
    model = torch.nn.DataParallel(RAFT(args))

    from fastvideo.eval.models import ensure_checkpoint
    ckpt_path = ensure_checkpoint(
        "raft-things.pth",
        source="sbalani/raft-things",
        filename="raft-things.pth",
    )
    model.load_state_dict(torch.load(ckpt_path, map_location="cpu"))
    model = model.module
    model.to(self.device)
    model.eval()
    self._model = model

`fastvideo.eval.metrics.vbench.dynamic_degree.metric.DynamicDegreeMetric.compute`

compute(sample: dict) -> MetricResult

Source

@torch.no_grad()
def compute(self, sample: dict) -> MetricResult:
    from vbench.third_party.RAFT.core.utils_core.utils import InputPadder

    video = sample["video"]  # (T, C, H, W) [0, 1]
    T, _, H, W = video.shape

    # fps controls the temporal sampling stride for optical flow.
    # vbench computes flow at 8fps (interval = round(fps/8)). The metric
    # cannot auto-derive fps from a tensor, so a missing fps would silently
    # use a wrong stride and produce a wrong score. Skip explicitly.
    if "fps" not in sample:
        return self._skip(sample, "missing 'fps' (required to set the "
                          "8fps optical-flow sampling stride)")
    fps = float(sample["fps"])
    interval = max(1, round(fps / 8.0))

    video_255 = video * 255.0
    chunk = self._chunk_size or 16

    # Cap chunk so the RAFT correlation volume doesn't overflow int32.
    # RAFT downsamples 8x in the feature encoder; CorrBlock's tensor is
    # shape (B*H1*W1, 1, H2, W2) with H1=H2=H/8, W1=W2=W/8. Its element
    # count is B*(H/8)^2*(W/8)^2 — F.avg_pool2d's index space starts to
    # overflow int32 around 2^31. Safety factor 2x.
    h_red = max(1, H // 8)
    w_red = max(1, W // 8)
    max_chunk = max(1, (1 << 30) // (h_red * h_red * w_red * w_red))
    chunk = min(chunk, max_chunk)

    indices = list(range(0, T, interval))
    n = len(indices)
    all_img1 = [video_255[indices[i]] for i in range(n - 1)]
    all_img2 = [video_255[indices[i + 1]] for i in range(n - 1)]

    scores: list[float] = []
    for start in range(0, len(all_img1), chunk):
        end = min(start + chunk, len(all_img1))
        img1_batch = torch.stack(all_img1[start:end]).to(self.device)
        img2_batch = torch.stack(all_img2[start:end]).to(self.device)
        padder = InputPadder(img1_batch.shape)
        img1p, img2p = padder.pad(img1_batch, img2_batch)
        _, flow = self._model(img1p, img2p, iters=20, test_mode=True)
        for i in range(flow.shape[0]):
            scores.append(self._get_score(flow[i]))

    scale = min(H, W)
    thres = 6.0 * (scale / 256.0)
    count_needed = round(4 * (n / 16.0))
    count_above = sum(1 for s in scores if s > thres)
    is_dynamic = 1.0 if count_above >= count_needed else 0.0

    return MetricResult(
        name=self.name,
        score=is_dynamic,
        details={
            "per_pair_magnitude": scores,
            "threshold": thres,
            "count_above": count_above,
            "count_needed": count_needed,
            "fps": fps,
            "interval": interval,
            "n_frames_used": n
        },
    )