Skip to content
FastVideo
Esc
↑↓navigate↵open⌘Jpreview
On this page

metric

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=&#123;&#125;)</code></pre></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(&quot;audio&quot;)
missing = &#91;&#93;
if video_path is None:
    missing.append(&quot;&#x27;video_path&#x27; or &#x27;video&#x27; with a path source&quot;)
if audio_path is None:
    missing.append(&quot;&#x27;audio&#x27;&quot;)
if missing:
    return self._skip(sample, f&quot;missing &#123;&#x27;, &#x27;.join(missing)&#125;&quot;)

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 ``&quot;native&quot;`` and pytorchvideo
# crashes on the resulting NDArray. Re-set it on this thread.
decord.bridge.set_bridge(&quot;torch&quot;)
with _IB_DECODE_LOCK:
    inputs = &#123;
        ModalityType.VISION: ib_data.load_and_transform_video_data(&#91;video_path&#93;, self.device),
        ModalityType.AUDIO: ib_data.load_and_transform_audio_data(&#91;audio_path&#93;, self.device),
    &#125;
embeds = self._model(inputs)
score = F.cosine_similarity(embeds&#91;ModalityType.VISION&#93;, embeds&#91;ModalityType.AUDIO&#93;, dim=-1)&#91;0&#93;.item()
return MetricResult(name=self.name, score=float(score), details=&#123;&#125;)</code></pre></details>

Was this page helpful?