Basic Video Generation Tutorial¶

from fastvideo import VideoGenerator

# from fastvideo.api.sampling_param import SamplingParam

OUTPUT_PATH = "video_samples"


def main():
    # FastVideo will automatically use the optimal default arguments for the
    # model.
    # If a local path is provided, FastVideo will make a best effort
    # attempt to identify the optimal arguments.
    generator = VideoGenerator.from_pretrained(
        "Wan-AI/Wan2.1-T2V-1.3B-Diffusers",
        # FastVideo will automatically handle distributed setup
        num_gpus=1,
        use_fsdp_inference=False,  # set to True if GPU is out of memory
        dit_cpu_offload=False,
        vae_cpu_offload=False,
        text_encoder_cpu_offload=True,
        pin_cpu_memory=True,  # set to false if low CPU RAM or hit obscure "CUDA error: Invalid argument"
        # image_encoder_cpu_offload=False,
    )

    # sampling_param = SamplingParam.from_pretrained("Wan-AI/Wan2.1-T2V-1.3B-Diffusers")
    # sampling_param.num_frames = 45
    # sampling_param.image_path = "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/astronaut.jpg"
    # Generate videos with the same simple API, regardless of GPU count
    prompt = ("A curious raccoon peers through a vibrant field of yellow sunflowers, its eyes "
              "wide with interest. The playful yet serene atmosphere is complemented by soft "
              "natural light filtering through the petals. Mid-shot, warm and cheerful tones.")
    video = generator.generate_video(prompt, output_path=OUTPUT_PATH, save_video=True)
    # video = generator.generate_video(prompt, sampling_param=sampling_param, output_path="wan_t2v_videos/")

    # Generate another video with a different prompt, without reloading the
    # model!
    prompt2 = ("A majestic lion strides across the golden savanna, its powerful frame "
               "glistening under the warm afternoon sun. The tall grass ripples gently in "
               "the breeze, enhancing the lion's commanding presence. The tone is vibrant, "
               "embodying the raw energy of the wild. Low angle, steady tracking shot, "
               "cinematic.")
    video2 = generator.generate_video(prompt2, output_path=OUTPUT_PATH, save_video=True)


if __name__ == "__main__":
    main()

# SPDX-License-Identifier: Apache-2.0
from fastvideo import VideoGenerator
from fastvideo.api.sampling_param import SamplingParam


def main():
    # Point this to your local diffusers model dir (or replace with a HF model ID).
    model_path = "KyleShao/Cosmos-Predict2.5-2B-Diffusers"

    generator = VideoGenerator.from_pretrained(
        model_path,
        num_gpus=1,
        use_fsdp_inference=False,  # set True if GPU is out of memory
        dit_cpu_offload=False,
        vae_cpu_offload=False,
        text_encoder_cpu_offload=True,
        pin_cpu_memory=True,
    )

    sampling_param = SamplingParam.from_pretrained(model_path)

    # image2world example from official repo
    image_path = "assets/images/bus_terminal.jpg"

    prompt = (
        "A nighttime city bus terminal gradually shifts from stillness to subtle movement. "
        "At first, multiple double-decker buses are parked under the glow of overhead lights, "
        "with a central bus labeled '87D' facing forward and stationary. "
        "As the video progresses, the bus in the middle moves ahead slowly, its headlights brightening the surrounding area "
        "and casting reflections onto adjacent vehicles. "
        "The motion creates space in the lineup, signaling activity within the otherwise quiet station. "
        "It then comes to a smooth stop, resuming its position in line. "
        "Overhead signage in Chinese characters remains illuminated, enhancing the vibrant, urban night scene.")

    generator.generate_video(
        prompt,
        sampling_param=sampling_param,
        image_path=str(image_path),
        num_cond_frames=1,
        output_path="outputs_video/cosmos2_5_i2w.mp4",
        save_video=True,
    )

    generator.shutdown()


if __name__ == "__main__":
    main()

from fastvideo import VideoGenerator
from fastvideo.api.sampling_param import SamplingParam


def main():
    # Point this to your local diffusers model dir (or replace with a HF model ID).
    model_path = "KyleShao/Cosmos-Predict2.5-2B-Diffusers"

    generator = VideoGenerator.from_pretrained(
        model_path,
        num_gpus=1,
        use_fsdp_inference=False,  # set True if GPU is out of memory
        dit_cpu_offload=False,
        vae_cpu_offload=False,
        text_encoder_cpu_offload=True,
        pin_cpu_memory=True,
    )

    # Load default sampling parameters (negative_prompt, resolution, steps, etc.)
    sampling_param = SamplingParam.from_pretrained(model_path)

    prompt = (
        "A high-definition video captures the precision of robotic welding in an industrial setting. "
        "The first frame showcases a robotic arm, equipped with a welding torch, positioned over a large metal structure. "
        "The welding process is in full swing, with bright sparks and intense light illuminating the scene, "
        "creating a vivid display of blue and white hues. "
        "A significant amount of smoke billows around the welding area, partially obscuring the view but emphasizing the heat and activity. "
        "The background reveals parts of the workshop environment, including a ventilation system and various pieces of machinery, "
        "indicating a busy and functional industrial workspace. "
        "As the video progresses, the robotic arm maintains its steady position, continuing the welding process and moving to its left. "
        "The welding torch consistently emits sparks and light, and the smoke continues to rise, diffusing slightly as it moves upward. "
        "The metal surface beneath the torch shows ongoing signs of heating and melting. "
        "The scene retains its industrial ambiance, with the welding sparks and smoke dominating the visual field, "
        "underscoring the ongoing nature of the welding operation.")

    generator.generate_video(
        prompt,
        sampling_param=sampling_param,
        output_path="outputs_video/cosmos2_5_t2w.mp4",
        save_video=True,
    )

    generator.shutdown()


if __name__ == "__main__":
    main()

# SPDX-License-Identifier: Apache-2.0
from fastvideo import VideoGenerator
from fastvideo.api.sampling_param import SamplingParam


