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

metric

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

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

Move metric (and its internal models) to device.

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

reset() -> None

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

fastvideo.eval.metrics.physics_iq.metric.BaseMetric.accumulate

accumulate(sample: dict) -> None

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

fastvideo.eval.metrics.physics_iq.metric.BaseMetric.finalize

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

fastvideo.eval.metrics.physics_iq.metric.BaseMetric.merge_from

merge_from(other: BaseMetric) -> None

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

fastvideo.eval.metrics.physics_iq.metric.PhysicsIQMSEMetric

PhysicsIQMSEMetric(**kwargs: Any)

fastvideo.eval.metrics.physics_iq.metric.PhysicsIQMSEMetric.compute

compute(sample: dict) -> MetricResult

fastvideo.eval.metrics.physics_iq.metric.SpatialIoUMetric

SpatialIoUMetric(**kwargs: Any)

fastvideo.eval.metrics.physics_iq.metric.SpatialIoUMetric.compute

compute(sample: dict) -> MetricResult

fastvideo.eval.metrics.physics_iq.metric.SpatiotemporalIoUMetric

SpatiotemporalIoUMetric(**kwargs: Any)

fastvideo.eval.metrics.physics_iq.metric.SpatiotemporalIoUMetric.compute

compute(sample: dict) -> MetricResult

fastvideo.eval.metrics.physics_iq.metric.WeightedSpatialIoUMetric

WeightedSpatialIoUMetric(**kwargs: Any)

fastvideo.eval.metrics.physics_iq.metric.WeightedSpatialIoUMetric.compute

compute(sample: dict) -> MetricResult

fastvideo.eval.metrics.physics_iq.metric.mean

mean(values: list[float] | tuple[float, ...]) -> float

fastvideo.eval.metrics.physics_iq.metric.prepare_pair_inputs

prepare_pair_inputs(generated: Any, reference: Any, *, generated_mask: Any | None = None, reference_mask: Any | None = None, target_fps: int = DEFAULT_TARGET_FPS, duration_seconds: int = DEFAULT_DURATION_SECONDS, video_time_selection: str = 'first', threshold: int = 10, alpha: float = 0.3, roundtrip_generated_masks: bool = True) -> PreparedPhysicsIQPair

fastvideo.eval.metrics.physics_iq.metric.prepare_triplet_inputs

prepare_triplet_inputs(generated: Any, reference: Any, reference_take2: Any, *, generated_mask: Any | None = None, reference_mask: Any | None = None, reference_take2_mask: Any | None = None, target_fps: int = DEFAULT_TARGET_FPS, duration_seconds: int = DEFAULT_DURATION_SECONDS, video_time_selection: str = 'first', threshold: int = 10, alpha: float = 0.3, roundtrip_generated_masks: bool = True) -> PreparedPhysicsIQTriplet

fastvideo.eval.metrics.physics_iq.metric.register

register(name: str)

Decorator to register a metric class.

Usage::

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

fastvideo.eval.metrics.physics_iq.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.physics_iq.metric.PhysicsIQMetric

PhysicsIQMetric(*, target_fps: int = DEFAULT_TARGET_FPS, duration_seconds: int = DEFAULT_DURATION_SECONDS, video_time_selection: str = 'first', threshold: int = 10, alpha: float = 0.3, roundtrip_generated_masks: bool = True)
Source
@register("physics_iq")
class PhysicsIQMetric(BaseMetric):
    name = "physics_iq"
    requires_reference = True
    higher_is_better = True

def __init__(
    self,
    *,
    target_fps: int = DEFAULT_TARGET_FPS,
    duration_seconds: int = DEFAULT_DURATION_SECONDS,
    video_time_selection: str = "first",
    threshold: int = 10,
    alpha: float = 0.3,
    roundtrip_generated_masks: bool = True,
) -> None:
    super().__init__()
    self._prep_kwargs = {
        "target_fps": target_fps,
        "duration_seconds": duration_seconds,
        "video_time_selection": video_time_selection,
        "threshold": threshold,
        "alpha": alpha,
        "roundtrip_generated_masks": roundtrip_generated_masks,
    }
    self._mse = PhysicsIQMSEMetric(**self._prep_kwargs)
    self._spatiotemporal_iou = SpatiotemporalIoUMetric(**self._prep_kwargs)
    self._spatial_iou = SpatialIoUMetric(**self._prep_kwargs)
    self._weighted_spatial_iou = WeightedSpatialIoUMetric(**self._prep_kwargs)

