[feat] eval: Add common.fvd - Fréchet Video Distance metric

fastvideo/eval/README.md

@@ -1,9 +1,9 @@
 # `fastvideo.eval`
 
 In-process evaluation suite for video generations. Includes pixel
-metrics (SSIM, PSNR, LPIPS), optical-flow comparisons, the full VBench
-suite, Physics-IQ, and a VLM scorer (VideoScore-2) behind a single
-registry-driven API.
+metrics (SSIM, PSNR, LPIPS), FVD (Fréchet Video Distance), optical-flow
+comparisons, the full VBench suite, Physics-IQ, and a VLM scorer
+(VideoScore-2) behind a single registry-driven API.
 
 ## Install
 
@@ -78,7 +78,7 @@ fastvideo/
 │   ├── datasets/                  # prompt corpora (vbench, physics_iq)
 │   └── metrics/
 │       ├── base.py                # BaseMetric + @register contract
-│       ├── common/                # SSIM, PSNR, LPIPS
+│       ├── common/                # SSIM, PSNR, LPIPS, FVD
 │       ├── optical_flow/          # gt_optical_flow, synthetic_optical_flow
 │       ├── videoscore2/           # VideoScore-2 (Qwen2.5-VL)
 │       ├── physics_iq/            # PhysicsIQ + sub-metrics
@@ -170,6 +170,7 @@ ${FASTVIDEO_CACHE_ROOT}/eval/
 ├── models/      # URL-fetched checkpoints (LAION head, AMT, GRiT)
 ├── torch/       # redirected TORCH_HOME (DINO via torch.hub, lpips)
 ├── clip/        # passed as download_root= to clip.load callsites
