"""MiniMax-H3 `ref2va`, split deployment — the denoising half. This Space holds the `transformer_ref` partition and the two autoencoders, unquantized bfloat16. Text encoding runs in [`qwen3vl-conditioner`](https://huggingface.co/spaces/multimodalart/qwen3vl-conditioner), which this one calls over the gradio API for every request; `reference_encoder` stays here, next to the autoencoders it runs. """ from __future__ import annotations import json import os import tempfile import time import traceback from functools import cache # Before anything that could initialize CUDA: `import spaces` patches `torch.cuda` so the 72 GiB load can happen at # startup rather than on GPU time. 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") # `lazy` moves all 72.16 GiB onto the card on the first GPU call and leaves it there; `offload` hands placement to # `ComponentsManager.enable_auto_cpu_offload`. Startup placement is not an option here — see `load_models`. PLACEMENT = os.environ.get("H3_PLACEMENT", "lazy").lower() # cuDNN's fused attention is 10-20% faster than the SDPA default on this pool and needs nothing installed. # flash-attention 3 is sm90-only and this card is sm120 (the `zero-a10g` flavour name is legacy). ATTENTION = os.environ.get("H3_ATTENTION", "_native_cudnn").lower() GPU_SIZE = os.environ.get("H3_GPU_SIZE", "xlarge") # Bounds on what `get_duration` may reserve. The pool reserves whatever number it is given, so a flat ceiling for every # request is what makes an account hit "too many ZeroGPU credits allocated to running tasks". MIN_GPU_DURATION = int(os.environ.get("H3_GPU_DURATION_MIN", "120")) MAX_GPU_DURATION = int(os.environ.get("H3_GPU_DURATION_MAX", "1500")) # Must stay identical to the conditioner's table: the *label* goes over the wire, so a canvas that half does not know # is rejected there and surfaces as a failure here. CANVASES = { # 16:9 "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), # 9:16 "544x960 · 9:16 fast": (960, 544), "640x1152 · 9:16": (1152, 640), "768x1344 · 9:16 full": (1344, 768), # 1:1 "544x544 · 1:1 fast": (544, 544), "768x768 · 1:1 full": (768, 768), # 4:3 / 3:4 "768x576 · 4:3 fast": (576, 768), "1024x768 · 4:3 full": (768, 1024), "576x768 · 3:4 fast": (768, 576), "768x1024 · 3:4 full": (1024, 768), # 21:9 "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 # It is the *snapped* frame count the ceiling has to hold for: 15 s is 360 frames, which rounds up to 362, i.e. # 15.083 s, and is refused. 14 is the last whole second that survives the snap. MAX_UI_DURATION = 14 MIN_DURATION = 2 # A reference video shorter than 2 s gives the model almost no motion to read. MIN_REFERENCE_VIDEO, MAX_REFERENCE_VIDEO = 2.0, 15.0 # `MINIMAX_H3_MAX_REFERENCE_IMAGES`. The slots are built up front and revealed one at a time, because a demo asking # for two subjects should not open with nine boxes. MAX_IMAGE_SLOTS, OPEN_IMAGE_SLOTS = 9, 2 # How many LoRA slots the UI offers, and the range each strength slider covers. LORA_SLOTS = 3 LORA_MIN_SCALE, LORA_MAX_SCALE = -2.0, 2.0 # Pre-wired Turbo LoRAs from `larryvrh/MiniMax-H3-Turbo-Lora`: a few-step distillation that renders joint video + # soundtrack in 4–8 steps instead of the usual ~20. Each entry is `(repo reference, recommended steps, blurb)`. The # reference is the `owner/repo/filename.safetensors` form `resolve_lora` accepts, so it downloads on first use and is # cached by `huggingface_hub` thereafter — nothing is bundled in this Space. LORA_PRESETS = { "Turbo v4 (step 600) · 6–8 steps": ( "larryvrh/MiniMax-H3-Turbo-Lora/minimax_h3_turbo_v4_step600.safetensors", 8, "Recommended for most work. Strong static / small-motion, good micro-detail, no over-sharpening. " "Use 6–8 steps; 4 steps can smear on heavy motion.", ), "Turbo v1 (ckpt 850) · 4 steps": ( "larryvrh/MiniMax-H3-Turbo-Lora/minimax_h3_turbo_4step_ckpt850.safetensors", 4, "The friendlier pick for 4-step heavy / fast motion, where v4 can trail. Over-sharpens at higher step counts, " "so keep it at 4 steps.", ), } # The lowest step count the model's own schedulers accept; the Turbo LoRAs are tuned for 4. MIN_STEPS = 4 # Seconds of GPU one request needs, from the packed sequence it is about to denoise: linear in the rows for the # matmuls, quadratic for the attention, against the AoTI block package this Space runs. STEP_LINEAR, STEP_QUADRATIC, SAFETY = 1.1745e-4, 3.8396e-9, 1.3 # The lazy 72.16 GiB `PIPE.to("cuda")` a cold worker pays inside its first GPU call; every request carries it, because # nothing here knows whether the worker it lands on is cold. PLACEMENT_ALLOWANCE = int(os.environ.get("H3_PLACEMENT_ALLOWANCE", "90")) AUDIO_LATENTS_PER_SECOND, AUDIO_CHANNELS = 40, 2 REFERENCE_IMAGE_SHORT_EDGE, CANVAS_MULTIPLE = 2048, 32 DECODE_BASE, DECODE_PER_DEFAULT_CANVAS, DEFAULT_CANVAS_PIXELS = 15, 25, 960 * 544 * 124 # Reading one adapter off local disk and injecting it across the 33B transformer's linear layers. LORA_ALLOWANCE = 12 def snap_frames(seconds: float) -> int: """The frame count MiniMax-H3's video VAE can decode: the next `17 * n + 5` at 24 fps.""" frames = max(1, round(float(seconds) * FPS)) while frames % FRAMES_PER_CHUNK != LATENTS_PER_CHUNK: frames += 1 return frames def lower_duration_floor(seconds: float = MIN_DURATION) -> None: """Let the pipeline generate below its 5 s floor. 56 frames (2.33 s) is fine on the released checkpoint.""" from diffusers.modular_pipelines.minimax_h3.modular_pipeline import MiniMaxH3ModularPipeline MiniMaxH3ModularPipeline.min_duration = property(lambda self: float(seconds)) def video_latent_frames(num_frames: int) -> int: """`17 * n + 5` frames become `5 * n + 2` video latents.""" return 5 * ((num_frames - LATENTS_PER_CHUNK) // FRAMES_PER_CHUNK) + 2 def target_rows(height: int, width: int, num_frames: int) -> int: """The generated rows of the packed sequence: video patched `(1, 2, 2)`, plus two audio rows per latent.""" video = video_latent_frames(num_frames) * (height // CANVAS_MULTIPLE) * (width // CANVAS_MULTIPLE) return video + round(num_frames / FPS * AUDIO_LATENTS_PER_SECOND) * AUDIO_CHANNELS def reference_rows(references: list[tuple[str, str]], num_frames: int) -> int: """The rows the reference blocks add, from metadata alone — no decode. An image is resized to a 2048 pixel short edge and encoded as a single frame; a video is put on the canvas *its own* aspect ratio resolves to, truncated to the generated frame count and snapped **down** to a `17 * n + 5` the VAE encodes without padding; a soundtrack contributes two rows per 1/40 s. """ from PIL import Image from diffusers.modular_pipelines.minimax_h3.modular_pipeline import resolve_canvas_size rows = 0 for kind, path in references: if kind == "image": width, height = Image.open(path).size scale = REFERENCE_IMAGE_SHORT_EDGE / min(width, height) resolved = [ max(CANVAS_MULTIPLE, round(edge * scale / CANVAS_MULTIPLE) * CANVAS_MULTIPLE) for edge in (height, width) ] rows += (resolved[0] // CANVAS_MULTIPLE) * (resolved[1] // CANVAS_MULTIPLE) continue video_seconds, audio_seconds = probe(path) if kind == "video" and video_seconds is not None: import av with av.open(path) as container: stream = container.streams.video[0] source_height, source_width = stream.height, stream.width canvas_height, canvas_width = resolve_canvas_size(source_width, source_height, CANVAS_MULTIPLE) frames = min(round(video_seconds * FPS), num_frames) snapped = max(1, (frames - LATENTS_PER_CHUNK) // FRAMES_PER_CHUNK) * FRAMES_PER_CHUNK + LATENTS_PER_CHUNK rows += ( video_latent_frames(snapped) * (canvas_height // CANVAS_MULTIPLE) * (canvas_width // CANVAS_MULTIPLE) ) if audio_seconds is not None: seconds = min(audio_seconds, num_frames / FPS) rows += round(seconds * AUDIO_LATENTS_PER_SECOND) * AUDIO_CHANNELS return rows def get_duration( prompt_embeds, text_token_tags, references, height, width, num_frames, steps, seed, loras=(), **_ ): """Seconds of GPU to reserve for one request. Takes the arguments of the `@spaces.GPU` function it decorates, and tolerates the `gr.Progress` `spaces` injects.""" sequence = int(text_token_tags.shape[0]) + reference_rows(references, num_frames) + target_rows( height, width, num_frames ) denoise = int(steps) * (STEP_LINEAR * sequence + STEP_QUADRATIC * sequence**2) * SAFETY # The two reference encoders ahead of the loop, and the two decoders plus the mux after it. Both scale with what # they are handed rather than with the step count. encode = 5 + reference_rows(references, num_frames) * 1e-3 decode = DECODE_BASE + DECODE_PER_DEFAULT_CANVAS * (height * width * num_frames) / DEFAULT_CANVAS_PIXELS total = PLACEMENT_ALLOWANCE + encode + denoise + decode + 10 + LORA_ALLOWANCE * len(loras or ()) duration = max(MIN_GPU_DURATION, min(MAX_GPU_DURATION, int(total))) print(f"[ref2va] S={sequence} -> reserving {duration}s ({denoise:.0f}s of denoise at {steps} steps)", flush=True) return duration PIPE = None MANAGER = None LOAD_ERROR: str | None = None def load_models() -> str | None: """Load the denoising half at startup, but *not* onto the card. `MiniMaxH3Ref2VAGeneratorBlocks` declares `transformer_ref`, `vae`, `audio_vae`, the two schedulers and `video_processor`, so `load_components` fetches exactly those subfolders — `text_encoder/` and the `transformer/` partition are never touched. Both autoencoders carry `_keep_in_fp32_modules` over every module and stay float32: a bfloat16 audio VAE decodes the soundtrack roughly 20 dB too quiet. Nothing moves onto the card here, for storage rather than memory: `spaces`' startup `torch.pack()` writes every startup-resident CUDA tensor to a second copy on disk, and 77.3 GB of weights plus its pack busts the 150 GB quota (`OSError: [Errno 28] No space left on device` out of `os.posix_fallocate`, mid-pack). """ global PIPE, MANAGER, LOAD_ERROR 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 MiniMaxH3Ref2VAGeneratorBlocks lower_duration_floor() manager = ComponentsManager() blocks = MiniMaxH3Ref2VAGeneratorBlocks() print(f"[ref2va] 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) # Both VAEs first, and explicitly. `set_attention_backend` also sets the registry's *global* backend, which # every processor that was not stamped falls through to, and the float32 audio VAE has no cuDNN kernel: # `RuntimeError: No available kernel. Aborting execution.` in its causal encoder attention, which only a # reference soundtrack ever reaches. pipe.vae.set_attention_backend("native") pipe.audio_vae.set_attention_backend("native") pipe.transformer_ref.set_attention_backend(ATTENTION) # Still startup, still free: an AoTI package carries no weights and opens its archive lazily inside the GPU # worker. Off unless `H3_AOTI=1`. It is the *same* package the `transformer/` partition runs — the two configs # are identical field for field and the compiled code carries no weights of either. import h3_aoti h3_aoti.maybe_load(pipe.transformer_ref) if PLACEMENT == "offload": manager.enable_auto_cpu_offload(device="cuda") _arm_decode_hooks(pipe) PIPE, MANAGER = pipe, manager print(f"[ref2va] ready in {time.time() - started:.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 _arm_decode_hooks(pipe): """Make the offload hooks fire for the two VAEs. `enable_auto_cpu_offload` wraps `forward`, and the reference-encoder and decode blocks call `vae.encode/decode(...)` directly, so the hook never runs and the VAE is still on the host when the latents arrive on the card. """ for name in ("vae", "audio_vae"): module = getattr(pipe, name) for method in ("encode", "decode"): inner = getattr(module, method) def armed(*args, _module=module, _inner=inner, **kwargs): hook = getattr(_module, "_hf_hook", None) if hook is not None: hook.pre_forward(_module) return _inner(*args, **kwargs) setattr(module, method, armed) # ---------------------------------------------------------------------------------------------------------------- # LoRA # ---------------------------------------------------------------------------------------------------------------- # There is no `MiniMaxH3LoraLoaderMixin` in the diffusers integration, so adapters are attached at the *model* level, # through the `PeftAdapterMixin` the transformer carries. That is the whole API this needs: `load_lora_adapter` for # each file and one `set_adapters` call to give them their strengths. Here the model is `transformer_ref`, so the # adapters have to be trained against the `transformer_ref/` partition — a `transformer/` adapter is a different # partition and will not match. def _hub_url_parts(url: str) -> tuple[str, str]: """Split a huggingface.co `blob`/`resolve` URL into its repo id and the file path inside it.""" from urllib.parse import unquote, urlparse parts = unquote(urlparse(url).path).strip("/").split("/") if len(parts) < 5 or parts[2] not in ("resolve", "blob"): raise gr.Error(f"Не разпознавам този адрес като файл в Hugging Face: `{url}`") return "/".join(parts[:2]), "/".join(parts[4:]) def resolve_lora(reference: str) -> str: """Turn what the user typed into a local `.safetensors` path. Accepts a local path, a huggingface.co file URL, `owner/repo/path/to/file.safetensors`, or a bare `owner/repo` whose single `.safetensors` is then picked for them. Runs outside the GPU call, so the download costs no GPU time. """ from huggingface_hub import hf_hub_download, list_repo_files reference = (reference or "").strip() if not reference: return "" if os.path.exists(reference): return reference if reference.startswith(("http://", "https://")): repo_id, filename = _hub_url_parts(reference) return hf_hub_download(repo_id, filename) parts = [part for part in reference.split("/") if part] if len(parts) > 2 and parts[-1].endswith(".safetensors"): return hf_hub_download("/".join(parts[:2]), "/".join(parts[2:])) if len(parts) != 2: raise gr.Error( f"`{reference}` не е нито съществуващ файл, нито `автор/хранилище`, нито адрес към Hugging Face." ) candidates = [name for name in list_repo_files(reference) if name.endswith(".safetensors")] if not candidates: raise gr.Error(f"В `{reference}` няма `.safetensors` файл.") if len(candidates) > 1: preferred = [name for name in candidates if "lora" in name.lower()] if len(preferred) != 1: listed = ", ".join(f"`{name}`" for name in sorted(candidates)[:8]) raise gr.Error(f"`{reference}` съдържа няколко файла. Напиши `{reference}/име.safetensors`. Има: {listed}") candidates = preferred return hf_hub_download(reference, candidates[0]) def _lora_prefix(state_dict) -> str | None: """The prefix `load_lora_adapter` has to strip before the keys match the transformer's own module names.""" key = next(iter(state_dict)) for prefix in ("model.diffusion_model", "diffusion_model", "transformer_ref", "transformer"): if key.startswith(f"{prefix}."): return prefix return None def _is_comfyui_lora(state_dict) -> bool: """Whether a LoRA state dict is in ComfyUI's MiniMax-H3 naming rather than diffusers'. ComfyUI names the block stack `blocks.N.*` and the token refiner `token_refiner.blocks.N.*`; diffusers names them `transformer_blocks.N.*` and `token_refiner.refiner_blocks.N.*`. A key starting with `blocks.` is the tell. """ for key in state_dict: if key.startswith(("blocks.", "token_refiner.blocks.", "final_layer.")): return True return False def _convert_comfyui_lora(state_dict) -> dict: """Remap a ComfyUI-format MiniMax-H3 Turbo LoRA to the diffusers `transformer_ref` module names. The Turbo LoRA ([`larryvrh/MiniMax-H3-Turbo-Lora`](https://huggingface.co/larryvrh/MiniMax-H3-Turbo-Lora)) is trained against the ComfyUI checkpoint, whose module names differ from diffusers' in four ways: * the block stack is `blocks.N` in ComfyUI but `transformer_blocks.N` in diffusers, * the token refiner is `token_refiner.blocks.N` but `token_refiner.refiner_blocks.N`, * the final AdaLN is `final_layer.adaln_proj.linear` but `norm_out.linear`, * attention QKV is one fused `attn.qkv_proj` in ComfyUI but three separate `attn.to_q` / `to_k` / `to_v` in diffusers, and the output projection is `attn.out_proj` but `attn.to_out.0`, * the feed-forward is `mlp.fc1` / `mlp.fc2` but `ff.fc1` / `ff.fc2`. The fused QKV `lora_B` is `[3 * inner_dim, rank]`; splitting it into three along dim 0 gives the three separate `lora_B` matrices, and `lora_A` (which is `[rank, hidden_size]`) is shared verbatim across the three. The metadata says `W_eff = W + lora_B @ lora_A` with alpha = rank, so the scaling is 1.0 and no alpha key is added. """ import torch converted = {} for key, value in state_dict.items(): # `blocks.N.` -> `transformer_blocks.N.` if key.startswith("blocks."): new_key = "transformer_blocks." + key[len("blocks."):] elif key.startswith("token_refiner.blocks."): new_key = "token_refiner.refiner_blocks." + key[len("token_refiner.blocks."):] elif key.startswith("final_layer.adaln_proj.linear."): new_key = "norm_out.linear." + key[len("final_layer.adaln_proj.linear."):] else: converted[key] = value continue # At this point `new_key` is a diffusers block path. Remap the leaf module names. if ".attn.qkv_proj." in new_key: # Fused QKV: split `lora_B` along dim 0 into q/k/v, duplicate `lora_A` verbatim. leaf = new_key.split(".attn.qkv_proj.")[-1] # `lora_A.weight` or `lora_B.weight` stem = new_key[: new_key.index(".attn.qkv_proj.")] if leaf == "lora_A.weight": for proj in ("to_q", "to_k", "to_v"): converted[f"{stem}.attn.{proj}.lora_A.weight"] = value else: # lora_B.weight q_b, k_b, v_b = value.chunk(3, dim=0) converted[f"{stem}.attn.to_q.lora_B.weight"] = q_b converted[f"{stem}.attn.to_k.lora_B.weight"] = k_b converted[f"{stem}.attn.to_v.lora_B.weight"] = v_b elif ".attn.out_proj." in new_key: converted[new_key.replace(".attn.out_proj.", ".attn.to_out.0.")] = value elif ".mlp." in new_key: converted[new_key.replace(".mlp.", ".ff.")] = value else: # `adaln_proj.linear` and the token refiner's attention/ff already match diffusers' names after the # block-prefix rename above. converted[new_key] = value return converted def apply_loras(transformer, loras) -> list[str]: """Attach `loras` (local path, strength) to `transformer` and give each its strength, replacing whatever was on it. Every adapter already on the model is removed first, so a request is never affected by the one before it — which matters when a worker is reused rather than forked fresh. A LoRA in ComfyUI's MiniMax-H3 naming is remapped to diffusers' module names on the fly, so the Turbo LoRA works without a separate conversion step. """ import torch from safetensors.torch import load_file for name in list(getattr(transformer, "peft_config", None) or {}): transformer.delete_adapters(name) names, scales = [], [] for index, (path, scale) in enumerate(loras): state_dict = load_file(path) if _is_comfyui_lora(state_dict): state_dict = _convert_comfyui_lora(state_dict) name = f"lora{index}" transformer.load_lora_adapter(state_dict, adapter_name=name, prefix=_lora_prefix(state_dict)) names.append(name) scales.append(float(scale)) if not names: return [] # PEFT builds the new layers on its own default device/dtype; the base weights are the truth here, under either # placement mode (`offload` keeps them on the host and moves whole modules by hook). base = next(param for key, param in transformer.named_parameters() if ".lora_" not in key) with torch.no_grad(): for key, param in transformer.named_parameters(): if ".lora_" in key and (param.device != base.device or param.dtype != base.dtype): param.data = param.data.to(device=base.device, dtype=base.dtype) transformer.set_adapters(names, scales) return names def collect_loras(lora_fields, progress) -> tuple[list[tuple[str, float]], list[str]]: """Resolve the UI's `reference, strength, reference, strength, ...` into `(local path, strength)` pairs. Resolved before the booking: a download that happens inside `@spaces.GPU` is billed as GPU time. """ loras, labels = [], [] for reference, scale in zip(lora_fields[::2], lora_fields[1::2]): reference = (reference or "").strip() if not reference or abs(float(scale)) < 1e-6: continue progress(0.0, desc=f"Fetching LoRA {reference} ...") loras.append((resolve_lora(reference), float(scale))) labels.append(f"{os.path.basename(reference)} @ {float(scale):g}") if loras and os.environ.get("H3_AOTI") == "1": raise gr.Error("LoRA не може да се приложи върху AoTI компилиран трансформър. Изключи `H3_AOTI`.") return loras, labels @cache def conditioner(): """The other half, over the gradio API. `gradio_client` attaches the caller's own ZeroGPU token per call, so the conditioner's booking is billed to whoever asked for the video.""" from gradio_client import Client return Client(CONDITIONER_SPACE) def probe(path: str) -> tuple[float | None, float | None]: """`(video seconds, audio seconds)` of a media file, either being `None` when the stream is absent.""" import av def seconds(stream, container): if stream.duration is not None and stream.time_base is not None: return float(stream.duration * stream.time_base) return None if container.duration is None else container.duration / av.time_base with av.open(path) as container: video = seconds(container.streams.video[0], container) if container.streams.video else None audio = seconds(container.streams.audio[0], container) if container.streams.audio else None return video, audio def collect(image_paths, audio_path, video_path) -> list[tuple[str, str]]: """The `(kind, path)` references of a request, **in the order the model reads them**. That order numbers the labels of MiniMax-H3's prompt presentation and advances the shared audio/video rotary clock, so the same references in a different order are a different request. """ ordered = [("image", path) for path in image_paths if path] if audio_path: ordered.append(("audio", audio_path)) if video_path: ordered.append(("video", video_path)) return ordered def build_references(references: list[tuple[str, str]]): """The `(kind, path)` references of a request as decoded reference dataclasses, in packed order. `from_file` brings the rates along: a video its own frame rate and soundtrack, a clip its sample rate.""" from diffusers.modular_pipelines.minimax_h3 import ( MiniMaxH3AudioReference, MiniMaxH3ImageReference, MiniMaxH3VideoReference, ) classes = {"image": MiniMaxH3ImageReference, "video": MiniMaxH3VideoReference, "audio": MiniMaxH3AudioReference} return [classes[kind].from_file(path) for kind, path in references] def audio_bearing(references: list[tuple[str, str]]) -> list[tuple[str, float]]: """The references that carry a waveform, and how long it is. A video reference brings its own soundtrack.""" carried = [] for kind, path in references: if kind == "image": continue _, audio_seconds = probe(path) if audio_seconds is not None: carried.append((kind, audio_seconds)) return carried def duration_controls(audio_path, video_path, match: bool): """Show the duration slider unless a single soundtrack can set it, which is when MiniMax-H3 lets it be left out.""" try: carried = audio_bearing(collect([], audio_path, video_path)) except Exception: carried = [] # Exactly one soundtrack, long enough to be a duration MiniMax-H3 generates; anything else is ambiguous or out of # range and the slider stays. derivable = len(carried) == 1 and MIN_DURATION <= snap_frames(carried[0][1]) / FPS <= MAX_REFERENCE_VIDEO return gr.update(visible=derivable), gr.update(visible=not (derivable and match)) def check(prompt: str, references: list[tuple[str, str]]) -> None: """The model's own rules, before anything is uploaded or a card is allocated.""" if not prompt or not prompt.strip(): raise gr.Error("MiniMax-H3 always takes a prompt, references or not.") if not references: raise gr.Error("Add at least one reference — an image or a video for the model to condition on.") if {kind for kind, _ in references} == {"audio"}: raise gr.Error("An audio reference needs an image or a video alongside it; it cannot go on its own.") for kind, path in references: if kind != "video": continue video_seconds, _ = probe(path) if video_seconds is None: raise gr.Error("That reference video has no video stream. Drop it in the audio slot instead.") if not MIN_REFERENCE_VIDEO <= video_seconds <= MAX_REFERENCE_VIDEO: raise gr.Error( f"The reference video is {video_seconds:.1f} s. Use a clip between " f"{MIN_REFERENCE_VIDEO:g} and {MAX_REFERENCE_VIDEO:g} seconds." ) def encode_remote(prompt, references, canvas, num_frames, rewrite_prompt=False): """`/encode_ref2va` on the conditioner Space: a safetensors file holding `prompt_embeds` + `text_token_tags`, with the resolved `height` / `width` / `num_frames` in its metadata, plus the plan. `canvas` is the label. `media` and `kinds` are parallel and ordered, and the references go over because `ref2va`'s presentation puts a vision block in front of the prompt for every image and every merged video frame pair. """ from gradio_client import handle_file from safetensors import safe_open path, plan = conditioner().predict( prompt=prompt, media=[handle_file(path) for _, path in references], kinds=",".join(kind for kind, _ in references), canvas=canvas, num_frames=num_frames, rewrite_prompt=bool(rewrite_prompt), api_name="/encode_ref2va", ) with safe_open(path, framework="pt") as handle: return handle.get_tensor("prompt_embeds"), handle.get_tensor("text_token_tags"), handle.metadata(), plan @spaces.GPU(duration=get_duration, size=GPU_SIZE) def _generate(prompt_embeds, text_token_tags, references, height, width, num_frames, steps, seed, loras=()): """The only thing on GPU time: the two reference encoders, the packed-sequence denoise loop and the decoders. References cross as paths and are decoded here; only the three generated outputs come back. A `@spaces.GPU` argument crosses a process boundary by pickling, a 5 s 1344x768 reference video is 370 MB of expanded frames, and the full `PipelineState` still holds the packed latents and the rotary grid on the card. The adapters are attached here rather than in the caller: `spaces` runs this body in its own worker, so the transformer the request sees is the one that has to carry them. """ import torch if PLACEMENT == "lazy": PIPE.to("cuda") apply_loras(PIPE.transformer_ref, loras or ()) state = PIPE( prompt_embeds=prompt_embeds.to("cuda"), text_token_tags=text_token_tags, references=build_references(references), height=height, width=width, num_frames=num_frames, num_inference_steps=int(steps), generator=torch.Generator("cpu").manual_seed(int(seed)), ) return state.get("videos")[0], state.get("audio")[0].cpu(), state.get("sampling_rate") def generate( # Every parameter after `prompt` has a default, and the newest ones sit at the end, so a positional API client # written against an older signature keeps working. prompt, image_1=None, audio_path=None, video_path=None, canvas=DEFAULT_CANVAS, image_2=None, image_3=None, image_4=None, image_5=None, image_6=None, image_7=None, image_8=None, image_9=None, match=True, duration=5, steps=28, seed=42, upsample=False, *lora_fields, progress=gr.Progress(track_tqdm=True), ): """One request. The LoRA fields are last and default to empty, so a positional API client that predates them is unaffected. `lora_fields` arrives as `reference, strength, reference, strength, ...`.""" if LOAD_ERROR: raise gr.Error(LOAD_ERROR) if PIPE is None: raise gr.Error("The denoiser is still loading.") from diffusers.utils import encode_video images = [image_1, image_2, image_3, image_4, image_5, image_6, image_7, image_8, image_9] references = collect(images, audio_path, video_path) check(prompt, references) # `0` is "leave it to the references" over the wire, which MiniMax-H3 accepts when exactly one of them carries a # soundtrack. The conditioner resolves it either way and this Space pins whatever comes back. derivable = len(audio_bearing(references)) == 1 requested = 0 if (match and derivable) else snap_frames(duration) loras, lora_labels = collect_loras(lora_fields, progress) progress(0.0, desc="Upsampling the prompt ..." if upsample else "Reading the prompt and references ...") conditioned = time.time() try: prompt_embeds, text_token_tags, metadata, plan = encode_remote( prompt, references, canvas, requested, rewrite_prompt=upsample ) except gr.Error: raise except Exception as error: # gradio only puts the exception *type* on the wire, so the useful half of a conditioner-side failure is in # that Space's logs. traceback.print_exc() raise gr.Error( f"The conditioner ({CONDITIONER_SPACE}) failed with `{type(error).__name__}: {error}`. " "Its logs carry the full traceback." ) from error 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 "" progress(0.1, desc=f"Generating {num_frames / FPS:.1f} s at {width}x{height} ...") started = time.time() frames, audio, sampling_rate = _generate( prompt_embeds, text_token_tags, references, height, width, num_frames, steps, seed, loras ) generate_seconds = time.time() - started directory = os.path.join(tempfile.gettempdir(), "h3-outputs") os.makedirs(directory, exist_ok=True) path = os.path.join(directory, f"h3-ref2va-{int(time.time() * 1000)}.mp4") encode_video(frames, fps=FPS, output_path=path, audio=audio, audio_sample_rate=sampling_rate) print( f"[ref2va] {[kind for kind, _ in references]} · `{width}x{height}`, {num_frames} frames " f"({num_frames / FPS:.3f} s), {int(steps)} steps · conditioner {condition_seconds:.0f}s " f"({plan['num_text_tokens']} tokens{', upsampled' if refined else ''}) · " f"denoise + decode {generate_seconds:.0f}s " f"({generate_seconds / int(steps):.1f} s/step) · seed {int(seed)}" f"{' · LoRA ' + ', '.join(lora_labels) if lora_labels else ''}", flush=True, ) return path, refined, gr.update(visible=bool(refined)) # ---------------------------------------------------------------------------------------------------------------- # Settings file # ---------------------------------------------------------------------------------------------------------------- # Everything typed rather than uploaded, so a session can be picked up where it was left off. The references # themselves are deliberately left out: gradio hands them over as paths into a per-session temporary directory that # is gone by the next visit, so a saved path would restore as a dead file rather than as the image. SETTINGS_VERSION = 1 SETTINGS_KEYS = ( ["prompt", "upsample", "canvas", "match", "duration", "steps", "seed"] + [f"lora_{slot + 1}" for slot in range(LORA_SLOTS)] + [f"lora_{slot + 1}_scale" for slot in range(LORA_SLOTS)] ) def save_settings(*values): """Write the current controls to a `.json` and reveal it for download.""" payload = {"version": SETTINGS_VERSION, "saved": time.strftime("%Y-%m-%d %H:%M:%S")} payload.update(dict(zip(SETTINGS_KEYS, values))) directory = os.path.join(tempfile.gettempdir(), "h3-settings") os.makedirs(directory, exist_ok=True) path = os.path.join(directory, f"h3-settings-{int(time.time())}.json") with open(path, "w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=False, indent=2, default=str) return gr.update(value=path, visible=True) def load_settings(path): """Restore the controls from a `.json`. A key the file does not carry leaves its control alone, so a settings file written by an older version of this Space still loads.""" if not path: return [gr.update() for _ in SETTINGS_KEYS] try: with open(path, encoding="utf-8") as handle: payload = json.load(handle) except Exception as error: raise gr.Error(f"Файлът с настройки не се чете: `{type(error).__name__}: {error}`") if not isinstance(payload, dict): raise gr.Error("Това не е файл с настройки на този Space.") updates = [] for key in SETTINGS_KEYS: value = payload.get(key) # An unknown canvas label would be rejected by the conditioner, which is the wrong place to find out. if value is None or (key == "canvas" and value not in CANVASES): updates.append(gr.update()) else: updates.append(gr.update(value=value)) return updates def _fill_lora_slots(files, *current): """Drop `.safetensors` files on the uploader and their paths land in the first free slots, so a local adapter needs no typing at all.""" slots = list(current) for path in files or []: for index, value in enumerate(slots): if not (value or "").strip(): slots[index] = path break return [gr.update(value=value) for value in slots] def _add_preset_lora(preset, *current): """Fill the first free LoRA slot with a preset adapter, set its strength to 1.0 and move the steps slider to the preset's recommended count. The Turbo presets are tuned for a specific step range, so the steps slider is moved along with the slot — it is the one output beyond the LoRA fields. A slot already holding the same reference is a no-op, so the button can be pressed twice without duplicating, and a full set of slots is left untouched. """ reference, steps, _ = LORA_PRESETS[preset] slots = list(current[:LORA_SLOTS]) scales = list(current[LORA_SLOTS:]) if reference not in [(value or "").strip() for value in slots]: for index, value in enumerate(slots): if not (value or "").strip(): slots[index] = reference scales[index] = 1.0 break return [*slots, *scales, steps] load_models() INTRO = """# MiniMax-H3 Reference Custom Lora
[ model ]   [ blog ]   [ text / image to video ]
**MiniMax-H3** is a 33B parameter state of the art video generation model that produces video and a fully synchronized soundtrack (ambience, foley, speech). Bring your own subject, voice or camera move as a reference. """ LORA_HELP = """Each slot takes a Hugging Face repo (`owner/repo`), a file inside one (`owner/repo/name.safetensors`), a file URL, or a local path — or just drop the files below. A strength of `0` switches a slot off without clearing it. Adapters have to be trained against the `transformer_ref/` partition. **Turbo LoRA presets** — from [`larryvrh/MiniMax-H3-Turbo-Lora`](https://huggingface.co/larryvrh/MiniMax-H3-Turbo-Lora), a few-step distillation that renders joint video + soundtrack in **4–8 steps** instead of the usual ~20 (a ~5× speedup). Pick one from the dropdown and **Add to a free slot** to fill a slot, set its strength to `1.0` and move the steps slider to the recommended count. Keep strength at `1.0`; only nudge it if a specific clip misbehaves (smear → up, over-sharp → down). """ SETTINGS_HELP = """Saves the prompt, the canvas, the sliders and the LoRA slots — everything typed rather than uploaded. Images, audio and video are not saved: gradio keeps them in a temporary folder that is gone by the next visit, so a saved path would come back as a dead file. """ CSS = """ .main.fillable { max-width: 1250px !important; } .dark .gradio-container { color: var(--body-text-color); } """ with gr.Blocks(title="MiniMax-H3 Reference Custom Lora") as demo: gr.Markdown(INTRO) with gr.Row(): with gr.Column(): prompt = gr.Textbox( label="Prompt", lines=3, value="The character walks through a neon-lit street in the rain, humming to themselves", ) upsample = gr.Checkbox(label="Upsample prompt", value=False) # One tab per modality, in the order the model reads them. A reference left in a tab that is not the open # one is still part of the request. with gr.Tabs(): with gr.Tab("Images"): # One `gr.Row`, so gradio splits the width evenly and wraps at `min_width` rather than leaving a # hole where a hidden slot used to be. with gr.Row(): images = [ gr.Image( label="Subject, style or scene", type="filepath", min_width=180, # Fixed, so a row that wraps to a single slot stays the size of a full one. height=210, visible=index < OPEN_IMAGE_SLOTS, ) for index in range(MAX_IMAGE_SLOTS) ] add_image = gr.Button("+ Add another image", size="sm", variant="secondary") with gr.Tab("Audio"): audio = gr.Audio(label="A voice or a piece of music", type="filepath") with gr.Tab("Video"): video = gr.Video(label="Motion & camera, 2–15 s. Its soundtrack comes along.") run = gr.Button("Generate", variant="primary") with gr.Accordion("LoRA", open=False): gr.Markdown(LORA_HELP) with gr.Row(): lora_preset = gr.Dropdown( label="Turbo LoRA presets", choices=list(LORA_PRESETS), value=list(LORA_PRESETS)[0], scale=4, ) lora_preset_add = gr.Button("Add to a free slot", size="sm", variant="secondary", scale=1) lora_references, lora_scales = [], [] for slot in range(LORA_SLOTS): with gr.Row(): lora_references.append( gr.Textbox(label=f"LoRA {slot + 1}", placeholder="owner/repo", scale=3) ) lora_scales.append( gr.Slider( label="Strength", minimum=LORA_MIN_SCALE, maximum=LORA_MAX_SCALE, step=0.05, value=1.0, scale=2, ) ) lora_upload = gr.File( label="Drop .safetensors here to fill the slots", file_count="multiple", file_types=[".safetensors"], type="filepath", ) with gr.Accordion("Advanced options", open=False): canvas = gr.Dropdown(label="Canvas", choices=list(CANVASES), value=DEFAULT_CANVAS) match = gr.Checkbox(label="Match the reference soundtrack", value=True, visible=False) duration = gr.Slider( label="Duration (s)", minimum=MIN_DURATION, maximum=MAX_UI_DURATION, step=1, value=5 ) steps = gr.Slider(label="Steps", minimum=MIN_STEPS, maximum=40, step=1, value=28) seed = gr.Number(label="Seed", value=42, precision=0) with gr.Accordion("Settings file", open=False): gr.Markdown(SETTINGS_HELP) save = gr.Button("Save settings to .json", size="sm") settings_download = gr.File(label="Your settings", visible=False, interactive=False) settings_upload = gr.File( label="Load a settings .json", file_types=[".json"], type="filepath" ) with gr.Column(): result = gr.Video(label="Video + soundtrack") # An output, so it can be revealed only for a request that asked for a rewrite. with gr.Accordion("Upsampled prompt", open=False, visible=False) as upsampled_panel: upsampled = gr.Textbox(show_label=False, lines=8, interactive=False) open_slots = gr.State(OPEN_IMAGE_SLOTS) def reveal_image_slot(open_count): open_count = min(open_count + 1, MAX_IMAGE_SLOTS) return [ open_count, *[gr.update(visible=index < open_count) for index in range(MAX_IMAGE_SLOTS)], gr.update(visible=open_count < MAX_IMAGE_SLOTS), ] add_image.click(reveal_image_slot, open_slots, [open_slots, *images, add_image], api_name=False) for control in (audio, video, match): control.change( duration_controls, [audio, video, match], [match, duration], show_progress="hidden", api_name=False ) # `reference, strength, reference, strength, ...`, which is how `generate` unpacks them. lora_inputs = [field for pair in zip(lora_references, lora_scales) for field in pair] lora_upload.upload(_fill_lora_slots, [lora_upload, *lora_references], lora_references, api_name=False) lora_preset_add.click( _add_preset_lora, [lora_preset, *lora_references, *lora_scales], [*lora_references, *lora_scales, steps], api_name=False, ) # Same order as `SETTINGS_KEYS`. settings_fields = [prompt, upsample, canvas, match, duration, steps, seed, *lora_references, *lora_scales] save.click(save_settings, settings_fields, settings_download, api_name=False) settings_upload.upload(load_settings, settings_upload, settings_fields, api_name=False) # Same order as `generate`'s signature: the five leading columns first, then the remaining image slots, then the # LoRA fields the `*lora_fields` tail collects. request = [ prompt, images[0], audio, video, canvas, *images[1:], match, duration, steps, seed, upsample, *lora_inputs ] run.click(generate, request, [result, upsampled, upsampled_panel], api_name="generate") if __name__ == "__main__": demo.launch(show_error=True, theme=gr.themes.Citrus(), css=CSS)