def main():
    # Point this to your local diffusers model dir (or replace with a HF model ID).
    model_path = "KyleShao/Cosmos-Predict2.5-2B-Diffusers"

    generator = VideoGenerator.from_pretrained(
        model_path,
        num_gpus=1,
        use_fsdp_inference=False,  # set True if GPU is out of memory
        dit_cpu_offload=False,
        vae_cpu_offload=False,
        text_encoder_cpu_offload=True,
        pin_cpu_memory=True,
    )

    sampling_param = SamplingParam.from_pretrained(model_path)

    # video2world example from official repo
    video_path = "assets/videos/robot_pouring.mp4"

    prompt = (
        "A robotic arm, primarily white with black joints and cables, is shown in a clean, modern indoor setting with a white tabletop. "
        "The arm, equipped with a gripper holding a small, light green pitcher, is positioned above a clear glass containing a reddish-brown liquid and a spoon. "
        "The robotic arm is in the process of pouring a transparent liquid into the glass. "
        "To the left of the pitcher, there is an opened jar with a similar reddish-brown substance visible through its transparent body. "
        "In the background, a vase with white flowers and a brown couch are partially visible, adding to the contemporary ambiance. "
        "The lighting is bright, casting soft shadows on the table. "
        "The robotic arm's movements are smooth and controlled, demonstrating precision in its task. "
        "As the video progresses, the robotic arm completes the pour, leaving the glass half-filled with the reddish-brown liquid. "
        "The jar remains untouched throughout the sequence, and the spoon inside the glass remains stationary. "
        "The other robotic arm on the right side also stays stationary throughout the video. "
        "The final frame captures the robotic arm with the pitcher finishing the pour, with the glass now filled to a higher level, while the pitcher is slightly tilted but still held securely by the gripper."
    )

    generator.generate_video(
        prompt,
        sampling_param=sampling_param,
        video_path=str(video_path),
        num_cond_frames=1,
        output_path="outputs_video/cosmos2_5_v2w.mp4",
        save_video=True,
    )

    generator.shutdown()


if __name__ == "__main__":
    main()

import os
import time
from fastvideo import VideoGenerator

from fastvideo.api.sampling_param import SamplingParam

OUTPUT_PATH = "video_samples_dmd2"


def main():
    os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "VIDEO_SPARSE_ATTN"

    load_start_time = time.perf_counter()
    model_name = "FastVideo/FastWan2.1-T2V-1.3B-Diffusers"
    generator = VideoGenerator.from_pretrained(
        model_name,
        # FastVideo will automatically handle distributed setup
        num_gpus=1,
        use_fsdp_inference=False,  # set to True if GPU is out of memory
        # Adjust these offload parameters if you have < 32GB of VRAM
        text_encoder_cpu_offload=True,
        pin_cpu_memory=True,  # set to false if low CPU RAM or hit obscure "CUDA error: Invalid argument"
        dit_cpu_offload=False,
        vae_cpu_offload=False,
        VSA_sparsity=0.8,
    )
    load_end_time = time.perf_counter()
    load_time = load_end_time - load_start_time

    sampling_param = SamplingParam.from_pretrained(model_name)
    sampling_param.num_frames = 81

    prompt = (
        "A neon-lit alley in futuristic Tokyo during a heavy rainstorm at night. The puddles reflect glowing signs in kanji, advertising ramen, karaoke, and VR arcades. A woman in a translucent raincoat walks briskly with an LED umbrella. Steam rises from a street food cart, and a cat darts across the screen. Raindrops are visible on the camera lens, creating a cinematic bokeh effect."
    )
    start_time = time.perf_counter()
    video = generator.generate_video(prompt, output_path=OUTPUT_PATH, save_video=True, sampling_param=sampling_param)
    end_time = time.perf_counter()
    gen_time = end_time - start_time

    # Generate another video with a different prompt, without reloading the
    # model!
    prompt2 = ("A majestic lion strides across the golden savanna, its powerful frame "
               "glistening under the warm afternoon sun. The tall grass ripples gently in "
               "the breeze, enhancing the lion's commanding presence. The tone is vibrant, "
               "embodying the raw energy of the wild. Low angle, steady tracking shot, "
               "cinematic.")
    start_time = time.perf_counter()
    video2 = generator.generate_video(prompt2, output_path=OUTPUT_PATH, save_video=True, num_frames=81)
    end_time = time.perf_counter()
    gen_time2 = end_time - start_time

    print(f"Time taken to load model: {load_time} seconds")
    print(f"Time taken to generate video: {gen_time} seconds")
    print(f"Time taken to generate video2: {gen_time2} seconds")


if __name__ == "__main__":
    main()

import os
import time

from fastvideo import VideoGenerator
from fastvideo.api import (
    EngineConfig,
    GenerationRequest,
    GeneratorConfig,
    OffloadConfig,
    OutputConfig,
    PipelineSelection,
)

OUTPUT_PATH = "video_samples_dmd2_typed"


def main():
    os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "VIDEO_SPARSE_ATTN"

    model_name = "FastVideo/FastWan2.1-T2V-1.3B-Diffusers"
    generator_config = GeneratorConfig(
        model_path=model_name,
        engine=EngineConfig(
            num_gpus=1,
            use_fsdp_inference=False,
            offload=OffloadConfig(
                text_encoder=True,
                pin_cpu_memory=True,
                dit=False,
                vae=False,
            ),
        ),
        # PR 2 still routes a few advanced inference knobs through the
        # compatibility bridge until they get first-class typed fields.
        pipeline=PipelineSelection(experimental={
            "VSA_sparsity": 0.8,
        }, ),
    )

    load_start_time = time.perf_counter()
    generator = VideoGenerator.from_config(generator_config)
    load_end_time = time.perf_counter()
    load_time = load_end_time - load_start_time

    prompt = ("A neon-lit alley in futuristic Tokyo during a heavy rainstorm at night. "
              "The puddles reflect glowing signs in kanji, advertising ramen, karaoke, "
              "and VR arcades. A woman in a translucent raincoat walks briskly with an "
              "LED umbrella. Steam rises from a street food cart, and a cat darts "
              "across the screen. Raindrops are visible on the camera lens, creating "
              "a cinematic bokeh effect.")
    request = GenerationRequest(
        prompt=prompt,
        output=OutputConfig(
            output_path=OUTPUT_PATH,
            save_video=True,
            return_frames=False,
        ),
    )

    start_time = time.perf_counter()
    result = generator.generate(request)
    end_time = time.perf_counter()
    gen_time = end_time - start_time

    prompt2 = ("A majestic lion strides across the golden savanna, its powerful frame "
               "glistening under the warm afternoon sun. The tall grass ripples gently "
               "in the breeze, enhancing the lion's commanding presence. The tone is "
               "vibrant, embodying the raw energy of the wild. Low angle, steady "
               "tracking shot, cinematic.")
    request2 = GenerationRequest(
        prompt=prompt2,
        output=OutputConfig(
            output_path=OUTPUT_PATH,
            save_video=True,
            return_frames=False,
        ),
    )

    start_time = time.perf_counter()
    result2 = generator.generate(request2)
    end_time = time.perf_counter()
    gen_time2 = end_time - start_time

    print(f"Time taken to load model: {load_time} seconds")
    print(f"Time taken to generate video: {gen_time} seconds")
    print(f"First output written to: {result.video_path}")
    print(f"Time taken to generate video2: {gen_time2} seconds")
    print(f"Second output written to: {result2.video_path}")