@staticmethod
def _extract_payload(result: MetricResult | Mapping[str, Any]) -> Mapping[str, Any]:
    if isinstance(result, MetricResult):
        return result.details
    return result

def _compute_pair_metrics(self, prepared_pair) -> dict[str, Any]:
    sample = {"_physics_iq_pair": prepared_pair}
    mse = self._mse.compute(sample)
    st = self._spatiotemporal_iou.compute(sample)
    spatial = self._spatial_iou.compute(sample)
    weighted = self._weighted_spatial_iou.compute(sample)
    return {
        "mse_per_frame": mse.details["per_frame"],
        "spatiotemporal_iou_per_frame": st.details["per_frame"],
        "spatial_iou": float(spatial.score),
        "weighted_spatial_iou": float(weighted.score),
        "mse_mean": float(mse.score),
        "spatiotemporal_iou_mean": float(st.score),
    }

def compute_single(
    self,
    generated: Any,
    reference: Any,
    reference_take2: Any,
    *,
    generated_mask: Any | None = None,
    reference_mask: Any | None = None,
    reference_take2_mask: Any | None = None,
    scenario: str | None = None,
    view: str | None = None,
) -> dict[str, Any]:
    prepared = prepare_triplet_inputs(
        generated,
        reference,
        reference_take2,
        generated_mask=generated_mask,
        reference_mask=reference_mask,
        reference_take2_mask=reference_take2_mask,
        **self._prep_kwargs,
    )
    pair_metrics = self._compute_pair_metrics(prepared)
    variance_pair = prepare_pair_inputs(
        reference,
        reference_take2,
        generated_mask=reference_mask,
        reference_mask=reference_take2_mask,
        **self._prep_kwargs,
    )
    variance_metrics = self._compute_pair_metrics(variance_pair)
    details = {
        **pair_metrics,
        "pv_mse_per_frame": variance_metrics["mse_per_frame"],
        "pv_spatiotemporal_iou_per_frame": variance_metrics["spatiotemporal_iou_per_frame"],
        "pv_spatial_iou": variance_metrics["spatial_iou"],
        "pv_weighted_spatial_iou": variance_metrics["weighted_spatial_iou"],
        "pv_mse_mean": variance_metrics["mse_mean"],
        "pv_spatiotemporal_iou_mean": variance_metrics["spatiotemporal_iou_mean"],
    }
    if scenario is not None:
        details["scenario"] = scenario
    if view is not None:
        details["view"] = view
    return details

@classmethod
def aggregate(cls, results_list: Iterable[MetricResult | Mapping[str, Any]]) -> float:
    payloads = [cls._extract_payload(result) for result in results_list]
    if not payloads:
        raise ValueError("PhysicsIQMetric.aggregate requires at least one result.")

    a_mse = mean([value for payload in payloads for value in payload["mse_per_frame"]])
    a_st = mean([value for payload in payloads for value in payload["spatiotemporal_iou_per_frame"]])
    a_s = mean([float(payload["spatial_iou"]) for payload in payloads])
    a_ws = mean([float(payload["weighted_spatial_iou"]) for payload in payloads])

    v_mse = mean([value for payload in payloads for value in payload["pv_mse_per_frame"]])
    v_st = mean([value for payload in payloads for value in payload["pv_spatiotemporal_iou_per_frame"]])
    v_s = mean([float(payload["pv_spatial_iou"]) for payload in payloads])
    v_ws = mean([float(payload["pv_weighted_spatial_iou"]) for payload in payloads])

    score = 100.0 * ((((a_st / v_st) + (a_s / v_s) + (a_ws / v_ws)) / 3.0) - (a_mse - v_mse))
    return round(float(np.clip(score, 0.0, 100.0)), 2)

@classmethod
def aggregate_components(cls, results_list: Iterable[MetricResult | Mapping[str, Any]]) -> dict[str, float]:
    payloads = [cls._extract_payload(result) for result in results_list]
    return {
        "physics_iq": cls.aggregate(payloads),
        "a_mse": mean([value for payload in payloads for value in payload["mse_per_frame"]]),
        "a_st": mean([value for payload in payloads for value in payload["spatiotemporal_iou_per_frame"]]),
        "a_s": mean([float(payload["spatial_iou"]) for payload in payloads]),
        "a_ws": mean([float(payload["weighted_spatial_iou"]) for payload in payloads]),
        "v_mse": mean([value for payload in payloads for value in payload["pv_mse_per_frame"]]),
        "v_st": mean([value for payload in payloads for value in payload["pv_spatiotemporal_iou_per_frame"]]),
        "v_s": mean([float(payload["pv_spatial_iou"]) for payload in payloads]),
        "v_ws": mean([float(payload["pv_weighted_spatial_iou"]) for payload in payloads]),
    }

