Compare optical flow extracted from a generated video against optical flow synthesized analytically from per-frame actions.

The reference flow is not observed from a ground-truth video — it's predicted from the action stream via a third-person camera-kinematics model (Longuet-Higgins linearization + off-pivot translation correction; no depth). Observed flow comes from the same ptlflow model used by gt_optical_flow, and the two are compared with the identical metric set, so scores are directly comparable across the two metrics.

Required sample keys

video (B, T, C, H, W) float in [0, 1]. actions dict (or list-of-dicts of length B) with two np.ndarray keys:

* ``keyboard`` of shape ``(T, 6)`` — ``[W, S, A, D, turn_left, turn_right]``
* ``mouse`` of shape ``(T, 2)`` — ``[pitch, yaw]``

calibration Either a path to a ThirdPersonCalibration JSON file, or a dict of fitted parameters. May also be set once at construction time via calibration_path= and reused across samples.

Optional sample keys

mouse_pitch_sign +1 (default) or -1 if the dataset's mouse-pitch sign is flipped (mhuo's data carries this in metadata).

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

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

Move metric (and its internal models) to device.

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

reset() -> None

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

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.BaseMetric.accumulate`

accumulate(sample: dict) -> None

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

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.BaseMetric.finalize`

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

`fastvideo.eval.metrics.optical_flow.synthetic_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.synthetic_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.synthetic_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.synthetic_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.synthetic_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.synthetic_optical_flow.metric.ThirdPersonCalibration`

ThirdPersonCalibration(alpha_yaw: float, alpha_pitch: float, alpha_turn: float, beta_fwd: float, beta_strafe: float, focal_length: float, r_z: float = 0.0, r_y: float = 0.0, init_pitch: float = 0.0, notes: str = '', fit_metadata: dict = dict())

Fitted parameters for a third-person rig.

r_y defaults to 0 (avatar at camera height). focal_length is in pixels. init_pitch is the camera's rest pitch (radians, negative means tilted down — typical over-the-shoulder framing). User mouse-pitch input is integrated on top of this baseline.

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.ThirdPersonCalibration.to_json`

to_json(path: str | Path) -> None

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.ThirdPersonCalibration.from_dict`

from_dict(d: dict) -> ThirdPersonCalibration

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.ThirdPersonFlowGenerator`

ThirdPersonFlowGenerator(calibration: ThirdPersonCalibration, frame_shape: tuple[int, int], mouse_pitch_sign: int = +1)

Vectorized 3P synthetic-flow generator. No depth.

Parameters

calibration : ThirdPersonCalibration frame_shape : (H, W) mouse_pitch_sign : +1 or -1 — sample-level flag from metadata (mouse_pitch_flipped: true in mhuo's data ⇒ -1).

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.ThirdPersonFlowGenerator.generate_flow`

generate_flow(keyboard: np.ndarray, mouse: np.ndarray) -> np.ndarray

Synthesize HxWx2 flow for one frame's action.

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.ThirdPersonFlowGenerator.generate_flow_sequence`

generate_flow_sequence(actions: dict, n_pairs: int | None = None) -> list[np.ndarray]

Generate flow for each consecutive frame pair.

Returns n flows where n = n_pairs if supplied, else len(keyboard) - 1.

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.load_calibration`