if __name__ == "__main__":
    main()

import os

from fastvideo import VideoGenerator

OUTPUT_PATH = os.getenv("DREAMX_WORLD_OUTPUT_PATH", "video_samples_dreamx_world")


def _env_int(name: str, default: int) -> int:
    return int(os.getenv(name, str(default)))


def _env_float(name: str, default: float) -> float:
    return float(os.getenv(name, str(default)))


def main():
    model_name = os.getenv("DREAMX_WORLD_MODEL_DIR", "FastVideo/DreamX-World-5B-Cam-Diffusers")
    generator = VideoGenerator.from_pretrained(
        model_name,
        num_gpus=1,
        use_fsdp_inference=False,
        dit_cpu_offload=False,
        vae_cpu_offload=True,
        text_encoder_cpu_offload=True,
        pin_cpu_memory=False,
        override_pipeline_cls_name="DreamXWorldPipeline",
    )

    prompt = os.getenv(
        "DREAMX_WORLD_PROMPT",
        "A cinematic first-person drive through a futuristic coastal city at "
        "sunrise, reflective glass towers, clean streets, soft volumetric light.",
    )
    image_path = os.getenv(
        "DREAMX_WORLD_IMAGE_PATH",
        "https://huggingface.co/datasets/YiYiXu/testing-images/resolve/main/wan_i2v_input.JPG",
    )

    kwargs = {
        "output_path": OUTPUT_PATH,
        "save_video": os.getenv("DREAMX_WORLD_SAVE_VIDEO", "1") != "0",
        "height": _env_int("DREAMX_WORLD_HEIGHT", 480),
        "width": _env_int("DREAMX_WORLD_WIDTH", 832),
        "num_frames": _env_int("DREAMX_WORLD_NUM_FRAMES", 161),
        "num_inference_steps": _env_int("DREAMX_WORLD_STEPS", 30),
        "guidance_scale": _env_float("DREAMX_WORLD_GUIDANCE", 5.0),
        "action_list": os.getenv("DREAMX_WORLD_ACTIONS", "w,d,w").split(","),
        "action_speed_list":
        [float(value) for value in os.getenv("DREAMX_WORLD_ACTION_SPEEDS", "4.0,2.0,4.0").split(",")],
    }
    if image_path:
        kwargs["image_path"] = image_path

    try:
        generator.generate_video(prompt, **kwargs)
    finally:
        generator.shutdown()


if __name__ == "__main__":
    main()

# SPDX-License-Identifier: Apache-2.0
"""Few-step video+audio generation with the DMD2-distilled MiniMax H3 preview.

The default ``all`` profile reproduces the fastest measured FastH3 Preview
recipe on four GB200 GPUs. It runs the checkpoint's native five-point sigma
grid (exactly four DiT forwards), trained VSA policy, Blackwell sparse kernel,
regional fullgraph DiT compile, compiled/parallel video VAE, and inference-only
H3 fusions. One compile warmup is excluded before three measured requests.

Both regional compile and the default fusions can change floating-point
operation order, so ``all`` is a report-only performance profile.
``--profile strict`` disables the H3 fusions but preserves regional compile;
combine it with ``--no-inference-torch-compile`` for the eager strict route.
"""

from __future__ import annotations

import argparse
import importlib.util
import os
import statistics
import time
from collections.abc import Sequence
from pathlib import Path

from fastvideo import VideoGenerator
from fastvideo.api import (
    CompileConfig,
    ComponentConfig,
    EngineConfig,
    GenerationRequest,
    GeneratorConfig,
    OffloadConfig,
    OutputConfig,
    ParallelismConfig,
    PipelineSelection,
    SamplingConfig,
)

DEFAULT_MODEL = "FastVideo/FastVideo-Minimax-FastH3-Preview-v0.2"


