add app [— main]

app.py

@@ -0,0 +1,652 @@
+"""MiniMax-H3-Turbo-LoRA-Fast — the denoising half of a split MiniMax-H3 deployment.
+
+The 62 GiB Qwen3-VL text encoder runs in a separate conditioner Space; this Space loads the transformer and the
+VAEs, folds the selected Turbo/Style LoRA into the bf16 weights, and serves first/last-frame video generation up
+to 20 seconds with optional audio.
+"""
+
+from __future__ import annotations
+
+import math
+import os
+import shutil
+import subprocess
+import tempfile
+import time
+import traceback
+from functools import cache
+
+import spaces
+import gradio as gr
+
+MODEL_REPO = os.environ.get("H3_MODEL_REPO", "MiniMaxAI/MiniMax-H3")
+CONDITIONER_SPACE = os.environ.get("H3_CONDITIONER", "multimodalart/qwen3vl-conditioner")
+PLACEMENT = os.environ.get("H3_PLACEMENT", "pack").lower()
+ATTENTION = os.environ.get("H3_ATTENTION", "_native_cudnn").lower()
+GPU_SIZE = os.environ.get("H3_GPU_SIZE", "xlarge")
+
+CANVASES = {
+    "960x544 · 16:9 fast": (544, 960),
+    "1024x576 · 16:9 fast": (576, 1024),
+    "1152x640 · 16:9": (640, 1152),
+    "1280x704 · 16:9": (704, 1280),
+    "1344x768 · 16:9 full": (768, 1344),
+    "544x960 · 9:16 fast": (960, 544),
+    "640x1152 · 9:16": (1152, 640),
+    "768x1344 · 9:16 full": (1344, 768),
+    "544x544 · 1:1 fast": (544, 544),
+    "768x768 · 1:1 full": (768, 768),
+    "768x576 · 4:3 fast": (576, 768),
+    "1024x768 · 4:3 full": (768, 1024),
+    "576x768 · 3:4 fast": (768, 576),
+    "768x1024 · 3:4 full": (1024, 768),
+    "1152x512 · 21:9 fast": (512, 1152),
+    "1536x672 · 21:9 full": (672, 1536),
+}
+DEFAULT_CANVAS = "960x544 · 16:9 fast"
+
+FPS, FRAMES_PER_CHUNK, LATENTS_PER_CHUNK = 24, 17, 5
+
+MIN_UI_DURATION = float(os.environ.get("H3_MIN_DURATION", "2"))
+MAX_UI_DURATION = float(os.environ.get("H3_MAX_DURATION", "20"))
+PIPELINE_MAX_DURATION = MAX_UI_DURATION + 1.0
+CHUNK_SECONDS = float(os.environ.get("H3_CHUNK_SECONDS", "14"))
+CONDITIONER_MAX_FRAMES = int(os.environ.get("H3_CONDITIONER_MAX_FRAMES", "345"))
+
+# Allowed GPU-duration choices (seconds) exposed in the Advanced panel.
+GPU_DURATION_CHOICES = (60, 90, 120, 150, 180, 250)
+DEFAULT_GPU_DURATION = 90
+
+OUTPUT_DIR = os.path.join(tempfile.gettempdir(), "h3-outputs")
+
+PIPE = None
+MUTE_PIPE = None
+MANAGER = None
+LOAD_ERROR: str | None = None
+MUTE_ERROR: str | None = None
+LOADED_IN: float | None = None
+LORA_STATUS: str | None = None
+
+
+def snap_frames(seconds: float) -> int:
+    frames = max(1, round(float(seconds) * FPS))
+    while frames % FRAMES_PER_CHUNK != LATENTS_PER_CHUNK:
+        frames += 1
+    return frames
+
+
+def patch_duration_window(minimum: float, maximum: float) -> None:
+    from diffusers.modular_pipelines.minimax_h3.modular_pipeline import MiniMaxH3ModularPipeline
+
+    MiniMaxH3ModularPipeline.min_duration = property(lambda self: float(minimum))
+    MiniMaxH3ModularPipeline.max_duration = property(lambda self: float(maximum))
+
+
+def _apply_lora(transformer) -> str | None:
+    try:
+        import h3_lora
+
+        return h3_lora.apply_lora(transformer)
+    except ImportError:
+        return "⚠️ `h3_lora.py` is missing — running without the Turbo LoRA. Raise the steps or add the file (README)"
+    except Exception as error:
+        traceback.print_exc()
+        return f"⚠️ LoRA loading failed ({type(error).__name__}: {error}) — running without"
+
+
+def _set_lora(transformer, name: str) -> str:
+    try:
+        import h3_lora
+
+        return h3_lora.set_active(transformer, name)
+    except Exception:
+        return "off"
+
+
+def status() -> str:
+    if LOAD_ERROR:
+        return LOAD_ERROR
+    if PIPE is None:
+        return f"Loading `{MODEL_REPO}` (transformer + VAEs, 77.3 GB). See the Space logs."
+    audio_note = "video-only pipeline active" if MUTE_PIPE is not None else f"video-only pipeline not available ({MUTE_ERROR})"
+    return (
+        f"Ready · transformer + VAEs **bfloat16, unquantized** · placement `{PLACEMENT}` · attention `{ATTENTION}` "
+        f"· {LORA_STATUS or 'no LoRA'} · {audio_note} · window {MIN_UI_DURATION:g}–{MAX_UI_DURATION:g}s "
+        f"· loaded in {LOADED_IN:.0f}s · conditioner `{CONDITIONER_SPACE}`"
+    )
+
+
+def load_models() -> str | None:
+    global PIPE, MUTE_PIPE, MANAGER, LOAD_ERROR, MUTE_ERROR, LOADED_IN, LORA_STATUS
+
+    if PIPE is not None or LOAD_ERROR is not None:
+        return LOAD_ERROR
+
+    started = time.time()
+    try:
+        import torch
+        from diffusers import ComponentsManager
+
+        from h3_split_blocks import MiniMaxH3GeneratorBlocks
+
+        patch_duration_window(MIN_UI_DURATION, PIPELINE_MAX_DURATION)
+        manager = ComponentsManager()
+        blocks = MiniMaxH3GeneratorBlocks()
+        print(f"[gen] loading {[c.name for c in blocks.expected_components]} from {MODEL_REPO} ...", flush=True)
+        pipe = blocks.init_pipeline(MODEL_REPO, components_manager=manager, collection="h3")
+        pipe.load_components(dtype=torch.bfloat16)
+
+        LORA_STATUS = _apply_lora(pipe.transformer)
+        if LORA_STATUS:
+            print(f"[gen] {LORA_STATUS}", flush=True)
+
+        pipe.transformer.set_attention_backend(ATTENTION)
+
+        if PLACEMENT == "pack":
+            pipe.transformer.to("cuda")
+        if PLACEMENT == "offload":
+            manager.enable_auto_cpu_offload(device="cuda")
+            _arm_decode_hooks(pipe)
+
+        PIPE, MANAGER = pipe, manager
+        MUTE_PIPE, MUTE_ERROR = _build_mute_pipe(pipe)
+        LOADED_IN = time.time() - started
+        print(f"[gen] ready in {LOADED_IN:.0f}s", flush=True)
+    except Exception as error:
+        traceback.print_exc()
+        LOAD_ERROR = (
+            f"**Loading `{MODEL_REPO}` failed** after {time.time() - started:.0f}s: "
+            f"`{type(error).__name__}: {error}`"
+        )
+    return LOAD_ERROR
+
+
+def _build_mute_pipe(pipe):
+    try:
+        from h3_split_blocks import MiniMaxH3MuteGeneratorBlocks
+
+        blocks = MiniMaxH3MuteGeneratorBlocks()
+        mute = blocks.init_pipeline(MODEL_REPO)
+        shared = {}
+        for spec in blocks.expected_components:
+            component = getattr(pipe, spec.name, None)
+            if component is not None:
+                shared[spec.name] = component
+        mute.update_components(**shared)
+        print(f"[gen] video-only pipeline over {sorted(shared)}", flush=True)
+        return mute, None
+    except Exception as error:
+        traceback.print_exc()
+        return None, f"{type(error).__name__}: {error}"
+
+
+def _arm_decode_hooks(pipe):
+    for name in ("vae", "audio_vae"):
+        module = getattr(pipe, name, None)
+        if module is None:
+            continue
+        inner = module.decode
+
+        def armed(*args, _module=module, _decode=inner, **kwargs):
+            hook = getattr(_module, "_hf_hook", None)
+            if hook is not None:
+                hook.pre_forward(_module)
+            return _decode(*args, **kwargs)
+
+        module.decode = armed
+
+
+@cache
+def conditioner():
+    from gradio_client import Client
+
+    return Client(CONDITIONER_SPACE)
+
+
+def conditioner_client(ip_token):
+    if not ip_token:
+        return conditioner()
+    from gradio_client import Client
+
+    return Client(CONDITIONER_SPACE, headers={"x-ip-token": ip_token})
+
+
+def encode_remote(prompt, image_path, last_image_path, canvas, num_frames, rewrite_prompt=False, ip_token=None):
+    from gradio_client import handle_file
+    from safetensors import safe_open
+
+    requested = int(num_frames)
+    wire_frames = min(requested, CONDITIONER_MAX_FRAMES)
+    path, plan = conditioner_client(ip_token).predict(
+        prompt=prompt,
+        image_path=handle_file(image_path) if image_path else None,
+        last_image_path=handle_file(last_image_path) if last_image_path else None,
+        canvas=canvas,
+        num_frames=wire_frames,
+        rewrite_prompt=bool(rewrite_prompt),
+        api_name="/encode",
+    )
+    with safe_open(path, framework="pt") as handle:
+        metadata = dict(handle.metadata())
+        metadata["num_frames"] = str(requested)
+        return handle.get_tensor("prompt_embeds"), handle.get_tensor("text_token_tags"), metadata, plan
+
+
+_DUR_B, _DUR_C = 1.1745e-4, 3.8396e-9
+_DECODE_BASE, _DECODE_PER_DEFAULT_CANVAS, _DEFAULT_CANVAS_PIXELS = 15, 15, 960 * 544 * 124
+_PLACEMENT_ALLOWANCE, _PAD = 12, 10
+
+
+def estimate_seconds(height, width, num_frames, steps, num_keyframes=0, with_audio=True) -> int:
+    height, width, num_frames, steps = int(height), int(width), int(num_frames), int(steps)
+    latent_frames = (num_frames - LATENTS_PER_CHUNK) // FRAMES_PER_CHUNK * LATENTS_PER_CHUNK + 2
+    patches = (height // 32) * (width // 32)
+    rows = latent_frames * patches + int(num_keyframes) * patches
+    denoise = steps * (_DUR_B * rows + _DUR_C * rows**2)
+    decode = _DECODE_BASE + _DECODE_PER_DEFAULT_CANVAS * (height * width * num_frames) / _DEFAULT_CANVAS_PIXELS
+    if not with_audio:
+        decode *= 0.85
+    return max(60, int(denoise + decode) + _PLACEMENT_ALLOWANCE + _PAD)
+
+
+def get_duration(
+    prompt_embeds, text_token_tags, image, last_image, height, width, num_frames, steps, seed,
+    lora="larry", with_audio=True, gpu_duration=DEFAULT_GPU_DURATION, *a, **k
+):
+    """Duration callback for `@spaces.GPU`.
+
+    Honors an explicit user override (`gpu_duration`) when it is one of the
+    allowed choices; otherwise falls back to the empirical estimate.
+    """
+    if gpu_duration:
+        try:
+            value = int(gpu_duration)
+        except (TypeError, ValueError):
+            value = 0
+        if value in GPU_DURATION_CHOICES:
+            return value
+    keyframes = int(image is not None) + int(last_image is not None)
+    return estimate_seconds(height, width, num_frames, steps, keyframes, with_audio)
+
+
+@spaces.GPU(duration=get_duration, size=GPU_SIZE)
+def _generate(
+    prompt_embeds, text_token_tags, image, last_image, height, width, num_frames, steps, seed,
+    lora="larry", with_audio=True, gpu_duration=DEFAULT_GPU_DURATION,
+):
+    import torch
+
+    active_lora = _set_lora(PIPE.transformer, lora)
+
+    pipe = PIPE if (with_audio or MUTE_PIPE is None) else MUTE_PIPE
+    decoded_audio = with_audio or MUTE_PIPE is None
+
+    if PLACEMENT == "lazy":
+        PIPE.to("cuda")
+    elif PLACEMENT == "pack":
+        PIPE.vae.to("cuda")
+        if decoded_audio:
+            PIPE.audio_vae.to("cuda")
+
+    state = pipe(
+        prompt_embeds=prompt_embeds.to("cuda"),
+        text_token_tags=text_token_tags,
+        image=image,
+        last_image=last_image,
+        height=int(height),
+        width=int(width),
+        num_frames=int(num_frames),
+        num_inference_steps=int(steps) + 1,
+        generator=torch.Generator("cpu").manual_seed(int(seed)),
+    )
+    frames = state.get("videos")[0]
+    if not decoded_audio:
+        return frames, None, None, active_lora
+    return frames, state.get("audio")[0].cpu(), state.get("sampling_rate"), active_lora
+
+
+def _fit_keyframe(image_path, current_canvas):
+    from PIL import Image as _Image
+
+    img = _Image.open(image_path)
+    aspect = img.width / img.height
+    fastest = {}
+    for label, (h, w) in CANVASES.items():
+        r = w / h
+        if r not in fastest or w * h < fastest[r][1][0] * fastest[r][1][1]:
+            fastest[r] = (label, (h, w))
+    ratio = min(fastest, key=lambda r: abs(r - aspect))
+    label, (h, w) = fastest[ratio]
+
+    cur_h, cur_w = CANVASES[current_canvas]
+    if abs(cur_w / cur_h - aspect) <= abs(ratio - aspect):
+        label = current_canvas
+        h, w = cur_h, cur_w
+
+    target = w / h
+    if abs(img.width / img.height - target) > 1e-3:
+        if img.width / img.height > target:
+            new_w = int(img.height * target)
+            left = (img.width - new_w) // 2
+            img = img.crop((left, 0, left + new_w, img.height))
+        else:
+            new_h = int(img.width / target)
+            top = (img.height - new_h) // 2
+            img = img.crop((0, top, img.width, top + new_h))
+        img.save(image_path)
+    return image_path, label
+
+
+LORA_NAMES = ("larry", "lightx", "lightx8", "realism", "joyfox", "H3-Facial-Realism-CloseUp", "off")
+
+
+def _resolve_lora(lora) -> str:
+    if not isinstance(lora, str) or not lora.strip():
+        return "larry"
+    value = lora.strip().lower()
+    lowered = {name.lower(): name for name in LORA_NAMES}
+    if value in lowered:
+        return lowered[value]
+    matches = [name for lower, name in lowered.items() if lower.startswith(value)]
+    if len(matches) == 1:
+        return matches[0]
+    raise gr.Error(f"Unknown LoRA `{lora}`. Allowed: {', '.join(LORA_NAMES)}.")
+
+
+def _as_path(value):
+    if isinstance(value, dict):
+        value = value.get("path") or (value.get("url") or "").removeprefix("/gradio_api/file=")
+    return value or None
+
+
+def keyframe(path):
+    from PIL import Image, ImageOps
+
+    return ImageOps.exif_transpose(Image.open(path)).convert("RGB") if path else None
+
+
+def _render_clip(prompt, first, last, canvas, duration, steps, seed, upsample, lora, with_audio, ip_token, work,
+                 gpu_duration=DEFAULT_GPU_DURATION):
+    from diffusers.utils import encode_video
+
+    num_frames = snap_frames(duration)
+
+    conditioned = time.time()
+    prompt_embeds, text_token_tags, metadata, plan = encode_remote(
+        prompt, first, last, canvas, num_frames, rewrite_prompt=upsample, ip_token=ip_token
+    )
+    condition_seconds = time.time() - conditioned
+    height, width, num_frames = (int(metadata[key]) for key in ("height", "width", "num_frames"))
+    refined = plan.get("refined_prompt") or ""
+
+    started = time.time()
+    frames, audio, sampling_rate, active_lora = _generate(
+        prompt_embeds, text_token_tags, keyframe(first), keyframe(last),
+        height, width, num_frames, steps, seed, lora, with_audio, gpu_duration,
+    )
+    generate_seconds = time.time() - started
+
+    path = os.path.join(work, f"clip-{int(time.time() * 1000)}.mp4")
+    if audio is not None:
+        encode_video(frames, fps=FPS, output_path=path, audio=audio, audio_sample_rate=sampling_rate)
+    else:
+        encode_video(frames, fps=FPS, output_path=path)
+
+    report = (
+        f"{width}x{height} · {num_frames} frames ({num_frames / FPS:.3f} s) · {int(steps)} steps · "
+        f"conditioner {condition_seconds:.0f}s ({plan['num_text_tokens']} tokens"
+        f"{', upsampled' if refined else ''}) · denoise+decode {generate_seconds:.0f}s "
+        f"({generate_seconds / int(steps):.1f} s/step) · LoRA {active_lora} · "
+        f"audio {'yes' if audio is not None else 'no'} · seed {int(seed)} · GPU booking {int(gpu_duration)}s"
+    )
+    print(f"[gen] {report}", flush=True)
+    return path, frames[-1], report, refined
+
+
+def _concat(paths, output_path):
+    listing = os.path.join(os.path.dirname(output_path), "concat.txt")
+    with open(listing, "w", encoding="utf-8") as handle:
+        for path in paths:
+            handle.write(f"file '{os.path.abspath(path)}'\n")
+
+    base = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-f", "concat", "-safe", "0", "-i", listing]
+    proc = subprocess.run(base + ["-c", "copy", output_path], capture_output=True, text=True)
+    if proc.returncode != 0:
+        proc = subprocess.run(
+            base + ["-r", str(FPS), "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac",
+                    "-movflags", "+faststart", output_path],
+            capture_output=True, text=True,
+        )
+        if proc.returncode != 0:
+            raise gr.Error(f"ffmpeg concat failed: {proc.stderr[-1500:]}")
+    return output_path
+
+
+def plan_clips(total: float, chunk: float) -> list[float]:
+    count = max(1, math.ceil(total / chunk))
+    each = total / count
+    if each < MIN_UI_DURATION:
+        count = max(1, int(total // MIN_UI_DURATION))
+        each = total / count
+    return [each] * count
+
+
+CHAIN_SUFFIX = (
+    "\n\nContinue seamlessly from the provided first frame. Preserve the same subject identity, clothing, "
+    "environment, lighting and camera style; do not cut to a new shot."
+)
+
+
+def generate(
+    prompt, first_frame=None, last_frame=None, canvas=DEFAULT_CANVAS, duration=5.0, mode="single",
+    steps=6, seed=42, upsample=False, lora="larry", with_audio=True, ip_token=None, progress=None,
+    gpu_duration=DEFAULT_GPU_DURATION,
+):
+    if LOAD_ERROR:
+        raise gr.Error(LOAD_ERROR.replace("**", "").replace("`", ""))
+    if PIPE is None:
+        raise gr.Error("The denoiser is still loading — check the Space logs and try again in a moment.")
+    if not prompt or not str(prompt).strip():
+        raise gr.Error("MiniMax-H3 always needs a prompt, with or without keyframes.")
+
+    prompt = str(prompt).strip()
+    lora = _resolve_lora(lora)
+    canvas = canvas if canvas in CANVASES else DEFAULT_CANVAS
+    duration = max(MIN_UI_DURATION, min(float(duration), MAX_UI_DURATION))
+    with_audio = bool(with_audio)
+
+    # Normalize GPU-duration override to one of the allowed choices.
+    try:
+        gpu_duration = int(gpu_duration)
+    except (TypeError, ValueError):
+        gpu_duration = DEFAULT_GPU_DURATION
+    if gpu_duration not in GPU_DURATION_CHOICES:
+        gpu_duration = min(GPU_DURATION_CHOICES, key=lambda v: abs(v - gpu_duration))
+
+    first, last = _as_path(first_frame), _as_path(last_frame)
+    if first:
+        first, canvas = _fit_keyframe(first, canvas)
+    if last:
+        last, canvas = _fit_keyframe(last, canvas)
+
+    os.makedirs(OUTPUT_DIR, exist_ok=True)
+    work = tempfile.mkdtemp(prefix="h3-", dir=OUTPUT_DIR)
+    durations = [duration] if mode == "single" else plan_clips(duration, CHUNK_SECONDS)
+
+    try:
+        segments, reports, refined_all = [], [], []
+        carry = first
+        for index, clip_duration in enumerate(durations):
+            is_last = index == len(durations) - 1
+            if progress is not None:
+                progress(index / len(durations), desc=f"Clip {index + 1}/{len(durations)} ({clip_duration:.1f}s)")
+            clip_prompt = prompt if index == 0 else prompt + CHAIN_SUFFIX
+            path, tail, report, refined = _render_clip(
+                clip_prompt, carry, last if is_last else None, canvas, clip_duration,
+                steps, int(seed) + index, upsample and index == 0, lora, with_audio, ip_token, work, gpu_duration,
+            )
+            segments.append(path)
+            reports.append(f"Clip {index + 1}: {report}")
+            if refined:
+                refined_all.append(refined)
+            if not is_last:
+                carry = os.path.join(work, f"handoff-{index}.png")
+                tail.save(carry)
+
+        final = os.path.join(OUTPUT_DIR, f"h3-{int(time.time() * 1000)}.mp4")
+        if len(segments) == 1:
+            shutil.move(segments[0], final)
+        else:
+            _concat(segments, final)
+
+        summary = (
+            f"**{mode}** · {len(durations)} clip(s) · target {duration:.2f}s · "
+            f"actual {sum(snap_frames(d) for d in durations) / FPS:.2f}s · GPU booking {gpu_duration}s/clip\n\n"
+            + "\n\n".join(reports)
+        )
+        if mode == "single" and duration > 15:
+            summary += "\n\n> Beyond 15 s is outside the H3 training distribution. Try chain mode if the video drifts."
+        return final, summary, "\n\n---\n\n".join(refined_all)
+    except gr.Error:
+        raise
+    except Exception as error:
+        message = str(error).lower()
+        if any(hint in message for hint in ("gpu limit", "quota", "could not allocate", "too many", "concurrent")):
+            raise gr.Error(
+                "The shared ZeroGPU pool is currently at its limit — this is not a problem with your inputs. "
+                "Wait a minute and try again."
+            ) from error
+        traceback.print_exc()
+        raise gr.Error(f"Generation failed: {type(error).__name__}: {error}") from error
+    finally:
+        shutil.rmtree(work, ignore_errors=True)
+
+
+def _caller_ip_token(request: gr.Request | None) -> str | None:
+    if request is None:
+        return None
+    headers = getattr(request, "headers", None)
+    return headers.get("x-ip-token") if headers is not None else None
+
+
+def run(prompt, first_frame, last_frame, canvas, duration, mode, steps, seed, upsample, lora, with_audio,
+        gpu_duration, request: gr.Request = None, progress=gr.Progress()):
+    return generate(
+        prompt, first_frame, last_frame, canvas, duration, mode, steps, seed, upsample, lora, with_audio,
+        ip_token=_caller_ip_token(request), progress=progress, gpu_duration=gpu_duration,
+    )
+
+
+def cost_hint(canvas, duration, mode, steps, first_frame, last_frame, with_audio, gpu_duration):
+    try:
+        height, width = CANVASES.get(canvas, CANVASES[DEFAULT_CANVAS])
+        keyframes = int(bool(first_frame)) + int(bool(last_frame))
+        durations = [float(duration)] if mode == "single" else plan_clips(float(duration), CHUNK_SECONDS)
+        estimated = sum(
+            estimate_seconds(height, width, snap_frames(d), int(steps), keyframes, bool(with_audio))
+            for d in durations
+        )
+        try:
+            gd = int(gpu_duration)
+        except (TypeError, ValueError):
+            gd = DEFAULT_GPU_DURATION
+        booking = gd * len(durations) if gd in GPU_DURATION_CHOICES else estimated
+        frames = sum(snap_frames(d) for d in durations)
+        parts = [
+            f"{width}x{height}",
+            f"{frames} frames ({frames / FPS:.2f}s)",
+            f"{len(durations)} pass(es)",
+            f"~{booking}s GPU booking" + (f" ({gd}s/clip override)" if gd in GPU_DURATION_CHOICES else " (auto-estimate)"),
+        ]
+        if not with_audio:
+            parts.append("audio VAE stays on the host")
+        if mode == "single" and float(duration) > 15:
+            parts.append("⚠️ >15s extrapolates")
+        return " · ".join(parts)
+    except Exception as error:
+        return f"(estimate unavailable: {error})"
+
+
+INTRO = f"""# MiniMax-H3-Turbo-LoRA-Fast
+
+Denoising half of a split deployment: the 62 GiB Qwen3-VL text encoder runs in
+[`{CONDITIONER_SPACE}`](https://huggingface.co/spaces/{CONDITIONER_SPACE}); this Space holds only the transformer
+and the VAEs.
+
+* **First/Last Frame** — both optional and usable individually. A last frame alone means "generate *towards* this image".
+* **Length** — `single` is one pass with a raised ceiling (H3 is trained up to 15 s; beyond that is experimental).
+  `chain` cuts the video into clips of at most {CHUNK_SECONDS:g}s and hands the last frame of each clip to the next.
+* **Audio** — when disabled, the audio VAE is never declared, never moved to the GPU and never decoded. The audio
+  rows themselves are always denoised by H3; they are part of the same packed sequence.
+* **GPU Duration** — override the ZeroGPU booking per clip in the Advanced panel
+  (choices: {', '.join(str(v) for v in GPU_DURATION_CHOICES)} s; default {DEFAULT_GPU_DURATION} s).
+"""
+
+
+with gr.Blocks(title="MiniMax-H3-Turbo-LoRA-Fast") as demo:
+    gr.Markdown(INTRO)
+    banner = gr.Markdown(status())
+
+    with gr.Row():
+        with gr.Column(scale=3):
+            prompt = gr.Textbox(
+                label="Prompt",
+                lines=4,
+                value="An extreme close-up shot focuses on a young Asian woman with auburn-tinted hair and silver ear piercings, gazing intensely toward the camera. She subtly blinks and shifts her deep brown eyes with a contemplative expression as soft, warm ambient light highlights fine skin texture, natural blemishes, and delicate facial micro-movements.",
+            )
+            with gr.Row():
+                first_frame = gr.Image(label="First Frame (optional)", type="filepath", height=200)
+                last_frame = gr.Image(label="Last Frame (optional)", type="filepath", height=200)
+            with gr.Row():
+                canvas = gr.Dropdown(label="Canvas", choices=list(CANVASES), value=DEFAULT_CANVAS)
+                mode = gr.Radio(
+                    label="Duration mode",
+                    choices=[("single (1 pass)", "single"), (f"chain (clips ≤ {CHUNK_SECONDS:g}s)", "chain")],
+                    value="single",
+                )
+            duration = gr.Slider(
+                MIN_UI_DURATION, MAX_UI_DURATION, value=5, step=0.5,
+                label="Length (s)", info="Rounded up to the next 17n+5 at 24 fps.",
+            )
+            with gr.Row():
+                with_audio = gr.Checkbox(label="Generate audio", value=True)
+                upsample = gr.Checkbox(label="Prompt upsampling (+15–50s)", value=False)
+            with gr.Accordion("Advanced", open=False):
+                steps = gr.Slider(2, 12, value=6, step=1, label="Steps")
+                lora = gr.Dropdown(label="Turbo/Style-LoRA", choices=list(LORA_NAMES), value="larry")
+                seed = gr.Number(label="Seed (clip i = seed + i)", value=42, precision=0)
+                gpu_duration = gr.Slider(
+                    minimum=GPU_DURATION_CHOICES[0],
+                    maximum=GPU_DURATION_CHOICES[-1],
+                    value=DEFAULT_GPU_DURATION,
+                    step=10,
+                    label="GPU Duration (s per clip)",
+                    info=f"Allowed: {', '.join(str(v) for v in GPU_DURATION_CHOICES)} s · "
+                         f"default {DEFAULT_GPU_DURATION} s · overrides ZeroGPU booking per clip.",
+                )
+            cost = gr.Markdown()
+            run_button = gr.Button("Generate video", variant="primary")
+
+        with gr.Column(scale=2):
+            output_video = gr.Video(label="Result", format="mp4", autoplay=True)
+            report = gr.Markdown(label="Report")
+            refined = gr.Textbox(label="Refined Prompt", lines=4)
+
+    cost_inputs = [canvas, duration, mode, steps, first_frame, last_frame, with_audio, gpu_duration]
+    for control in cost_inputs:
+        control.change(cost_hint, cost_inputs, cost)
+    demo.load(cost_hint, cost_inputs, cost)
+    demo.load(status, None, banner)
+
+    run_button.click(
+        run,
+        [prompt, first_frame, last_frame, canvas, duration, mode, steps, seed, upsample, lora, with_audio, gpu_duration],
+        [output_video, report, refined],
+        api_name="generate",
+    )
+
+
+load_models()
+
+if __name__ == "__main__":
+    demo.queue(default_concurrency_limit=1, max_size=8).launch(show_error=True, allowed_paths=[OUTPUT_DIR])

pre-requirements.txt

@@ -0,0 +1 @@
+pip>=26.1.2

requirements.txt

@@ -0,0 +1,30 @@
+# `diffusers` is installed from the canonical MiniMax-H3 pull request,
+# https://github.com/huggingface/diffusers/pull/14371 ("Minimax h3 follow up (review & refactor)"), pinned to a
+# **commit** rather than to its `minimax-h3-refactor` branch: the PR is a WIP and its head moves, and this Space's
+# blocks subclass its block classes. Re-pin — and re-check `h3_split_blocks.py` against the block names of the new
+# head — whenever the PR updates.
+#
+#   665f578278365ea4a3318cb8c9b66ce6c01204b9 = refs/pull/14371/head at the time of this deploy
+--extra-index-url https://download.pytorch.org/whl/cu130
+diffusers @ git+https://github.com/huggingface/diffusers.git@665f578278365ea4a3318cb8c9b66ce6c01204b9
+torch==2.11.0
+torchvision==0.26.0
+# The Qwen3-VL processor decides the vision patch count, so a different minor changes the conditioning.
+transformers==5.8.0
+accelerate==1.14.0
+# diffusers pins <2.
+huggingface-hub==1.24.0
+# Track Gradio's main branch (https://github.com/gradio-app/gradio) via the prebuilt preview wheel — installing
+# from git does not build the frontend assets, so the app never serves. a9ce60ae03123ed6374b739d862e62dcae2f451e
+# = refs/heads/main at the time of this deploy; re-pin to the new head (and bump `sdk_version` in README.md if the
+# version changed) to move past it.
+gradio
+spaces==0.51.1
+# No `kernels` pin on purpose: the Hub attention backends want `kernels>=0.12.3`, and that version breaks
+# transformers 5.8.0 at import.
+# PyAV muxes the generated soundtrack onto the frames (`encode_video`).
+av
+pillow
+numpy
+requests
+safetensors>=0.8.0