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

metric

Synchformer DeSync — average audio-video desynchronization in seconds.

Per-sample. Ports hkchengrex/av-benchmark's DeSync against the 24-01-04T16-39-21 Synchformer checkpoint (the one MMAudio reports against). argmax over the 21-class grid in [-2, +2] seconds for the first 14 and last 14 segments separately; lower is better, 0 = ideal.

Synchformer's transformer carries a fixed positional embedding sized for ~14 segments per modality, so keep input clips in the 3-10 s range — shorter clips raise in _segment_video and longer clips ignore the middle.

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

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

Move metric (and its internal models) to device.

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

reset() -> None

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

fastvideo.eval.metrics.audio.desync.metric.BaseMetric.accumulate

accumulate(sample: dict) -> None

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

fastvideo.eval.metrics.audio.desync.metric.BaseMetric.finalize

finalize() -> MetricResult

Compute the corpus-level result from buffered state.

fastvideo.eval.metrics.audio.desync.metric.BaseMetric.merge_from

merge_from(other: BaseMetric) -> None

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

fastvideo.eval.metrics.audio.desync.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.desync.metric.register

register(name: str)

Decorator to register a metric class.

Usage::

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

fastvideo.eval.metrics.audio.desync.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.desync.metric.DeSyncMetric

DeSyncMetric(src_fps: float | None = None)

Per-sample Synchformer DeSync magnitude in seconds (lower is better).

Source
@register("audio.desync")
class DeSyncMetric(BaseMetric):
    """Per-sample Synchformer DeSync magnitude in seconds (lower is better)."""

name = "audio.desync"
requires_reference = False
higher_is_better = False
needs_gpu = True
is_set_metric = False
backbone = "synchformer"
dependencies = ["librosa", "torchaudio"]

def __init__(self, src_fps: float | None = None) -> None:
    super().__init__()
    # Defaults to 25 fps (Synchformer's target) when the pool's decode fps is unknown.
    self._src_fps = src_fps
    self._model: Any = None
    self._mel: Any = None
    self._grid: torch.Tensor | None = None
    self._resample_cache: dict[int, Any] = {}

def to(self, device):
    super().to(device)
    if self._model is not None:
        self._model = self._model.to(self.device)
    if self._mel is not None:
        self._mel = self._mel.to(self.device)
    return self