def build_parser(description: str | None = None) -> argparse.ArgumentParser:
    """Build the shared FastH3 preview CLI used by full and LoRA checkpoints."""
    parser = argparse.ArgumentParser(description=description or __doc__)
    parser.add_argument("--model-path", default=DEFAULT_MODEL)
    # The HF repo may require authentication while the MiniMax H3 Community
    # License review completes. A local snapshot can be passed here instead.
    parser.add_argument("--prompt", required=True)
    parser.add_argument("--output", default="outputs/fasth3")
    parser.add_argument("--lazy-module-load",
                        action=argparse.BooleanOptionalAction,
                        default=None,
                        help="load each heavy component on first use and free it after the last stage that "
                        "needs it, so peak memory is the largest overlapping set instead of the sum of every "
                        "component. Omit for auto (on for unified-memory devices such as GB10; off on discrete "
                        "GPUs). Costs a reload per generation; pass --no-lazy-module-load to keep every "
                        "component resident")
    parser.add_argument("--profile",
                        choices=("all", "strict"),
                        default="all",
                        help="all enables the fastest measured, non-parity H3 fusions; strict disables only them")
    parser.add_argument("--height", type=int, default=768)
    parser.add_argument("--width", type=int, default=1344)
    parser.add_argument("--num-frames", type=int, default=124)
    # num_inference_steps counts sigma-GRID POINTS. The distilled schedule is
    # t=1000,750,500,250 -> 0: five points and exactly four DiT forwards.
    parser.add_argument("--steps",
                        type=int,
                        default=5,
                        help="sigma-grid points; N points run N-1 DiT forwards (the trained default is 5)")
    parser.add_argument("--seed", type=int, default=1000, help="seed reused for every measured request")
    parser.add_argument("--warmup-seed", type=int, default=999)
    parser.add_argument("--repeats", type=int, default=3, help="number of measured requests after warmup")
    parser.add_argument("--warmup",
                        action=argparse.BooleanOptionalAction,
                        default=True,
                        help="run one excluded request before timing")
    parser.add_argument("--num-gpus", type=int, default=4)
    parser.add_argument(
        "--execution-backend",
        choices=("mp", "ray"),
        default=None,
        help="mp for one node; ray for a Ray cluster (two DGX Sparks). "
        "Default: ray when RAY_ADDRESS is set, otherwise mp",
    )
    parser.add_argument("--vsa-sparsity",
                        type=float,
                        default=0.9,
                        help="run-level VSA sparsity in [0, 1); 0.9 is the checkpoint's trained policy")
    parser.add_argument("--vsa-tile-size",
                        type=int,
                        choices=(64, 256),
                        default=64,
                        help="VSA-H3 tile size; 64 is the checkpoint's trained and measured geometry")
    parser.add_argument("--vsa-kernel",
                        choices=("triton", "sm100a"),
                        default="sm100a",
                        help="tile-64 sparse kernel; sm100a is the measured GB200 route and requires a compatible "
                        "fastvideo-kernel build")
    parser.add_argument("--fa4",
                        action=argparse.BooleanOptionalAction,
                        default=True,
                        help="use FA4 for eligible non-VSA attention paths")
    parser.add_argument("--h3-fusions",
                        action=argparse.BooleanOptionalAction,
                        default=None,
                        help="override the profile's H3 fusion policy (changes model numerics when enabled)")
    parser.add_argument("--compile-vae",
                        action=argparse.BooleanOptionalAction,
                        default=True,
                        help="compile the video VAE decoder independently of the DiT")
    parser.add_argument("--parallel-vae",
                        action=argparse.BooleanOptionalAction,
                        default=True,
                        help="round-robin VAE temporal chunks across sequence-parallel ranks")
    parser.add_argument("--h3-sequential-load",
                        action=argparse.BooleanOptionalAction,
                        default=None,
                        help="encode with Qwen3-VL, release it, then load DiT/VAEs. Default auto: on for "
                        "unified-memory devices (GB10), off on discrete GPUs")
    parser.add_argument("--video-decode-backend",
                        choices=("h3-vae", "taeh3"),
                        default="h3-vae",
                        help="h3-vae is the full MiniMax VAE; taeh3 is the fast approximate preview decoder")
    parser.add_argument("--taeh3-checkpoint", default=None, help="local taeh3.safetensors; unset uses the pinned cache")
    parser.add_argument("--replicated-dit",
                        action=argparse.BooleanOptionalAction,
                        default=True,
                        help="replicate DiT weights instead of FSDP-sharding them")
    parser.add_argument("--pin-cpu-memory",
                        action=argparse.BooleanOptionalAction,
                        default=True,
                        help="pin CPU-offloaded text-encoder and VAE weights")
    parser.add_argument("--torch-compile",
                        action=argparse.BooleanOptionalAction,
                        default=False,
                        help="compile the whole DiT path (off in the fastest FastH3 profile)")
    parser.add_argument("--inference-torch-compile",
                        action=argparse.BooleanOptionalAction,
                        default=True,
                        help="regionally compile DiT blocks (enabled in the fastest FastH3 profile)")
    parser.add_argument("--ulysses-a2a",
                        choices=("off", "auto"),
                        default="off",
                        help="sequence-parallel all-to-all route; off reproduces the fastest FastH3 profile, while "
                        "auto opts into the fused NVLink kernel when the installed kernel package supports it")
    parser.add_argument("--compile-mode",
                        default=None,
                        help='whole-DiT torch.compile mode, e.g. "reduce-overhead"; requires '
                        "--no-inference-torch-compile")
    return parser


def validate_args(parser: argparse.ArgumentParser, args: argparse.Namespace) -> argparse.Namespace:
    if args.repeats < 1:
        parser.error("--repeats must be at least 1")
    if args.num_gpus < 1:
        parser.error("--num-gpus must be at least 1")
    if not 0.0 <= args.vsa_sparsity < 1.0:
        parser.error("--vsa-sparsity must be in [0, 1)")
    if args.compile_mode is not None and args.inference_torch_compile:
        parser.error("--compile-mode cannot be combined with regional compile; pass --no-inference-torch-compile")
    return args


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    parser = build_parser()
    return validate_args(parser, parser.parse_args(argv))


def _uses_vsa(args: argparse.Namespace) -> bool:
    """Full FastH3 checkpoints use VSA; LoRA previews may select dense attention."""
    return bool(getattr(args, "vsa", True))


def _h3_fusions_enabled(args: argparse.Namespace) -> bool:
    if args.h3_fusions is not None:
        return bool(args.h3_fusions)
    return args.profile == "all"


