"""Blob -> plan for the PreStage's MiniMax H3 branch: one still, generated by the video model. The other two pre-stage architectures are image models and are compiled by `compile_image.py`. This one is not: H3 is the video model the rest of the pack already loads, and a still from it is *a video generation whose first latent frame is decoded as a picture*. So this module compiles a pre-stage blob into exactly the request shape `compile.compile_request` already takes, and the render path reuses the video segment node verbatim — same conditioning, same reference ordering, same LoRA routing, same checkpoints. **Why the first latent frame is a still at all.** The H3 VAE is causal on the 17k+5 <-> 5k+2 grid: a 5-frame clip has two latent frames, and the first of them is a function of frame 0 alone. Encoding a single image produces that same one-frame latent (core's `downscale_ratio` returns 1 for a single frame), which is what the experimental T=1 image decoder was trained against. Decoding latent index 0 is therefore not a trick — it is the one temporal slice the image VAE was fitted to. **How long a clip, and which frame of it.** The DiT's trained range is 124-362 frames (`canvas.TRAINED_MIN_FRAMES`), so the short end of `STILL_LENGTHS` is off-distribution temporally even though the *latent* it produces is in distribution spatially. Both are pills rather than constants because the answer is a property of the weights, not of this code, and the weights keep changing. Pure, like its two neighbours: no torch, no ComfyUI, no disk. The canvas and duration rules are `canvas.py`'s, because they are the video model's. """ from dataclasses import dataclass from . import canvas from .compile import CHECKPOINTS, CompileError, active_loras, collect_triggers ARCH = "minimax" # The shortest legal clip: 5 frames, two latent frames, the cheapest sampler # pass H3 can be asked for. The default because a still only needs the first # latent frame and everything past it is paid for and thrown away. DEFAULT_FRAMES = 5 # What the length pill offers. Every one is a legal 17n+5 count; the top of the # list is the bottom of the trained range, so the pill spans "cheapest possible" # to "what the weights actually saw". STILL_LENGTHS = (5, 22, 39, 56, 90, 124) # Which latent frame is decoded. 0 is the causal first frame — the slice the # image VAE was trained on. Negative indexes from the end, so -1 is the last # frame of the clip and is how you ask for "the shot a moment later". DEFAULT_LATENT_INDEX = 0 # How the prompt reaches the DiT. "context-ir" is what a video render does: the # shot description is composed into the model's documented Context-IR form, # duration line and all. "plain" sends the typed sentence verbatim through # `prompt_override`, which is the same escape hatch the refiner uses — worth # having because a Context-IR block describes a 0.21-second video, and whether # that helps or hurts a still is exactly the kind of thing only a render says. PROMPT_MODES = ("context-ir", "plain") DEFAULT_PROMPT_MODE = "context-ir" def latent_frames(frames): """Frames -> how many latent frames the VAE packs them into. Mirror of core's `nodes_minimax_h3.video_latent_t`, kept here so an index can be refused before anything is queued. 5 -> 2, 22 -> 5, 124 -> 25. """ return 2 if frames <= 5 else ((frames - 5) // 17) * 5 + 2 def resolve_index(index, frames): """A latent index (possibly negative) -> a real one, or refuse it.""" total = latent_frames(frames) resolved = index if index >= 0 else total + index if not 0 <= resolved < total: raise CompileError( f"latent frame {index} does not exist in a {frames}-frame clip — " f"it packs into {total} latent frames (0..{total - 1})" ) return resolved @dataclass(frozen=True) class StillPlan: """One sampler pass, and the frame taken out of it.""" request: dict frames: int = DEFAULT_FRAMES index: int = DEFAULT_LATENT_INDEX prompt_override: str = None # Whether anything attached will be encoded as sound. False for almost every # still, and that is what leaves the audio VAE unloaded. audio: bool = False @property def payload(self): """The segment payload — the same shape `MiniMaxH3Creator` builds.""" payload = {"request": self.request, "continue": False, "continue_audio": False} if self.prompt_override: payload["prompt_override"] = self.prompt_override return payload def _checkpoint(request): """Which H3 checkpoint this still will route to. The same derivation `compile._resolve_checkpoint` makes, needed here only so the plain-prompt branch can collect the trigger words of the LoRAs that will actually be patched on. The authoritative answer is still the compiled request's, which is what the render path reads. """ route = ((request.get("models") or {}).get("route")) or "auto" if route in CHECKPOINTS: return route pin = request.get("checkpoint") or "auto" if pin in CHECKPOINTS: return pin refs = [a for a in request.get("assets") or [] if a.get("role", "reference") == "reference"] return "ref2va" if refs else "fl2va" def needs_audio(request): """Whether this still will reach for the audio VAE. A reference audio clip, or a reference video cited for its soundtrack. Most stills touch neither, and a loader in the graph is a file loaded whether or not anything reads it — so this is what decides the audio VAE is required. """ for asset in request.get("assets") or []: if asset.get("role", "reference") != "reference": continue if asset.get("kind") == "audio": return True if asset.get("kind") == "video" and asset.get("track") in ("picture+sound", "sound"): return True return False def _frames(raw): """A requested frame count -> the legal 17n+5 count actually generated.""" try: frames = int(raw) except (TypeError, ValueError): raise CompileError("the still length must be a whole number of frames") return canvas.frames_for_seconds(max(1, frames) / canvas.FPS) def _block(data): block = data.get(ARCH) return block if isinstance(block, dict) else {} def _request(block, frames): """The request one pass generates from: the still's own, at this length. A copy rather than the stored dict, because the length it is generated at is the still's setting rather than the request's. """ request = dict(block.get("request") or {}) if not isinstance(request, dict): raise CompileError("the still's request must be a JSON object") # The duration the video side speaks in. `frames` is already snapped, so # `frames_for_seconds` inside `compile_request` lands back on it exactly. request["duration_s"] = canvas.seconds_for_frames(frames) # Never two-pass: a still past the native edge upscales through the # single-image VAE decode, and the still graph has no refine pass to hand # a capped canvas to — left unpinned, `compile_request` would sample at # 768 and the slider above it would quietly do nothing. request["upscale"] = "direct" return request def _plain_prompt(request, checkpoint): """The typed sentence, trigger words in front, and nothing composed. Same construction as the video body's prefix — a word only counts if its LoRA is actually in this run — because that is the one part of the composed prompt an override would otherwise drop on the floor. """ prompt = str(request.get("prompt") or "").strip() triggers = collect_triggers(active_loras(request.get("loras"), checkpoint)) if triggers: prompt = f"{', '.join(triggers)}, {prompt}" if prompt else ", ".join(triggers) return prompt def compile_still(data): """A pre-stage blob (arch 'minimax') -> `StillPlan`. The request inside the blob is already in `compile.compile_request`'s shape — the pre-stage's H3 branch is driven by the Creator's own editor, which writes the Creator's own state — so this settles only what is the *still's*: how long a clip to sample, which of its latent frames to keep, and how the prompt reaches the DiT. Everything else is validated where it is for a shot, by the segment node compiling the request it is handed. """ if not isinstance(data, dict): raise CompileError("prestage_data must be a JSON object") block = _block(data) frames = _frames(block.get("frames", DEFAULT_FRAMES)) request = _request(block, frames) if not str(request.get("prompt") or "").strip() and not ( (request.get("refined") or {}).get("body") or "").strip(): raise CompileError("describe the still first — the prompt is empty") mode = block.get("prompt_mode", DEFAULT_PROMPT_MODE) if mode not in PROMPT_MODES: raise CompileError(f"unknown prompt mode {mode!r}") override = _plain_prompt(request, _checkpoint(request)) if mode == "plain" else None try: index = int(block.get("latent_index", DEFAULT_LATENT_INDEX)) except (TypeError, ValueError): raise CompileError("the latent frame index must be a whole number") return StillPlan( request=request, frames=frames, index=resolve_index(index, frames), prompt_override=override, audio=needs_audio(request), )