Upload folder using huggingface_hub

README.md

@@ -1,13 +1,60 @@
 ---
 title: BFS Best Face Swap Video
-emoji: 🐨
-colorFrom: green
+emoji: 🎭
+colorFrom: purple
 colorTo: blue
 sdk: gradio
-sdk_version: 6.14.0
-python_version: '3.13'
+sdk_version: "5.0.0"
 app_file: app.py
-pinned: false
+pinned: true
+license: other
+hardware: a100-large
+suggested_hardware: a100-large
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# BFS — Best Face Swap Video
+
+Hugging Face Space for the **V3 persistent-template** head-swap workflow built on [LTX-2.3](https://huggingface.co/Lightricks/LTX-2.3).
+
+## How it works
+
+1. Upload a **guide video** (the body/performance you want to keep)
+2. Upload a **reference face image** (the identity to transfer)
+3. Write or paste a **text prompt** in the structured format below
+4. Click **Generate**
+
+The reference face is composited into a green chroma side-strip that persists across all frames during diffusion, giving the model continuous identity conditioning. The strip is cropped from the final output automatically.
+
+### Prompt format
+
+```
+head_swap:
+FACE: <detailed identity description from the reference image>
+ACTION: <performance description from the guide video — body, clothing, movement only>
+```
+
+**Example:**
+```
+head_swap:
+FACE: Female, fair skin, approximately 25-30 years old, long wavy auburn hair,
+bright green eyes, no facial hair, smooth skin, small silver earrings.
+
+ACTION: A person in a grey hoodie walks toward the camera in an indoor hallway,
+holds a coffee cup in their right hand, glances to the left.
+```
+
+## Models used
+
+| Component | Source |
+|-----------|--------|
+| Transformer (22B, FP8) | [Kijai/LTX2.3_comfy](https://huggingface.co/Kijai/LTX2.3_comfy) |
+| Video VAE | [Kijai/LTX2.3_comfy](https://huggingface.co/Kijai/LTX2.3_comfy) |
+| Text encoder (Gemma 3 12B, FP8) | [Comfy-Org/ltx-2](https://huggingface.co/Comfy-Org/ltx-2) |
+| Motion LoRA | [Kijai/LTX2.3_comfy](https://huggingface.co/Kijai/LTX2.3_comfy) |
+| BFS head-swap LoRA | [Alissonerdx/BFS-Best-Face-Swap-Video](https://huggingface.co/Alissonerdx/BFS-Best-Face-Swap-Video) |
+
+## Ethical use
+
+This tool is intended for **filmmakers, VFX artists, and researchers**.
+You must have explicit consent from any person whose likeness you process.
+See the [full model card](https://huggingface.co/Alissonerdx/BFS-Best-Face-Swap-Video) for the complete terms.

app.py

@@ -0,0 +1,227 @@
+"""
+BFS — Best Face Swap Video  ·  Hugging Face Space
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+
+import gradio as gr
+import numpy as np
+from PIL import Image
+
+from composer import compose_frames, crop_reserved_region
+from video_utils import (
+    compute_target_size,
+    extract_audio,
+    frames_for_duration,
+    load_video_frames,
+    resize_frames,
+    save_video,
+)
+
+# ---------------------------------------------------------------------------
+# GPU decorator — no-op locally, activates the ZeroGPU grant on HF Spaces
+# ---------------------------------------------------------------------------
+try:
+    import spaces
+    GPU = spaces.GPU
+except ImportError:
+    def GPU(fn=None, **kwargs):  # type: ignore
+        return fn if fn is not None else lambda f: f
+
+# ---------------------------------------------------------------------------
+# Global model state (loaded once per worker)
+# ---------------------------------------------------------------------------
+_pipeline_state: dict | None = None
+
+REGION_SIZE = 256
+DEFAULT_FPS = 24.0
+DEFAULT_DURATION = 5.0
+DEFAULT_RESOLUTION = 768
+
+# ---------------------------------------------------------------------------
+# Core processing function
+# ---------------------------------------------------------------------------
+
+@GPU(duration=300)
+def generate(
+    guide_video_path: str,
+    face_image: Image.Image,
+    prompt: str,
+    duration: float,
+    fps: float,
+    lora_strength: float,
+    seed: int,
+    hf_token: str = "",
+    progress: gr.Progress = gr.Progress(track_tqdm=True),
+) -> tuple[str, str]:
+    """
+    Full head-swap pipeline:
+      1. Load + resize guide video frames
+      2. Compose chroma face strip (ReservedRegionFrameComposer)
+      3. Run LTX-2.3 diffusion
+      4. Crop face strip from output
+      5. Mux original audio back in
+
+    Returns (output_video_path, status_message).
+    """
+    global _pipeline_state
+
+    # ---- validate inputs early ----
+    if guide_video_path is None:
+        return "", "Please upload a guide video."
+    if face_image is None:
+        return "", "Please upload a reference face image."
+    if not prompt.strip():
+        return "", "Please enter a text prompt."
+
+    # ---- lazy model load ----
+    if _pipeline_state is None:
+        from pipeline import load_pipeline
+        progress(0, desc="Loading models (first run only — ~5 min)…")
+        _pipeline_state = load_pipeline(
+            token=hf_token.strip() or None,
+            progress_cb=lambda msg: progress(0, desc=msg),
+        )
+
+    progress(0.05, desc="Loading guide video…")
+    frames, source_fps = load_video_frames(guide_video_path)
+    if len(frames) == 0:
+        return "", "Could not read frames from the guide video."
+
+    # ---- extract audio before we do anything else ----
+    audio_tmp = tempfile.mktemp(suffix=".wav")
+    has_audio = extract_audio(guide_video_path, audio_tmp)
+
+    # ---- resize frames ----
+    progress(0.10, desc="Resizing frames…")
+    orig_h, orig_w = frames.shape[1], frames.shape[2]
+    target_w, target_h = compute_target_size(orig_w, orig_h, DEFAULT_RESOLUTION)
+    frames = resize_frames(frames, target_w, target_h)
+
+    # ---- trim / pad to requested duration ----
+    n_frames = frames_for_duration(fps, duration)
+    if len(frames) >= n_frames:
+        frames = frames[:n_frames]
+    else:
+        # loop last frame
+        pad = np.stack([frames[-1]] * (n_frames - len(frames)))
+        frames = np.concatenate([frames, pad], axis=0)
+
+    # ---- compose chroma strip ----
+    progress(0.15, desc="Compositing reference face strip…")
+    composed = compose_frames(
+        frames,
+        face_image,
+        region_position="left",
+        region_size_px=REGION_SIZE,
+    )
+
+    # ---- run diffusion ----
+    progress(0.20, desc="Running LTX-2.3 diffusion…")
+    from pipeline import run_inference
+    generated = run_inference(
+        _pipeline_state,
+        composed,
+        prompt=prompt,
+        fps=fps,
+        lora_strength=lora_strength,
+        seed=int(seed),
+        progress_cb=lambda msg: progress(0.20, desc=msg),
+    )
+
+    # ---- crop face strip from output ----
+    progress(0.90, desc="Cropping reserved region…")
+    cropped = crop_reserved_region(
+        generated,
+        region_position="left",
+        region_size_px=REGION_SIZE,
+        output_size=(target_w, target_h),
+    )
+
+    # ---- save output video with audio ----
+    progress(0.95, desc="Encoding output video…")
+    out_path = tempfile.mktemp(suffix=".mp4")
+    save_video(
+        cropped,
+        fps=fps,
+        output_path=out_path,
+        audio_path=audio_tmp if has_audio else None,
+        audio_duration=duration,
+    )
+
+    progress(1.0, desc="Done.")
+    return out_path, "Generation complete."
+
+
+# ---------------------------------------------------------------------------
+# Gradio UI
+# ---------------------------------------------------------------------------
+
+DESCRIPTION = """
+# BFS — Best Face Swap Video
+
+Swap the identity in any video using the **V3 persistent-template** technique.
+The reference face is placed in a green chroma side-strip that persists across
+all frames, giving the model continuous identity conditioning throughout generation.
+
+**Prompt format:**
+```
+head_swap:
+FACE: Female, fair skin, ~25 years old, long wavy auburn hair, green eyes…
+ACTION: A person in a grey hoodie walks toward the camera indoors…
+```
+"""
+
+EXAMPLES: list[list] = [
+    # [guide_video, face_image, prompt, duration, fps, lora_strength, seed]
+]
+
+with gr.Blocks(title="BFS — Best Face Swap Video") as demo:
+    gr.Markdown(DESCRIPTION)
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            guide_video = gr.Video(label="Guide Video", sources=["upload"])
+            face_image   = gr.Image(label="Reference Face", type="pil", sources=["upload"])
+            prompt       = gr.Textbox(
+                label="Text Prompt",
+                placeholder="head_swap:\nFACE: ...\nACTION: ...",
+                lines=6,
+            )
+
+            with gr.Accordion("Parameters", open=False):
+                duration       = gr.Slider(1, 15, value=DEFAULT_DURATION, step=0.5,  label="Duration (seconds)")
+                fps            = gr.Slider(8, 30,  value=DEFAULT_FPS,      step=1.0,  label="FPS")
+                lora_strength  = gr.Slider(0.5, 1.5, value=1.2,            step=0.05, label="Face Swap Strength")
+                seed           = gr.Number(value=42, label="Seed", precision=0)
+                hf_token       = gr.Textbox(
+                    label="HF Token (optional)",
+                    type="password",
+                    placeholder="hf_…  — only needed if the Space owner's token has no access to a gated model",
+                )
+
+            run_btn = gr.Button("Generate", variant="primary")
+
+        with gr.Column(scale=1):
+            output_video  = gr.Video(label="Result", interactive=False)
+            status_text   = gr.Textbox(label="Status", interactive=False)
+
+    run_btn.click(
+        fn=generate,
+        inputs=[guide_video, face_image, prompt, duration, fps, lora_strength, seed, hf_token],
+        outputs=[output_video, status_text],
+    )
+
+    gr.Markdown("""
+---
+**Hardware:** A100 80 GB GPU required.
+**Model:** [Alissonerdx/BFS-Best-Face-Swap-Video](https://huggingface.co/Alissonerdx/BFS-Best-Face-Swap-Video) · Built on [LTX-2.3](https://huggingface.co/Lightricks/LTX-2.3)
+**License:** For research and professional VFX use only. You must have explicit consent for any likeness you process.
+""")
+
+
+if __name__ == "__main__":
+    demo.launch()

composer.py

@@ -0,0 +1,212 @@
+"""
+ReservedRegionFrameComposer — pure-Python port of the ComfyUI-BFSNodes node.
+
+Adds a chroma-key side strip to every frame, placing the reference face inside
+it so the LTX-2.3 model can use it as a persistent identity template throughout
+generation.  After generation, call crop_reserved_region() to remove the strip.
+"""
+
+import math
+from typing import Literal
+
+import numpy as np
+from PIL import Image
+
+
+# ---------------------------------------------------------------------------
+# Low-level helpers (ported from ComfyUI-BFSNodes/util.py)
+# ---------------------------------------------------------------------------
+
+def _fit_inside(src_w: int, src_h: int, max_w: int, max_h: int) -> tuple[int, int]:
+    if src_w <= 0 or src_h <= 0:
+        return 1, 1
+    scale = min(max_w / src_w, max_h / src_h)
+    return max(1, int(round(src_w * scale))), max(1, int(round(src_h * scale)))
+
+
+def _aligned_offset(container: int, content: int, align: str) -> int:
+    if align == "start":
+        return 0
+    if align == "end":
+        return max(0, container - content)
+    return max(0, (container - content) // 2)
+
+
+def _paste_with_alpha(dst: Image.Image, src: Image.Image, xy: tuple[int, int]) -> None:
+    if src.mode == "RGBA":
+        dst.paste(src, xy, src.split()[-1])
+    else:
+        dst.paste(src, xy)
+
+
+def _add_padding(img: Image.Image, pad: int = 16) -> Image.Image:
+    canvas = Image.new("RGBA", (img.width + pad * 2, img.height + pad * 2), (255, 255, 255, 255))
+    canvas.paste(img.convert("RGBA"), (pad, pad))
+    return canvas
+
+
+# ---------------------------------------------------------------------------
+# Face layout helpers
+# ---------------------------------------------------------------------------
+
+def _layout_faces(
+    faces: list[Image.Image],
+    region_w: int,
+    region_h: int,
+    scale_pct: float,
+    padding: int,
+    gap: int,
+    stack: str,
+    align_main: str,
+    align_cross: str,
+) -> Image.Image:
+    """Composite all faces into a single region_w x region_h RGBA tile."""
+    canvas = Image.new("RGBA", (region_w, region_h), (0, 0, 0, 0))
+    if not faces:
+        return canvas
+
+    n = len(faces)
+    avail_w = region_w - 2 * padding
+    avail_h = region_h - 2 * padding
+
+    # Determine grid shape
+    if stack == "horizontal" or (stack == "auto" and region_w >= region_h):
+        cols, rows = n, 1
+    elif stack == "vertical" or (stack == "auto" and region_h > region_w):
+        cols, rows = 1, n
+    else:  # grid
+        cols = math.ceil(math.sqrt(n))
+        rows = math.ceil(n / cols)
+
+    cell_w = max(1, (avail_w - gap * (cols - 1)) // cols)
+    cell_h = max(1, (avail_h - gap * (rows - 1)) // rows)
+
+    for i, face in enumerate(faces):
+        col, row = i % cols, i // cols
+        fw, fh = _fit_inside(face.width, face.height, int(cell_w * scale_pct / 100), int(cell_h * scale_pct / 100))
+        resized = face.resize((fw, fh), Image.LANCZOS)
+
+        cx = padding + col * (cell_w + gap) + _aligned_offset(cell_w, fw, align_cross)
+        cy = padding + row * (cell_h + gap) + _aligned_offset(cell_h, fh, align_main)
+        _paste_with_alpha(canvas, resized, (cx, cy))
+
+    return canvas
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def compose_frames(
+    frames: np.ndarray,
+    face_image: Image.Image,
+    region_position: Literal["left", "right", "top", "bottom"] = "left",
+    region_size_px: int = 256,
+    face_scale_pct: float = 100.0,
+    face_padding_px: int = 12,
+    face_gap_px: int = 12,
+    face_align_main: str = "center",
+    face_align_cross: str = "center",
+    chroma_rgb: tuple[int, int, int] = (0, 255, 0),
+) -> np.ndarray:
+    """
+    Args:
+        frames:         uint8 numpy array [N, H, W, 3]
+        face_image:     PIL Image — the reference face
+        region_*:       strip geometry and face placement
+        chroma_rgb:     background colour of the reserved strip
+
+    Returns:
+        uint8 numpy array [N, H, W, 3] with the chroma strip composited in.
+        The original content is shrunk to fill the remaining area; total
+        resolution is unchanged (matches the input WxH).
+    """
+    N, H, W, C = frames.shape
+    face_pil = _add_padding(face_image.convert("RGBA"), 16)
+
+    vertical = region_position in ("top", "bottom")
+    if vertical:
+        content_h = H - region_size_px
+        content_w = W
+        region_w, region_h = W, region_size_px
+    else:
+        content_w = W - region_size_px
+        content_h = H
+        region_w, region_h = region_size_px, H
+
+    # Pre-render the face tile (same for every frame)
+    face_tile = _layout_faces(
+        [face_pil],
+        region_w, region_h,
+        face_scale_pct, face_padding_px, face_gap_px,
+        "auto", face_align_main, face_align_cross,
+    )
+    chroma_bg = Image.new("RGB", (region_w, region_h), chroma_rgb)
+    chroma_bg.paste(face_tile, (0, 0), face_tile)  # alpha-composite face onto chroma
+
+    out = np.empty_like(frames)
+
+    for i in range(N):
+        frame_pil = Image.fromarray(frames[i], "RGB")
+
+        # Resize original content to fit the non-reserved area
+        content_pil = frame_pil.resize((content_w, content_h), Image.LANCZOS)
+
+        # Build full-size canvas
+        canvas = Image.new("RGB", (W, H))
+        if region_position == "left":
+            canvas.paste(chroma_bg, (0, 0))
+            canvas.paste(content_pil, (region_size_px, 0))
+        elif region_position == "right":
+            canvas.paste(content_pil, (0, 0))
+            canvas.paste(chroma_bg, (content_w, 0))
+        elif region_position == "top":
+            canvas.paste(chroma_bg, (0, 0))
+            canvas.paste(content_pil, (0, region_size_px))
+        else:  # bottom
+            canvas.paste(content_pil, (0, 0))
+            canvas.paste(chroma_bg, (0, content_h))
+
+        out[i] = np.array(canvas)
+
+    return out
+
+
+def crop_reserved_region(
+    frames: np.ndarray,
+    region_position: Literal["left", "right", "top", "bottom"] = "left",
+    region_size_px: int = 256,
+    output_size: tuple[int, int] | None = None,
+) -> np.ndarray:
+    """
+    Remove the reserved strip from generated frames and resize back to
+    output_size (W, H).  If output_size is None, resize to fill the full
+    original frame dimensions.
+
+    Args:
+        frames:       uint8 [N, H, W, 3]
+        output_size:  (W, H) to resize to after cropping, or None for original size
+
+    Returns:
+        uint8 [N, out_H, out_W, 3]
+    """
+    N, H, W, _ = frames.shape
+    target_w = output_size[0] if output_size else W
+    target_h = output_size[1] if output_size else H
+
+    if region_position == "left":
+        crop = frames[:, :, region_size_px:, :]
+    elif region_position == "right":
+        crop = frames[:, :, :W - region_size_px, :]
+    elif region_position == "top":
+        crop = frames[:, region_size_px:, :, :]
+    else:  # bottom
+        crop = frames[:, :H - region_size_px, :, :]
+
+    if crop.shape[2] == target_w and crop.shape[1] == target_h:
+        return crop
+
+    out = np.empty((N, target_h, target_w, 3), dtype=np.uint8)
+    for i in range(N):
+        out[i] = np.array(Image.fromarray(crop[i]).resize((target_w, target_h), Image.LANCZOS))
+    return out

pipeline.py

@@ -0,0 +1,201 @@
+"""
+LTX-2.3 inference pipeline for BFS head-swap.
+
+Model sources:
+  Transformer : Kijai/LTX2.3_comfy  — diffusion_models/ltx-2.3-22b-distilled_transformer_only_fp8_input_scaled_v3.safetensors
+  Video VAE   : Kijai/LTX2.3_comfy  — vae/LTX23_video_vae_bf16.safetensors
+  Pipeline cfg: Lightricks/LTX-Video (text encoder, scheduler, tokenizer via from_pretrained)
+  LoRA motion : Kijai/LTX2.3_comfy  — loras/ltx-2.3-22b-distilled-lora-dynamic_fro09_avg_rank_105_bf16.safetensors
+  LoRA BFS    : Alissonerdx/BFS-Best-Face-Swap-Video — ltx-2.3/head_swap_v3_rank_adaptive_fro_098.safetensors
+"""
+
+from __future__ import annotations
+
+import gc
+import os
+import tempfile
+from pathlib import Path
+from typing import Callable
+
+import numpy as np
+import torch
+from huggingface_hub import hf_hub_download
+from PIL import Image
+
+# ---------------------------------------------------------------------------
+# Model file specs
+# ---------------------------------------------------------------------------
+
+_HF_CACHE = os.environ.get("HF_HOME", os.path.expanduser("~/.cache/huggingface"))
+
+MODELS = {
+    "transformer": ("Kijai/LTX2.3_comfy", "diffusion_models/ltx-2.3-22b-distilled_transformer_only_fp8_input_scaled_v3.safetensors"),
+    "video_vae":   ("Kijai/LTX2.3_comfy", "vae/LTX23_video_vae_bf16.safetensors"),
+    "lora_motion": ("Kijai/LTX2.3_comfy", "loras/ltx-2.3-22b-distilled-lora-dynamic_fro09_avg_rank_105_bf16.safetensors"),
+    "lora_bfs":    ("Alissonerdx/BFS-Best-Face-Swap-Video", "ltx-2.3/head_swap_v3_rank_adaptive_fro_098.safetensors"),
+}
+
+# Distilled sigmas from the workflow (BasicScheduler bong_tangent, 8 steps)
+DISTILLED_SIGMAS = [1.0, 0.99375, 0.9875, 0.98125, 0.975, 0.909375, 0.725, 0.421875, 0.0]
+
+NEGATIVE_PROMPT = (
+    "pc game, console game, video game, cartoon, childish, ugly, "
+    "artifacts, low resolution, blurry, jagged edges"
+)
+
+
+# ---------------------------------------------------------------------------
+# Model download helpers
+# ---------------------------------------------------------------------------
+
+def _download(key: str, token: str | None = None) -> str:
+    repo, filename = MODELS[key]
+    return hf_hub_download(repo_id=repo, filename=filename, token=token)
+
+
+def _maybe_download_all(
+    token: str | None = None,
+    progress_cb: Callable[[str], None] | None = None,
+) -> dict[str, str]:
+    paths = {}
+    for key in MODELS:
+        if progress_cb:
+            progress_cb(f"Downloading {key}…")
+        paths[key] = _download(key, token=token)
+    return paths
+
+
+# ---------------------------------------------------------------------------
+# Pipeline construction
+# ---------------------------------------------------------------------------
+
+def load_pipeline(
+    device: str = "cuda",
+    token: str | None = None,
+    progress_cb: Callable[[str], None] | None = None,
+) -> dict:
+    """
+    Download (if needed) and load all model components.
+    Returns a dict of loaded objects cached for reuse.
+
+    LoRAs are loaded but NOT fused so that lora_strength can be adjusted
+    per-request in run_inference() via set_adapters().
+    """
+    from diffusers import (
+        AutoencoderKLLTXVideo,
+        LTXImageToVideoPipeline,
+        LTXVideoTransformer3DModel,
+    )
+
+    effective_token = token or os.environ.get("HF_TOKEN")
+    paths = _maybe_download_all(token=effective_token, progress_cb=progress_cb)
+
+    if progress_cb:
+        progress_cb("Loading transformer…")
+    # fp8_e4m3fn weights are loaded into bfloat16 compute precision to maximise
+    # diffusers compatibility; the weight file on disk is fp8-quantised.
+    transformer = LTXVideoTransformer3DModel.from_single_file(
+        paths["transformer"],
+        torch_dtype=torch.bfloat16,
+    ).to(device)
+
+    if progress_cb:
+        progress_cb("Loading video VAE…")
+    video_vae = AutoencoderKLLTXVideo.from_single_file(
+        paths["video_vae"],
+        torch_dtype=torch.bfloat16,
+    ).to(device)
+
+    if progress_cb:
+        progress_cb("Building pipeline…")
+    pipe = LTXImageToVideoPipeline.from_pretrained(
+        "Lightricks/LTX-Video",
+        transformer=transformer,
+        vae=video_vae,
+        torch_dtype=torch.bfloat16,
+        token=effective_token,
+    ).to(device)
+
+    if progress_cb:
+        progress_cb("Loading LoRAs…")
+    pipe.load_lora_weights(paths["lora_motion"], adapter_name="motion")
+    pipe.load_lora_weights(paths["lora_bfs"],    adapter_name="bfs")
+    # Default weights — overridden per-request in run_inference()
+    pipe.set_adapters(["motion", "bfs"], adapter_weights=[1.0, 1.0])
+
+    return {"pipe": pipe, "device": device}
+
+
+# ---------------------------------------------------------------------------
+# Inference
+# ---------------------------------------------------------------------------
+
+def run_inference(
+    state: dict,
+    composed_frames: np.ndarray,
+    prompt: str,
+    fps: float = 24.0,
+    lora_strength: float = 1.0,
+    seed: int = 42,
+    num_inference_steps: int = 8,
+    guidance_scale: float = 1.0,
+    region_size_px: int = 256,
+    progress_cb: Callable[[str], None] | None = None,
+) -> np.ndarray:
+    """
+    Run LTX-2.3 image-to-video inference on the composed frames.
+
+    Args:
+        state:           dict returned by load_pipeline()
+        composed_frames: uint8 [N, H, W, 3] with chroma strip added
+        prompt:          text prompt (head_swap: format)
+        fps:             target frame rate
+        lora_strength:   multiplier applied on top of default LoRA weights
+        seed:            RNG seed
+        region_size_px:  strip width to set guide frame crop
+
+    Returns:
+        uint8 [N, H, W, 3] — generated frames (strip still present, call
+        composer.crop_reserved_region() to remove it)
+    """
+    pipe = state["pipe"]
+    device = state["device"]
+
+    N, H, W, _ = composed_frames.shape
+    first_frame = Image.fromarray(composed_frames[0])
+
+    # Always set adapter weights — LoRAs are not fused, so strength is dynamic.
+    # motion LoRA is kept at 1.0; only the BFS identity LoRA is user-adjustable.
+    pipe.set_adapters(["motion", "bfs"], adapter_weights=[1.0, lora_strength])
+
+    generator = torch.Generator(device=device).manual_seed(seed)
+
+    if progress_cb:
+        progress_cb("Running diffusion…")
+
+    with torch.inference_mode():
+        result = pipe(
+            image=first_frame,
+            prompt=prompt,
+            negative_prompt=NEGATIVE_PROMPT,
+            width=W,
+            height=H,
+            num_frames=N,
+            frame_rate=fps,
+            guidance_scale=guidance_scale,
+            num_inference_steps=num_inference_steps,
+            generator=generator,
+            decode_timestep=0.05,
+            decode_noise_scale=0.025,
+            output_type="pt",
+        )
+
+    # result.frames is [1, N, C, H, W] float in [0,1]
+    frames_pt = result.frames[0]  # [N, C, H, W]
+    frames_np = (frames_pt.permute(0, 2, 3, 1).cpu().float().numpy() * 255).clip(0, 255).astype(np.uint8)
+
+    gc.collect()
+    if torch.cuda.is_available():
+        torch.cuda.empty_cache()
+
+    return frames_np

requirements.txt

@@ -0,0 +1,19 @@
+torch>=2.4.0
+torchvision
+torchaudio
+diffusers>=0.32.0
+transformers>=4.50.0
+accelerate>=1.0.0
+huggingface_hub>=0.26.0
+safetensors>=0.4.0
+gradio>=5.0.0
+spaces
+opencv-python-headless
+pillow>=10.0.0
+numpy>=1.24.0
+ffmpeg-python
+imageio
+imageio-ffmpeg
+decord
+einops
+kornia

video_utils.py

@@ -0,0 +1,185 @@
+"""
+Video I/O utilities: load frames + audio from a video file, save frames back
+to video with audio, and trim audio to match video duration.
+"""
+
+import os
+import subprocess
+import tempfile
+from pathlib import Path
+
+import numpy as np
+from PIL import Image
+
+
+# ---------------------------------------------------------------------------
+# Loading
+# ---------------------------------------------------------------------------
+
+def load_video_frames(
+    path: str,
+    fps: float = 24.0,
+    max_frames: int | None = None,
+) -> tuple[np.ndarray, float]:
+    """
+    Decode video frames to a uint8 numpy array [N, H, W, 3].
+
+    Returns (frames, actual_fps).
+    Uses decord when available; falls back to opencv.
+    """
+    try:
+        import decord
+        decord.bridge.set_bridge("native")
+        vr = decord.VideoReader(path, ctx=decord.cpu(0))
+        actual_fps = float(vr.get_avg_fps())
+        total = len(vr)
+        if max_frames is not None:
+            total = min(total, max_frames)
+        indices = list(range(total))
+        frames = vr.get_batch(indices).asnumpy()  # [N, H, W, 3]
+        return frames, actual_fps
+    except ImportError:
+        pass
+
+    import cv2
+    cap = cv2.VideoCapture(path)
+    actual_fps = cap.get(cv2.CAP_PROP_FPS) or fps
+    frames = []
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+        frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
+        if max_frames is not None and len(frames) >= max_frames:
+            break
+    cap.release()
+    return np.stack(frames, axis=0), actual_fps
+
+
+def extract_audio(video_path: str, output_path: str) -> bool:
+    """Extract audio track from video to a WAV file.  Returns False if no audio."""
+    result = subprocess.run(
+        [
+            "ffprobe", "-v", "quiet", "-select_streams", "a",
+            "-show_entries", "stream=codec_type",
+            "-of", "csv=p=0", video_path,
+        ],
+        capture_output=True, text=True,
+    )
+    if "audio" not in result.stdout:
+        return False
+
+    subprocess.run(
+        [
+            "ffmpeg", "-y", "-i", video_path,
+            "-vn", "-acodec", "pcm_s16le",
+            "-ar", "44100", "-ac", "2", output_path,
+        ],
+        capture_output=True, check=True,
+    )
+    return True
+
+
+# ---------------------------------------------------------------------------
+# Saving
+# ---------------------------------------------------------------------------
+
+def save_video(
+    frames: np.ndarray,
+    fps: float,
+    output_path: str,
+    audio_path: str | None = None,
+    audio_duration: float | None = None,
+    crf: int = 19,
+) -> str:
+    """
+    Encode frames [N, H, W, 3] uint8 to an mp4 file.
+    Optionally mux audio_path (trimmed to audio_duration seconds if provided).
+
+    Returns the path to the written file.
+    """
+    N, H, W, _ = frames.shape
+    tmp_video = output_path + ".noaudio.mp4"
+
+    # Write raw video with ffmpeg via stdin pipe
+    cmd = [
+        "ffmpeg", "-y",
+        "-f", "rawvideo",
+        "-vcodec", "rawvideo",
+        "-s", f"{W}x{H}",
+        "-pix_fmt", "rgb24",
+        "-r", str(fps),
+        "-i", "pipe:0",
+        "-vcodec", "libx264",
+        "-pix_fmt", "yuv420p",
+        "-crf", str(crf),
+        "-preset", "fast",
+        tmp_video,
+    ]
+    proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    for frame in frames:
+        proc.stdin.write(frame.tobytes())
+    proc.stdin.close()
+    proc.wait()
+
+    if audio_path and os.path.exists(audio_path):
+        duration_flag = ["-t", str(audio_duration)] if audio_duration else []
+        subprocess.run(
+            [
+                "ffmpeg", "-y",
+                "-i", tmp_video,
+                "-i", audio_path,
+                *duration_flag,
+                "-c:v", "copy",
+                "-c:a", "aac", "-b:a", "192k",
+                "-shortest",
+                output_path,
+            ],
+            capture_output=True, check=True,
+        )
+        os.remove(tmp_video)
+    else:
+        os.rename(tmp_video, output_path)
+
+    return output_path
+
+
+# ---------------------------------------------------------------------------
+# Resolution helpers
+# ---------------------------------------------------------------------------
+
+def align_to(value: int, multiple: int = 32) -> int:
+    """Round value up to the nearest multiple."""
+    return ((value + multiple - 1) // multiple) * multiple
+
+
+def compute_target_size(
+    orig_w: int,
+    orig_h: int,
+    base_resolution: int = 768,
+    multiple: int = 32,
+) -> tuple[int, int]:
+    """
+    Scale the longer edge to base_resolution, preserving aspect ratio,
+    then align both dimensions to `multiple`.
+    """
+    scale = base_resolution / max(orig_w, orig_h)
+    new_w = align_to(int(orig_w * scale), multiple)
+    new_h = align_to(int(orig_h * scale), multiple)
+    return new_w, new_h
+
+
+def resize_frames(frames: np.ndarray, target_w: int, target_h: int) -> np.ndarray:
+    """Resize [N, H, W, 3] frames to target_w x target_h."""
+    if frames.shape[2] == target_w and frames.shape[1] == target_h:
+        return frames
+    out = np.empty((len(frames), target_h, target_w, 3), dtype=np.uint8)
+    for i, f in enumerate(frames):
+        out[i] = np.array(Image.fromarray(f).resize((target_w, target_h), Image.LANCZOS))
+    return out
+
+
+def frames_for_duration(fps: float, duration: float) -> int:
+    """Return frame count aligned to LTX-2.3 requirements: ((n * fps) // 8) * 8 + 1."""
+    raw = int(duration * fps)
+    return ((raw // 8) * 8) + 1