def profile_environment(args: argparse.Namespace) -> dict[str, str | None]:
    """Return the complete boot-time environment for this profile.

    ``None`` means the variable must be removed. Values are explicit even for
    disabled features so a shell's inherited experiment settings cannot
    silently change the advertised profile.
    """
    use_vsa = _uses_vsa(args)
    return {
        "FASTVIDEO_ATTENTION_BACKEND": "VIDEO_SPARSE_ATTN_H3" if use_vsa else "FLASH_ATTN",
        "FASTVIDEO_VSA_SM100A": "1" if use_vsa and args.vsa_kernel == "sm100a" else "0",
        "FASTVIDEO_VSA_CUTEDSL": "0",
        # A non-empty output path enables the diagnostic probe.
        "FASTVIDEO_H3_VSA_PROBE": None,
        "FASTVIDEO_DISABLE_ATTENTION_COMPILE": "0",
        "FASTVIDEO_FA4": "1" if args.fa4 else "0",
        "FASTVIDEO_NVFP4_FA4": "0",
        "FASTVIDEO_MINIMAX_H3_FA4_PACKED_VARLEN": "0",
        "FASTVIDEO_MINIMAX_H3_FUSIONS": "all" if _h3_fusions_enabled(args) else "0",
        "FASTVIDEO_INFERENCE_TORCH_COMPILE": "1" if args.inference_torch_compile else "0",
        "FASTVIDEO_VAE_PARALLEL_DECODE": "1" if args.parallel_vae else "0",
        "FASTVIDEO_VAE_PARALLEL_ENCODE": "0",
        "FASTVIDEO_VAE_PARALLEL_DECODE_STRATEGY": "gather",
        "FASTVIDEO_ULYSSES_A2A": args.ulysses_a2a,
        "FASTVIDEO_STAGE_LOGGING": "1",
    }


def configure_environment(args: argparse.Namespace) -> dict[str, str | None]:
    environment = profile_environment(args)
    for name, value in environment.items():
        if value is None:
            os.environ.pop(name, None)
        else:
            os.environ[name] = value
    return environment


def _fa4_is_installed() -> bool:
    try:
        return importlib.util.find_spec("flash_attn.cute") is not None
    except (ImportError, ModuleNotFoundError):
        return False


def _sm100a_kernel_is_installed() -> bool:
    try:
        from fastvideo_kernel import block_sparse_attn_sm100a
    except ImportError:
        return False
    return bool(getattr(block_sparse_attn_sm100a, "_HAS_VSA_SM100A", False))


def validate_profile_dependencies(args: argparse.Namespace) -> None:
    """Fail before model loading when the selected measured route is absent."""
    if args.fa4 and not _fa4_is_installed():
        raise RuntimeError(
            "FastH3's FA4 profile requires the pinned flash-attn-4 package. Install it with "
            "`UV_TORCH_BACKEND=cu130 uv pip install -e \".[fasth3]\"`, or pass --no-fa4.")
    if _uses_vsa(args) and args.vsa_kernel == "sm100a" and not _sm100a_kernel_is_installed():
        raise RuntimeError(
            "FastH3's sm100a profile requires fastvideo-kernel 0.3.4 built with the Blackwell VSA extension. "
            "Install this checkout with `UV_TORCH_BACKEND=cu130 uv pip install -e \".[fasth3]\"` (or run "
            "`cd fastvideo-kernel && ./build.sh`), or pass --vsa-kernel triton.")


def _execution_backend(args: argparse.Namespace) -> str:
    if args.execution_backend is not None:
        return args.execution_backend
    return "ray" if os.environ.get("RAY_ADDRESS") else "mp"


def build_generator_config(args: argparse.Namespace) -> GeneratorConfig:
    use_vsa = _uses_vsa(args)
    experimental: dict[str, object] = {
        "attention_backend": "VIDEO_SPARSE_ATTN_H3" if use_vsa else "FLASH_ATTN",
        "inference_torch_compile": args.inference_torch_compile,
        "vae_parallel_decode": args.parallel_vae,
        "vae_parallel_decode_strategy": "gather",
    }
    if args.h3_sequential_load is not None:
        experimental["h3_sequential_load"] = args.h3_sequential_load
    if args.video_decode_backend != "h3-vae":
        experimental["video_decode_backend"] = args.video_decode_backend
    if args.taeh3_checkpoint is not None:
        experimental["taeh3_checkpoint"] = args.taeh3_checkpoint
    if use_vsa:
        experimental.update({
            "VSA_sparsity": args.vsa_sparsity,
            "VSA_tile_size": args.vsa_tile_size,
        })
    return GeneratorConfig(
        model_path=args.model_path,
        pipeline=PipelineSelection(
            components=ComponentConfig(
                lora_path=getattr(args, "lora_path", None),
                lora_strength=float(getattr(args, "lora_strength", 1.0)),
            ),
            experimental=experimental,
        ),
        engine=EngineConfig(
            num_gpus=args.num_gpus,
            execution_backend=_execution_backend(args),
            use_fsdp_inference=args.num_gpus > 1 and not args.replicated_dit,
            parallelism=ParallelismConfig(tp_size=1, sp_size=args.num_gpus),
            offload=OffloadConfig(
                dit=False,
                dit_layerwise=False,
                text_encoder=True,
                vae=True,
                pin_cpu_memory=args.pin_cpu_memory,
                lazy_module_load=args.lazy_module_load,
            ),
            compile=CompileConfig(
                enabled=args.torch_compile,
                mode=args.compile_mode,
                vae_enabled=args.compile_vae,
            ),
        ),
    )


def build_request(args: argparse.Namespace, output_path: Path, seed: int) -> GenerationRequest:
    return GenerationRequest(
        prompt=args.prompt,
        negative_prompt="",
        sampling=SamplingConfig(
            height=args.height,
            width=args.width,
            num_frames=args.num_frames,
            fps=24,
            num_inference_steps=args.steps,
            # MiniMax-H3 is guidance-distilled; FastH3 inherits that contract.
            guidance_scale=1.0,
            batch_cfg=False,
            seed=seed,
        ),
        output=OutputConfig(
            output_path=str(output_path),
            save_video=True,
            return_frames=False,
        ),
    )


def _actual_output_path(result: object, requested: Path) -> Path:
    video_path = getattr(result, "video_path", None)
    return Path(video_path) if video_path else requested


def _denoise_seconds(result: object) -> float | None:
    stages = getattr(getattr(result, "logging_info", None), "stages", None)
    if not stages:
        return None
    for stage_name, metrics in stages.items():
        if "denois" not in stage_name.lower():
            continue
        execution_time = metrics.get("execution_time")
        return float(execution_time) if execution_time is not None else None
    return None