def _per_video_score(self, details: Mapping[str, Any]) -> float:
    score = 100.0 * (
        ((mean(details["spatiotemporal_iou_per_frame"]) / mean(details["pv_spatiotemporal_iou_per_frame"])) +
         (float(details["spatial_iou"]) / float(details["pv_spatial_iou"])) +
         (float(details["weighted_spatial_iou"]) / float(details["pv_weighted_spatial_iou"]))) / 3.0 -
        (mean(details["mse_per_frame"]) - mean(details["pv_mse_per_frame"])))
    return round(float(np.clip(score, 0.0, 100.0)), 2)

def compute(self, sample: dict) -> MetricResult:
    if "reference" not in sample:
        raise KeyError("PhysicsIQMetric requires sample['reference'].")

    take2_key = None
    for candidate in ("reference_take2", "real_take2", "take2"):
        if candidate in sample:
            take2_key = candidate
            break
    if take2_key is None:
        raise KeyError("PhysicsIQMetric requires sample['reference_take2'] or an alias.")

    video = sample["video"]
    reference = sample["reference"]
    reference_take2 = sample[take2_key]
    generated_mask = sample.get("video_mask")
    reference_mask = sample.get("reference_mask")
    reference_take2_mask = sample.get("reference_take2_mask")

    scenario = sample.get("scenario")
    view = sample.get("view")

    details = self.compute_single(
        video,
        reference,
        reference_take2,
        generated_mask=generated_mask,
        reference_mask=reference_mask,
        reference_take2_mask=reference_take2_mask,
        scenario=scenario,
        view=view,
    )
    return MetricResult(name=self.name, score=self._per_video_score(details), details=details)</code></pre></details>


fastvideo.eval.metrics.physics_iq.metric.PhysicsIQMetric.compute_single


compute_single(generated: Any, reference: Any, reference_take2: Any, *, generated_mask: Any | None = None, reference_mask: Any | None = None, reference_take2_mask: Any | None = None, scenario: str | None = None, view: str | None = None) -> dict[str, Any]


Source
def compute_single(
    self,
    generated: Any,
    reference: Any,
    reference_take2: Any,
    *,
    generated_mask: Any | None = None,
    reference_mask: Any | None = None,
    reference_take2_mask: Any | None = None,
    scenario: str | None = None,
    view: str | None = None,
) -> dict[str, Any]:
    prepared = prepare_triplet_inputs(
        generated,
        reference,
        reference_take2,
        generated_mask=generated_mask,
        reference_mask=reference_mask,
        reference_take2_mask=reference_take2_mask,
        **self._prep_kwargs,
    )
    pair_metrics = self._compute_pair_metrics(prepared)
    variance_pair = prepare_pair_inputs(
        reference,
        reference_take2,
        generated_mask=reference_mask,
        reference_mask=reference_take2_mask,
        **self._prep_kwargs,
    )
    variance_metrics = self._compute_pair_metrics(variance_pair)
    details = {
        **pair_metrics,
        "pv_mse_per_frame": variance_metrics["mse_per_frame"],
        "pv_spatiotemporal_iou_per_frame": variance_metrics["spatiotemporal_iou_per_frame"],
        "pv_spatial_iou": variance_metrics["spatial_iou"],
        "pv_weighted_spatial_iou": variance_metrics["weighted_spatial_iou"],
        "pv_mse_mean": variance_metrics["mse_mean"],
        "pv_spatiotemporal_iou_mean": variance_metrics["spatiotemporal_iou_mean"],
    }
    if scenario is not None:
        details["scenario"] = scenario
    if view is not None:
        details["view"] = view
    return details


fastvideo.eval.metrics.physics_iq.metric.PhysicsIQMetric.aggregate


aggregate(results_list: Iterable[MetricResult | Mapping[str, Any]]) -> float


