Frechet Audio Distance over PaSST embeddings (FD_PaSST).

Corpus-vs-corpus Fréchet distance between Gaussian moments of two PaSST 768-d embedding sets. Ports av_bench.metrics.fad.compute_fd 1:1.

References are supplied per-sample via reference_audio / role="reference", or once from a .pt cache at $FASTVIDEO_FAD_REF_FEATURES. Skips when either side has fewer than two finite embeddings.

`fastvideo.eval.metrics.audio.frechet_distance.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.audio.frechet_distance.metric.BaseMetric.to`

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

Move metric (and its internal models) to device.

`fastvideo.eval.metrics.audio.frechet_distance.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.audio.frechet_distance.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.audio.frechet_distance.metric.BaseMetric.reset`

reset() -> None

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

`fastvideo.eval.metrics.audio.frechet_distance.metric.BaseMetric.accumulate`

accumulate(sample: dict) -> None

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

`fastvideo.eval.metrics.audio.frechet_distance.metric.BaseMetric.finalize`

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

`fastvideo.eval.metrics.audio.frechet_distance.metric.BaseMetric.merge_from`

merge_from(other: BaseMetric) -> None

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

`fastvideo.eval.metrics.audio.frechet_distance.metric.register`

register(name: str)

Decorator to register a metric class.

Usage::

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

`fastvideo.eval.metrics.audio.frechet_distance.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.audio.frechet_distance.metric.FrechetAudioDistanceMetric`

FrechetAudioDistanceMetric()

Corpus-vs-corpus Frechet Audio Distance with PaSST embeddings.

Source

