app.py

import os, sys, subprocess

# FramePack's sampling helpers aren't a pip package: fetch them at startup.
if not os.path.isdir("FramePack"):
    subprocess.run(["git", "clone", "--depth", "1",
                    "https://github.com/lllyasviel/FramePack", "FramePack"], check=True)
sys.path.insert(0, os.path.abspath("FramePack"))

import torch, numpy as np, gradio as gr
from diffusers import AutoencoderKLHunyuanVideo
from transformers import (LlamaModel, CLIPTextModel, LlamaTokenizerFast, CLIPTokenizer,
                          SiglipImageProcessor, SiglipVisionModel)
from diffusers_helper.hunyuan import encode_prompt_conds, vae_decode, vae_encode
from diffusers_helper.utils import (save_bcthw_as_mp4, crop_or_pad_yield_mask, soft_append_bcthw,
                                    resize_and_center_crop, generate_timestamp)
from diffusers_helper.models.hunyuan_video_packed import HunyuanVideoTransformer3DModelPacked
from diffusers_helper.pipelines.k_diffusion_hunyuan import sample_hunyuan
from diffusers_helper.clip_vision import hf_clip_vision_encode
from diffusers_helper.bucket_tools import find_nearest_bucket

try:
    import spaces
    GPU = spaces.GPU(duration=120)
except Exception:  # dedicated GPU hardware
    GPU = lambda f: f

MODEL_ID = "Suparious/FP-image-to-video-FLUX.1-HV-bf16"
BASE = "hunyuanvideo-community/HunyuanVideo"
dev = "cuda"

# Everything fits on a single 80GB GPU (ZeroGPU H200 / A100-80GB) without offloading.
text_encoder = LlamaModel.from_pretrained(BASE, subfolder="text_encoder", torch_dtype=torch.float16).to(dev).eval()
text_encoder_2 = CLIPTextModel.from_pretrained(BASE, subfolder="text_encoder_2", torch_dtype=torch.float16).to(dev).eval()
tokenizer = LlamaTokenizerFast.from_pretrained(BASE, subfolder="tokenizer")
tokenizer_2 = CLIPTokenizer.from_pretrained(BASE, subfolder="tokenizer_2")
vae = AutoencoderKLHunyuanVideo.from_pretrained(BASE, subfolder="vae", torch_dtype=torch.float16).to(dev).eval()
vae.enable_slicing(); vae.enable_tiling()
feature_extractor = SiglipImageProcessor.from_pretrained("lllyasviel/flux_redux_bfl", subfolder="feature_extractor")
image_encoder = SiglipVisionModel.from_pretrained("lllyasviel/flux_redux_bfl", subfolder="image_encoder",
                                                  torch_dtype=torch.float16).to(dev).eval()
transformer = HunyuanVideoTransformer3DModelPacked.from_pretrained(MODEL_ID, torch_dtype=torch.bfloat16).to(dev).eval()
transformer.high_quality_fp32_output_for_inference = True
for m in (text_encoder, text_encoder_2, vae, image_encoder, transformer):
    m.requires_grad_(False)

os.makedirs("outputs", exist_ok=True)