Source
@classmethod
def aggregate(cls, results_list: Iterable[MetricResult | Mapping[str, Any]]) -> float:
    payloads = [cls._extract_payload(result) for result in results_list]
    if not payloads:
        raise ValueError("PhysicsIQMetric.aggregate requires at least one result.")

a_mse = mean(&#91;value for payload in payloads for value in payload&#91;&quot;mse_per_frame&quot;&#93;&#93;)
a_st = mean(&#91;value for payload in payloads for value in payload&#91;&quot;spatiotemporal_iou_per_frame&quot;&#93;&#93;)
a_s = mean(&#91;float(payload&#91;&quot;spatial_iou&quot;&#93;) for payload in payloads&#93;)
a_ws = mean(&#91;float(payload&#91;&quot;weighted_spatial_iou&quot;&#93;) for payload in payloads&#93;)

v_mse = mean(&#91;value for payload in payloads for value in payload&#91;&quot;pv_mse_per_frame&quot;&#93;&#93;)
v_st = mean(&#91;value for payload in payloads for value in payload&#91;&quot;pv_spatiotemporal_iou_per_frame&quot;&#93;&#93;)
v_s = mean(&#91;float(payload&#91;&quot;pv_spatial_iou&quot;&#93;) for payload in payloads&#93;)
v_ws = mean(&#91;float(payload&#91;&quot;pv_weighted_spatial_iou&quot;&#93;) for payload in payloads&#93;)

score = 100.0 * ((((a_st / v_st) + (a_s / v_s) + (a_ws / v_ws)) / 3.0) - (a_mse - v_mse))
return round(float(np.clip(score, 0.0, 100.0)), 2)</code></pre></details>


fastvideo.eval.metrics.physics_iq.metric.PhysicsIQMetric.aggregate_components


aggregate_components(results_list: Iterable[MetricResult | Mapping[str, Any]]) -> dict[str, float]


Source
@classmethod
def aggregate_components(cls, results_list: Iterable[MetricResult | Mapping[str, Any]]) -> dict[str, float]:
    payloads = [cls._extract_payload(result) for result in results_list]
    return {
        "physics_iq": cls.aggregate(payloads),
        "a_mse": mean([value for payload in payloads for value in payload["mse_per_frame"]]),
        "a_st": mean([value for payload in payloads for value in payload["spatiotemporal_iou_per_frame"]]),
        "a_s": mean([float(payload["spatial_iou"]) for payload in payloads]),
        "a_ws": mean([float(payload["weighted_spatial_iou"]) for payload in payloads]),
        "v_mse": mean([value for payload in payloads for value in payload["pv_mse_per_frame"]]),
        "v_st": mean([value for payload in payloads for value in payload["pv_spatiotemporal_iou_per_frame"]]),
        "v_s": mean([float(payload["pv_spatial_iou"]) for payload in payloads]),
        "v_ws": mean([float(payload["pv_weighted_spatial_iou"]) for payload in payloads]),
    }


fastvideo.eval.metrics.physics_iq.metric.PhysicsIQMetric.compute


compute(sample: dict) -> MetricResult


Source
def compute(self, sample: dict) -> MetricResult:
    if "reference" not in sample:
        raise KeyError("PhysicsIQMetric requires sample['reference'].")

take2_key = None
for candidate in (&quot;reference_take2&quot;, &quot;real_take2&quot;, &quot;take2&quot;):
    if candidate in sample:
        take2_key = candidate
        break
if take2_key is None:
    raise KeyError(&quot;PhysicsIQMetric requires sample&#91;&#x27;reference_take2&#x27;&#93; or an alias.&quot;)

video = sample&#91;&quot;video&quot;&#93;
reference = sample&#91;&quot;reference&quot;&#93;
reference_take2 = sample&#91;take2_key&#93;
generated_mask = sample.get(&quot;video_mask&quot;)
reference_mask = sample.get(&quot;reference_mask&quot;)
reference_take2_mask = sample.get(&quot;reference_take2_mask&quot;)

scenario = sample.get(&quot;scenario&quot;)
view = sample.get(&quot;view&quot;)

details = self.compute_single(
    video,
    reference,
    reference_take2,
    generated_mask=generated_mask,
    reference_mask=reference_mask,
    reference_take2_mask=reference_take2_mask,
    scenario=scenario,
    view=view,
)
return MetricResult(name=self.name, score=self._per_video_score(details), details=details)</code></pre></details>

Was this page helpful?