Skip to content

preprocess_ltx2_overfit

Preprocess LTX-2 overfit data into parquet format.

Encodes videos with the LTX-2 causal video VAE and captions with the Gemma text encoder (feature extractor + embedding connector) into the t2v parquet schema expected by the training framework.

The stored text embeddings are POST-connector: the connector replaces pad positions with learnable registers and returns an all-valid mask, so the parquet collate's ones/zeros mask stays semantically correct and training needs no text encoder at all. Captions are encoded via the encoder's forward() (the exact inference path), which handles both LTX-2.0 (shared 3840-d features) and LTX-2.3 (separate 4096-d video / 2048-d audio feature extractors).

Videos are resampled to TRAIN_FPS and the preprocessed clip is also saved as an mp4 next to the parquet so overfit tests can use it as the SSIM reference.

Usage

CUDA_VISIBLE_DEVICES=0 python fastvideo/pipelines/preprocess/preprocess_ltx2_overfit.py

Functions:

fastvideo.pipelines.preprocess.preprocess_ltx2_overfit.load_video 

load_video(path: str, num_frames: int, target_fps: float, height: int, width: int) -> tuple[Tensor, ndarray]

Load a video as [1, C, T, H, W] in [-1, 1], resampled to target_fps.

Also returns the uint8 RGB frames [T, H, W, C] for reference-video export.

Source code in fastvideo/pipelines/preprocess/preprocess_ltx2_overfit.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def load_video(path: str, num_frames: int, target_fps: float, height: int,
               width: int) -> tuple[torch.Tensor, np.ndarray]:
    """Load a video as [1, C, T, H, W] in [-1, 1], resampled to target_fps.

    Also returns the uint8 RGB frames [T, H, W, C] for reference-video export.
    """
    with av.open(path) as container:
        if not container.streams.video:
            raise RuntimeError(f"No video stream found in {path}")
        stream = container.streams.video[0]
        native_fps = float(stream.average_rate or target_fps)
        decoded = [torch.from_numpy(frame.to_ndarray(format="rgb24")) for frame in container.decode(video=0)]
    if not decoded:
        raise RuntimeError(f"Could not read any frames from {path}")
    raw = torch.stack(decoded).permute(0, 3, 1, 2)  # [T, C, H, W] uint8
    step = native_fps / target_fps
    wanted = [min(int(round(i * step)), raw.shape[0] - 1) for i in range(num_frames)]

    frames = raw[wanted].float()  # [T, C, H, W] in [0, 255]
    src_h, src_w = frames.shape[2], frames.shape[3]
    scale = max(height / src_h, width / src_w)
    new_h, new_w = int(round(src_h * scale)), int(round(src_w * scale))
    frames = torch.nn.functional.interpolate(frames, size=(new_h, new_w), mode="bilinear", antialias=True)
    top = (new_h - height) // 2
    left = (new_w - width) // 2
    frames = frames[:, :, top:top + height, left:left + width]

    frames_np = (frames.permute(0, 2, 3, 1).clamp(0, 255).round().to(torch.uint8).numpy())
    video = frames / 127.5 - 1.0  # [0,255] -> [-1,1]
    video = video.permute(1, 0, 2, 3).unsqueeze(0)  # [1,C,T,H,W]
    return video, frames_np