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

metric

Compare optical flow extracted from a generated video against optical flow extracted from a ground-truth reference video.

Both flows are produced by the same ptlflow model (default dpflow/things). The resulting per-pixel / per-frame / temporal metric set is identical to synthetic_optical_flow — only the way the reference flow is constructed differs.

fastvideo.eval.metrics.optical_flow.gt_optical_flow.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.optical_flow.gt_optical_flow.metric.BaseMetric.to

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

Move metric (and its internal models) to device.

fastvideo.eval.metrics.optical_flow.gt_optical_flow.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.optical_flow.gt_optical_flow.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.optical_flow.gt_optical_flow.metric.BaseMetric.reset

reset() -> None

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

fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.BaseMetric.accumulate

accumulate(sample: dict) -> None

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

fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.BaseMetric.finalize

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.BaseMetric.merge_from

merge_from(other: BaseMetric) -> None

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

fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.aggregate_temporal

aggregate_temporal(per_frame: list[dict[str, float]]) -> dict[str, float | int | None]

Aggregate per-frame metric dicts into mean/std/max/auc/onset summaries.

Port of mhuo's compute_temporal_metrics.

fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.compute_frame_metrics

compute_frame_metrics(flow_gt: np.ndarray, flow_gen: np.ndarray, grid_size: int = 8, min_mag: float = 0.5, max_mag_pct: float = 80.0) -> dict[str, float]

Per-frame comparison metrics between two HxWx2 flow fields.

Port of mhuo's compute_frame_metrics — see ptlflow_validation.py.

fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.extract_video_flows

extract_video_flows(model, video: torch.Tensor, *, chunk: int, device: torch.device) -> list[np.ndarray]

Run model on every consecutive frame pair in video.

Returns a list of HxWx2 flow arrays of length T - 1.

fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.load_ptlflow_model

load_ptlflow_model(model_name: str, ckpt: str, device: torch.device)

Load a ptlflow model on device in eval mode.

fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.register

register(name: str)

Decorator to register a metric class.

Usage::

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

fastvideo.eval.metrics.optical_flow.gt_optical_flow.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.optical_flow.gt_optical_flow.metric.GtOpticalFlowMetric

GtOpticalFlowMetric(model_name: str = 'dpflow', ckpt: str = 'things', min_mag: float = 0.5, max_mag_pct: float = 80.0, grid_size: int = 8)

Per-pixel / per-frame / temporal flow comparison vs. a reference video.

The headline score is pixel_epe_mean_mean (lower is better); every other scalar lives in details so downstream consumers can pick whichever one they care about.

Source
@register("optical_flow.gt_optical_flow")
class GtOpticalFlowMetric(BaseMetric):
    """Per-pixel / per-frame / temporal flow comparison vs. a reference video.

The headline ``score`` is ``pixel_epe_mean_mean`` (lower is better);
every other scalar lives in ``details`` so downstream consumers can
pick whichever one they care about.
"""

name = "optical_flow.gt_optical_flow"
requires_reference = True
higher_is_better = False
needs_gpu = True
backbone = "optical_flow"
dependencies = ["ptlflow"]

def __init__(
    self,
    model_name: str = "dpflow",
    ckpt: str = "things",
    min_mag: float = 0.5,
    max_mag_pct: float = 80.0,
    grid_size: int = 8,
) -> None:
    super().__init__()
    self.model_name = model_name
    self.ckpt = ckpt
    self.min_mag = min_mag
    self.max_mag_pct = max_mag_pct
    self.grid_size = grid_size
    self._model = None
    # One frame-pair per DPFlow forward: the cost volume is ~4 GB
    # per pair at 1080p, so batching OOMs. Bump for low-res inputs.
    self._chunk_size = 1

def to(self, device: str | torch.device) -> GtOpticalFlowMetric:
    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
    self._model = load_ptlflow_model(self.model_name, self.ckpt, self.device)

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

    gen_video = sample["video"].float()  # (T, C, H, W)
    ref_video = sample["reference"].float()
    n = min(gen_video.shape[0], ref_video.shape[0])
    gen_video, ref_video = gen_video[:n], ref_video[:n]
    if n < 2:
        raise ValueError("Need at least 2 frames to compute optical flow")

    chunk = self._chunk_size or 16
    gen_flows = extract_video_flows(
        self._model,
        gen_video,
        chunk=chunk,
        device=self.device,
    )
    ref_flows = extract_video_flows(
        self._model,
        ref_video,
        chunk=chunk,
        device=self.device,
    )
    per_frame = [
        compute_frame_metrics(
            rf,
            gf,
            grid_size=self.grid_size,
            min_mag=self.min_mag,
            max_mag_pct=self.max_mag_pct,
        ) for rf, gf in zip(ref_flows, gen_flows, strict=False)
    ]
    summary = aggregate_temporal(per_frame)
    score = summary.get("pixel_epe_mean_mean")
    details = dict(summary)
    details["per_frame_metrics"] = per_frame
    return MetricResult(
        name=self.name,
        score=float(score) if score is not None else None,
        details=details,
    )</code></pre></details>


fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.GtOpticalFlowMetric.to


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


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


fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.GtOpticalFlowMetric.setup


setup() -> None


Source
def setup(self) -> None:
    if self._model is not None:
        return
    self._model = load_ptlflow_model(self.model_name, self.ckpt, self.device)


fastvideo.eval.metrics.optical_flow.gt_optical_flow.metric.GtOpticalFlowMetric.compute


compute(sample: dict) -> MetricResult


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

gen_video = sample&#91;&quot;video&quot;&#93;.float()  # (T, C, H, W)
ref_video = sample&#91;&quot;reference&quot;&#93;.float()
n = min(gen_video.shape&#91;0&#93;, ref_video.shape&#91;0&#93;)
gen_video, ref_video = gen_video&#91;:n&#93;, ref_video&#91;:n&#93;
if n &lt; 2:
    raise ValueError(&quot;Need at least 2 frames to compute optical flow&quot;)

chunk = self._chunk_size or 16
gen_flows = extract_video_flows(
    self._model,
    gen_video,
    chunk=chunk,
    device=self.device,
)
ref_flows = extract_video_flows(
    self._model,
    ref_video,
    chunk=chunk,
    device=self.device,
)
per_frame = &#91;
    compute_frame_metrics(
        rf,
        gf,
        grid_size=self.grid_size,
        min_mag=self.min_mag,
        max_mag_pct=self.max_mag_pct,
    ) for rf, gf in zip(ref_flows, gen_flows, strict=False)
&#93;
summary = aggregate_temporal(per_frame)
score = summary.get(&quot;pixel_epe_mean_mean&quot;)
details = dict(summary)
details&#91;&quot;per_frame_metrics&quot;&#93; = per_frame
return MetricResult(
    name=self.name,
    score=float(score) if score is not None else None,
    details=details,
)</code></pre></details>

Was this page helpful?