ImageBind audio↔video cosine similarity (IB-Score).

Per-sample. Reads sample["video_path"] (or sample["video"].source for a :class:Video wrapper) and sample["audio"]; ImageBind decodes its own clips so the path is required, not the pool-decoded tensor.

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

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

Move metric (and its internal models) to device.

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

reset() -> None

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

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

accumulate(sample: dict) -> None

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

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

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

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

merge_from(other: BaseMetric) -> None

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

`fastvideo.eval.metrics.audio.imagebind_score.metric.ensure_checkpoint`

ensure_checkpoint(name: str, source: str, filename: str | None = None) -> str

Resolve a model checkpoint path, downloading on miss.

See module docstring for the full source contract. name is used only as the local cache filename for URL sources; ignored otherwise.

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

register(name: str)

Decorator to register a metric class.

Usage::

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

`fastvideo.eval.metrics.audio.imagebind_score.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.imagebind_score.metric.Video`

Video(source: Any, fps: float | None = None, frames: Any = None, audio: Any = None, audio_sr: int | None = None)

Path-backed media handle. The :class:VideoPool populates frames (and optionally audio) before the metric loop sees the sample.

`fastvideo.eval.metrics.audio.imagebind_score.metric.Video.has_frames`

has_frames() -> bool

`fastvideo.eval.metrics.audio.imagebind_score.metric.Video.has_audio`

has_audio() -> bool

`fastvideo.eval.metrics.audio.imagebind_score.metric.ImageBindScoreMetric`

ImageBindScoreMetric()

ImageBind audio↔video cosine similarity, per-sample.

Source

@register("audio.imagebind_score")
class ImageBindScoreMetric(BaseMetric):
    """ImageBind audio↔video cosine similarity, per-sample."""

    name = "audio.imagebind_score"
    requires_reference = False
    higher_is_better = True
    needs_gpu = True
    is_set_metric = False
    backbone = "imagebind"
    dependencies = ["imagebind", "decord", "pytorchvideo"]

    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 imagebind.models import imagebind_model
        # Route the state-dict through ``ensure_checkpoint``; the stock
        # ``imagebind_huge(pretrained=True)`` hardcodes ``.checkpoints/`` in cwd.
        model = imagebind_model.imagebind_huge(pretrained=False)
        ckpt = ensure_checkpoint(_IMAGEBIND_NAME, _IMAGEBIND_URL)
        model.load_state_dict(torch.load(ckpt, weights_only=True, map_location="cpu"))
        self._model = model.eval().to(self.device)

    @torch.no_grad()
    def compute(self, sample: dict) -> MetricResult:
        if self._model is None:
            self.setup()

        video_path = _video_source(sample)
        audio_path = sample.get("audio")
        missing = []
        if video_path is None:
            missing.append("'video_path' or 'video' with a path source")
        if audio_path is None:
            missing.append("'audio'")
        if missing:
            return self._skip(sample, f"missing {', '.join(missing)}")

        import decord
        from imagebind import data as ib_data
        from imagebind.models.imagebind_model import ModalityType

        # decord.bridge is thread-local with a missing-``global`` upstream
        # bug; worker threads otherwise see ``"native"`` and pytorchvideo
        # crashes on the resulting NDArray. Re-set it on this thread.
        decord.bridge.set_bridge("torch")
        with _IB_DECODE_LOCK:
            inputs = {
                ModalityType.VISION: ib_data.load_and_transform_video_data([video_path], self.device),
                ModalityType.AUDIO: ib_data.load_and_transform_audio_data([audio_path], self.device),
            }
        embeds = self._model(inputs)
        score = F.cosine_similarity(embeds[ModalityType.VISION], embeds[ModalityType.AUDIO], dim=-1)[0].item()
        return MetricResult(name=self.name, score=float(score), details={})

`fastvideo.eval.metrics.audio.imagebind_score.metric.ImageBindScoreMetric.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.imagebind_score.metric.ImageBindScoreMetric.setup`

setup() -> None

Source

def setup(self) -> None:
    if self._model is not None:
        return
    from imagebind.models import imagebind_model
    # Route the state-dict through ``ensure_checkpoint``; the stock
    # ``imagebind_huge(pretrained=True)`` hardcodes ``.checkpoints/`` in cwd.
    model = imagebind_model.imagebind_huge(pretrained=False)
    ckpt = ensure_checkpoint(_IMAGEBIND_NAME, _IMAGEBIND_URL)
    model.load_state_dict(torch.load(ckpt, weights_only=True, map_location="cpu"))
    self._model = model.eval().to(self.device)

`fastvideo.eval.metrics.audio.imagebind_score.metric.ImageBindScoreMetric.compute`

compute(sample: dict) -> MetricResult

Source

@torch.no_grad()
def compute(self, sample: dict) -> MetricResult:
    if self._model is None:
        self.setup()

    video_path = _video_source(sample)
    audio_path = sample.get("audio")
    missing = []
    if video_path is None:
        missing.append("'video_path' or 'video' with a path source")
    if audio_path is None:
        missing.append("'audio'")
    if missing:
        return self._skip(sample, f"missing {', '.join(missing)}")

    import decord
    from imagebind import data as ib_data
    from imagebind.models.imagebind_model import ModalityType

    # decord.bridge is thread-local with a missing-``global`` upstream
    # bug; worker threads otherwise see ``"native"`` and pytorchvideo
    # crashes on the resulting NDArray. Re-set it on this thread.
    decord.bridge.set_bridge("torch")
    with _IB_DECODE_LOCK:
        inputs = {
            ModalityType.VISION: ib_data.load_and_transform_video_data([video_path], self.device),
            ModalityType.AUDIO: ib_data.load_and_transform_audio_data([audio_path], self.device),
        }
    embeds = self._model(inputs)
    score = F.cosine_similarity(embeds[ModalityType.VISION], embeds[ModalityType.AUDIO], dim=-1)[0].item()
    return MetricResult(name=self.name, score=float(score), details={})