metric
PaSST KL divergence KL(gt || pred) on AudioSet-527 logits.
Ports av_bench.metrics.kl.compute_kl 1:1. The primary score is
the softmax variant (reported as "MKL" / "KL_PaSST" by V2A papers);
the sigmoid variant is exposed in details["kl_sigmoid"].
fastvideo.eval.metrics.audio.kl_divergence.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.kl_divergence.metric.BaseMetric.to
to(device: str | torch.device) -> BaseMetric
Move metric (and its internal models) to device.
fastvideo.eval.metrics.audio.kl_divergence.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.kl_divergence.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.kl_divergence.metric.BaseMetric.reset
reset() -> None
Clear accumulator state at the start of each evaluate() call.
fastvideo.eval.metrics.audio.kl_divergence.metric.BaseMetric.accumulate
accumulate(sample: dict) -> None
Buffer per-sample features for a corpus-level metric.
fastvideo.eval.metrics.audio.kl_divergence.metric.BaseMetric.finalize
finalize() -> MetricResult
Compute the corpus-level result from buffered state.
fastvideo.eval.metrics.audio.kl_divergence.metric.BaseMetric.merge_from
merge_from(other: BaseMetric) -> None
Multi-GPU: fold another worker's accumulator state into this one.
fastvideo.eval.metrics.audio.kl_divergence.metric.register
register(name: str)
Decorator to register a metric class.
Usage::
@register("ssim")
class SSIMMetric(BaseMetric):
...
fastvideo.eval.metrics.audio.kl_divergence.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.kl_divergence.metric.KLDivergenceMetric
KLDivergenceMetric()
PaSST KL divergence KL(gt || pred) on AudioSet-527 logits.
Per-sample. Requires sample["audio"] (generated) and
sample["reference_audio"] (ground truth).
Source
@register("audio.kl_divergence")
class KLDivergenceMetric(BaseMetric):
"""PaSST KL divergence ``KL(gt || pred)`` on AudioSet-527 logits.
Per-sample. Requires ``sample["audio"]`` (generated) and
``sample["reference_audio"]`` (ground truth).
"""
name = "audio.kl_divergence"
requires_reference = True
higher_is_better = False
needs_gpu = True
is_set_metric = False
dependencies = ["hear21passt", "librosa", "pyloudnorm"]
def __init__(self) -> None:
super().__init__()
self._model: Any = None
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 not None:
return
from hear21passt.base import get_basic_model
model = get_basic_model(mode="logits")
model.eval()
model = model.to(self.device)
self._model = model
@torch.no_grad()
def compute(self, sample: dict) -> MetricResult:
if self._model is None:
self.setup()
gen_path = sample.get("audio")
ref_path = sample.get("reference_audio")
if gen_path is None or ref_path is None:
missing = [k for k, v in (("audio", gen_path), ("reference_audio", ref_path)) if v is None]
return self._skip(sample, f"missing {', '.join(repr(k) for k in missing)}")
gt_logits = _collect_logits(self._model, ref_path, self.device)
pred_logits = _collect_logits(self._model, gen_path, self.device)
if not torch.isfinite(gt_logits).all() or not torch.isfinite(pred_logits).all():
which = []
if not torch.isfinite(gt_logits).all():
which.append("reference_audio")
if not torch.isfinite(pred_logits).all():
which.append("audio")
return self._skip(
sample,
f"non-finite PaSST logits on {'+'.join(which)} (silent or corrupt audio?)",
)
kl_sm = _kl_softmax(pred_logits, gt_logits)
kl_sg = _kl_sigmoid(pred_logits, gt_logits)
return MetricResult(
name=self.name,
score=kl_sm,
details={
"kl_softmax": kl_sm,
"kl_sigmoid": kl_sg
},
)</code></pre></details>
fastvideo.eval.metrics.audio.kl_divergence.metric.KLDivergenceMetric.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.kl_divergence.metric.KLDivergenceMetric.setup
setup() -> None
Source
def setup(self) -> None:
if self._model is not None:
return
from hear21passt.base import get_basic_model
model = get_basic_model(mode="logits")
model.eval()
model = model.to(self.device)
self._model = model
fastvideo.eval.metrics.audio.kl_divergence.metric.KLDivergenceMetric.compute
compute(sample: dict) -> MetricResult
Source
@torch.no_grad()
def compute(self, sample: dict) -> MetricResult:
if self._model is None:
self.setup()
gen_path = sample.get("audio")
ref_path = sample.get("reference_audio")
if gen_path is None or ref_path is None:
missing = [k for k, v in (("audio", gen_path), ("reference_audio", ref_path)) if v is None]
return self._skip(sample, f"missing {', '.join(repr(k) for k in missing)}")
gt_logits = _collect_logits(self._model, ref_path, self.device)
pred_logits = _collect_logits(self._model, gen_path, self.device)
if not torch.isfinite(gt_logits).all() or not torch.isfinite(pred_logits).all():
which = []
if not torch.isfinite(gt_logits).all():
which.append("reference_audio")
if not torch.isfinite(pred_logits).all():
which.append("audio")
return self._skip(
sample,
f"non-finite PaSST logits on {'+'.join(which)} (silent or corrupt audio?)",
)
kl_sm = _kl_softmax(pred_logits, gt_logits)
kl_sg = _kl_sigmoid(pred_logits, gt_logits)
return MetricResult(
name=self.name,
score=kl_sm,
details={
"kl_softmax": kl_sm,
"kl_sigmoid": kl_sg
},
)</code></pre></details>