"""MiniMax-H3 inpainting, split deployment — the denoising half. Paint over part of a clip and MiniMax-H3 repaints it, keeping everything else. What the mask preserves is handed to the model as *conditioning* — clean content pinned at the `0.999` timestep its keyframe anchors ride at — rather than as the source re-noised to the step's sigma, which is a level the checkpoint has never seen a target row claim. The soundtrack is masked on the same footing and is kept whole by default, which is what makes the model animate to the words already there. This Space holds the `transformer_ref` partition and the two autoencoders. Text encoding runs in [`qwen3vl-conditioner`](https://huggingface.co/spaces/multimodalart/qwen3vl-conditioner), called over the gradio API. """ from __future__ import annotations 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 load can happen at startup # rather than on GPU time. import spaces import gradio as gr import numpy as np from PIL import Image MODEL_REPO = os.environ.get("H3_MODEL_REPO", "multimodalart/MiniMax-H3-Pruned") # The pruned DiT's AoTI package is keyed apart from the released one: its AdaLN projections read 8 inputs instead of # 2688, and `LazyAOTIModel` binds weights by fully qualified name, so loading the released package against these # blocks is a SIGSEGV rather than an error. Set before `h3_aoti` is imported anywhere. os.environ.setdefault("H3_WIDTH", "bf16-pruned") os.environ.setdefault("H3_AOTI_KEY", "bf16-pruned/torch2.11/sm120/dynamic") CONDITIONER_SPACE = os.environ.get("H3_CONDITIONER", "multimodalart/qwen3vl-conditioner") # The same Qwen3-VL, asked a different question: one plain instruction becomes the noun phrase to segment and a # description of the finished shot. Keeping it next to the conditioner means this Space never loads a 62 GiB encoder # just to read a sentence. PLANNER_SPACE = os.environ.get("H3_PLANNER", "linoyts/qwen3vl-inpaint-planner") 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. ATTENTION = os.environ.get("H3_ATTENTION", "_native_cudnn").lower() # `xlarge` is billed at *twice* the credit of the default slice — a request `get_duration` sizes at 222s is # quoted as 444s in a quota refusal — and the measured peak of a full run is only ~56 GiB, so the default # slice looks like a free halving. It is not available: on the default slice this workload dies with # `RuntimeError: NVML_SUCCESS == r INTERNAL ASSERT FAILED` from inside torch, before the first step, with # `size` passed as None *and* with the argument omitted entirely. Nothing here calls NVML. Set # `H3_GPU_SIZE=` to retry it if the pool changes. GPU_SIZE = os.environ.get("H3_GPU_SIZE", "xlarge").strip() or None # A few-step LoRA trained for the reference partition. It carries attention and feed-forward modules only — no AdaLN — # which is why it loads onto the *pruned* DiT, whose AdaLN projections are a different shape from the released one. TURBO_REPO = os.environ.get("H3_TURBO_REPO", "lightx2v/Minimax-h3-Turbo") TURBO_FILE = os.environ.get("H3_TURBO_FILE", "minimax_h3_ref2v_turbo_4step_v0.1_bf16.safetensors") MIN_GPU_DURATION = int(os.environ.get("H3_GPU_DURATION_MIN", "120")) MAX_GPU_DURATION = int(os.environ.get("H3_GPU_DURATION_MAX", "1500")) # The conditioner resolves a canvas from this *label*, so the table has to stay identical to its own. Inpainting # generates on the crop's canvas rather than on any of these — the `ref2va` text presentation reads only the prompt # and the references, never the target canvas — but the label still has to be one the other half knows, so the # request picks whichever of these is closest in aspect ratio. 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), "1024x1024 · 1:1 max": (1024, 1024), "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), } 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 and is # refused. 14 is the last whole second that survives the snap. MAX_UI_DURATION, MIN_DURATION = 14, 2 MIN_REFERENCE_VIDEO, MAX_REFERENCE_VIDEO = 2.0, 15.0 MAX_IMAGE_SLOTS, OPEN_IMAGE_SLOTS = 9, 4 # With the few-step LoRA on, eight steps sits within reach of a 28-step run at roughly a third of the wall clock # (77 s against 221 s at 768x768x124). Six is fine on a roomy mask and starts to lag on a tight one, so the default # buys the margin. DEFAULT_STEPS, DEFAULT_STEPS_FULL = 8, 28 DEFAULT_GROW, DEFAULT_WINDOW = 24, 5 # Quality/memory budget for the crop, as (label, short edge, max pixels). Inpainting is where the community workflows # spend all their effort on resolution, because a repaint is pasted back into a plate that is already sharp. BUDGETS = { "Full · 768 short edge": (768, 768 * 1344), "Balanced · 640": (640, 640 * 1120), "Fast · 512": (512, 512 * 896), } # Full by default. A plate generated below its own size is pasted back through an upsample, and an upsampled repaint # crossfading into a sharp plate is exactly what shows as a band along the mask. DEFAULT_BUDGET = "Full · 768 short edge" # 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 PLACEMENT_ALLOWANCE = int(os.environ.get("H3_PLACEMENT_ALLOWANCE", "90")) AUDIO_LATENTS_PER_SECOND, AUDIO_CHANNELS = 40, 2 CANVAS_MULTIPLE = 32 # `resolve_canvas_size` needs the checkpoint's canvas budget as well as the multiple. These mirror # `transformer.config.canvas_short_edge` / `canvas_max_pixels` for the released model, and are only ever used # to *estimate* a booking — the pipeline itself reads the real values off the loaded components. CANVAS_SHORT_EDGE, CANVAS_MAX_PIXELS = 768, 768 * 1344 # A reference image is encoded at its own short edge and its rows ride through *every* denoising step, so this is one # of the largest levers on cost. The pipeline's own default is 2048, which is 4,096 rows per step for a square photo. # At 768 it is 576, and on the worked example the two are indistinguishable — same subject, same pose, detail 4.75 # against 4.77 — so the smaller one is the default and the larger is a choice. REFERENCE_EDGES = {"Balanced · 768": 768, "Detailed · 1024": 1024, "Maximum · 2048": 2048} DEFAULT_REFERENCE_EDGE = "Balanced · 768" REFERENCE_IMAGE_SHORT_EDGE = 2048 DECODE_BASE, DECODE_PER_DEFAULT_CANVAS, DEFAULT_CANVAS_PIXELS = 15, 25, 960 * 544 * 124 # Inpainting encodes a whole clip through the video VAE before the loop, which `ref2va` alone never does. Same shape # as the decode term and cheaper per pixel, since the encoder is the CNN half and the decoder is the 36-layer ViT. SOURCE_ENCODE_PER_DEFAULT_CANVAS = 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, image_edge: int = 768) -> int: """The rows the reference blocks add, from metadata alone — no decode.""" 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 = min(1.0, image_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, CANVAS_SHORT_EDGE, CANVAS_MAX_PIXELS ) 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(request, steps, seed, **_): """Seconds of GPU to reserve. Takes the arguments of the `@spaces.GPU` function it decorates.""" height, width, num_frames = request["height"], request["width"], request["num_frames"] references = request["references"] edge = request.get("reference_edge", 768) sequence = ( int(request["text_token_tags"].shape[0]) + reference_rows(references, num_frames, edge) + target_rows(height, width, num_frames) ) denoise = int(steps) * (STEP_LINEAR * sequence + STEP_QUADRATIC * sequence**2) * SAFETY encode = 5 + reference_rows(references, num_frames, edge) * 1e-3 canvas_share = (height * width * num_frames) / DEFAULT_CANVAS_PIXELS # The source clip and its soundtrack go through the encoders before the loop; the result comes back through the # decoders after it. source = SOURCE_ENCODE_PER_DEFAULT_CANVAS * canvas_share decode = DECODE_BASE + DECODE_PER_DEFAULT_CANVAS * canvas_share total = PLACEMENT_ALLOWANCE + encode + source + denoise + decode + 10 duration = max(MIN_GPU_DURATION, min(MAX_GPU_DURATION, int(total))) print(f"[inpaint] 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 TURBO_READY = False def check_prompt(prompt: str) -> None: """The NCII guard. Every request here carries real footage and repaints part of it, so unlike the reference Space — where it gates only requests that brought a photo — this runs on all of them. It sits before the conditioner call and the denoise booking, so a refused prompt costs no GPU time on either half.""" import ncii_guard flag = ncii_guard.classify(prompt) if flag["label"] == "ncii": print(f"[guard] prompt refused (ncii {flag['score']:.2f}): {prompt!r}", flush=True) raise gr.Error("This prompt was flagged by a content filter and wasn't run.") def load_models() -> str | None: """Load the denoising half at startup, but *not* onto the card. `MiniMaxH3Ref2VAInpaintGeneratorBlocks` 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 stay float32: a bfloat16 audio VAE decodes roughly 20 dB too quiet. """ 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_inpaint_blocks import MiniMaxH3Ref2VAInpaintGeneratorBlocks lower_duration_floor() manager = ComponentsManager() blocks = MiniMaxH3Ref2VAInpaintGeneratorBlocks() print(f"[inpaint] 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") # The pruned DiT is served as remote code, reached through the `AutoModel` type hint in # `modular_model_index.json`. `load_components` forwards `trust_remote_code` only to components in the # pipeline's own repo, so the VAEs and schedulers — still pointed at `MiniMaxAI/MiniMax-H3` — never see it. pipe.load_components(dtype=torch.bfloat16, trust_remote_code=True) # Both VAEs first, and explicitly. `set_attention_backend` also sets the registry's *global* backend, and the # float32 audio VAE has no cuDNN kernel: `RuntimeError: No available kernel.` in its causal encoder # attention — which inpainting reaches on *every* request, because it always encodes a soundtrack. 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`. import h3_aoti h3_aoti.maybe_load(pipe.transformer_ref) # The LoRA is loaded once and switched on per request, because a request that wants maximum quality turns it # off rather than reloading the checkpoint. global TURBO_READY try: from huggingface_hub import hf_hub_download pipe.load_lora_weights(hf_hub_download(TURBO_REPO, TURBO_FILE), adapter_name="turbo") TURBO_READY = True print(f"[inpaint] turbo LoRA ready ({TURBO_REPO})", flush=True) except Exception as error: print(f"[inpaint] turbo LoRA unavailable ({type(error).__name__}: {error}) — full steps only", flush=True) if PLACEMENT == "offload": manager.enable_auto_cpu_offload(device="cuda") _arm_decode_hooks(pipe) PIPE, MANAGER = pipe, manager print(f"[inpaint] 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 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) def _remote(space: str): """A client for one of the two halves that live in their own Space. Anonymous by default so that ZeroGPU bills the caller and not this Space's owner; see `conditioner` for why that stopped being a problem. Both halves have to be *public* for an anonymous client to reach them at all. """ from gradio_client import Client if os.environ.get("H3_REMOTE_TOKEN", "").strip() in ("1", "true", "yes"): token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") if token: return Client(space, token=token) return Client(space) @cache def conditioner(): """The other half, over the gradio API. Tokenless on purpose. `gradio_client` forwards the caller's own ZeroGPU token per call, so the remote booking bills whoever asked for the video rather than whoever owns this Space — which is the only sane arrangement once the Space is public. The one thing that broke it was gradio's example caching, which runs outside any request and so has no token to forward, and the conditioner answered `Expired ZeroGPU proxy token`; `cache_examples=False` removed that path. `H3_REMOTE_TOKEN=1` puts the owner's token back for a private deployment. """ return _remote(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) if container.duration is not None: return float(container.duration / av.time_base) return None with av.open(path) as container: video = next((s for s in container.streams if s.type == "video"), None) audio = next((s for s in container.streams if s.type == "audio"), None) return ( None if video is None else seconds(video, container), None if audio is None else seconds(audio, container), ) # -------------------------------------------------------------------------------------------------------------- # Source clip # -------------------------------------------------------------------------------------------------------------- def read_clip(path: str, start_seconds: float, seconds: float): """The clip on MiniMax-H3's own 24 fps grid, plus its soundtrack. Resampled here rather than inside the pipeline so every later step — the mask, the crop, the paste — is 1:1 with the frames that come back. Whole frames are held and dropped, never blended, the way the reference implementation resamples a clip. """ import av import torch from h3_inpaint_blocks import resample_frame_indices with av.open(path) as container: stream = container.streams.video[0] source_fps = float(stream.average_rate or FPS) frames = np.stack([frame.to_ndarray(format="rgb24") for frame in container.decode(video=0)]) indices = resample_frame_indices(frames.shape[0], source_fps, FPS) start = int(round(start_seconds * FPS)) indices = indices[start : start + int(round(seconds * FPS))] if indices.size == 0: raise gr.Error("The chosen start time is past the end of the clip.") waveform, sample_rate = None, None with av.open(path) as container: audio_stream = next((s for s in container.streams if s.type == "audio"), None) if audio_stream is not None: chunks = [frame.to_ndarray() for frame in container.decode(audio=0)] if chunks: sample_rate = audio_stream.rate waveform = torch.from_numpy(np.concatenate(chunks, axis=-1)).float() if waveform.shape[0] == 1 and audio_stream.channels == 2: waveform = waveform.reshape(2, -1) offset = int(round(start_seconds * sample_rate)) waveform = waveform[:, offset : offset + int(round(seconds * sample_rate))] return frames[indices], waveform, sample_rate, source_fps, frames.shape[0] def build_mask(editor_value, mask_video_path, num_frames, soften: bool = False): """The per-frame mask, from a painted region or from an uploaded mask clip. A painted mask is one shape held for the whole clip. That is the honest limitation of painting on a still: it works when the box around the subject's whole travel is still worth cropping to, and it does not when the subject crosses the frame. A mask clip — from a segmenter, one frame per source frame — has no such limit. """ if mask_video_path: import av with av.open(mask_video_path) as container: painted = np.stack( [frame.to_ndarray(format="rgb24").mean(-1) for frame in container.decode(video=0)] ).astype(np.float32) / 255.0 if painted.shape[0] < num_frames: painted = np.concatenate([painted, np.repeat(painted[-1:], num_frames - painted.shape[0], axis=0)]) return _harden(painted[:num_frames], soften) if not editor_value: raise gr.Error("Paint over the part of the clip to repaint, or upload a mask clip.") alpha = _painted_alpha(editor_value) if alpha is None or alpha.max() < 0.02: raise gr.Error("Nothing was painted, so there is nothing to repaint.") return _harden(np.repeat((alpha / alpha.max())[None], num_frames, axis=0), soften) def _harden(mask: np.ndarray, soften: bool) -> np.ndarray: """Square the mask off unless a soft edge was asked for. A feathered mask puts its edge rows at intermediate timesteps, so they come back holding a *mixture* of source and repaint — lower contrast than either. Paste that through an upscale and crossfade it into a sharp plate and the result is a visible band along the mask. Squaring the mask off removes the mixture; the paste's own feather is what hides the join. """ return mask if soften else (mask > 0.5).astype(np.float32) def _painted_alpha(editor_value) -> np.ndarray | None: """What the brush covered, as a float mask. Strokes normally arrive as RGBA layers whose alpha *is* the mask. `layers=False` is not a promise that the list is populated, though, so a request that arrives without one falls back to where the composite differs from the background — which is the same region, given the fixed opaque brush. """ layers = editor_value.get("layers") or [] if layers: first = np.asarray(layers[0]) alpha = np.zeros(first.shape[:2], dtype=np.float32) for layer in layers: layer = np.asarray(layer) alpha = np.maximum(alpha, (layer[..., 3] if layer.shape[-1] == 4 else layer.mean(-1)).astype(np.float32)) if alpha.max() >= 0.02: return alpha composite, background = editor_value.get("composite"), editor_value.get("background") if composite is None or background is None: return None composite, background = np.asarray(composite), np.asarray(background) if composite.shape[:2] != background.shape[:2]: return None changed = np.abs(composite[..., :3].astype(np.float32) - background[..., :3].astype(np.float32)).max(-1) return (changed > 8).astype(np.float32) def nearest_canvas_label(height: int, width: int) -> str: """The conditioner's table only takes labels it knows. Only the aspect ratio can matter, so pick the closest.""" target = width / height return min(CANVASES, key=lambda label: abs(CANVASES[label][1] / CANVASES[label][0] - target)) def collect(images, audio_path, video_path) -> list[tuple[str, str]]: """References in the order MiniMax-H3 reads them: images, then video, then audio.""" references = [("image", path) for path in (images or []) if path] if video_path: references.append(("video", video_path)) if audio_path: references.append(("audio", audio_path)) return references def synthesise_reference(frames: np.ndarray, mask: np.ndarray) -> str: """A reference built out of the shot itself, for a request that brought no picture. `ref2va` always conditions on something, so an instruction like "replace the fox with a capybara" — which names no picture — has nothing to hand it. This hands it the plate: the first frame with the masked subject blurred away, so the reference reads as *this forest, empty* rather than as the animal being removed. The prompt does not cite it, and the model fills the hole from the description instead. Verified against a real reference run: a capybara, plate preserved at 1.6/255, 20.6x separation. """ from PIL import ImageFilter frame = frames[0] blurred = np.asarray(Image.fromarray(frame).filter(ImageFilter.GaussianBlur(28))) alpha = mask[0][..., None].astype(np.float32) synth = (frame * (1 - alpha) + blurred * alpha).round().clip(0, 255).astype(np.uint8) path = os.path.join(tempfile.mkdtemp(), "scene-reference.png") Image.fromarray(synth).save(path) return path def conditioning_signature(prompt: str, reference_paths, num_frames: int) -> str: """What the cached embeddings were built from. Anything here changing means they have to be built again.""" import hashlib material = "\u0000".join([(prompt or "").strip(), str(int(num_frames)), *[str(p) for p in reference_paths]]) return hashlib.sha256(material.encode()).hexdigest() @cache def planner(): """The planner Space, over the gradio API. Tokenless for the same reason the conditioner is.""" return _remote(PLANNER_SPACE) def plan_instruction(source_path, instruction, start, duration, references, progress=None): """One instruction into a segmentation target and a generation prompt, grounded in the clip's own first frame. H3 is not an edit model: it renders what its prompt describes, so "replace the fox with a capybara" describes nothing it can render. The planner looks at the frame and writes the finished shot instead, keeping the setting, the light and the camera move that the instruction did not ask to change. """ import tempfile from gradio_client import handle_file from PIL import Image if not source_path: raise gr.Error("Upload the clip you want to repaint first.") if not (instruction or "").strip(): raise gr.Error("Say what should change — for example 'replace the fox with a capybara'.") if progress is not None: progress(0.0, desc="Reading the shot ...") frames = _clip_window(source_path, start, duration) path = os.path.join(tempfile.mkdtemp(), "frame.png") Image.fromarray(frames[0]).save(path) try: plan = planner().predict( instruction=instruction.strip(), frame_path=handle_file(path), has_reference=bool(gallery_paths(references)), api_name="/plan_inpaint", ) except gr.Error: raise except Exception as error: traceback.print_exc() raise gr.Error( f"The planner ({PLANNER_SPACE}) failed with `{type(error).__name__}: {error}`. Its logs carry the full " "traceback. You can still fill the prompt in by hand and mask the clip yourself." ) from error target = (plan or {}).get("target") or "" prompt = (plan or {}).get("prompt") or "" if not target: raise gr.Error( f"The planner could not tell what to segment from {instruction.strip()!r}. Name the thing directly, or " "click it on the 'Point at it' tab." ) print(f"[inpaint] planned {instruction.strip()!r} -> segment {target!r}", flush=True) return target, prompt def _finish_mask(source_path, prompt, frames, clip, base, view, references, note, progress=None): """The half that is the same however the mask was made: settle the reference, encode, and cache the result. Both masking paths end here so that Repaint behaves identically whether the region was named in words or clicked on — including having the prompt already encoded. """ num_frames = frames.shape[0] supplied = gallery_paths(references) mask = _read_mask_clip(clip, num_frames) reference_paths = supplied or [synthesise_reference(frames, mask)] if progress is not None: progress(0.9, desc="Reading the prompt ...") prompt_embeds, text_token_tags, metadata, _ = encode_remote( prompt, collect(reference_paths, None, None), nearest_canvas_label(frames.shape[1], frames.shape[2]), num_frames, ) conditioning = { "signature": conditioning_signature(prompt, reference_paths, int(metadata["num_frames"])), "prompt_embeds": prompt_embeds, "text_token_tags": text_token_tags, "num_frames": int(metadata["num_frames"]), "reference_paths": reference_paths, "synthetic": not supplied, } if not supplied: note += " No reference was attached, so the shot itself is standing in as one." return prompt, clip, base, view, note, conditioning def plan_signature(source_path, instruction, start, duration, references) -> str: """What a plan was made from. The mask does not depend on `grow` — that is re-applied afterwards — so growing the mask must not invalidate it, and neither must anything downstream of the prompt.""" import hashlib material = "\u0000".join([ str(source_path), (instruction or "").strip(), f"{float(start):.2f}", f"{float(duration):.2f}", *[str(path) for path in gallery_paths(references)], ]) return hashlib.sha256(material.encode()).hexdigest() def prepare_if_needed(source_path, prompt, references, mask_video, instruction, start, duration, grow, conditioning, mask_base, cache=None, progress=gr.Progress()): """Fill in whatever Inpaint needs and does not already have, then hand off to it. This is a separate event from `generate` for one hard reason: `generate` runs under `gr.Progress(track_tqdm=True)` so the denoise loop can report itself, and gradio's patched tqdm cannot be used to iterate SAM 3's propagation — its `__next__` reads `self.iterables[-1]`, which is empty inside a ZeroGPU worker, and raises `IndexError` regardless of whether the bar is disabled. Segmentation therefore has to happen in a handler that does *not* track tqdm, which is what this is. """ # Every slot is given a real value rather than a bare `gr.update()`. Two of the outputs are `gr.State`, and a # State handed an update *dict* stores the dict — so passing through has to mean passing the value through. if (prompt or "").strip(): return gr.update(), mask_video, mask_base, gr.update(), gr.update(), conditioning, cache if not (instruction or "").strip(): raise gr.Error("Say what should change first — one line, like 'replace the fox with a capybara'.") if mask_video: # The region is already settled by an upload or by clicking, so only the prompt half of the plan is wanted. _, written = plan_instruction(source_path, instruction, start, duration, references, progress) return (gr.update(value=written, visible=True), mask_video, mask_base, gr.update(), f"Using the mask you gave. Prompt: _{written}_", conditioning, cache) written, clip, base, view, note, cached, fresh = plan_and_mask( source_path, instruction, start, duration, references, grow, cache, progress ) return gr.update(value=written, visible=True), clip, base, view, note, cached, fresh def plan_and_mask(source_path, instruction, start, duration, references, grow, cache=None, progress=gr.Progress()): """Plan, then segment what the plan named. Leaves the prompt and the mask on screen to be looked at. Returns the note as rendered markdown rather than the bare target, so this can be wired straight to the UI: a lambda in between would hand gradio an anonymous signature and the endpoint would lose its parameter names. """ wanted = plan_signature(source_path, instruction, start, duration, references) if isinstance(cache, dict) and cache.get("signature") == wanted: # Same clip, same instruction, same references: the prompt and the segmentation are already right. Only the # dilation can have moved, and that is arithmetic on the mask that is already in hand. import h3_sam3 frames = _clip_window(source_path, start, duration) masks = _read_mask_clip(cache["base"], frames.shape[0]) grown = h3_sam3.dilate(masks, int(grow)) clip, view = _mask_outputs(frames, grown) print("[inpaint] reusing the plan and the mask; only the dilation changed", flush=True) return (cache["prompt"], clip, cache["base"], view, f"Masking **{cache['target']}**. Prompt: _{cache['prompt']}_", cache["conditioning"], cache) target, prompt = plan_instruction(source_path, instruction, start, duration, references, progress) supplied = gallery_paths(references) frames = _clip_window(source_path, start, duration) num_frames = frames.shape[0] # With a real reference the conditioner has everything it needs already, so it runs while the GPU segments — # a remote HTTP call in a worker thread, never a `@spaces.GPU` booking, which must stay on the handler's thread. encoding = None if supplied: from concurrent.futures import ThreadPoolExecutor pool = ThreadPoolExecutor(1) encoding = pool.submit( encode_remote, prompt, collect(supplied, None, None), nearest_canvas_label(frames.shape[1], frames.shape[2]), num_frames, ) clip, base, view = find_mask(source_path, target, start, duration, grow, progress) # The prompt goes into the note as well as the box. The box is hidden in Simple until it is filled, and a # user pressing Inpaint straight away should still be able to read what is about to be rendered. note = f"Masking **{target}**. Prompt: _{prompt}_" if encoding is not None: # A real reference meant the conditioner could run alongside the segmentation; take what it returned. prompt_embeds, text_token_tags, metadata, _ = encoding.result() conditioning = { "signature": conditioning_signature(prompt, supplied, int(metadata["num_frames"])), "prompt_embeds": prompt_embeds, "text_token_tags": text_token_tags, "num_frames": int(metadata["num_frames"]), "reference_paths": supplied, "synthetic": False, } result = (prompt, clip, base, view, note, conditioning) else: # Without one the reference has to be built out of the mask first, so nothing could have started earlier. result = _finish_mask(source_path, prompt, frames, clip, base, view, references, note, progress) fresh = { "signature": wanted, "target": target, "prompt": result[0], "base": result[2], "conditioning": result[5], } return (*result, fresh) def check_references(references: list[tuple[str, str]]) -> None: """The `ref2va` rules, checked here so a bad request costs neither half a GPU booking. The first one is the one that bites: this Space runs the `transformer_ref` partition, which needs something to reference. Repainting from the prompt alone — covering an object over, say — is the `t2va` partition's job, and that is a second 38 GiB checkpoint this Space deliberately does not hold. """ if not references: raise gr.Error( "Add at least one reference — an image of what should go in the painted region is the usual one. " "This Space runs MiniMax-H3's reference partition, which always conditions on something." ) kinds = [kind for kind, _ in references] if set(kinds) == {"audio"}: raise gr.Error("An audio reference has to be paired with an image or a video reference.") for kind, path in references: if kind != "video": continue video_seconds, _ = probe(path) if video_seconds is None or not MIN_REFERENCE_VIDEO <= video_seconds <= MAX_REFERENCE_VIDEO: raise gr.Error( f"A reference video has to be between {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`.""" 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 GPU_KWARGS = {"duration": get_duration} | ({"size": GPU_SIZE} if GPU_SIZE else {}) # `size` has to be *absent* for the default slice, not None: `spaces.GPU(size=None)` gets far enough to # configure a worker and then dies with `RuntimeError: NVML_SUCCESS == r INTERNAL ASSERT FAILED`. @spaces.GPU(**GPU_KWARGS) def _generate(request, steps, seed): """The only thing on GPU time: the reference and source encoders, the denoise loop and the decoders. Everything crosses as one dict because a `@spaces.GPU` argument is pickled across a process boundary, and the cropped clip plus its mask is the bulk of it either way. """ import torch if PLACEMENT == "lazy": PIPE.to("cuda") # Switched per request rather than loaded per request: the weights are already resident either way. try: if request.get("turbo") and TURBO_READY: PIPE.enable_lora() PIPE.set_adapters(["turbo"], [1.0]) elif TURBO_READY: PIPE.disable_lora() except Exception as error: print(f"[inpaint] could not set the turbo adapter ({type(error).__name__}: {error})", flush=True) state = PIPE( prompt_embeds=request["prompt_embeds"].to("cuda"), text_token_tags=request["text_token_tags"], references=build_references(request["references"], request.get("reference_edge", 768)), source_video=request["source_video"], source_fps=FPS, mask=request["mask"], source_audio=request["source_audio"], source_audio_sample_rate=request["source_audio_sample_rate"], audio_mask=request["audio_mask"], height=request["height"], width=request["width"], num_frames=request["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 build_references(references, image_edge: int = 768): """Reference files into the in-memory dataclasses the pipeline takes. Decoded inside the GPU worker. Images are put on `image_edge` here rather than left to the pipeline's 2048, because those rows are paid on every step. The conditioner still describes the image at its own resolution and the two disagreeing costs nothing measurable — checked against a 2048 run on the same seed. """ from diffusers.modular_pipelines.minimax_h3.references import ( MiniMaxH3AudioReference, MiniMaxH3ImageReference, MiniMaxH3VideoReference, ) from PIL import Image as PILImage built = [] for kind, path in references: if kind == "image": image = PILImage.open(path).convert("RGB") scale = image_edge / min(image.size) if scale < 1.0: image = image.resize( (round(image.width * scale), round(image.height * scale)), PILImage.Resampling.LANCZOS ) built.append(MiniMaxH3ImageReference(image=np.asarray(image))) elif kind == "video": built.append(MiniMaxH3VideoReference.from_file(path)) else: built.append(MiniMaxH3AudioReference.from_file(path)) return built def generate( # The order here is the order of the UI's inputs list, positionally — gradio maps them that way, so a parameter # moving is a silent mis-wiring rather than an error. `painted` sits at the very end because the brush is no # longer a component: a mask painted on one still was the wrong tool for video, and the path only survives for # callers that pass one explicitly. source_path, prompt, image_1=None, # a gallery of reference images, in the order the model reads them mask_video=None, seed=42, turbo=True, steps=DEFAULT_STEPS, budget=DEFAULT_BUDGET, crop=True, crop_scale=0.5, reference_detail=DEFAULT_REFERENCE_EDGE, soften=False, keep_audio=True, start=0.0, duration=5, audio_path=None, video_path=None, as_pose=True, conditioning=None, painted=None, progress=gr.Progress(track_tqdm=True), ): """One request.""" if LOAD_ERROR: raise gr.Error(LOAD_ERROR) if PIPE is None: raise gr.Error("The denoiser is still loading.") if not source_path: raise gr.Error("Upload the clip you want to repaint.") if not (prompt or "").strip(): raise gr.Error( "There is no prompt yet. Say what should change and press **Generate mask** — it writes the prompt and " "masks the clip for you." ) from diffusers.utils import encode_video from h3_inpaint_crop import canvas_for_box, crop as crop_array, mask_bounding_box, paste_back check_prompt(prompt) progress(0.0, desc="Reading the clip ...") frames, waveform, sample_rate, source_fps, source_frames = read_clip(source_path, start, duration) num_frames = snap_frames(frames.shape[0] / FPS) if frames.shape[0] < num_frames: frames = np.concatenate([frames, np.repeat(frames[-1:], num_frames - frames.shape[0], axis=0)]) frames = frames[:num_frames] clip_height, clip_width = frames.shape[1], frames.shape[2] mask = build_mask(painted, mask_video, num_frames, soften=bool(soften)) if mask.shape[1:3] != (clip_height, clip_width): # A painted mask comes back at the size of the still that was shown, which is the clip's own size; a mask clip # can be anything. The reduction onto the latent grid is a maximum either way, so nearest is enough. from PIL import Image mask = np.stack( [ np.asarray( Image.fromarray((frame * 255).astype(np.uint8)).resize( (clip_width, clip_height), Image.Resampling.NEAREST ), dtype=np.float32, ) / 255.0 for frame in mask ] ) short_edge, max_pixels = BUDGETS[budget] if crop: box = mask_bounding_box(mask, clip_height, clip_width, crop_scale=float(crop_scale)) else: box = (0, 0, (clip_height // CANVAS_MULTIPLE) * CANVAS_MULTIPLE, (clip_width // CANVAS_MULTIPLE) * CANVAS_MULTIPLE) height, width = canvas_for_box(box[2], box[3], CANVAS_MULTIPLE, short_edge, max_pixels) cropped_frames, cropped_mask = crop_array(frames, box), crop_array(mask, box) # `image_1` is a gallery: one component holding the whole ordered list, since the order is what numbers them # ``, `` in the prompt. With none supplied the shot stands in for one, which is what lets a # request that only ever named a thing in words ("a capybara") run at all. supplied = gallery_paths(image_1) reference_paths = supplied or [synthesise_reference(frames, mask)] references = collect(reference_paths, audio_path, video_path) if video_path and as_pose: import h3_pose usable, why = h3_pose.available() if not usable: gr.Warning(f"Pose conversion is unavailable ({why}); using the motion clip as it is.") else: posed = h3_pose.to_pose_clip(_clip_window(video_path, 0, MAX_REFERENCE_VIDEO), FPS, progress) if posed is None: gr.Warning("No pose was found in the motion clip; using it as it is.") else: references = [(kind, posed if path == video_path else path) for kind, path in references] print("[inpaint] motion reference converted to a pose render", flush=True) check_references(references) conditioned = time.time() wanted = conditioning_signature(prompt, reference_paths, num_frames) cached = conditioning if isinstance(conditioning, dict) else None if cached and cached.get("signature") == wanted: # Planning already paid for this: the prompt, the references and the frame count are all unchanged, so the # embeddings on hand are the ones this request would have asked for. prompt_embeds = cached["prompt_embeds"] text_token_tags = cached["text_token_tags"] metadata = {"num_frames": cached["num_frames"]} print("[inpaint] reusing the conditioning the plan already fetched", flush=True) else: progress(0.05, desc="Reading the prompt and references ...") try: prompt_embeds, text_token_tags, metadata, _ = encode_remote( prompt, references, nearest_canvas_label(box[2], box[3]), num_frames ) except gr.Error: raise except Exception as error: 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 # Only `num_frames` is binding: it truncates reference soundtracks, so both halves have to agree on it. The canvas # the conditioner resolved is *not* — the `ref2va` presentation reads the prompt and the references and never the # target canvas — and this request generates on the crop's own canvas instead. num_frames = int(metadata["num_frames"]) request = { "prompt_embeds": prompt_embeds, "text_token_tags": text_token_tags, "references": references, "source_video": cropped_frames, "mask": cropped_mask, "source_audio": waveform if keep_audio else None, "source_audio_sample_rate": sample_rate, # Nothing is passed, so a kept soundtrack is preserved whole — the point of keeping it. "audio_mask": None, "height": height, "width": width, "num_frames": num_frames, "turbo": bool(turbo), "reference_edge": REFERENCE_EDGES.get(reference_detail, 768), } progress(0.1, desc=f"Repainting {num_frames / FPS:.1f} s at {width}x{height} ...") started = time.time() generated, audio, sampling_rate = _generate(request, steps, seed) generate_seconds = time.time() - started progress(0.95, desc="Pasting back ...") painted_frames = np.stack([np.asarray(frame) for frame in generated]) out = paste_back(frames, painted_frames, box, mask=cropped_mask) directory = os.path.join(tempfile.gettempdir(), "h3-outputs") os.makedirs(directory, exist_ok=True) path = os.path.join(directory, f"h3-inpaint-{int(time.time() * 1000)}.mp4") import torch # As a tensor rather than the `uint8` array itself: `encode_video`'s ndarray branch tests whether the values look # like `[0, 1]` floats and warns on every pasted plate, which these are not. encode_video(torch.from_numpy(out), fps=FPS, output_path=path, audio=audio, audio_sample_rate=sampling_rate) covered = float((mask > 0.02).mean()) report = ( f"**{width}x{height}** crop of a {clip_width}x{clip_height} clip " f"({box[2]}x{box[3]} box at {box[1]},{box[0]}) · {num_frames} frames, {num_frames / FPS:.2f} s · " f"{covered:.0%} of the frame repainted · {int(steps)} steps" f"{' · turbo' if (turbo and TURBO_READY) else ''} · " f"conditioner {condition_seconds:.0f}s · repaint {generate_seconds:.0f}s " f"({generate_seconds / int(steps):.1f} s/step) · seed {int(seed)}" ) print(f"[inpaint] {report}", flush=True) return path, report def load_first_frame(path): """The still the mask is painted on: the first frame of the chosen stretch.""" if not path: return gr.update(value=None) import av with av.open(path) as container: frame = next(container.decode(video=0)).to_ndarray(format="rgb24") return gr.update(value=frame) # -------------------------------------------------------------------------------------------------------------- # Masking # -------------------------------------------------------------------------------------------------------------- def _clip_window(source_path, start, duration): """The frames a request over this window will actually see, so a mask is built against the same footage.""" frames, _, _, _, _ = read_clip(source_path, start, duration) num_frames = snap_frames(frames.shape[0] / FPS) if frames.shape[0] < num_frames: frames = np.concatenate([frames, np.repeat(frames[-1:], num_frames - frames.shape[0], axis=0)]) return frames[:num_frames] def _mask_outputs(frames, masks): import h3_sam3 return h3_sam3.to_clip(masks, fps=FPS), h3_sam3.preview(frames, masks, fps=FPS) def _sam3_duration(frames, phrase, progress=None, **_): """Seconds to reserve for segmentation. Takes the arguments of `_segment`, the function it decorates — not `find_mask`'s — which is why it reads the frame count off the array rather than off a duration slider.""" return int(60 + len(frames) * 1.1) @spaces.GPU(duration=_sam3_duration) def _segment(frames, phrase, progress): import h3_sam3 return h3_sam3.segment(frames, phrase, progress) FILMSTRIP_FRAMES = 12 POINT_RADIUS = 9 # Below this share of the frame a track is almost certainly one part of the subject, not the subject. PART_COVERAGE_WARNING = 0.03 def _filmstrip_indices(num_frames: int) -> np.ndarray: return np.linspace(0, num_frames - 1, min(FILMSTRIP_FRAMES, num_frames)).round().astype(int) def _click_xy(index, width: int, height: int) -> tuple[int, int]: """A click index as `(x, y)`. Gradio documents `SelectData.index` as "a tuple if the component is two dimensional" without fixing the order for an image, so this reads it as `(x, y)` and falls back to the transposed reading when that one lands outside the frame. On a square frame both readings are in bounds and the first wins — which is why the click is drawn back onto the frame straight away: a transposed marker is visible immediately rather than silently segmenting the wrong thing. """ first, second = int(index[0]), int(index[1]) if first < width and second < height: return first, second if second < width and first < height: return second, first return min(first, width - 1), min(second, height - 1) def _draw_points(frame: np.ndarray, points, labels) -> np.ndarray: """The clicked frame with its points marked — filled for include, hollow for exclude.""" from PIL import Image, ImageDraw image = Image.fromarray(frame) draw = ImageDraw.Draw(image) for (x, y), label in zip(points or [], labels or []): box = (x - POINT_RADIUS, y - POINT_RADIUS, x + POINT_RADIUS, y + POINT_RADIUS) if label: draw.ellipse(box, fill=(64, 220, 180), outline=(10, 30, 30), width=3) else: draw.ellipse(box, fill=None, outline=(255, 90, 90), width=4) return np.asarray(image) def prepare_clip(source_path, start, duration): """On a new clip: the filmstrip, the first frame to click on, and everything else cleared.""" from PIL import Image if not source_path: return gr.update(value=None), gr.update(value=None), 0, [], [], gr.update(value=None) frames = _clip_window(source_path, start, duration) strip = [(Image.fromarray(frames[i]), f"{i / FPS:.1f}s") for i in _filmstrip_indices(frames.shape[0])] return strip, frames[0], 0, [], [], None def pick_frame(source_path, start, duration, event: gr.SelectData): """Choosing a filmstrip thumbnail moves the clicking to that frame and drops the points made on the old one.""" if not source_path: return gr.update(), 0, [], [] frames = _clip_window(source_path, start, duration) index = int(_filmstrip_indices(frames.shape[0])[int(event.index)]) return frames[index], index, [], [] def add_point(source_path, start, duration, frame_index, points, labels, mode, event: gr.SelectData): if not source_path: return gr.update(), points, labels frames = _clip_window(source_path, start, duration) frame = frames[max(0, min(int(frame_index), frames.shape[0] - 1))] x, y = _click_xy(event.index, frame.shape[1], frame.shape[0]) points = list(points or []) + [[x, y]] labels = list(labels or []) + [1 if mode == "Include" else 0] return _draw_points(frame, points, labels), points, labels def clear_points(source_path, start, duration, frame_index): if not source_path: return gr.update(), [], [] frames = _clip_window(source_path, start, duration) return frames[max(0, min(int(frame_index), frames.shape[0] - 1))], [], [] def _track_duration(frames, points, labels, frame_index, progress=None, **_): return int(45 + len(frames) * 0.5) @spaces.GPU(duration=_track_duration) def _track(frames, points, labels, frame_index, progress): import h3_sam3 return h3_sam3.segment_from_points(frames, points, labels, frame_index, progress) def track_points(source_path, start, duration, frame_index, points, labels, grow, instruction="", prompt="", references=None, progress=gr.Progress()): """Follow what was clicked through the whole clip, then write the prompt for it. Clicking says *where*; the instruction still says *what goes there*. So this ends in the same place the described path does — with a prompt on screen and its conditioning already fetched — rather than leaving a mask and an empty prompt box, which is the one thing the two paths used to disagree about. """ import h3_sam3 if not source_path: raise gr.Error("Upload the clip you want to repaint first.") if not points: raise gr.Error("Click the thing you want to repaint — one click on it is usually enough.") usable, why = h3_sam3.available() if not usable: raise gr.Error(f"Automatic masking is off: {why}. Upload a mask clip instead.") progress(0.0, desc="Reading the clip ...") frames = _clip_window(source_path, start, duration) masks = _track(frames, points, labels, int(frame_index), progress) if masks is None: raise gr.Error("Those clicks did not select anything. Try clicking nearer the middle of the subject.") base, _ = _mask_outputs(frames, masks) grown = h3_sam3.dilate(masks, int(grow)) clip, view = _mask_outputs(frames, grown) print(f"[inpaint] tracked {len(points)} click(s) on frame {frame_index} -> {masks.mean()*100:.0f}%, " f"grown to {grown.mean()*100:.0f}%", flush=True) # A click on a head or a leg comes back as the head or the leg — the part/whole ambiguity is the failure mode # here, not imprecision, and it is unmistakable in the coverage. Measured on the worked clip: a body click gives # 0.98 IoU against the phrase, a head click 0.003. if masks.mean() < PART_COVERAGE_WARNING: gr.Warning( f"That selected only {masks.mean()*100:.1f}% of the frame, which usually means a part of something rather " "than the whole of it. Add an Include click nearer the middle of the subject." ) written = (prompt or "").strip() if not written: if not (instruction or "").strip(): raise gr.Error( "The region is masked, but nothing says what should go there. Fill in **What should change?** above " "— 'a capybara' is enough — or write a prompt yourself." ) # Only the prompt half of the plan is wanted here: the region is already settled by the clicks. _, written = plan_instruction(source_path, instruction, start, duration, references, progress) note = f"Masking what you clicked, {grown.mean()*100:.0f}% of the frame. Prompt: _{written}_" return (*_finish_mask(source_path, written, frames, clip, base, view, references, note, progress), None) def gallery_paths(value) -> list[str]: """The file paths out of a gallery, whichever shape gradio hands them over in. A bare string is wrapped rather than iterated: an API caller passing one reference passes a path, and letting a string fall through to the loop below would take it apart one character at a time. """ if value is None: return [] if isinstance(value, (str, dict)): value = [value] paths = [] for item in value or []: if isinstance(item, (tuple, list)): item = item[0] while isinstance(item, dict): item = item.get("path") or item.get("image") or item.get("name") if item: paths.append(item) return paths def find_mask(source_path, phrase, start, duration, grow, progress=gr.Progress()): """Segment what the phrase names and track it through the window the request will use.""" import h3_sam3 if not source_path: raise gr.Error("Upload the clip you want to repaint first.") if not (phrase or "").strip(): raise gr.Error("Say what should change — a couple of words is enough, like 'the fox'.") usable, why = h3_sam3.available() if not usable: raise gr.Error(f"Automatic masking is off: {why}. Upload a mask clip instead.") progress(0.0, desc="Reading the clip ...") frames = _clip_window(source_path, start, duration) masks = _segment(frames, phrase.strip(), progress) if masks is None: raise gr.Error(f"Nothing matching '{phrase}' was found. Try plainer words — 'person', 'dog', 'car'.") base, _ = _mask_outputs(frames, masks) grown = h3_sam3.dilate(masks, int(grow)) clip, view = _mask_outputs(frames, grown) print(f"[inpaint] segmented '{phrase}' -> {masks.mean()*100:.0f}%, grown to {grown.mean()*100:.0f}%", flush=True) return clip, base, view def regrow_mask(source_path, base_path, grow, start, duration): """Re-apply the slider to the *undilated* segmentation, so dragging it does not compound.""" import h3_sam3 if not source_path or not base_path: return gr.update(), gr.update() frames = _clip_window(source_path, start, duration) masks = _read_mask_clip(base_path, frames.shape[0]) grown = h3_sam3.dilate(masks, int(grow)) clip, view = _mask_outputs(frames, grown) return clip, view def _read_mask_clip(path, num_frames): """A mask clip as booleans at the frame count of the window, padded by holding its last frame.""" import av with av.open(path) as container: masks = np.stack([f.to_ndarray(format="rgb24").mean(-1) for f in container.decode(video=0)]) > 127 if masks.shape[0] < num_frames: masks = np.concatenate([masks, np.repeat(masks[-1:], num_frames - masks.shape[0], axis=0)]) return masks[:num_frames] import ncii_guard ncii_guard.start() load_models() # Fetched at startup so the gated download never lands on GPU time. A Space without the token simply falls back to # the brush and the upload. try: import h3_sam3 _sam3_ok, _sam3_why = h3_sam3.available() if _sam3_ok: h3_sam3.load() print("[inpaint] SAM 3 ready", flush=True) else: print(f"[inpaint] SAM 3 off: {_sam3_why}", flush=True) except Exception as _error: print(f"[inpaint] SAM 3 unavailable ({type(_error).__name__}: {_error})", flush=True) # Reported at startup rather than discovered on a request: mediapipe is fussy about protobuf, and whether it works # here is decided by whatever pip resolved. try: import h3_pose _pose_ok, _pose_why = h3_pose.available() print(f"[inpaint] pose conversion {'ready' if _pose_ok else 'off: ' + _pose_why}", flush=True) except Exception as _error: print(f"[inpaint] pose conversion unavailable ({type(_error).__name__}: {_error})", flush=True) INTRO = """# MiniMax-H3 Inpainting [ model ]   [ modular blocks ]   [ text / image to video ] Inpaint videos with **MiniMax-H3** using [Modular Diffusers](https://huggingface.co/diffusers-modular/minimax-h3-inpainting) 🧨. Instruct inpainting with a simple prompt, click on the subject to inpaint or upload your own mask video. You can also add reference images and/or motion videos for pose transfer for additional edit control. """ CSS = """ .main.fillable { max-width: 1400px !important; } /* Simple | Advanced as a pill segmented control rather than two radio dots. Written defensively: if a gradio release changes the radio's internals this degrades to a plain radio, which still works. */ #viewmode { flex: none !important; } #viewmode .wrap, #viewmode fieldset > div, #pointmode .wrap, #pointmode fieldset > div { display: inline-flex !important; gap: 2px; padding: 3px; background: var(--background-fill-secondary); border: 1px solid var(--border-color-primary); border-radius: 999px; } #viewmode label, #pointmode label { border: none !important; background: transparent !important; box-shadow: none !important; border-radius: 999px !important; padding: 5px 18px !important; margin: 0 !important; font-weight: 600; cursor: pointer; transition: background .15s, color .15s; } #viewmode label:has(input:checked), #pointmode label:has(input:checked) { background: var(--button-primary-background-fill) !important; color: var(--button-primary-text-color) !important; } #viewmode input[type="radio"], #pointmode input[type="radio"] { display: none !important; } /* the click controls read as one panel rather than three loose widgets */ #pointmode { flex: none !important; } /* the filmstrip reads as a strip, not a grid */ #filmstrip .grid-wrap { overflow-x: auto; } #filmstrip .grid-container { grid-auto-flow: column; grid-template-columns: none !important; } #annotate img { cursor: crosshair; } .dark .gradio-container { color: var(--body-text-color); } """ def run_example(source_path, instruction, prompt, references, mask_video, seed, video_path, as_pose, progress=gr.Progress()): """One example row, start to finish, so gradio can cache the result and replay it on the next click. `gr.Examples` cannot cache anything without an `fn`: given `inputs` alone its rows are input presets and there is no output for gradio to store, which is why clicking a row a second time re-ran everything from scratch. The parameter order here IS the `inputs` order — gradio maps them positionally, same trap as `request_inputs`. The progress object deliberately does **not** track tqdm. This one handler does both halves of the work, and gradio's patched tqdm cannot iterate SAM 3's propagation — it raises `IndexError` from inside transformers regardless of whether the bar is disabled. The button path avoids that by splitting into two events; here the only way is to leave tqdm alone, which costs the denoise progress bar and nothing else. """ if not (prompt or "").strip() or not mask_video: prompt, mask_video, _, view, note, conditioning, _ = plan_and_mask( source_path, instruction, 0.0, DEFAULT_WINDOW, references, DEFAULT_GROW, None, progress ) else: # Everything this row needs is already in it, which is what makes the first example a single press. view, note, conditioning = None, "", None video, report = generate( source_path, prompt, references, mask_video, seed, video_path=video_path, as_pose=as_pose, conditioning=conditioning, progress=progress, ) return prompt, mask_video, view, note, video, report with gr.Blocks(title="MiniMax-H3 Inpainting") as demo: gr.Markdown(INTRO) mask_base = gr.State(None) conditioning = gr.State(None) # The plan and the segmentation that produced it, so pressing Generate mask twice on an unchanged request costs # nothing but the re-dilation. plan_cache = gr.State(None) frame_index = gr.State(0) points = gr.State([]) labels = gr.State([]) with gr.Row(): with gr.Column(scale=5): view_mode = gr.Radio( ["Simple", "Advanced"], value="Simple", show_label=False, container=False, elem_id="viewmode", ) source = gr.Video(label="Clip to repaint", height=240) instruction = gr.Textbox( label="Inpainting prompt", placeholder="replace the fox with a capybara", info="Instruct which subject to replace and with what. It works out the mask and writes the " "generation prompt for you.", ) # Declared here, immediately after the textbox, so gradio groups the two into one form and the # checkbox renders as part of the prompt block rather than as a stray line further down. It reads the # clip's own audio and hands it to the model as a *condition*, pinned the way unmasked pixels are, which # is what preserves it exactly instead of regenerating something similar. Silent clips are unaffected. keep_audio = gr.Checkbox(label="Keep the audio from the input video", value=True) # Choosing the region sits directly under the instruction, because that is the order the work happens in: # say what you want, say where, then act. Everything these produce appears below. with gr.Column(visible=False) as advanced_masking: with gr.Tabs(): with gr.Tab("Point at it"): with gr.Group(): filmstrip = gr.Gallery( show_label=False, columns=8, rows=1, height=92, allow_preview=False, object_fit="cover", interactive=False, elem_id="filmstrip", ) annotate = gr.Image( label="Click the middle of the subject", type="numpy", height=300, interactive=False, elem_id="annotate", ) with gr.Row(equal_height=True): point_mode = gr.Radio( ["Include", "Exclude"], value="Include", show_label=False, container=False, scale=3, elem_id="pointmode", ) clear = gr.Button("Clear", size="sm", variant="secondary", scale=1, min_width=80) track = gr.Button("Track it", variant="primary", scale=2, min_width=120) gr.Markdown( "Aim for the body — a click on a head or a leg selects only that part. Exclude " "carves off anything it took too much of.") with gr.Tab("Upload a mask"): mask_video = gr.Video( label="White repaints, one frame per source frame", height=300) gr.Markdown( "One frame per source frame, at any resolution. Anything a segmenter can export " "works.") with gr.Accordion("References (image, video, audio conditions)", open=False), gr.Tabs(): with gr.Tab("Images"): references = gr.Gallery( label="Up to 9, in the order the prompt numbers them", type="filepath", interactive=True, columns=4, rows=1, height=180, allow_preview=False, object_fit="cover", ) gr.Markdown( "Referred to as ``, `` … in upload order. Leave this empty and the " "shot itself stands in, which is enough for \"replace the fox with a capybara\".") with gr.Tab("Audio"): audio = gr.Audio(label="A voice or a piece of music to follow", type="filepath") with gr.Tab("Motion"): video = gr.Video(label="A clip whose motion, camera move and soundtrack to follow, 2–15 s") as_pose = gr.Checkbox( label="Use it as a pose skeleton", value=True, info="Sends a skeleton render instead of the footage, so the clip says how to move without " "also saying how to look. On by default: a motion clip is usually given in order to " "change the subject, and the raw footage re-supplies the old subject's appearance and " "fights the instruction. Turn it off to borrow the clip's look as well as its movement.", ) gr.Markdown( "The performance lever: the model reads motion and camera from the clip. Its rows ride " "through every step, so it is the most expensive reference by far.") with gr.Row(): # Hidden in Simple, which is the default: there, Inpaint is the only button and it does the # planning itself. Generating the mask as a separate step is an Advanced affordance, for # looking at the mask — and adjusting the dilation — before spending GPU time on it. plan = gr.Button("Generate mask", variant="secondary", visible=False) run = gr.Button("Inpaint", variant="primary") planned_target = gr.Markdown() mask_view = gr.Video(label="What will be repainted", height=230, interactive=False) grow = gr.Slider( label="Grow the mask", minimum=0, maximum=96, step=4, value=DEFAULT_GROW, info="Replacing a subject rarely keeps its outline, and anything left outside the mask is preserved " "and shows. Give it room when you are swapping something out.", ) prompt = gr.Textbox( label="Prompt the model will render", lines=3, visible=False, info="Written for you from the instruction above. Edit it and press Inpaint to keep your wording.", ) with gr.Accordion("Settings", open=False), gr.Tabs(): with gr.Tab("Generation"): turbo = gr.Checkbox( label="Few-step LoRA", value=True, info="Eight steps instead of twenty-eight, for about a third of the wall clock. Turn it off " "and raise the steps for the last of the quality — it still wins on a tight mask.", ) steps = gr.Slider(label="Steps", minimum=3, maximum=40, step=1, value=DEFAULT_STEPS) budget = gr.Dropdown(label="Quality budget", choices=list(BUDGETS), value=DEFAULT_BUDGET) reference_detail = gr.Dropdown( label="Reference detail", choices=list(REFERENCE_EDGES), value=DEFAULT_REFERENCE_EDGE, info="Reference rows ride through every step, so this is one of the biggest levers on cost.", ) with gr.Row(): start = gr.Slider(label="Start (s)", minimum=0, maximum=60, step=0.5, value=0) duration = gr.Slider(label="Duration (s)", minimum=MIN_DURATION, maximum=MAX_UI_DURATION, step=1, value=DEFAULT_WINDOW) seed = gr.Number(label="Seed", value=42, precision=0) with gr.Tab("Framing"): crop = gr.Checkbox( label="Crop to the masked region", value=True, info="Repaints a box around the mask instead of the whole frame. Much cheaper on a small " "subject in a large plate.", ) crop_scale = gr.Slider(label="Context around the mask", minimum=0.0, maximum=1.5, step=0.05, value=0.5) soften = gr.Checkbox( label="Soft mask edge", value=False, info="Off by default — a feathered mask can show as a band along the mask.", ) with gr.Column(scale=6): result = gr.Video(label="Inpainted clip", height=520) report = gr.Markdown() def toggle_mode(choice, written): """Advanced reveals the region tabs and the separate Generate mask step; Simple keeps one box to fill and one button to press. Settings is not gated either way — it is collapsed by default, so it costs nothing in Simple, and someone who wants the step count or the seed should not have to change mode to reach it. The prompt is an *output* of planning, not a second thing to type, so it stays hidden in Simple until there is one — and once there is, switching back does not hide the wording away again. Generate mask goes the same way: pressing Inpaint plans whatever is missing anyway, so in Simple the button is a second name for the same work, and the caching means skipping it costs nothing — an unchanged clip, instruction and reference set reuse the plan and the embeddings rather than paying for them twice. """ advanced = choice == "Advanced" return ( gr.update(visible=advanced), gr.update(visible=advanced or bool((written or "").strip())), gr.update(visible=advanced), ) view_mode.change(toggle_mode, [view_mode, prompt], [advanced_masking, prompt, plan], api_name=False) def show_prompt(written): """Reveal the prompt box once there is a prompt in it. Chained rather than folded into the masking handlers, so those keep returning a plain string over the API instead of a gradio update.""" return gr.update(visible=bool((written or "").strip())) window = [source, start, duration] source.change(prepare_clip, window, [filmstrip, annotate, frame_index, points, labels, mask_view], api_name=False) source.change(lambda: (None, None), None, [conditioning, plan_cache], api_name=False) filmstrip.select(pick_frame, window, [annotate, frame_index, points, labels], api_name=False) annotate.select(add_point, window + [frame_index, points, labels, point_mode], [annotate, points, labels], api_name=False) clear.click(clear_points, window + [frame_index], [annotate, points, labels], api_name=False) track.click( track_points, window + [frame_index, points, labels, grow, instruction, prompt, references], [prompt, mask_video, mask_base, mask_view, planned_target, conditioning, plan_cache], api_name="track_points", ).then(show_prompt, prompt, prompt, api_name=False) plan.click( plan_and_mask, [source, instruction, start, duration, references, grow, plan_cache], [prompt, mask_video, mask_base, mask_view, planned_target, conditioning, plan_cache], api_name="plan_and_mask", ).then(show_prompt, prompt, prompt, api_name=False) # Regrowing is arithmetic on a mask that already exists, so it never books GPU. It writes `mask_video` too, and # deliberately has no `.change` handler on that component: one would fire here and overwrite the undilated base # with the dilated result, so every drag of the slider would grow the mask again. grow.release(regrow_mask, [source, mask_base, grow, start, duration], [mask_video, mask_view], api_name=False) request_inputs = [source, prompt, references, mask_video, seed, turbo, steps, budget, crop, crop_scale, reference_detail, soften, keep_audio, start, duration, audio, video, as_pose, conditioning] gr.Examples( examples=[ # Ready to run: the mask and the prompt are already there, so this one is a single press. [ "examples/plate.mp4", "replace the fox with the man from the picture", " the smiling man from the picture, wearing a thick winter parka, walking upright " "toward the camera through deep snow in a pine forest", ["examples/subject.png"], "examples/plate_mask.mp4", 7, None, True, ], # The Simple path with nothing prepared: the instruction alone has to find the man, mask him and write # the prompt. The clip keeps its own soundtrack, so what replaces him speaks the words already there. [ "examples/platform.mp4", "replace the man with the woman from the picture", "", ["examples/wednesday.png"], None, 7, None, True, ], # The pose path, which nothing else demonstrates: the clip is its own motion reference, converted to a # skeleton. The mask has to contain the movement — one drawn around a standing person cannot hold a # dance — so the dancer's own footage is the only sane plate, and a skeleton reference then says "move # like this" without also saying "look like this", which is what lets the subject actually change. [ "examples/dance.mp4", "replace the dancer with a rusty humanoid robot", "", [], None, 7, # A 2.5s motion clip rather than the whole 5.17s plate: a reference video's rows ride through # every step and outnumber the target's, so the full-length version books 1050s of ZeroGPU credit # against a 1500s daily allowance — 70% of a day for one press. Half the reference, 756s. "examples/dance_motion.mp4", True, ], ], inputs=[source, instruction, prompt, references, mask_video, seed, video, as_pose], fn=run_example, outputs=[prompt, mask_video, mask_view, planned_target, result, report], # Lazy, not eager: eager caching runs every example at build time, outside any request, where there is no # ZeroGPU token to forward and the conditioner answers `Expired ZeroGPU proxy token`. Lazy caches the first # real click instead, so the second click on a row replays it instantly. cache_examples=True, cache_mode="lazy", examples_per_page=3, label="Examples — click one to run it", ) # Two events, because they need different progress modes: planning and segmenting must not track tqdm (gradio's # patched tqdm cannot iterate SAM 3's propagation), while the denoise loop wants to. run.click( prepare_if_needed, [source, prompt, references, mask_video, instruction, start, duration, grow, conditioning, mask_base, plan_cache], [prompt, mask_video, mask_base, mask_view, planned_target, conditioning, plan_cache], api_name=False, # `prepare_if_needed` writes the prompt but cannot reveal the box: it returns plain values so the API keeps its # shape, and in Simple the box starts hidden. Without this link a direct press of Inpaint plans, fills the prompt # and then renders it invisibly, which is the half of the earlier report that filling the value alone did not fix. ).then(show_prompt, prompt, prompt, api_name=False).then( generate, request_inputs, [result, report], api_name="generate" ) if __name__ == "__main__": demo.launch(show_error=True, theme=gr.themes.Citrus(), css=CSS, max_threads=1000)