@GPU
@torch.no_grad()
def generate(image, prompt, n_prompt, seconds, steps, seed, gs, use_teacache, crf, progress=gr.Progress()):
    if image is None:
        raise gr.Error("Please upload an image.")
    latent_window_size, cfg, rs = 9, 1.0, 0.0
    sections = max(int(round(seconds * 30 / (latent_window_size * 4))), 1)
    job = generate_timestamp()

    llama_vec, clip_pool = encode_prompt_conds(prompt, text_encoder, text_encoder_2, tokenizer, tokenizer_2)
    llama_vec_n, clip_pool_n = torch.zeros_like(llama_vec), torch.zeros_like(clip_pool)
    llama_vec, llama_mask = crop_or_pad_yield_mask(llama_vec, length=512)
    llama_vec_n, llama_mask_n = crop_or_pad_yield_mask(llama_vec_n, length=512)

    H, W, _ = image.shape
    height, width = find_nearest_bucket(H, W, resolution=640)
    img_np = resize_and_center_crop(image, target_width=width, target_height=height)
    img_pt = (torch.from_numpy(img_np).float() / 127.5 - 1).permute(2, 0, 1)[None, :, None]
    start_latent = vae_encode(img_pt, vae)
    img_hidden = hf_clip_vision_encode(img_np, feature_extractor, image_encoder).last_hidden_state

    dt = transformer.dtype
    llama_vec, llama_vec_n = llama_vec.to(dt), llama_vec_n.to(dt)
    clip_pool, clip_pool_n = clip_pool.to(dt), clip_pool_n.to(dt)
    img_hidden = img_hidden.to(dt)

    rnd = torch.Generator("cpu").manual_seed(int(seed))
    num_frames = latent_window_size * 4 - 3
    history_latents = torch.zeros((1, 16, 1 + 2 + 16, height // 8, width // 8), dtype=torch.float32).cpu()
    history_pixels, total = None, 0
    paddings = list(reversed(range(sections)))
    if sections > 4:
        paddings = [3] + [2] * (sections - 3) + [1, 0]

    out = None
    for i, pad in enumerate(paddings):
        last = pad == 0
        pad_size = pad * latent_window_size
        idx = torch.arange(0, 1 + pad_size + latent_window_size + 1 + 2 + 16).unsqueeze(0)
        (pre_i, _, lat_i, post_i, c2x_i, c4x_i) = idx.split([1, pad_size, latent_window_size, 1, 2, 16], dim=1)
        clean_i = torch.cat([pre_i, post_i], dim=1)
        post, c2x, c4x = history_latents[:, :, :19].split([1, 2, 16], dim=2)
        clean = torch.cat([start_latent.to(history_latents), post], dim=2)

        transformer.initialize_teacache(enable_teacache=use_teacache, num_steps=steps)

        def cb(d, i=i):
            progress((i * steps + d["i"] + 1) / (len(paddings) * steps),
                     desc=f"Section {i + 1}/{len(paddings)} · step {d['i'] + 1}/{steps}")

        gen = sample_hunyuan(
            transformer=transformer, sampler="unipc", width=width, height=height, frames=num_frames,
            real_guidance_scale=cfg, distilled_guidance_scale=gs, guidance_rescale=rs,
            num_inference_steps=steps, generator=rnd,
            prompt_embeds=llama_vec, prompt_embeds_mask=llama_mask, prompt_poolers=clip_pool,
            negative_prompt_embeds=llama_vec_n, negative_prompt_embeds_mask=llama_mask_n,
            negative_prompt_poolers=clip_pool_n,
            device=dev, dtype=torch.bfloat16, image_embeddings=img_hidden,
            latent_indices=lat_i, clean_latents=clean, clean_latent_indices=clean_i,
            clean_latents_2x=c2x, clean_latent_2x_indices=c2x_i,
            clean_latents_4x=c4x, clean_latent_4x_indices=c4x_i, callback=cb)

        if last:
            gen = torch.cat([start_latent.to(gen), gen], dim=2)
        total += int(gen.shape[2])
        history_latents = torch.cat([gen.to(history_latents), history_latents], dim=2)
        real = history_latents[:, :, :total]

        if history_pixels is None:
            history_pixels = vae_decode(real, vae).cpu()
        else:
            n = latent_window_size * 2 + (1 if last else 0)
            cur = vae_decode(real[:, :, :n], vae).cpu()
            history_pixels = soft_append_bcthw(cur, history_pixels, latent_window_size * 4 - 3)

        out = os.path.join("outputs", f"{job}_{total}.mp4")
        save_bcthw_as_mp4(history_pixels, out, fps=30, crf=int(crf))
        yield out
    yield out


with gr.Blocks(title="FramePack Image-to-Video") as demo:
    gr.Markdown("# 🎞️ FramePack Image-to-Video\nUpload an image, describe the motion, and generate a video with "
                f"[`{MODEL_ID}`](https://huggingface.co/{MODEL_ID}). Videos are built section by section "
                "(~1s each) and the preview updates as they finish.")
    with gr.Row():
        with gr.Column():
            image = gr.Image(type="numpy", label="Start image", height=360)
            prompt = gr.Textbox(label="Prompt", placeholder="The girl dances gracefully, with clear movements.")
            seconds = gr.Slider(1, 5, value=1, step=1, label="Video length (seconds)")
            with gr.Accordion("Advanced", open=False):
                n_prompt = gr.Textbox(label="Negative prompt (unused at CFG 1)", value="")
                steps = gr.Slider(10, 50, value=25, step=1, label="Steps")
                gs = gr.Slider(1, 32, value=10, step=0.5, label="Distilled CFG scale")
                seed = gr.Number(value=31337, precision=0, label="Seed")
                teacache = gr.Checkbox(value=True, label="TeaCache (faster, slightly lower quality)")
                crf = gr.Slider(0, 51, value=16, step=1, label="MP4 compression (lower = better)")
            go = gr.Button("Generate", variant="primary")
        with gr.Column():
            video = gr.Video(label="Result", autoplay=True, loop=True)
    go.click(generate, [image, prompt, n_prompt, seconds, steps, seed, gs, teacache, crf], video)

demo.queue().launch()