Skip to content

wan_s2v_pipeline

Wan2.2-S2V speech-to-video pipeline.

Audio-driven generation: a reference image fixes who the subject is, the prompt sets the scene, and the speech track drives the motion. The audio is encoded once up front (it is the same for every denoising step) and cross-attended to inside 12 of the transformer's 40 blocks.

Long requests run as several clips, the way the official runner does: each clip denoises fresh noise conditioned on the same reference image, its slice of the audio, and the previous clip's last motion_frames pixels re-encoded as motion latents. The clips are concatenated and cut to num_frames.

Classes

Functions:

fastvideo.pipelines.basic.wan.wan_s2v_pipeline.align_audio 

align_audio(audio: ndarray, sample_rate: int, fps: int, num_frames: int) -> ndarray

Cut the source track to the generated span.

Output frame j was generated for audio frame j + WARMUP_FRAMES (the first clip drops its warm-up frames but the audio window starts at 0), so the track starts WARMUP_FRAMES / fps seconds in and lasts exactly num_frames / fps seconds. The MP4 muxer trims to the shorter stream, so a track that ends early simply ends early.

Source code in fastvideo/pipelines/basic/wan/wan_s2v_pipeline.py 
51
52
53
54
55
56
57
58
59
60
61
62
def align_audio(audio: np.ndarray, sample_rate: int, fps: int, num_frames: int) -> np.ndarray:
    """Cut the source track to the generated span.

    Output frame ``j`` was generated for audio frame ``j + WARMUP_FRAMES`` (the
    first clip drops its warm-up frames but the audio window starts at 0), so
    the track starts ``WARMUP_FRAMES / fps`` seconds in and lasts exactly
    ``num_frames / fps`` seconds. The MP4 muxer trims to the shorter stream, so
    a track that ends early simply ends early.
    """
    start = int(round(WARMUP_FRAMES / fps * sample_rate))
    stop = start + int(round(num_frames / fps * sample_rate))
    return audio[..., start:stop]

fastvideo.pipelines.basic.wan.wan_s2v_pipeline.roll_motion_history 

roll_motion_history(history: Tensor | None, frames: Tensor, motion_frames: int) -> Tensor

Shift the newest frames ([B, 3, T, H, W]) into a motion_frames-long history.

The official runner starts from a zero history and slides the newest clip in, so a clip shorter than motion_frames keeps older frames in front.

Source code in fastvideo/pipelines/basic/wan/wan_s2v_pipeline.py 
39
40
41
42
43
44
45
46
47
48
def roll_motion_history(history: torch.Tensor | None, frames: torch.Tensor, motion_frames: int) -> torch.Tensor:
    """Shift the newest ``frames`` ([B, 3, T, H, W]) into a ``motion_frames``-long history.

    The official runner starts from a zero history and slides the newest clip
    in, so a clip shorter than ``motion_frames`` keeps older frames in front.
    """
    if history is None:
        history = frames.new_zeros(frames.shape[0], frames.shape[1], motion_frames, *frames.shape[3:])
    keep = min(motion_frames, frames.shape[2])
    return torch.cat([history[:, :, keep:], frames[:, :, -keep:]], dim=2)