@register("audio.frechet_distance")
class FrechetAudioDistanceMetric(BaseMetric):
    """Corpus-vs-corpus Frechet Audio Distance with PaSST embeddings."""

    name = "audio.frechet_distance"
    requires_reference = True
    higher_is_better = False
    needs_gpu = True
    is_set_metric = True
    backbone = "passt"
    dependencies = ["hear21passt", "librosa", "pyloudnorm"]

    def __init__(self) -> None:
        super().__init__()
        self._model: Any = None
        self._gen_buf: list[np.ndarray] = []
        self._ref_buf: list[np.ndarray] = []
        self._cached_ref_path: str | None = envs.FASTVIDEO_FAD_REF_FEATURES.get()
        self._cached_ref_mu: np.ndarray | None = None
        self._cached_ref_sigma: np.ndarray | None = None
        self._n_cached_ref: int = 0

    def to(self, device):
        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 None:
            from hear21passt.base import get_basic_model

            model = get_basic_model(mode="all")
            model.eval()
            model = model.to(self.device)
            self._model = model
        if self._cached_ref_path and self._cached_ref_mu is None:
            self._load_cached_ref()

    def _load_cached_ref(self) -> None:
        assert self._cached_ref_path is not None
        path = Path(self._cached_ref_path)
        if not path.exists():
            raise FileNotFoundError(f"{REF_FEATURES_ENV} set to {path}, but the file does not exist. "
                                    f"Either unset the env var to use sample-supplied references or "
                                    f"pre-compute the reference features file.")
        ref = torch.load(str(path), weights_only=True, map_location="cpu")
        if hasattr(ref, "numpy"):
            ref = ref.numpy()
        ref = np.asarray(ref)
        if ref.ndim != 2 or ref.shape[1] != PASST_EMBED_DIM:
            raise ValueError(f"Expected a 2-D tensor of shape (N, {PASST_EMBED_DIM}) at {path}; got shape {ref.shape}.")
        finite = np.isfinite(ref).all(axis=1)
        ref = ref[finite]
        if ref.shape[0] < 2:
            raise ValueError(f"Cached reference features at {path} have only {ref.shape[0]} finite rows; need >= 2.")
        self._cached_ref_mu = ref.mean(axis=0)
        self._cached_ref_sigma = np.cov(ref, rowvar=False)
        self._n_cached_ref = int(ref.shape[0])

    def reset(self) -> None:
        self._gen_buf.clear()
        self._ref_buf.clear()

    def accumulate(self, sample: dict) -> None:
        if self._model is None:
            self.setup()
        if sample.get("role") == "reference":
            ref_path = sample.get("audio") or sample.get("reference_audio")
            if ref_path is not None:
                self._ref_buf.append(_passt_embed(self._model, ref_path, self.device))
            return
        gen_path = sample.get("audio")
        if gen_path is not None:
            self._gen_buf.append(_passt_embed(self._model, gen_path, self.device))
        if self._cached_ref_mu is not None:
            return
        ref_path = sample.get("reference_audio")
        if ref_path is not None:
            self._ref_buf.append(_passt_embed(self._model, ref_path, self.device))

    def merge_from(self, other: FrechetAudioDistanceMetric) -> None:
        self._gen_buf.extend(other._gen_buf)
        self._ref_buf.extend(other._ref_buf)

    def finalize(self) -> MetricResult:
        gen_all = np.stack(self._gen_buf) if self._gen_buf else np.empty((0, ))
        gen = gen_all[np.isfinite(gen_all).all(axis=1)] if gen_all.size else gen_all
        n_gen = int(gen.shape[0])
        n_gen_dropped = len(self._gen_buf) - n_gen

        if self._cached_ref_mu is not None:
            mu_r = self._cached_ref_mu
            sigma_r = self._cached_ref_sigma
            n_ref = self._n_cached_ref
            n_ref_dropped = 0
            ref_source = "cached"
        else:
            ref_all = np.stack(self._ref_buf) if self._ref_buf else np.empty((0, ))
            ref = ref_all[np.isfinite(ref_all).all(axis=1)] if ref_all.size else ref_all
            n_ref = int(ref.shape[0])
            n_ref_dropped = len(self._ref_buf) - n_ref
            ref_source = "samples"
            if n_ref < 2 or n_gen < 2:
                return MetricResult(
                    name=self.name,
                    score=None,
                    details={
                        "skipped":
                        f"FAD needs >=2 finite-embed samples per side "
                        f"(got n_gen={n_gen} valid of {len(self._gen_buf)}, "
                        f"n_ref={n_ref} valid of {len(self._ref_buf)})",
                        "n_gen":
                        n_gen,
                        "n_ref":
                        n_ref,
                        "n_gen_dropped_nonfinite":
                        n_gen_dropped,
                        "n_ref_dropped_nonfinite":
                        n_ref_dropped,
                        "ref_source":
                        ref_source,
                    },
                )
            mu_r, sigma_r = ref.mean(axis=0), np.cov(ref, rowvar=False)

        if n_gen < 2:
            return MetricResult(
                name=self.name,
                score=None,
                details={
                    "skipped": f"FAD needs >=2 finite-embed gen samples (got {n_gen} valid of {len(self._gen_buf)})",
                    "n_gen": n_gen,
                    "n_ref": n_ref,
                    "n_gen_dropped_nonfinite": n_gen_dropped,
                    "n_ref_dropped_nonfinite": n_ref_dropped,
                    "ref_source": ref_source,
                },
            )

        mu_g, sigma_g = gen.mean(axis=0), np.cov(gen, rowvar=False)
        fd = _frechet_distance(mu_g, sigma_g, mu_r, sigma_r)
        return MetricResult(
            name=self.name,
            score=fd,
            details={
                "n_gen": n_gen,
                "n_ref": n_ref,
                "n_gen_dropped_nonfinite": n_gen_dropped,
                "n_ref_dropped_nonfinite": n_ref_dropped,
                "ref_source": ref_source,
            },
        )

`fastvideo.eval.metrics.audio.frechet_distance.metric.FrechetAudioDistanceMetric.to`

to(device)

Source

def to(self, device):
    super().to(device)
    if self._model is not None:
        self._model = self._model.to(self.device)
    return self

`fastvideo.eval.metrics.audio.frechet_distance.metric.FrechetAudioDistanceMetric.setup`

setup() -> None

Source

def setup(self) -> None:
    if self._model is None:
        from hear21passt.base import get_basic_model

        model = get_basic_model(mode="all")
        model.eval()
        model = model.to(self.device)
        self._model = model
    if self._cached_ref_path and self._cached_ref_mu is None:
        self._load_cached_ref()

`fastvideo.eval.metrics.audio.frechet_distance.metric.FrechetAudioDistanceMetric.reset`