+├── fvd/         # common.fvd: real_features.pt (cached I3D reference features)
 └── datasets/    # auto-fetched dataset assets, one subdir per benchmark
                  # (e.g. datasets/physics_iq/{split-videos,switch-frames,...})
 ```
@@ -206,9 +207,8 @@ the metric's docstring if it matters.
 
 - **MIND** metrics. Depend on a separate `vipe` upstream submodule.
 - **VBench-2.0**. Sibling vbench2 package; needs its own port.
-- **FVD as a registered metric**. Currently still at `benchmarks/fvd/`.
-  FVD is a set-vs-set distribution distance and does not fit the
-  per-sample `BaseMetric.compute` API without a stateful accumulator;
-  conversion is a designed follow-up.
+- **FVD** is now available as `common.fvd` (see `metrics/common/fvd/`).
+  Uses the `is_set_metric` accumulate/finalize interface; requires
+  ≥ 256 videos for statistically reliable scores (standard: 2048).
 - **Training-time eval callback** (`EvalCallback`) and the
   `RolloutEvaluator` helper.

fastvideo/eval/__init__.py

@@ -21,7 +21,7 @@ def _redirect_third_party_caches() -> None:
 
 _redirect_third_party_caches()
 
-from fastvideo.eval.types import MetricResult  # noqa: E402
+from fastvideo.eval.types import EvalResults, MetricResult, Video  # noqa: E402
 from fastvideo.eval.metrics.base import BaseMetric  # noqa: E402
 from fastvideo.eval.registry import register, list_metrics, get_metric  # noqa: E402
 from fastvideo.eval.api import evaluate  # noqa: E402
@@ -34,7 +34,9 @@ def _redirect_third_party_caches() -> None:
     "evaluate",
     "Evaluator",
     "create_evaluator",
+    "EvalResults",
     "MetricResult",
+    "Video",
     "BaseMetric",
     "register",
     "list_metrics",

fastvideo/eval/evaluator.py

@@ -3,22 +3,24 @@
 Layering (mirrors FastVideo's VideoGenerator → Worker pattern, but
 in-process)::
 
-    Evaluator               ← user-facing; round-robins samples across workers
+    Evaluator               ← user-facing
       └── EvalWorker × N    ← single-GPU; owns metric replicas
+      └── VideoPool         ← async path-→-tensor prefetch (per evaluate call)
 
 The constructor builds one :class:`EvalWorker` per GPU and loads every
 metric on every worker eagerly. :meth:`evaluate` is the single entry
 point: pass kwargs for one sample, or pass a list of sample dicts to
-fan-out across GPU replicas — same method, return type follows the
-input shape.
+fan-out across GPU replicas with pipelined decoding — same method,
+return type follows the input shape.
 """
 from __future__ import annotations
 
-from concurrent.futures import ThreadPoolExecutor
+import threading
 from collections.abc import Iterable
+from typing import Any
 
 from fastvideo.eval.registry import (list_metrics, missing_dependencies, resolve_group)
-from fastvideo.eval.types import MetricResult
+from fastvideo.eval.types import EvalResults, MetricResult
 from fastvideo.eval.worker import EvalWorker
 from fastvideo.logger import init_logger
 
@@ -38,6 +40,22 @@ class Evaluator:
         Number of GPU replicas. Each gets its own :class:`EvalWorker`.
     compile : bool
         Apply :func:`torch.compile` to each metric's ``_model``.
+    loader_threads : int
+        Background decode threads in the :class:`VideoPool`. Default 1
+        (hide decode behind compute). Bump for I/O-heavy benchmark sets
+        where one loader can't keep up with the workers.
+    prefetch_factor : int
+        ``pool max_size = prefetch_factor * num_workers``. Default 2 —
+        one sample being consumed, one prefetched per worker.
+    pre_upload : bool
+        When ``True`` (default), the worker performs a single
+        host→device upload of ``video`` / ``reference`` per sample
+        before the metric loop, and every metric reads from that
+        shared GPU-resident tensor. Without it, each metric pays its
+        own ``.to(self.device)`` — N transfers of the same clip for N
+        metrics, which dominates at high resolution. Set ``False`` for
+        training-time eval, where keeping a clip resident on GPU
+        across the metric loop would fight the training step for VRAM.
     """
 
     def __init__(
@@ -46,13 +64,20 @@ def __init__(
         device: str = "cuda:0",
         num_gpus: int = 1,
         compile: bool = False,
+        *,
+        loader_threads: int = 1,
+        prefetch_factor: int = 2,
+        pre_upload: bool = True,
     ) -> None:
         names = _resolve_metric_names(metrics)
         if num_gpus > 1:
-            self._workers = [EvalWorker(names, f"cuda:{i}", compile=compile) for i in range(num_gpus)]
+            self._workers = [
+                EvalWorker(names, f"cuda:{i}", compile=compile, pre_upload=pre_upload) for i in range(num_gpus)
+            ]
         else:
-            self._workers = [EvalWorker(names, device, compile=compile)]
-        self._pool = (ThreadPoolExecutor(max_workers=num_gpus) if num_gpus > 1 else None)
+            self._workers = [EvalWorker(names, device, compile=compile, pre_upload=pre_upload)]
+        self._loader_threads = max(1, loader_threads)
+        self._prefetch_factor = max(1, prefetch_factor)
 
     @property
     def num_gpus(self) -> int:
@@ -66,46 +91,101 @@ def evaluate(
         self,
         samples: Iterable[dict] | None = None,
         **kwargs,
-    ) -> dict[str, MetricResult] | list[dict[str, MetricResult]]:
+    ) -> dict[str, MetricResult] | EvalResults:
         """Score one sample (kwargs form) or many samples (list form).
 
-        ``video`` and ``reference`` may be either a pre-loaded
-        ``(T, C, H, W)`` tensor or a path-like (``str`` / ``Path``).
-        Paths are decoded inside the worker thread that picks up the
-        sample, so memory stays bounded by ``num_gpus`` even when
-        thousands of paths are queued — see ``score_folder.py`` for
-        the canonical pattern.
+        Both forms go through the same :class:`VideoPool` pipeline;
+        ``video`` / ``reference`` paths are decoded asynchronously,
+        ``(1, T, C, H, W)`` tensors are squeezed.
 
         One sample::
 
             ev.evaluate(video=tensor, text_prompt="...", fps=24.0)
             ev.evaluate(video="path/to/clip.mp4", fps=24.0)
 
-        Many samples — fan out across GPU replicas, results in input order::
+        Returns a ``dict[str, MetricResult]``.
+
+        Many samples — pipelined decode + work-stealing across replicas::
 
             ev.evaluate(samples=[
                 {"video": "a.mp4", "reference": "ref_a.mp4"},
                 {"video": "b.mp4", "reference": "ref_b.mp4"},
                 ...
             ])
 
-        Multi-GPU dispatch fires automatically iff ``num_gpus > 1`` *and*
-        the list form is used. The kwargs form always runs on worker 0;
-        if you have a single sample but multiple GPUs, wrap it in a
-        one-element list to use the pool, or just accept that a single
-        call uses one GPU — that's fine.
+        Returns an :class:`EvalResults` (list-of-dict subclass): per-sample
+        dicts in input order, with set-metric scores under ``.corpus``.
         """
-        if samples is None:
-            return self._workers[0].evaluate(**kwargs)
+        single = samples is None
+        sample_list: list[dict] = [kwargs] if samples is None else list(samples)
+        if not sample_list:
+            return EvalResults(samples=[], corpus={})
+
+        per_sample, corpus = self._run(sample_list)
+
+        if single:
+            return per_sample[0]
+        return EvalResults(samples=per_sample, corpus=corpus)
 
-        samples = list(samples)
-        if self._pool is None or len(samples) <= 1:
-            return [self._workers[0].evaluate(**s) for s in samples]
+    def _run(self, samples: list[dict]) -> tuple[list[dict[str, MetricResult]], dict[str, MetricResult]]:
+        """Pool-driven sample pipeline + set-metric finalize.
 
-        n = len(self._workers)
-        # Round-robin: worker i handles samples i, i+n, i+2n, ...
-        futures = [self._pool.submit(self._workers[idx % n].evaluate, **sample) for idx, sample in enumerate(samples)]
-        return [f.result() for f in futures]
+        Returns ``(per_sample_results, corpus_results)``.
+        """
+        from fastvideo.eval.pool import VideoPool
+
+        # Reset every worker's set-metric buffers — per-call isolation.
+        for w in self._workers:
+            w.reset_set_metrics()
+
+        n_workers = len(self._workers)
+        max_size = self._prefetch_factor * n_workers
+        per_sample: list[Any] = [None] * len(samples)
+
+        with VideoPool(samples, loader_threads=self._loader_threads, max_size=max_size) as pool:
+            if n_workers == 1:
+                while True:
+                    item = pool.get()
+                    if item is None:
+                        break
+                    idx, decoded = item
+                    per_sample[idx] = self._workers[0].evaluate(**decoded)
+            else:
+                # Multi-GPU: every worker drains the shared pool (work-stealing).
+                errors: list[BaseException] = []
+                threads: list[threading.Thread] = []
+                for w in self._workers:
+                    t = threading.Thread(target=self._consumer_loop, args=(w, pool, per_sample, errors), daemon=True)
+                    t.start()
+                    threads.append(t)
+                for t in threads:
+                    t.join()
+                if errors:
+                    raise errors[0]
+
+        # Finalize set metrics. With multiple workers, fold per-worker
+        # accumulator state into worker 0 first, then finalize once.
+        corpus: dict[str, MetricResult] = {}
+        base_set = self._workers[0].set_metrics()
+        if base_set:
+            for w in self._workers[1:]:
+                for name, m in w.set_metrics().items():
+                    base_set[name].merge_from(m)
+            corpus = {name: m.finalize() for name, m in base_set.items()}
+
+        return per_sample, corpus
+
+    @staticmethod
+    def _consumer_loop(worker: EvalWorker, pool: Any, results: list, errors: list) -> None:
+        try:
+            while True:
+                item = pool.get()
+                if item is None:
+                    return
+                idx, decoded = item
+                results[idx] = worker.evaluate(**decoded)
+        except BaseException as e:  # noqa: BLE001 — surface to parent thread via shared list
+            errors.append(e)
 
     def release_cuda_memory(self) -> None:
         """Free CUDA caches on every replica without dropping models."""
@@ -123,10 +203,7 @@ def reload(self) -> None:
             w.reload()
 
     def shutdown(self) -> None:
-        """Tear down the worker thread pool. Idempotent."""
-        if self._pool is not None:
-            self._pool.shutdown(wait=True)
-            self._pool = None
+        """No-op; kept for API compatibility with older callers."""
 
 
 def create_evaluator(

fastvideo/eval/metrics/base.py

@@ -1,23 +1,31 @@
 from __future__ import annotations
 
-from abc import ABC, abstractmethod
-
 import torch
 
 from fastvideo.eval.types import MetricResult
 
 
-class BaseMetric(ABC):
+class BaseMetric:
     """Abstract base class for all eval metrics.
 
-    Subclasses must implement :meth:`compute`. Optionally override
-    :meth:`setup` to eagerly load models.
+    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.
 
-    Metrics that need to chunk along the time dimension (frames or frame
-    pairs) for memory reasons should 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`` therefore receives a single sample, not a batch.
+    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.
     """
 
     name: str = ""
@@ -26,6 +34,7 @@ class BaseMetric(ABC):
     dependencies: list[str] = []
     needs_gpu: bool = False
     backbone: str | None = None
+    is_set_metric: bool = False
 
     # Default time-dim chunk size for metrics that batch internally over
     # frames or frame-pairs. Override in subclass __init__ if needed
@@ -56,14 +65,27 @@ def _skip(self, sample: dict, reason: str) -> MetricResult:
         """Return a skipped result (``score=None`` + reason in details)."""
         return MetricResult(name=self.name, score=None, details={"skipped": reason})
 
-    @abstractmethod
     def compute(self, sample: dict) -> MetricResult:
-        """Compute the metric on a single sample.
+        """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.
-
-        If required inputs are missing, return ``self._skip(sample, reason)``
-        instead of raising.
+        ``sample["reference"]`` (if used) has the same shape. Return
+        ``self._skip(sample, reason)`` for missing inputs.
         """
-        ...
+        raise NotImplementedError(f"{type(self).__name__}.compute is not implemented")
+
+    # --- set-vs-set protocol (only invoked when is_set_metric=True) ---
+
+    def reset(self) -> None:  # noqa: B027 - intentionally optional override
+        """Clear accumulator state at the start of each evaluate() call."""
+
+    def accumulate(self, sample: dict) -> None:
+        """Buffer per-sample features for a corpus-level metric."""
+        raise NotImplementedError(f"{type(self).__name__}.accumulate is not implemented")
+
+    def finalize(self) -> MetricResult:
+        """Compute the corpus-level result from buffered state."""
+        raise NotImplementedError(f"{type(self).__name__}.finalize is not implemented")
+
+    def merge_from(self, other: BaseMetric) -> None:  # noqa: B027 - intentionally optional override
+        """Multi-GPU: fold another worker's accumulator state into this one."""