def setup(self) -> None:
    if self._model is not None:
        return
    import torchaudio
    from fastvideo.third_party.eval.synchformer import Synchformer, make_class_grid

    ckpt = ensure_checkpoint(_SYNCHFORMER_NAME, _SYNCHFORMER_URL)
    model = Synchformer().to(self.device).eval()
    model.load_state_dict(torch.load(ckpt, weights_only=True, map_location=self.device))
    self._model = model

    self._mel = torchaudio.transforms.MelSpectrogram(
        sample_rate=_AUDIO_SR,
        win_length=400,
        hop_length=160,
        n_fft=1024,
        n_mels=128,
    ).to(self.device)
    self._grid = make_class_grid(_GRID_LOW, _GRID_HIGH, _GRID_SIZE)

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

    video = sample.get("video")
    audio_path = sample.get("audio")
    if video is None or audio_path is None:
        missing = [k for k, v in (("video", video), ("audio", audio_path)) if v is None]
        return self._skip(sample, f"missing {', '.join(repr(k) for k in missing)}")

    # Prefer the per-sample fps the pool decoded at; fall back to the
    # constructor override. Without either, the audio/video windows can't
    # be aligned, so skip rather than guess.
    sample_fps = sample.get("fps")
    src_fps = float(sample_fps) if sample_fps is not None else self._src_fps
    if src_fps is None:
        return self._skip(sample, "missing 'fps' (set sample fps or pass src_fps=)")

    # Video preprocessing → (S, T_seg, C, 224, 224)
    frames = video.float().to(self.device)
    frames = _resample_video(frames, _SYNC_FPS, src_fps)
    frames = _video_transform(frames)
    vsegs = _segment_video(frames).unsqueeze(0)  # (B=1, S, T_seg, C, H, W)
    # Synchformer's extract_vfeats wants (B, S, T_seg, C, H, W); fold (B, S) → (B*S, 1, ...).
    b, s = vsegs.shape[:2]
    from einops import rearrange
    vsegs_for_model = rearrange(vsegs, "b s t c h w -> (b s) 1 t c h w")
    vfeats = self._model.extract_vfeats(vsegs_for_model)
    vfeats = rearrange(vfeats, "(b s) 1 t d -> b s t d", b=b)  # (1, S, t_v, D)

    # Audio preprocessing → Synchformer audio features (1, S, t_a, D)
    waveform = _load_audio_waveform(audio_path, self._resample_cache).to(self.device)
    assert self._mel is not None
    afeats = _encode_audio_segments(self._model, waveform, self._mel)

    # Compare first-14 and last-14 segments; argmax over the 21-class grid;
    # |grid|-value → average across the two directions.
    sync_grid = self._grid.to(self.device) if self._grid is not None else None
    assert sync_grid is not None
    # Synchformer's transformer carries a fixed 198-token positional embedding
    # (~14 segments × (tv+ta) + 2 special tokens). Anything shorter would
    # shape-mismatch on the pos_emb add inside compare_v_a, so skip.
    n_vsegs = int(vfeats.shape[1])
    n_asegs = int(afeats.shape[1])
    s_used = min(n_vsegs, n_asegs)
    if s_used < _NUM_SEG_PER_DIRECTION:
        return self._skip(
            sample,
            f"too few segments for Synchformer pos_emb "
            f"(need {_NUM_SEG_PER_DIRECTION}, got v={n_vsegs} a={n_asegs}); "
            f"use clips of at least ~5 s",
        )
    s_used = _NUM_SEG_PER_DIRECTION
    front_logits = self._model.compare_v_a(vfeats[:, :s_used], afeats[:, :s_used])
    back_logits = self._model.compare_v_a(vfeats[:, -s_used:], afeats[:, -s_used:])
    front_id = int(torch.argmax(front_logits, dim=-1).item())
    back_id = int(torch.argmax(back_logits, dim=-1).item())
    front_d = abs(float(sync_grid[front_id].item()))
    back_d = abs(float(sync_grid[back_id].item()))
    desync = (front_d + back_d) / 2.0
    return MetricResult(
        name=self.name,
        score=desync,
        details={
            "front_desync_s": front_d,
            "back_desync_s": back_d,
            "num_segments_used": s_used,
        },
    )</code></pre></details>


fastvideo.eval.metrics.audio.desync.metric.DeSyncMetric.to


to(device)


Source
def to(self, device):
    super().to(device)
    if self._model is not None:
        self._model = self._model.to(self.device)
    if self._mel is not None:
        self._mel = self._mel.to(self.device)
    return self


fastvideo.eval.metrics.audio.desync.metric.DeSyncMetric.setup


setup() -> None


Source
def setup(self) -> None:
    if self._model is not None:
        return
    import torchaudio
    from fastvideo.third_party.eval.synchformer import Synchformer, make_class_grid

ckpt = ensure_checkpoint(_SYNCHFORMER_NAME, _SYNCHFORMER_URL)
model = Synchformer().to(self.device).eval()
model.load_state_dict(torch.load(ckpt, weights_only=True, map_location=self.device))
self._model = model

self._mel = torchaudio.transforms.MelSpectrogram(
    sample_rate=_AUDIO_SR,
    win_length=400,
    hop_length=160,
    n_fft=1024,
    n_mels=128,
).to(self.device)
self._grid = make_class_grid(_GRID_LOW, _GRID_HIGH, _GRID_SIZE)</code></pre></details>


fastvideo.eval.metrics.audio.desync.metric.DeSyncMetric.compute


compute(sample: dict) -> MetricResult


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

