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: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.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
            },
        )

`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
        },
    )