metric

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.

@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,
    )</code></pre></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(&quot;actions&quot;)
if actions is None:
    return self._skip(sample, &quot;missing &#x27;actions&#x27; (keyboard + mouse)&quot;)

cal_obj = sample.get(&quot;calibration&quot;)
cal = self._calibration if cal_obj is None else _resolve_calibration(cal_obj)
if cal is None:
    return self._skip(
        sample,
        &quot;missing &#x27;calibration&#x27; (pass calibration_path= at construction &quot;
        &quot;or sample&#91;&#x27;calibration&#x27;&#93; per call)&quot;,
    )

video = sample&#91;&quot;video&quot;&#93;.float()  # (T, C, H, W)
T, _, H, W = video.shape
if T &lt; 2:
    raise ValueError(&quot;Need at least 2 frames to compute optical flow&quot;)
n_pairs = T - 1

mouse_pitch_sign = int(sample.get(&quot;mouse_pitch_sign&quot;, 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 = &#91;
    compute_frame_metrics(
        predicted&#91;i&#93;,
        observed&#91;i&#93;,
        grid_size=self.grid_size,
        min_mag=self.min_mag,
        max_mag_pct=self.max_mag_pct,
    ) for i in range(n)
&#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>