video = sample.get(&quot;video&quot;)
audio_path = sample.get(&quot;audio&quot;)
if video is None or audio_path is None:
    missing = &#91;k for k, v in ((&quot;video&quot;, video), (&quot;audio&quot;, audio_path)) if v is None&#93;
    return self._skip(sample, f&quot;missing &#123;&#x27;, &#x27;.join(repr(k) for k in missing)&#125;&quot;)

# Prefer the per-sample fps the pool decoded at; fall back to the
# constructor override. Without either, the audio/video windows can&#x27;t
# be aligned, so skip rather than guess.
sample_fps = sample.get(&quot;fps&quot;)
src_fps = float(sample_fps) if sample_fps is not None else self._src_fps
if src_fps is None:
    return self._skip(sample, &quot;missing &#x27;fps&#x27; (set sample fps or pass src_fps=)&quot;)

# Video preprocessing → (S, T_seg, C, 224, 224)
frames = video.float().to(self.device)
frames = _resample_video(frames, _SYNC_FPS, src_fps)
frames = _video_transform(frames)
vsegs = _segment_video(frames).unsqueeze(0)  # (B=1, S, T_seg, C, H, W)
# Synchformer&#x27;s extract_vfeats wants (B, S, T_seg, C, H, W); fold (B, S) → (B*S, 1, ...).
b, s = vsegs.shape&#91;:2&#93;
from einops import rearrange
vsegs_for_model = rearrange(vsegs, &quot;b s t c h w -&gt; (b s) 1 t c h w&quot;)
vfeats = self._model.extract_vfeats(vsegs_for_model)
vfeats = rearrange(vfeats, &quot;(b s) 1 t d -&gt; b s t d&quot;, b=b)  # (1, S, t_v, D)

# Audio preprocessing → Synchformer audio features (1, S, t_a, D)
waveform = _load_audio_waveform(audio_path, self._resample_cache).to(self.device)
assert self._mel is not None
afeats = _encode_audio_segments(self._model, waveform, self._mel)

# Compare first-14 and last-14 segments; argmax over the 21-class grid;
# |grid|-value → average across the two directions.
sync_grid = self._grid.to(self.device) if self._grid is not None else None
assert sync_grid is not None
# Synchformer&#x27;s transformer carries a fixed 198-token positional embedding
# (~14 segments × (tv+ta) + 2 special tokens). Anything shorter would
# shape-mismatch on the pos_emb add inside compare_v_a, so skip.
n_vsegs = int(vfeats.shape&#91;1&#93;)
n_asegs = int(afeats.shape&#91;1&#93;)
s_used = min(n_vsegs, n_asegs)
if s_used &lt; _NUM_SEG_PER_DIRECTION:
    return self._skip(
        sample,
        f&quot;too few segments for Synchformer pos_emb &quot;
        f&quot;(need &#123;_NUM_SEG_PER_DIRECTION&#125;, got v=&#123;n_vsegs&#125; a=&#123;n_asegs&#125;); &quot;
        f&quot;use clips of at least ~5 s&quot;,
    )
s_used = _NUM_SEG_PER_DIRECTION
front_logits = self._model.compare_v_a(vfeats&#91;:, :s_used&#93;, afeats&#91;:, :s_used&#93;)
back_logits = self._model.compare_v_a(vfeats&#91;:, -s_used:&#93;, afeats&#91;:, -s_used:&#93;)
front_id = int(torch.argmax(front_logits, dim=-1).item())
back_id = int(torch.argmax(back_logits, dim=-1).item())
front_d = abs(float(sync_grid&#91;front_id&#93;.item()))
back_d = abs(float(sync_grid&#91;back_id&#93;.item()))
desync = (front_d + back_d) / 2.0
return MetricResult(
    name=self.name,
    score=desync,
    details=&#123;
        &quot;front_desync_s&quot;: front_d,
        &quot;back_desync_s&quot;: back_d,
        &quot;num_segments_used&quot;: s_used,
    &#125;,
)</code></pre></details>

Was this page helpful?