def run(args: argparse.Namespace) -> list[float]:
    output_dir = Path(args.output)
    output_dir.mkdir(parents=True, exist_ok=True)
    environment = configure_environment(args)
    validate_profile_dependencies(args)

    print(f"Profile: {args.profile} ({'non-parity fusions' if _h3_fusions_enabled(args) else 'fusions off'})")
    print(f"Output directory: {output_dir.resolve()}")
    print("Denoising contract: 5 sigma points = 4 DiT forwards" if args.steps == 5 else
          f"Denoising contract override: {args.steps} sigma points = {args.steps - 1} DiT forwards")
    print("Profile environment: " + " ".join(f"{key}={value if value is not None else '<unset>'}"
                                                  for key, value in environment.items()))
    print(f"Execution backend: {_execution_backend(args)}")

    generator = VideoGenerator.from_config(build_generator_config(args))
    measured_wall_times: list[float] = []
    measured_denoise_times: list[float] = []
    try:
        if args.warmup:
            warmup_path = output_dir / "_fasth3_warmup.mp4"
            print(f"[warmup] generating (excluded from timing summary): {warmup_path}")
            started = time.perf_counter()
            warmup_result = generator.generate(build_request(args, warmup_path, args.warmup_seed))
            warmup_wall = time.perf_counter() - started
            actual_warmup_path = _actual_output_path(warmup_result, warmup_path)
            print(f"[warmup] wall={warmup_wall:.3f}s (excluded)")
            print(f"Warmup output written to: {actual_warmup_path}")

        for index in range(1, args.repeats + 1):
            requested_path = output_dir / f"fasth3_{args.profile}_run_{index:02d}.mp4"
            print(f"[measured {index}/{args.repeats}] generating: {requested_path}")
            started = time.perf_counter()
            result = generator.generate(build_request(args, requested_path, args.seed))
            wall = time.perf_counter() - started
            measured_wall_times.append(wall)
            actual_path = _actual_output_path(result, requested_path)
            print(f"Output written to: {actual_path}")
            print(f"E2E wall time: {wall:.3f}s")
            generation_time = getattr(result, "generation_time", None)
            if generation_time is not None:
                print(f"Generation time: {float(generation_time):.3f}s")
            denoise_time = _denoise_seconds(result)
            if denoise_time is not None:
                measured_denoise_times.append(denoise_time)
                print(f"Denoising time: {denoise_time:.3f}s")

        median = statistics.median(measured_wall_times)
        print(f"Measured E2E wall times (n={len(measured_wall_times)}, warmup excluded): "
              f"{[round(value, 3) for value in measured_wall_times]}")
        print(f"Median E2E wall time: {median:.3f}s")
        if measured_denoise_times:
            print(f"Median denoising time: {statistics.median(measured_denoise_times):.3f}s")
        return measured_wall_times
    finally:
        generator.shutdown()


def main() -> None:
    run(parse_args())


if __name__ == "__main__":
    main()

# SPDX-License-Identifier: Apache-2.0
"""Run a FastH3 four-step Preview LoRA with the measured FastVideo defaults.

This is the LoRA counterpart of ``basic_fasth3.py``. Both routes share the
same performance profile: four DiT forwards, regional fullgraph DiT compile,
H3 fusions, compiled and sequence-parallel video VAE decode, replicated DiT,
pinned CPU offload, FA4, and the sm100a tile-64 kernel for VSA adapters.

The FastH3 adapters include low-rank factors plus exact dense deltas. Some also
provide the VSA compression gate that is absent from the base checkpoint. Pass
the adapter at construction so all three payload types receive the same
``--lora-strength``. The attention backend is inferred from that payload unless
``--vsa`` or ``--no-vsa`` is specified explicitly.
"""

from __future__ import annotations

import argparse
import math
from collections.abc import Sequence

try:
    from . import basic_fasth3
except ImportError:
    # Direct script execution puts this directory, rather than ``examples``, on
    # sys.path. Keep both ``python file.py`` and module/importlib use working.
    import basic_fasth3  # type: ignore[no-redef]

BASE_MODEL = "MiniMaxAI/MiniMax-H3"


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    parser = basic_fasth3.build_parser(description=__doc__)
    parser.set_defaults(model_path=BASE_MODEL, output="outputs/fasth3_lora_preview")
    parser.add_argument(
        "--lora-path",
        required=True,
        help="FastH3 adapter safetensors file or local adapter directory",
    )
    parser.add_argument(
        "--lora-strength",
        type=float,
        default=1.0,
        help="adapter strength; 0 zeros its weights but keeps its backend, and 1 applies its published scale",
    )
    parser.add_argument(
        "--vsa",
        action=argparse.BooleanOptionalAction,
        default=None,
        help="select VSA explicitly; by default it is inferred from the adapter's compression-gate payload",
    )
    args = basic_fasth3.validate_args(parser, parser.parse_args(argv))
    if not math.isfinite(args.lora_strength):
        parser.error("--lora-strength must be finite")
    return _resolve_attention_backend(parser, args)


def _resolve_attention_backend(parser: argparse.ArgumentParser, args: argparse.Namespace) -> argparse.Namespace:
    # Header-only inspection keeps the payload on disk. A replacement compression
    # gate is an unambiguous VSA requirement; adapters without one default to dense.
    from fastvideo.models.loader.lora_patch import DenseLoRAPatch

    patch = DenseLoRAPatch.from_adapter(args.lora_path, strength=args.lora_strength)
    needs_vsa = bool(patch and any("gate_compress" in name for name in patch.replacement_parameters))
    if args.vsa is None:
        args.vsa = needs_vsa
    elif needs_vsa and not args.vsa:
        parser.error(f"{args.lora_path} provides to_gate_compress and must run with VSA; drop --no-vsa")
    return args


def main() -> None:
    args = parse_args()
    print(f"FastH3 adapter: {args.lora_path}")
    print(f"LoRA strength: {args.lora_strength:g}")
    print(f"Attention: {'VSA-H3' if args.vsa else 'dense FA4'}")
    basic_fasth3.run(args)


if __name__ == "__main__":
    main()

# SPDX-License-Identifier: Apache-2.0
"""Generate 5-second FastH3 videos with a supported Preview adapter."""

