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

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

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