load_calibration(path: str | Path) -> ThirdPersonCalibration

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.register`

register(name: str)

Decorator to register a metric class.

Usage::

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

`fastvideo.eval.metrics.optical_flow.synthetic_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.synthetic_optical_flow.metric.SyntheticOpticalFlowMetric`

SyntheticOpticalFlowMetric(model_name: str = 'dpflow', ckpt: str = 'things', calibration_path: str | Path | None = None, min_mag: float = 0.5, max_mag_pct: float = 80.0, grid_size: int = 8)

Action-driven synthetic flow vs. video-extracted observed flow.

Pass calibration_path at construction to bind the calibration once across all samples; otherwise supply sample["calibration"] per call. Missing actions or calibration produce a skipped result (score=None) rather than raising.

Source

@register("optical_flow.synthetic_optical_flow")
class SyntheticOpticalFlowMetric(BaseMetric):
    """Action-driven synthetic flow vs. video-extracted observed flow.

    Pass ``calibration_path`` at construction to bind the calibration
    once across all samples; otherwise supply ``sample["calibration"]``
    per call. Missing actions or calibration produce a skipped result
    (``score=None``) rather than raising.
    """

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

    def __init__(
        self,
        model_name: str = "dpflow",
        ckpt: str = "things",
        calibration_path: str | Path | None = None,
        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._calibration: ThirdPersonCalibration | None = (_resolve_calibration(calibration_path)
                                                            if calibration_path else None)
        self._model = None
        # One frame-pair per DPFlow forward; see ``gt_optical_flow``.
        self._chunk_size = 1

    def to(self, device: str | torch.device) -> SyntheticOpticalFlowMetric:
        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()

        actions = sample.get("actions")
        if actions is None:
            return self._skip(sample, "missing 'actions' (keyboard + mouse)")

        cal_obj = sample.get("calibration")
        cal = self._calibration if cal_obj is None else _resolve_calibration(cal_obj)
        if cal is None:
            return self._skip(
                sample,
                "missing 'calibration' (pass calibration_path= at construction "
                "or sample['calibration'] per call)",
            )

        video = sample["video"].float()  # (T, C, H, W)
        T, _, H, W = video.shape
        if T < 2:
            raise ValueError("Need at least 2 frames to compute optical flow")
        n_pairs = T - 1

        mouse_pitch_sign = int(sample.get("mouse_pitch_sign", 1))
        chunk = self._chunk_size or 16

        observed = extract_video_flows(
            self._model,
            video,
            chunk=chunk,
            device=self.device,
        )
        predictor = ThirdPersonFlowGenerator(
            calibration=cal,
            frame_shape=(H, W),
            mouse_pitch_sign=mouse_pitch_sign,
        )
        predicted = predictor.generate_flow_sequence(actions, n_pairs=n_pairs)

        n = min(len(observed), len(predicted))
        per_frame = [
            compute_frame_metrics(
                predicted[i],
                observed[i],
                grid_size=self.grid_size,
                min_mag=self.min_mag,
                max_mag_pct=self.max_mag_pct,
            ) for i in range(n)
        ]
        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,
        )

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.SyntheticOpticalFlowMetric.to`

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

Source

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

`fastvideo.eval.metrics.optical_flow.synthetic_optical_flow.metric.SyntheticOpticalFlowMetric.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.synthetic_optical_flow.metric.SyntheticOpticalFlowMetric.compute`

compute(sample: dict) -> MetricResult

Source

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

    actions = sample.get("actions")
    if actions is None:
        return self._skip(sample, "missing 'actions' (keyboard + mouse)")

    cal_obj = sample.get("calibration")
    cal = self._calibration if cal_obj is None else _resolve_calibration(cal_obj)
    if cal is None:
        return self._skip(
            sample,
            "missing 'calibration' (pass calibration_path= at construction "
            "or sample['calibration'] per call)",
        )

    video = sample["video"].float()  # (T, C, H, W)
    T, _, H, W = video.shape
    if T < 2:
        raise ValueError("Need at least 2 frames to compute optical flow")
    n_pairs = T - 1

    mouse_pitch_sign = int(sample.get("mouse_pitch_sign", 1))
    chunk = self._chunk_size or 16

    observed = extract_video_flows(
        self._model,
        video,
        chunk=chunk,
        device=self.device,
    )
    predictor = ThirdPersonFlowGenerator(
        calibration=cal,
        frame_shape=(H, W),
        mouse_pitch_sign=mouse_pitch_sign,
    )
    predicted = predictor.generate_flow_sequence(actions, n_pairs=n_pairs)

    n = min(len(observed), len(predicted))
    per_frame = [
        compute_frame_metrics(
            predicted[i],
            observed[i],
            grid_size=self.grid_size,
            min_mag=self.min_mag,
            max_mag_pct=self.max_mag_pct,
        ) for i in range(n)
    ]
    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,
    )