video_io.py

"""Video I/O compatible with the V-JEPA 2.1 reference pipeline.

Two functions, both transcribed from `src/datasets/video_dataset.py` in
`facebookresearch/vjepa2`:

    clip_indices(...)   which frames a clip is built from
    decode_frames(...)  what those frame indices decode to, as RGB uint8

Drop this into your project, or use it as the specification for your own
implementation. `tests/test_frame_sampling.py` checks either against the
reference:

    VJEPA21_USER_SAMPLER=video_io:clip_indices \\
    VJEPA21_USER_DECODER=video_io:decode_frames \\
    python -m pytest tests/test_frame_sampling.py -s -q
"""

from __future__ import annotations

import math
from typing import Sequence

import numpy as np


# --- sampling ---------------------------------------------------------------


def clip_indices(
    video_len: int,
    frames_per_clip: int,
    frame_step: int,
    num_clips: int = 1,
    allow_clip_overlap: bool = False,
    random_clip_sampling: bool = False,
    rng: np.random.Generator | None = None,
) -> list[np.ndarray]:
    """Frame indices for `num_clips` clips, as the reference pipeline picks them.

    The video is split into `num_clips` equal partitions and one clip is taken
    from each. Note that the frames within a clip are placed with `np.linspace`
    over a window of `frames_per_clip * frame_step`, so the effective stride is
    `fpc*fstp/(fpc-1)` rather than `frame_step`: at 16 frames with step 4 this
    differs from `range(0, 64, 4)` on 12 of the 16 indices.

    Args:
        video_len: Total number of frames in the video.
        frames_per_clip: Frames to return per clip.
        frame_step: Nominal stride; sets the window length, see above.
        num_clips: Number of clips, one per partition.
        allow_clip_overlap: When a partition is shorter than a clip, allow
            successive clips to overlap instead of padding within the partition.
        random_clip_sampling: Random window inside the partition. Leave False
            for evaluation and for a reproducible feature store.
        rng: Optional generator, used only when `random_clip_sampling` is True.

    Returns:
        One int64 array of length `frames_per_clip` per clip.
    """
    if video_len <= 0:
        raise ValueError(f"video_len must be positive, got {video_len}")
    if frames_per_clip <= 0 or frame_step <= 0:
        raise ValueError("frames_per_clip and frame_step must be positive")

    fpc, fstp = frames_per_clip, frame_step
    clip_len = int(fpc * fstp)
    partition_len = video_len // num_clips
    draw = rng if rng is not None else np.random

    out = []
    for i in range(num_clips):
        if partition_len > clip_len:
            end = clip_len
            if random_clip_sampling:
                end = int(draw.integers(clip_len, partition_len)) if rng is not None \
                    else int(np.random.randint(clip_len, partition_len))
            start = end - clip_len
            idx = np.linspace(start, end, num=fpc)
            idx = np.clip(idx, start, end - 1).astype(np.int64) + i * partition_len
        elif not allow_clip_overlap:
            idx = np.linspace(0, partition_len, num=partition_len // fstp)
            idx = np.concatenate((idx, np.ones(fpc - partition_len // fstp) * partition_len))
            idx = np.clip(idx, 0, partition_len - 1).astype(np.int64) + i * partition_len
        else:
            sample_len = min(clip_len, video_len) - 1
            idx = np.linspace(0, sample_len, num=sample_len // fstp)
            idx = np.concatenate((idx, np.ones(fpc - sample_len // fstp) * sample_len))
            idx = np.clip(idx, 0, sample_len - 1).astype(np.int64)
            step = (video_len - clip_len) // (num_clips - 1) if video_len > clip_len else 0
            idx = idx + i * step
        out.append(idx)
    return out


def frame_step_for_fps(video_fps: float, target_fps: int) -> int:
    """Stride that samples a video at `target_fps`.

    The reference rounds the source rate *up* before the integer division, so
    29.97 fps behaves like 30 rather than like 29.
    """
    step = math.ceil(video_fps) // target_fps
    if step < 1:
        raise ValueError(
            f"target_fps={target_fps} exceeds the source rate {video_fps}; step would be 0"
        )
    return step


# --- decoding ---------------------------------------------------------------


def available_backends() -> list[str]:
    backends = []
    for name, module in (("decord", "decord"), ("pyav", "av"), ("opencv", "cv2")):
        try:
            __import__(module)
            backends.append(name)
        except ImportError:
            continue
    return backends


def decode_frames(path: str, indices: Sequence[int], backend: str = "auto") -> np.ndarray:
    """Decode the given frame indices as RGB.

    Args:
        path: Video file.
        indices: Frame indices, as returned by `clip_indices`. Repeats are
            allowed and preserved, which matters because the reference pads
            short clips by repeating the last frame.
        backend: "decord", "pyav", "opencv", or "auto".

    Returns:
        `(len(indices), H, W, 3)` uint8 in **RGB** order.

    The colour order is the point of this function. `cv2.VideoCapture` hands
    back BGR; feeding that to the model swaps red and blue, nothing raises, and
    the features are quietly wrong.
    """
    wanted = [int(i) for i in indices]
    if not wanted:
        raise ValueError("indices is empty")
    if min(wanted) < 0:
        raise ValueError(f"negative frame index: {min(wanted)}")

    if backend == "auto":
        found = available_backends()
        if not found:
            raise RuntimeError(
                "no video backend available; install one of: "
                "decord, av (PyAV), opencv-python"
            )
        backend = found[0]

    if backend == "decord":
        return _decode_decord(path, wanted)
    if backend == "pyav":
        return _decode_pyav(path, wanted)
    if backend == "opencv":
        return _decode_opencv(path, wanted)
    raise ValueError(f"unknown backend {backend!r}")


def _decode_decord(path: str, wanted: list[int]) -> np.ndarray:
    from decord import VideoReader, cpu

    reader = VideoReader(path, num_threads=-1, ctx=cpu(0))
    reader.seek(0)  # the reference rewinds before sampling
    return reader.get_batch(wanted).asnumpy()


def _decode_pyav(path: str, wanted: list[int]) -> np.ndarray:
    import av

    needed = set(wanted)
    frames: dict[int, np.ndarray] = {}
    with av.open(path) as container:
        stream = container.streams.video[0]
        stream.thread_type = "AUTO"
        for position, frame in enumerate(container.decode(stream)):
            if position in needed:
                frames[position] = frame.to_ndarray(format="rgb24")
                if len(frames) == len(needed):
                    break
    _check_complete(frames, wanted, path)
    return np.stack([frames[i] for i in wanted])


def _decode_opencv(path: str, wanted: list[int]) -> np.ndarray:
    import cv2

    needed = set(wanted)
    frames: dict[int, np.ndarray] = {}
    capture = cv2.VideoCapture(path)
    try:
        position = 0
        # Sequential decoding: CAP_PROP_POS_FRAMES snaps to keyframes on several
        # codecs, which silently returns a neighbouring frame.
        while len(frames) < len(needed):
            ok, frame = capture.read()
            if not ok:
                break
            if position in needed:
                frames[position] = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            position += 1
    finally:
        capture.release()
    _check_complete(frames, wanted, path)
    return np.stack([frames[i] for i in wanted])


def _check_complete(frames: dict, wanted: list[int], path: str) -> None:
    missing = sorted(set(wanted) - set(frames))
    if missing:
        raise RuntimeError(
            f"{path}: could not decode frames {missing[:8]}"
            f"{'...' if len(missing) > 8 else ''}; the video has {len(frames)} "
            "decodable frames at or before those positions"
        )


__all__ = [
    "clip_indices",
    "dense_clip_indices",
    "frame_step_for_fps",
    "decode_frames",
    "available_backends",
    "aggregate_predictions",
    "temporal_coverage",
    "clip_scores_to_frame_scores",
]


# --- multi-clip: aggregation, coverage, dense grids --------------------------


def aggregate_predictions(logits_per_view: Sequence[np.ndarray]) -> np.ndarray:
    """Combine the predictions of several clips or spatial views of one video.

    Matches the reference evaluation loop, which averages **softmax
    probabilities**, not logits:
    `sum(F.softmax(o, dim=1) for o in outputs) / len(outputs)`
    (`evals/video_classification_frozen/eval.py`).

    Averaging logits instead is a different estimator — it is a geometric rather
    than an arithmetic mean over probabilities — and it lets one confident view
    dominate. The two agree only when the views agree.

    Args:
        logits_per_view: One `(batch, num_classes)` array per clip or view.

    Returns:
        `(batch, num_classes)` probabilities that sum to one.
    """
    if not logits_per_view:
        raise ValueError("logits_per_view is empty")
    probabilities = []
    for logits in logits_per_view:
        logits = np.asarray(logits, dtype=np.float64)
        shifted = logits - logits.max(axis=-1, keepdims=True)
        exponentiated = np.exp(shifted)
        probabilities.append(exponentiated / exponentiated.sum(axis=-1, keepdims=True))
    return np.mean(probabilities, axis=0)


def temporal_coverage(clips: Sequence[np.ndarray], video_len: int) -> dict:
    """How much of a video a set of clips actually looks at.

    The partitioned sampling used by the reference evaluation was designed for
    short single-label videos. On a long one it leaves most of the timeline
    unseen, and an event shorter than `max_gap` can fall entirely between two
    clips without any of it being observed.

    Returns:
        `covered_fraction`, `max_gap` and `num_gaps`, in frames.
    """
    seen = np.zeros(video_len, dtype=bool)
    for clip in clips:
        clip = np.asarray(clip)
        seen[int(clip.min()) : int(clip.max()) + 1] = True

    gaps, run = [], 0
    for visible in seen:
        if visible:
            if run:
                gaps.append(run)
            run = 0
        else:
            run += 1
    if run:
        gaps.append(run)

    return {
        "covered_fraction": float(seen.mean()),
        "max_gap": int(max(gaps)) if gaps else 0,
        "num_gaps": len(gaps),
    }


def dense_clip_indices(
    video_len: int,
    frames_per_clip: int,
    frame_step: int,
    stride: int | None = None,
) -> list[np.ndarray]:
    """A sliding grid of clips covering the whole video.

    Use this instead of `clip_indices` when you need a score per position in
    time rather than one prediction per video — temporal anomaly detection,
    action localisation, anything scored frame by frame. `clip_indices`
    partitions the video and samples one clip per partition, which is the right
    thing for classifying a short video and the wrong thing here.

    Args:
        stride: Frames between the start of consecutive clips. Defaults to the
            clip window `frames_per_clip * frame_step`, giving contiguous
            non-overlapping clips. A smaller value overlaps them, which raises
            temporal resolution at proportional cost.

    Returns:
        Clips in temporal order, the last one clamped to the end of the video.
    """
    window = frames_per_clip * frame_step
    stride = window if stride is None else stride
    if stride <= 0:
        raise ValueError(f"stride must be positive, got {stride}")

    starts = list(range(0, max(video_len - window, 0) + 1, stride))
    if not starts:
        starts = [0]
    if starts[-1] + window < video_len:
        starts.append(video_len - window)

    clips = []
    for start in starts:
        idx = np.linspace(start, start + window, num=frames_per_clip)
        clips.append(np.clip(idx, 0, video_len - 1).astype(np.int64))
    return clips


def clip_scores_to_frame_scores(
    clips: Sequence[np.ndarray],
    scores: Sequence[float],
    video_len: int,
    reduce: str = "max",
) -> np.ndarray:
    """Spread clip-level scores back over frames, for frame-level metrics.

    Frame-level AUC on UCF-Crime and average precision on XD-Violence are
    computed per frame, so a clip score has to be assigned to the frames the
    clip covers. Overlapping clips give a frame several scores; `reduce` picks
    between them. Frames covered by no clip keep the score of the nearest
    covered frame, so the output is dense.

    Args:
        reduce: "max" (an anomaly anywhere in the window marks the window),
            "mean" (smoother, blunter) or "first".
    """
    if len(clips) != len(scores):
        raise ValueError(f"{len(clips)} clips but {len(scores)} scores")

    accumulated = np.zeros(video_len, dtype=np.float64)
    counts = np.zeros(video_len, dtype=np.int64)
    assigned = np.zeros(video_len, dtype=bool)

    for clip, score in zip(clips, scores):
        clip = np.asarray(clip)
        lo, hi = int(clip.min()), int(clip.max()) + 1
        if reduce == "max":
            accumulated[lo:hi] = np.where(
                assigned[lo:hi], np.maximum(accumulated[lo:hi], score), score
            )
        elif reduce == "mean":
            accumulated[lo:hi] += score
        elif reduce == "first":
            accumulated[lo:hi] = np.where(assigned[lo:hi], accumulated[lo:hi], score)
        else:
            raise ValueError(f"unknown reduce {reduce!r}")
        counts[lo:hi] += 1
        assigned[lo:hi] = True

    if reduce == "mean":
        accumulated[counts > 0] /= counts[counts > 0]

    if not assigned.all():
        covered = np.flatnonzero(assigned)
        if covered.size == 0:
            raise ValueError("no frame was covered by any clip")
        nearest = covered[np.abs(np.subtract.outer(np.arange(video_len), covered)).argmin(axis=1)]
        accumulated = np.where(assigned, accumulated, accumulated[nearest])
    return accumulated