metric
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:MetricResultper 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:resetto clear buffers and :meth:merge_fromto 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) -> BaseMetricMove metric (and its internal models) to device.
fastvideo.eval.metrics.audio.frechet_distance.metric.BaseMetric.setup
setup() -> NoneEagerly 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) -> MetricResultPer-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() -> NoneClear accumulator state at the start of each evaluate() call.
fastvideo.eval.metrics.audio.frechet_distance.metric.BaseMetric.accumulate
accumulate(sample: dict) -> NoneBuffer per-sample features for a corpus-level metric.
fastvideo.eval.metrics.audio.frechet_distance.metric.BaseMetric.finalize
finalize() -> MetricResultCompute the corpus-level result from buffered state.
fastvideo.eval.metrics.audio.frechet_distance.metric.BaseMetric.merge_from
merge_from(other: BaseMetric) -> NoneMulti-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,
},
)</code></pre></details>
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()</code></pre></details>
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,
},
)</code></pre></details>