reset() -> None

Source

def reset(self) -> None:
    self._gen_buf.clear()
    self._ref_buf.clear()

`fastvideo.eval.metrics.audio.frechet_distance.metric.FrechetAudioDistanceMetric.accumulate`

accumulate(sample: dict) -> None

Source

def accumulate(self, sample: dict) -> None:
    if self._model is None:
        self.setup()
    if sample.get("role") == "reference":
        ref_path = sample.get("audio") or sample.get("reference_audio")
        if ref_path is not None:
            self._ref_buf.append(_passt_embed(self._model, ref_path, self.device))
        return
    gen_path = sample.get("audio")
    if gen_path is not None:
        self._gen_buf.append(_passt_embed(self._model, gen_path, self.device))
    if self._cached_ref_mu is not None:
        return
    ref_path = sample.get("reference_audio")
    if ref_path is not None:
        self._ref_buf.append(_passt_embed(self._model, ref_path, self.device))

`fastvideo.eval.metrics.audio.frechet_distance.metric.FrechetAudioDistanceMetric.merge_from`

merge_from(other: FrechetAudioDistanceMetric) -> None

Source

def merge_from(self, other: FrechetAudioDistanceMetric) -> None:
    self._gen_buf.extend(other._gen_buf)
    self._ref_buf.extend(other._ref_buf)

`fastvideo.eval.metrics.audio.frechet_distance.metric.FrechetAudioDistanceMetric.finalize`

finalize() -> MetricResult

Source

def finalize(self) -> MetricResult:
    gen_all = np.stack(self._gen_buf) if self._gen_buf else np.empty((0, ))
    gen = gen_all[np.isfinite(gen_all).all(axis=1)] if gen_all.size else gen_all
    n_gen = int(gen.shape[0])
    n_gen_dropped = len(self._gen_buf) - n_gen

    if self._cached_ref_mu is not None:
        mu_r = self._cached_ref_mu
        sigma_r = self._cached_ref_sigma
        n_ref = self._n_cached_ref
        n_ref_dropped = 0
        ref_source = "cached"
    else:
        ref_all = np.stack(self._ref_buf) if self._ref_buf else np.empty((0, ))
        ref = ref_all[np.isfinite(ref_all).all(axis=1)] if ref_all.size else ref_all
        n_ref = int(ref.shape[0])
        n_ref_dropped = len(self._ref_buf) - n_ref
        ref_source = "samples"
        if n_ref < 2 or n_gen < 2:
            return MetricResult(
                name=self.name,
                score=None,
                details={
                    "skipped":
                    f"FAD needs >=2 finite-embed samples per side "
                    f"(got n_gen={n_gen} valid of {len(self._gen_buf)}, "
                    f"n_ref={n_ref} valid of {len(self._ref_buf)})",
                    "n_gen":
                    n_gen,
                    "n_ref":
                    n_ref,
                    "n_gen_dropped_nonfinite":
                    n_gen_dropped,
                    "n_ref_dropped_nonfinite":
                    n_ref_dropped,
                    "ref_source":
                    ref_source,
                },
            )
        mu_r, sigma_r = ref.mean(axis=0), np.cov(ref, rowvar=False)

    if n_gen < 2:
        return MetricResult(
            name=self.name,
            score=None,
            details={
                "skipped": f"FAD needs >=2 finite-embed gen samples (got {n_gen} valid of {len(self._gen_buf)})",
                "n_gen": n_gen,
                "n_ref": n_ref,
                "n_gen_dropped_nonfinite": n_gen_dropped,
                "n_ref_dropped_nonfinite": n_ref_dropped,
                "ref_source": ref_source,
            },
        )

    mu_g, sigma_g = gen.mean(axis=0), np.cov(gen, rowvar=False)
    fd = _frechet_distance(mu_g, sigma_g, mu_r, sigma_r)
    return MetricResult(
        name=self.name,
        score=fd,
        details={
            "n_gen": n_gen,
            "n_ref": n_ref,
            "n_gen_dropped_nonfinite": n_gen_dropped,
            "n_ref_dropped_nonfinite": n_ref_dropped,
            "ref_source": ref_source,
        },
    )