from __future__ import annotations

import argparse
import os
import statistics
import sys
import time
from pathlib import Path

from huggingface_hub import hf_hub_download

FASTVIDEO_ROOT = Path(__file__).resolve().parents[3]
sys.path.insert(0, str(FASTVIDEO_ROOT))

from fastvideo import VideoGenerator  # noqa: E402
from fastvideo.api import (  # noqa: E402
    CompileConfig,
    ComponentConfig,
    EngineConfig,
    GenerationRequest,
    GeneratorConfig,
    OffloadConfig,
    OutputConfig,
    ParallelismConfig,
    PipelineSelection,
    SamplingConfig,
)

# FastVideo exposes these three backend switches only through environment variables.
os.environ.update({
    "FASTVIDEO_FA4": "1",
    "FASTVIDEO_MINIMAX_H3_FUSIONS": "all",
    "FASTVIDEO_VSA_SM100A": "0",
})
os.environ.pop("FASTVIDEO_INFERENCE_TORCH_COMPILE", None)

VARIANT_BACKENDS = {
    "dense-datafree": "FLASH_ATTN",
    "vsa-datafree": "VIDEO_SPARSE_ATTN_H3",
    "vsa-synthetic-step1300": "VIDEO_SPARSE_ATTN_H3",
    "vsa-synthetic-step1900": "VIDEO_SPARSE_ATTN_H3",
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("variant", choices=VARIANT_BACKENDS)
    parser.add_argument("--prompt", required=True)
    return parser.parse_args()


def main() -> None:
    """Run one compile warmup and three measured generations with one fixed recipe."""
    args = parse_args()

    attention_backend = VARIANT_BACKENDS[args.variant]
    experimental = {
        "attention_backend": attention_backend,
        "inference_torch_compile": attention_backend == "FLASH_ATTN",
        "vae_parallel_decode": True,
        "vae_parallel_decode_strategy": "gather",
    }
    if attention_backend == "VIDEO_SPARSE_ATTN_H3":
        experimental.update({
            "VSA_sparsity": 0.9,
            "VSA_tile_size": 64,
        })

    adapter_path = hf_hub_download(
        repo_id="FastVideo/FastVideo-FastH3-4-step-Preview-v1-LoRA",
        filename=f"{args.variant}/adapter_model.safetensors",
    )
    output_dir = FASTVIDEO_ROOT / "outputs/fasth3_lora_preview" / args.variant
    output_dir.mkdir(parents=True, exist_ok=True)

    generator = VideoGenerator.from_config(
        GeneratorConfig(
            model_path="MiniMaxAI/MiniMax-H3",
            pipeline=PipelineSelection(
                components=ComponentConfig(lora_path=adapter_path, lora_strength=1.0),
                experimental=experimental,
            ),
            engine=EngineConfig(
                num_gpus=4,
                parallelism=ParallelismConfig(tp_size=1, sp_size=4),
                offload=OffloadConfig(dit=False, dit_layerwise=False),
                compile=CompileConfig(vae_enabled=True),
            ),
        )
    )

    generation_count = 4
    warmup_count = 1
    measured_count = generation_count - warmup_count
    measured_seconds: list[float] = []

    print(f"Variant: {args.variant} ({attention_backend})")
    print(f"Output directory: {output_dir}")
    try:
        for generation_index in range(generation_count):
            measured = generation_index >= warmup_count
            if measured:
                measured_index = generation_index - warmup_count + 1
                label = f"measured {measured_index}/{measured_count}"
                output_path = output_dir / f"fasth3_all_run_{measured_index:02d}.mp4"
            else:
                warmup_index = generation_index + 1
                label = f"warmup {warmup_index}/{warmup_count}"
                output_path = output_dir / f"_fasth3_warmup_{warmup_index:02d}.mp4"

            request = GenerationRequest(
                prompt=args.prompt,
                negative_prompt="",
                sampling=SamplingConfig(
                    height=768,
                    width=1344,
                    num_frames=345, # <-- Change video length: 5sec: 124; 10sec: 243; 15sec: 345.
                    fps=24,
                    num_inference_steps=5,
                    guidance_scale=1.0,
                    batch_cfg=False,
                    seed=1000,
                ),
                output=OutputConfig(
                    output_path=str(output_path),
                    save_video=True,
                    return_frames=False,
                ),
            )
            started = time.perf_counter()
            result = generator.generate(request)
            elapsed = time.perf_counter() - started
            if measured:
                measured_seconds.append(elapsed)
            suffix = "" if measured else " (excluded from median)"
            print(f"[{label}] {result.video_path or output_path}: {elapsed:.3f}s{suffix}")
    finally:
        generator.shutdown()

    print(f"Median E2E wall time: {statistics.median(measured_seconds):.3f}s")


if __name__ == "__main__":
    main()

# SPDX-License-Identifier: Apache-2.0
"""Profile one FastH3 Preview generation with NVTX after warmup."""

from __future__ import annotations

import argparse
from pathlib import Path

import torch
from huggingface_hub import hf_hub_download

from fastvideo import VideoGenerator
from fastvideo.api import (
    CompileConfig,
    ComponentConfig,
    EngineConfig,
    GenerationRequest,
    GeneratorConfig,
    OffloadConfig,
    OutputConfig,
    ParallelismConfig,
    PipelineSelection,
    QuantizationConfig,
    SamplingConfig,
)
from fastvideo.profiler import nvtx_range

VARIANT_BACKENDS = {
    "dense-datafree": "FLASH_ATTN",
    "vsa-datafree": "VIDEO_SPARSE_ATTN_H3",
    "vsa-synthetic-step1300": "VIDEO_SPARSE_ATTN_H3",
    "vsa-synthetic-step1900": "VIDEO_SPARSE_ATTN_H3",
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("variant", choices=VARIANT_BACKENDS)
    parser.add_argument("--prompt", required=True)
    parser.add_argument("--output", required=True)
    parser.add_argument("--num-frames", type=int, default=345)
    parser.add_argument("--warmup-runs", type=int, default=3)
    return parser.parse_args()


def main() -> None:
    """Warm the selected FastH3 recipe, then profile one identical generation."""
    args = parse_args()
    output_dir = Path(args.output)
    output_dir.mkdir(parents=True, exist_ok=True)

    attention_backend = VARIANT_BACKENDS[args.variant]
    experimental = {
        "attention_backend": attention_backend,
        "inference_torch_compile": attention_backend == "FLASH_ATTN",
        "vae_parallel_decode": True,
        "vae_parallel_decode_strategy": "gather",
    }
    if attention_backend == "VIDEO_SPARSE_ATTN_H3":
        experimental.update({
            "VSA_sparsity": 0.9,
            "VSA_tile_size": 64,
        })

    adapter_path = hf_hub_download(
        repo_id="FastVideo/FastVideo-FastH3-4-step-Preview-v1-LoRA",
        filename=f"{args.variant}/adapter_model.safetensors",
    )
    generator = VideoGenerator.from_config(
        GeneratorConfig(
            model_path="MiniMaxAI/MiniMax-H3",
            pipeline=PipelineSelection(
                components=ComponentConfig(lora_path=adapter_path, lora_strength=1.0),
                experimental=experimental,
            ),
            engine=EngineConfig(
                num_gpus=4,
                parallelism=ParallelismConfig(tp_size=1, sp_size=4),
                offload=OffloadConfig(dit=False, dit_layerwise=False),
                compile=CompileConfig(vae_enabled=True),
                quantization=QuantizationConfig(transformer_quant="MXFP8"),
            ),
        )
    )

    request = GenerationRequest(
        prompt=args.prompt,
        negative_prompt="",
        sampling=SamplingConfig(
            height=768,
            width=1344,
            num_frames=args.num_frames,
            fps=24,
            num_inference_steps=5,
            guidance_scale=1.0,
            batch_cfg=False,
            seed=1000,
        ),
        output=OutputConfig(
            output_path=str(output_dir / "fasth3_profile.mp4"),
            save_video=True,
            return_frames=False,
        ),
    )

    try:
        for warmup_index in range(args.warmup_runs):
            warmup_result = generator.generate(request)
            print(f"Warmup {warmup_index + 1}/{args.warmup_runs}: {warmup_result.video_path}")

        torch.cuda.profiler.start()
        try:
            with nvtx_range("fasth3.profiled_generation"):
                measured_result = generator.generate(request)
        finally:
            torch.cuda.profiler.stop()

        print(f"Profiled output: {measured_result.video_path}")
        if measured_result.generation_time is not None:
            print(f"Generation time: {measured_result.generation_time:.2f}s")
    finally:
        generator.shutdown()


if __name__ == "__main__":
    main()

# SPDX-License-Identifier: Apache-2.0
"""Run full Flux2 text-to-image generation through FastVideo.

User story:
    "I have a local or HF Diffusers-format full Flux2 checkpoint and want a
    minimal text-to-image generation command that uses embedded guidance."
"""
import argparse
import os
from pathlib import Path

from fastvideo import VideoGenerator
from fastvideo.api import (
    ComponentConfig,
    EngineConfig,
    GenerationRequest,
    GeneratorConfig,
    OffloadConfig,
    OutputConfig,
    ParallelismConfig,
    PipelineSelection,
    SamplingConfig,
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run full Flux2 text-to-image generation.")
    parser.add_argument(
        "--model-path",
        default="black-forest-labs/FLUX.2-dev",
        help="HF id or local diffusers-format full Flux2 weights directory.",
    )
    parser.add_argument(
        "--output",
        default="outputs/flux2/flux2.png",
        help="Output PNG path.",
    )
    parser.add_argument(
        "--prompt",
        default="a photo of a banana on a wooden table, studio lighting",
        help="Text prompt.",
    )
    parser.add_argument("--height", type=int, default=1024)
    parser.add_argument("--width", type=int, default=1024)
    parser.add_argument("--steps", type=int, default=50)
    parser.add_argument("--guidance-scale", type=float, default=4.0)
    parser.add_argument("--max-sequence-length", type=int, default=None)
    parser.add_argument("--seed", type=int, default=0)
    parser.add_argument("--num-gpus", type=int, default=1)
    parser.add_argument("--tp-size", type=int, default=None)
    parser.add_argument("--sp-size", type=int, default=None)
    parser.add_argument(
        "--backend",
        default=None,
        help="Set FASTVIDEO_ATTENTION_BACKEND, for example TORCH_SDPA.",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    if args.backend:
        os.environ["FASTVIDEO_ATTENTION_BACKEND"] = args.backend

    output = Path(args.output)
    output.parent.mkdir(parents=True, exist_ok=True)
    tp_size = args.tp_size if args.tp_size is not None else (args.num_gpus if args.num_gpus > 1 else 1)
    sp_size = args.sp_size if args.sp_size is not None else (1 if args.num_gpus > 1 else args.num_gpus)

    generator_config = GeneratorConfig(
        model_path=args.model_path,
        engine=EngineConfig(
            num_gpus=args.num_gpus,
            parallelism=ParallelismConfig(tp_size=tp_size, sp_size=sp_size),
            use_fsdp_inference=False,
            offload=OffloadConfig(
                dit=False,
                vae=True,
                text_encoder=True,
                pin_cpu_memory=False,
            ),
        ),
        pipeline=PipelineSelection(
            workload_type="t2i",
            components=ComponentConfig(override_pipeline_cls_name="Flux2Pipeline"),
        ),
    )

    generator = VideoGenerator.from_config(generator_config)
    try:
        sampling = SamplingConfig(
            height=args.height,
            width=args.width,
            num_frames=1,
            fps=1,
            num_inference_steps=args.steps,
            guidance_scale=args.guidance_scale,
            seed=args.seed,
        )
        extensions = {}
        if args.max_sequence_length is not None:
            extensions["max_sequence_length"] = args.max_sequence_length

        request = GenerationRequest(
            prompt=args.prompt,
            sampling=sampling,
            output=OutputConfig(
                output_path=str(output),
                save_video=True,
                return_frames=False,
            ),
            extensions=extensions,
        )
        generator.generate(request)
    finally:
        generator.shutdown()


if __name__ == "__main__":
    main()