minimax_h3_pipeline

@dataclass
class GenerationResult:
    video_path: str | None
    frames: np.ndarray | None
    waveform: np.ndarray | None
    sample_rate: int
    timings: dict[str, float] = field(default_factory=dict)
    peak_memory_gib: dict[str, float] = field(default_factory=dict)
    vsa: dict[str, Any] = field(default_factory=dict)
    video_decode_backend: str = "h3-vae"
    prefix_kv: dict[str, Any] = field(default_factory=dict)

@dataclass(frozen=True)
class FastTemporalPlan:
    """Sparse video geometry for RIFE fast mode with full-duration audio."""
target_frames: int
source_frames: int
factor: int
video_temporal_scale: float</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.plan_fast_temporal
plan_fast_temporal(target_frames: int, factor: int = 2) -> FastTemporalPlan
Choose the smallest H3-valid source sequence that covers the target timeline.
Source
def plan_fast_temporal(target_frames: int, factor: int = 2) -> FastTemporalPlan:
    """Choose the smallest H3-valid source sequence that covers the target timeline."""
    target_frames = align_num_frames(target_frames)
    if factor < 2:
        raise ValueError(f"fast factor must be at least 2, got {factor}.")
    ideal_source_frames = math.ceil((target_frames - 1) / factor) + 1
    source_frames = align_num_frames(ideal_source_frames)
    if source_frames >= target_frames:
        raise ValueError(f"fast factor {factor} does not reduce the H3-aligned target of {target_frames} frames.")
source_grid = temporal_position_grid(video_latent_num_frames(source_frames), 0.0)
target_grid = temporal_position_grid(video_latent_num_frames(target_frames), 0.0)
video_temporal_scale = float(target_grid&#91;-1&#93; / source_grid&#91;-1&#93;)
return FastTemporalPlan(
    target_frames=target_frames,
    source_frames=source_frames,
    factor=factor,
    video_temporal_scale=video_temporal_scale,
)</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.FastSpatialPlan
FastSpatialPlan(target_height: int, target_width: int, stage1_height: int, stage1_width: int, canvas_height: int, canvas_width: int, scale: int, upsample_mode: str, sharpen: float)
Reduced-canvas geometry for spatial fast mode (RIFE's spatial twin).
Source
@dataclass(frozen=True)
class FastSpatialPlan:
    """Reduced-canvas geometry for spatial fast mode (RIFE's spatial twin)."""
target_height: int
target_width: int
stage1_height: int
stage1_width: int
canvas_height: int
canvas_width: int
scale: int
upsample_mode: str
sharpen: float</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.plan_fast_spatial
plan_fast_spatial(height: int, width: int, *, scale: int = 2, upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE, sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN) -> FastSpatialPlan
Choose the smallest H3-valid canvas that covers target / scale.
H3 geometry rounds up to the 32px model grid and center-crops after
decode — the same convention plain 720p generation uses via
_model_canvas_size — so no size the full-resolution path accepts is
rejected here. The return trip to the target size runs in pixel space
after the VAE decode, never on latents; see
:mod:fastvideo.mlx_runtime.frame_upsample for why.
Source
def plan_fast_spatial(
    height: int,
    width: int,
    *,
    scale: int = 2,
    upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE,
    sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN,
) -> FastSpatialPlan:
    """Choose the smallest H3-valid canvas that covers ``target / scale``.
H3 geometry rounds *up* to the 32px model grid and center-crops after
decode — the same convention plain 720p generation uses via
``_model_canvas_size`` — so no size the full-resolution path accepts is
rejected here. The return trip to the target size runs in pixel space
after the VAE decode, never on latents; see
:mod:`fastvideo.mlx_runtime.frame_upsample` for why.
&quot;&quot;&quot;
if scale &lt; 2:
    raise ValueError(f&quot;fast-spatial scale must be at least 2, got &#123;scale&#125;.&quot;)
if upsample_mode not in PIXEL_UPSAMPLE_MODES:
    raise ValueError(f&quot;Unsupported upsample mode: &#123;upsample_mode!r&#125; &quot;
                     f&quot;(expected one of &#123;&#x27;, &#x27;.join(PIXEL_UPSAMPLE_MODES)&#125;)&quot;)
if sharpen &lt; 0:
    raise ValueError(f&quot;fast_spatial_sharpen must be non-negative, got &#123;sharpen&#125;.&quot;)
target_canvas_height, target_canvas_width = _model_canvas_size(height, width)
stage1_height = math.ceil(height / scale)
stage1_width = math.ceil(width / scale)
canvas_height, canvas_width = _model_canvas_size(stage1_height, stage1_width)
if canvas_height * canvas_width &gt;= target_canvas_height * target_canvas_width:
    raise ValueError(
        f&quot;fast-spatial scale &#123;scale&#125; does not reduce the H3 canvas for &#123;height&#125;x&#123;width&#125; &quot;
        f&quot;(stage-1 canvas &#123;canvas_width&#125;x&#123;canvas_height&#125; vs &#123;target_canvas_width&#125;x&#123;target_canvas_height&#125;).&quot;)
return FastSpatialPlan(
    target_height=height,
    target_width=width,
    stage1_height=stage1_height,
    stage1_width=stage1_width,
    canvas_height=canvas_height,
    canvas_width=canvas_width,
    scale=scale,
    upsample_mode=upsample_mode,
    sharpen=sharpen,
)</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.prompt_cache_path
prompt_cache_path(cache_dir: str | Path, model_root: str | Path, prompt: str) -> Path
Source
def prompt_cache_path(cache_dir: str | Path, model_root: str | Path, prompt: str) -> Path:
    digest = hashlib.sha256(
        f"{MINIMAX_H3_PROMPT_CACHE_VERSION}::{Path(model_root)}::{prompt}".encode()).hexdigest()[:24]
    return Path(cache_dir) / f"prompt_embeds_{digest}.npz"
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline
MiniMaxH3MLXPipeline(*, model_root: str | Path, mlx_dit_checkpoint: str | Path, vae_dtype: str = 'fp32', vae_decoder_path: str | Path | None = None, vae_encoder_path: str | Path | None = None, vae_decoder_quantization: str | None = None, prompt_cache_dir: str | Path | None = None, conditioner_dir: str | Path | None = None, tokenizer_dir: str | Path | None = None, metal_wired_limit_gib: float | None = None, mlx_cache_limit_gib: float | None = None, quant_gemm: str | None = None, video_decode_backend: str = 'h3-vae', taeh3_checkpoint: str | Path | None = None, taeh3_chunk_size: int = 5, conditioner_mode: str = 'auto', resident: bool = False)
Text-to-video-with-audio generation through the native MLX runtime.
generate also takes an ordered references list for the
reference-conditioned (Ref2VA / OmniRef) student, which prepends the
encoded references to the packed sequence and pins them as conditioning.
Source
class MiniMaxH3MLXPipeline:
    """Text-to-video-with-audio generation through the native MLX runtime.
``generate`` also takes an ordered ``references`` list for the
reference-conditioned (Ref2VA / OmniRef) student, which prepends the
encoded references to the packed sequence and pins them as conditioning.
&quot;&quot;&quot;

# Defaults for pipelines built without __init__ (unit tests use __new__).
resident = False
conditioner_mode = &quot;auto&quot;
vae_decoder_path: Path | None = None
vae_encoder_path: Path | None = None
vae_decoder_quantization: str | None = None

def __init__(
    self,
    *,
    model_root: str | Path,
    mlx_dit_checkpoint: str | Path,
    vae_dtype: str = &quot;fp32&quot;,
    vae_decoder_path: str | Path | None = None,
    vae_encoder_path: str | Path | None = None,
    vae_decoder_quantization: str | None = None,
    prompt_cache_dir: str | Path | None = None,
    conditioner_dir: str | Path | None = None,
    tokenizer_dir: str | Path | None = None,
    metal_wired_limit_gib: float | None = None,
    mlx_cache_limit_gib: float | None = None,
    quant_gemm: str | None = None,
    video_decode_backend: str = &quot;h3-vae&quot;,
    taeh3_checkpoint: str | Path | None = None,
    taeh3_chunk_size: int = 5,
    conditioner_mode: str = &quot;auto&quot;,
    resident: bool = False,
) -&gt; None:
    import mlx.core as mx

    self.model_root = Path(model_root)
    if conditioner_mode not in (&quot;auto&quot;, &quot;streamed&quot;, &quot;nvfp4&quot;):
        raise ValueError(f&quot;Unknown H3 conditioner mode: &#123;conditioner_mode&#125;&quot;)
    if resident and video_decode_backend != &quot;h3-vae&quot;:
        raise ValueError(&quot;Resident H3 generation requires the H3 video VAE.&quot;)
    self.conditioner_mode = conditioner_mode
    self.resident = resident
    self._resident_components: dict&#91;str, Any&#93; = &#123;&#125;
    self.dit_checkpoint = Path(mlx_dit_checkpoint)
    self.vae_dtype = vae_dtype
    self.vae_decoder_path = Path(vae_decoder_path) if vae_decoder_path else None
    self.vae_encoder_path = Path(vae_encoder_path) if vae_encoder_path else None
    self.vae_decoder_quantization = vae_decoder_quantization
    if video_decode_backend not in (&quot;h3-vae&quot;, &quot;taeh3&quot;):
        raise ValueError(f&quot;Unknown H3 video decoder: &#123;video_decode_backend&#125;&quot;)
    if taeh3_chunk_size &lt; 1:
        raise ValueError(&quot;taeh3_chunk_size must be positive.&quot;)
    if taeh3_checkpoint is not None and video_decode_backend != &quot;taeh3&quot;:
        raise ValueError(&quot;taeh3_checkpoint requires video_decode_backend=&#x27;taeh3&#x27;.&quot;)
    self.video_decode_backend = video_decode_backend
    self.taeh3_checkpoint = taeh3_checkpoint
    self.taeh3_chunk_size = taeh3_chunk_size
    self.prompt_cache_dir = Path(prompt_cache_dir) if prompt_cache_dir else None
    self.conditioner_dir = Path(conditioner_dir) if conditioner_dir else self.model_root / &quot;text_encoder&quot;
    self.tokenizer_dir = Path(tokenizer_dir) if tokenizer_dir else self.model_root / &quot;tokenizer&quot;
    self._validate_inputs_before_loading()
    manifest = json.loads((self.dit_checkpoint / H3_MANIFEST_FILENAME).read_text())
    dit_config = manifest&#91;&quot;config&quot;&#93;
    patch_size = dit_config&#91;&quot;patch_size&quot;&#93;
    if len(patch_size) != 3:
        raise ValueError(f&quot;H3 DiT patch_size must have three dimensions, got &#123;patch_size&#125;.&quot;)
    self._dit_patch_size = (int(patch_size&#91;0&#93;), int(patch_size&#91;1&#93;), int(patch_size&#91;2&#93;))
    self._dit_in_channels = int(dit_config&#91;&quot;in_channels&quot;&#93;)
    self.last_dit_forward_s = 0.0
    self.last_vsa_stats: dict&#91;str, Any&#93; | None = None
    self._apply_quant_gemm(quant_gemm)
    self._previous_memory_limit, self._previous_wired_limit = _configure_metal_memory_limits(mx,
                                                                                             metal_wired_limit_gib,
                                                                                             resident=resident)
    self._previous_cache_limit = _configure_mlx_cache_limit(mx, mlx_cache_limit_gib)

def _apply_quant_gemm(self, quant_gemm: str | None) -&gt; None:
    &quot;&quot;&quot;An explicit mode overrides FASTVIDEO_MLX_QUANT_GEMM for this pipeline; None leaves it in charge.&quot;&quot;&quot;
    self._quant_gemm_set = quant_gemm is not None
    self._previous_quant_gemm = set_quant_gemm_mode(quant_gemm) if self._quant_gemm_set else None

def _restore_quant_gemm(self) -&gt; None:
    if getattr(self, &quot;_quant_gemm_set&quot;, False):
        set_quant_gemm_mode(self._previous_quant_gemm)
        self._quant_gemm_set = False

def _load_video_vae(self, *, include_encoder: bool, include_decoder: bool = True):
    &quot;&quot;&quot;Load the H3 video VAE, swapping in standalone decoder / encoder files when configured.&quot;&quot;&quot;
    from fastvideo.mlx_runtime import minimax_h3_video_vae as video_vae

    vae_dir = self.model_root / &quot;vae&quot;
    standalone = (self.vae_decoder_path is not None or self.vae_encoder_path is not None
                  or self.vae_decoder_quantization is not None)
    if not standalone:
        return video_vae.mlx_h3_video_vae_from_dir(vae_dir,
                                                   include_encoder=include_encoder,
                                                   include_decoder=include_decoder,
                                                   storage_dtype=self.vae_dtype)
    has_base = vae_dir.is_dir() and (any(vae_dir.glob(&quot;*.safetensors&quot;)) or (vae_dir / &quot;config.json&quot;).exists())
    return video_vae.mlx_h3_video_vae_from_files(
        vae_dir=vae_dir if has_base else None,
        decoder_path=self.vae_decoder_path,
        encoder_path=self.vae_encoder_path,
        include_encoder=include_encoder,
        include_decoder=include_decoder,
        storage_dtype=self.vae_dtype,
        decoder_quantization=self.vae_decoder_quantization,
    )

# -- input validation (before anything heavy loads) -------------------

def _validate_inputs_before_loading(self) -&gt; bool:
    missing = &#91;&#93;
    if not self.dit_checkpoint.exists():
        missing.append(str(self.dit_checkpoint))
    vae_dir = self.model_root / &quot;vae&quot;
    audio_dir = self.model_root / &quot;audio_vae&quot;
    # Standalone decoder + encoder files (LightVAE, FastH3 decoder) replace the base VAE shards entirely.
    standalone = self.vae_decoder_path is not None and self.vae_encoder_path is not None
    if (self.video_decode_backend == &quot;h3-vae&quot; and not standalone
            and not (vae_dir.exists() and any(vae_dir.glob(&quot;*.safetensors&quot;)))):
        missing.append(str(vae_dir))
    for path in (self.vae_decoder_path, self.vae_encoder_path):
        if path is not None and not path.is_file():
            missing.append(str(path))
    if not (audio_dir.exists() and any(audio_dir.glob(&quot;*.safetensors&quot;))):
        missing.append(str(audio_dir))
    if missing:
        raise FileNotFoundError(f&quot;Missing required H3 components: &#123;missing&#125;&quot;)
    return True

@staticmethod
def resolve_geometry(
    height: int,
    width: int,
    num_frames: int,
    *,
    enforce_duration: bool = True,
) -&gt; dict&#91;str, int&#93;:
    &quot;&quot;&quot;Explicit canvases pass through (positive multiples of 32); the
    aspect-ratio resolver only applies when dimensions are omitted.&quot;&quot;&quot;
    if height &lt;= 0 or width &lt;= 0 or height % 32 or width % 32:
        raise ValueError(f&quot;H3 canvas must be positive multiples of 32, got &#123;height&#125;x&#123;width&#125;.&quot;)
    aligned_frames = align_num_frames(num_frames)
    latent_frames = video_latent_num_frames(aligned_frames)
    duration = aligned_frames / MINIMAX_H3_FPS
    if enforce_duration and not MINIMAX_H3_MIN_ALIGNED_FRAMES &lt;= aligned_frames &lt;= MINIMAX_H3_MAX_ALIGNED_FRAMES:
        raise ValueError(f&quot;H3 generates &#123;MINIMAX_H3_MIN_DURATION:g&#125;-&#123;MINIMAX_H3_MAX_DURATION:g&#125; s at &quot;
                         f&quot;&#123;MINIMAX_H3_FPS&#125; fps; &#123;aligned_frames&#125; frames is &#123;duration:.2f&#125; s, outside the &quot;
                         f&quot;accepted &#123;MINIMAX_H3_MIN_ALIGNED_FRAMES&#125;-&#123;MINIMAX_H3_MAX_ALIGNED_FRAMES&#125; frame range.&quot;)
    return &#123;
        &quot;height&quot;: height,
        &quot;width&quot;: width,
        &quot;num_frames&quot;: aligned_frames,
        &quot;latent_frame_count&quot;: latent_frames,
        &quot;latent_height&quot;: height // 16,
        &quot;latent_width&quot;: width // 16,
    &#125;

# -- phase 1: conditioning -------------------------------------------

def prepare_resident(self) -&gt; None:
    &quot;&quot;&quot;Load the encoder, DiT, and both decoders once, before timed requests.&quot;&quot;&quot;
    if not self.resident or self._resident_components:
        return
    from fastvideo.mlx_runtime.minimax_h3_conditioner import ResidentNVFP4MiniMaxH3TextConditioner
    from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir
    import mlx.core as mx

    try:
        conditioner = self._load_conditioner()
        if not isinstance(conditioner, ResidentNVFP4MiniMaxH3TextConditioner):
            conditioner.close()
            raise ValueError(&quot;All-resident generation requires the packed NVFP4 text encoder.&quot;)
        self._resident_components&#91;&quot;conditioner&quot;&#93; = conditioner
        self._resident_components&#91;&quot;dit&quot;&#93; = _load_resident_dit(self.dit_checkpoint)
        self._resident_components&#91;&quot;video_vae&quot;&#93; = self._load_video_vae(include_encoder=False)
        self._resident_components&#91;&quot;audio_vae&quot;&#93; = mlx_h3_audio_vae_from_dir(self.model_root / &quot;audio_vae&quot;,
                                                                           include_encoder=False)
        mx.eval(list(self._resident_components&#91;&quot;audio_vae&quot;&#93;.weights.values()))
        logger.info(&quot;H3 components resident: %.2f GiB active MLX memory&quot;, mx.get_active_memory() / 2**30)
    except Exception:
        self.close()
        raise

def close(self) -&gt; None:
    self._restore_quant_gemm()
    conditioner = self._resident_components.get(&quot;conditioner&quot;)
    if conditioner is not None:
        conditioner.close()
    self._resident_components.clear()
    _cleanup_mlx()
    previous = getattr(self, &quot;_previous_wired_limit&quot;, None)
    if previous is not None:
        import mlx.core as mx

        _restore_metal_wired_limit(mx, previous)
        self._previous_wired_limit = None
    previous_memory = getattr(self, &quot;_previous_memory_limit&quot;, None)
    if previous_memory is not None:
        import mlx.core as mx

        _restore_metal_memory_limit(mx, previous_memory)
        self._previous_memory_limit = None
    previous_cache = getattr(self, &quot;_previous_cache_limit&quot;, None)
    if previous_cache is not None:
        import mlx.core as mx

        mx.set_cache_limit(previous_cache)
        self._previous_cache_limit = None

def encode_prompt(self, prompt: str) -&gt; tuple&#91;np.ndarray, np.ndarray&#93;:
    &quot;&quot;&quot;Returns (hidden states (S, hidden), token tags). Uses the cache or
    the streamed conditioner.&quot;&quot;&quot;
    cache_key = self._prompt_cache_key(prompt)
    if cache_key is not None and cache_key.exists():
        data = np.load(cache_key)
        logger.info(&quot;Loaded prompt embeddings from cache %s&quot;, cache_key)
        return data&#91;&quot;hidden_states&quot;&#93;, data&#91;&quot;token_tags&quot;&#93;

    if self.resident:
        self.prepare_resident()
        conditioner = self._resident_components&#91;&quot;conditioner&quot;&#93;
    else:
        conditioner = self._load_conditioner()
    hidden, tags = conditioner.encode_prompt(prompt)
    if not self.resident:
        conditioner.close()
    _cleanup_mlx()
    if cache_key is not None:
        cache_key.parent.mkdir(parents=True, exist_ok=True)
        tmp_cache = cache_key.with_name(f&quot;.&#123;cache_key.name&#125;.tmp&quot;)
        try:
            with tmp_cache.open(&quot;wb&quot;) as handle:
                np.savez(handle, hidden_states=hidden, token_tags=tags)
            tmp_cache.replace(cache_key)
        finally:
            tmp_cache.unlink(missing_ok=True)
    return hidden, tags

def _prompt_cache_key(self, prompt: str) -&gt; Path | None:
    &quot;&quot;&quot;Cache path bound to the effective encoder and its files; None skips the cache.&quot;&quot;&quot;
    if self.prompt_cache_dir is None:
        return None
    fingerprint = &#123;
        &quot;conditioner&quot;: self._conditioner_kind(),
        &quot;text_encoder&quot;: text_encoder_fingerprint(self.conditioner_dir),
        &quot;tokenizer&quot;: text_encoder_fingerprint(self.tokenizer_dir),
    &#125;
    if not (fingerprint&#91;&quot;text_encoder&quot;&#93;&#91;&quot;complete&quot;&#93; and fingerprint&#91;&quot;tokenizer&quot;&#93;&#91;&quot;complete&quot;&#93;):
        return None
    return prompt_cache_path(self.prompt_cache_dir, fingerprint_digest(fingerprint), prompt)

def load_prompt_cache(self, path: str | Path) -&gt; tuple&#91;np.ndarray, np.ndarray&#93;:
    data = np.load(path)
    return data&#91;&quot;hidden_states&quot;&#93;, data&#91;&quot;token_tags&quot;&#93;

def prepare_references(
    self,
    references: Sequence&#91;MiniMaxH3Reference&#93;,
    num_frames: int,
    *,
    max_pixels: int | None = None,
) -&gt; list&#91;MiniMaxH3PreparedReference&#93;:
    &quot;&quot;&quot;Validate and decode the ordered reference list.

    Decoding happens once, before anything is loaded, so a bad reference
    fails before the DiT and the encoders are resident.
    &quot;&quot;&quot;
    validated = validate_references(list(references))
    target_sample_rate = self.audio_sample_rate()
    prepared = &#91;
        prepare_reference(reference, num_frames, target_sample_rate, max_pixels=max_pixels)
        for reference in validated
    &#93;
    logger.info(&quot;Prepared %d reference(s): %s&quot;, len(prepared),
                &quot;, &quot;.join(reference.media_type for reference in prepared))
    return prepared

def audio_sample_rate(self) -&gt; int:
    &quot;&quot;&quot;The audio VAE&#x27;s sample rate, which reference soundtracks are resampled to.&quot;&quot;&quot;
    config_path = self.model_root / &quot;audio_vae&quot; / &quot;config.json&quot;
    if not config_path.is_file():
        return 32000
    return int(json.loads(config_path.read_text(encoding=&quot;utf-8&quot;)).get(&quot;sampling_rate&quot;, 32000))

def encode_reference_prompt(
    self,
    prompt: str,
    references: list&#91;MiniMaxH3PreparedReference&#93;,
) -&gt; tuple&#91;np.ndarray, np.ndarray&#93;:
    &quot;&quot;&quot;Encode the prompt with its ordered references as Qwen3-VL condition rows.&quot;&quot;&quot;
    from fastvideo.mlx_runtime.minimax_h3_ref_conditioning import encode_reference_prompt, load_vision_limits
    from fastvideo.mlx_runtime.minimax_h3_vision import Qwen3VLVisionConfig

    if self.resident:
        self.prepare_resident()
        conditioner = self._resident_components&#91;&quot;conditioner&quot;&#93;
    else:
        conditioner = self._load_conditioner()
    try:
        config = Qwen3VLVisionConfig.from_config_json(self.conditioner_dir / &quot;config.json&quot;)
        min_pixels, max_pixels = load_vision_limits(self.processor_dir())
        video_min_pixels, video_max_pixels = load_vision_limits(self.processor_dir(), video=True)
        hidden, tags = encode_reference_prompt(
            conditioner,
            prompt,
            references,
            vision_config=config,
            min_pixels=min_pixels,
            max_pixels=max_pixels,
            video_min_pixels=video_min_pixels,
            video_max_pixels=video_max_pixels,
        )
    finally:
        if not self.resident:
            conditioner.close()
    _cleanup_mlx()
    return hidden, tags

def processor_dir(self) -&gt; Path:
    &quot;&quot;&quot;Where the Qwen3-VL processor config lives, defaulting to the tokenizer dir.&quot;&quot;&quot;
    for candidate in (self.model_root / &quot;processor&quot;, self.tokenizer_dir):
        if (candidate / &quot;preprocessor_config.json&quot;).is_file():
            return candidate
    return self.tokenizer_dir

def has_conditioner_weights(self) -&gt; bool:
    marker = self.conditioner_dir / &quot;model.safetensors.index.json&quot;
    single = self.conditioner_dir / &quot;model.safetensors&quot;
    return marker.exists() or single.exists()

def _load_conditioner(self):
    from fastvideo.mlx_runtime.minimax_h3_conditioner import (
        ResidentNVFP4MiniMaxH3TextConditioner,
        StreamedMiniMaxH3TextConditioner,
    )

    if self._conditioner_kind() == &quot;nvfp4&quot;:
        return ResidentNVFP4MiniMaxH3TextConditioner(self.conditioner_dir, self.tokenizer_dir)
    return StreamedMiniMaxH3TextConditioner(self.conditioner_dir, self.tokenizer_dir)

def _conditioner_kind(self) -&gt; str:
    &quot;&quot;&quot;The encoder that conditioner_mode selects for these weights: &#x27;nvfp4&#x27; or &#x27;streamed&#x27;.&quot;&quot;&quot;
    config = json.loads((self.conditioner_dir / &quot;config.json&quot;).read_text())
    packed = str(config.get(&quot;quantization_config&quot;, &#123;&#125;).get(&quot;quant_method&quot;, &quot;&quot;)).lower() == &quot;nvfp4&quot;
    if self.conditioner_mode == &quot;nvfp4&quot; or (self.conditioner_mode == &quot;auto&quot; and packed):
        return &quot;nvfp4&quot;
    if packed:
        raise ValueError(&quot;The streamed conditioner requires BF16 weights; use conditioner_mode=&#x27;nvfp4&#x27;.&quot;)
    return &quot;streamed&quot;

# -- phase 2: denoise --------------------------------------------------

def denoise(
    self,
    text_rows: np.ndarray,
    token_tags: np.ndarray,
    *,
    height: int,
    width: int,
    num_frames: int,
    audio_num_frames: int | None = None,
    video_temporal_scale: float = 1.0,
    seed: int,
    num_steps: int = 4,
    dit: Any | None = None,
    vsa_config: MiniMaxH3VSAConfig | None = None,
    inter_step_cooldown_s: float = 0.0,
    references: list&#91;MiniMaxH3PreparedReference&#93; | None = None,
    condition_video_rows: np.ndarray | None = None,
    condition_audio_rows: np.ndarray | None = None,
    reuse_prefix_kv: bool = False,
    prefix_kv_refresh_every: int = 0,
    prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB,
    prefix_kv_storage: str = &quot;bf16&quot;,
) -&gt; tuple&#91;np.ndarray, np.ndarray&#93;:
    &quot;&quot;&quot;Denoise joint latents; returns (normalized target video rows, target audio rows).

    With ``references`` the packed sequence is reference-prefixed and the
    encoder rows ride along in every forward, but only the target rows are
    stepped and returned: the references are conditioning, not output.

    ``reuse_prefix_kv`` (Ref2VA only, lossy) records the text/reference
    rows&#x27; per-block K/V on the first step and runs later steps on the
    target rows alone; ``prefix_kv_refresh_every=R`` re-records every R
    steps (0 never). The cache is refused above ``prefix_kv_cache_max_gib``.
    &quot;&quot;&quot;
    import mlx.core as mx

    self.last_prefix_kv_stats: dict&#91;str, object&#93; = &#123;&quot;enabled&quot;: False&#125;
    self.last_denoise_step_times = &#91;&#93;
    if prefix_kv_refresh_every &lt; 0:
        raise ValueError(f&quot;prefix_kv_refresh_every must be &gt;= 0, got &#123;prefix_kv_refresh_every&#125;.&quot;)
    if reuse_prefix_kv and references is None:
        raise ValueError(&quot;Prefix K/V reuse needs Ref2VA references (a reference-prefixed packed layout).&quot;)
    validate_prefix_kv_storage(prefix_kv_storage)

    geometry = self.resolve_geometry(height, width, num_frames, enforce_duration=audio_num_frames is None)
    audio_frames = geometry&#91;&quot;num_frames&quot;&#93; if audio_num_frames is None else align_num_frames(audio_num_frames)

    resident_dit = dit is None and self.resident
    if resident_dit:
        _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
        self.prepare_resident()
        dit = self._resident_components&#91;&quot;dit&quot;&#93;
    owned_dit = dit is None
    if owned_dit:
        _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
        t0 = time.perf_counter()
        dit = load_mlx_h3_checkpoint(self.dit_checkpoint)
        logger.info(&quot;Loaded MLX H3 DiT from %s in %.1fs&quot;, self.dit_checkpoint, time.perf_counter() - t0)

    schedule = resolve_h3_denoise_schedule(self.model_root, num_steps)
    video_scheduler = schedule.video
    audio_scheduler = schedule.audio
    union = schedule.adaln_timesteps
    num_steps = schedule.num_steps
    # A FastH3 PDD export runs one fused head block per trained block; its
    # plan supplies the block partition, the per-modality node sigmas
    # (already on the schedulers) and the integration weights.
    pdd_plan = schedule.pdd_plan
    # The keyframe-noise timestep (0.999) is only exercised by FL2VA/Ref2VA
    # conditioning rows. A Ref2VA PDD ladder already carries it, so a
    # converted DiT&#x27;s baked cache matches; this append is then a no-op.
    if references is not None:
        if video_temporal_scale != 1.0:
            # The reference-prefixed builder stretches only the target video
            # rows, which the CUDA packing never does either; refusing beats
            # silently ignoring the request.
            raise ValueError(&quot;Reference-conditioned packing has no temporal-scale stretch; &quot;
                             &quot;run Ref2VA without fast mode.&quot;)
        union = np.unique(np.concatenate(&#91;union, np.array(&#91;MINIMAX_H3_KEYFRAME_NOISE_AUG&#93;,
                                                          dtype=np.float32)&#93;)).astype(np.float32)
        if condition_video_rows is None or condition_audio_rows is None:
            condition_video_rows, condition_audio_rows = self._encode_reference_conditions(
                references, dit.patch_size, video_scheduler, seed)

    cache = getattr(dit, &quot;_adaln_cache&quot;, None)
    if cache is None or not np.array_equal(cache.timesteps.astype(np.float32), union):
        if cache is not None:
            extra = np.setdiff1d(union, cache.timesteps)
            logger.info(&quot;Recomputing AdaLN cache for %d-step ladder (extra timesteps %s).&quot;, num_steps, extra)
        if _adaln_weights_dropped(dit):
            # An earlier request on this DiT already released the AdaLN projections.
            if not resident_dit:
                raise ValueError(&quot;This H3 DiT dropped its AdaLN weights for another step ladder; &quot;
                                 &quot;pass a freshly loaded DiT for a different num_steps.&quot;)
            logger.info(&quot;Reloading the resident H3 DiT for the %d-step ladder.&quot;, num_steps)
            cache = dit = None
            self._resident_components.pop(&quot;dit&quot;, None)
            _cleanup_mlx()
            dit = self._resident_components&#91;&quot;dit&quot;&#93; = _load_resident_dit(self.dit_checkpoint)
        dit.precompute_adaln(union, drop_weights=True)

    if resident_dit:
        # The resident DiT keeps the previous request&#x27;s VSA mode unless reset to the default.
        vsa_config = vsa_config or MiniMaxH3VSAConfig()
    if vsa_config is not None:
        dit.configure_vsa(vsa_config)
    if hasattr(dit, &quot;reset_vsa_stats&quot;):
        dit.reset_vsa_stats()

    if references is not None:
        layout = build_ref2va_packed_layout(
            np.asarray(token_tags, dtype=np.int64),
            references,
            geometry&#91;&quot;latent_frame_count&quot;&#93;,
            geometry&#91;&quot;latent_height&quot;&#93;,
            geometry&#91;&quot;latent_width&quot;&#93;,
            audio_latent_num_frames(audio_frames),
            patch_size=dit.patch_size,
        )
    else:
        layout = build_packed_layout(
            len(token_tags),
            geometry&#91;&quot;latent_frame_count&quot;&#93;,
            geometry&#91;&quot;latent_height&quot;&#93;,
            geometry&#91;&quot;latent_width&quot;&#93;,
            audio_latent_num_frames(audio_frames),
            patch_size=dit.patch_size,
            text_token_tags=np.asarray(token_tags, dtype=np.int64),
            video_temporal_scale=video_temporal_scale,
        )
    if getattr(dit, &quot;vsa_config&quot;, None) is not None and dit.vsa_config.enabled:
        dit.prepare_vsa_geometry(layout)

    video_key, audio_key = mx.random.split(mx.random.key(seed))
    target_video_rows = int(layout.video_indices.shape&#91;0&#93; - layout.num_condition_video_rows)
    target_audio_rows = int(layout.audio_indices.shape&#91;0&#93; - layout.num_condition_audio_rows)
    x_v = mx.random.normal((target_video_rows, dit.patch_dim), key=video_key)
    x_a = mx.random.normal((target_audio_rows, dit.audio_in_channels), key=audio_key)
    if condition_video_rows is not None:
        if condition_video_rows.shape != (layout.num_condition_video_rows, dit.patch_dim):
            raise ValueError(f&quot;Video condition rows &#123;condition_video_rows.shape&#125; do not match the layout&#x27;s &quot;
                             f&quot;&#123;layout.num_condition_video_rows&#125; condition rows of width &#123;dit.patch_dim&#125;.&quot;)
        x_v = mx.concatenate(&#91;mx.array(condition_video_rows), x_v&#93;, axis=0)
    elif layout.num_condition_video_rows:
        raise ValueError(&quot;A reference-prefixed layout needs its encoded video condition rows.&quot;)
    if condition_audio_rows is not None:
        if condition_audio_rows.shape != (layout.num_condition_audio_rows, dit.audio_in_channels):
            raise ValueError(f&quot;Audio condition rows &#123;condition_audio_rows.shape&#125; do not match the layout&#x27;s &quot;
                             f&quot;&#123;layout.num_condition_audio_rows&#125; condition rows.&quot;)
        x_a = mx.concatenate(&#91;mx.array(condition_audio_rows), x_a&#93;, axis=0)
    elif layout.num_condition_audio_rows:
        raise ValueError(&quot;A reference-prefixed layout needs its encoded audio condition rows.&quot;)
    text = mx.array(text_rows.astype(np.float32))
    dit_forward_s = 0.0
    prefix_kv = None
    if reuse_prefix_kv:
        frozen_rows = frozen_prefix_rows(layout)
        estimated = estimate_prefix_kv_bytes(len(dit.blocks), frozen_rows, dit.num_heads, dit.head_dim,
                                             prefix_kv_storage)
        check_prefix_kv_budget(estimated, prefix_kv_cache_max_gib, rows=frozen_rows)
        prefix_kv = MiniMaxH3PrefixKVCache(len(dit.blocks), frozen_rows, dit.num_heads, dit.head_dim,
                                           prefix_kv_storage)
        logger.info(&quot;Prefix K/V reuse: %d frozen rows, ~%.2f GiB cache (%s), refresh_every=%d&quot;, frozen_rows,
                    estimated / 2**30, prefix_kv_storage, prefix_kv_refresh_every)
    recorded_steps: list&#91;int&#93; = &#91;&#93;
    reused_steps: list&#91;int&#93; = &#91;&#93;

    if pdd_plan is not None:
        # PDD targets start at the first node&#x27;s noise level (0.999 * eps),
        # as the students were trained; conditioning rows stay untouched.
        x_v = mx.concatenate(&#91;
            x_v&#91;:layout.num_condition_video_rows&#93;,
            x_v&#91;layout.num_condition_video_rows:&#93; * float(pdd_plan.node_sigmas&#91;&quot;video&quot;&#93;&#91;0&#93;),
        &#93;,
                             axis=0)
        x_a = mx.concatenate(&#91;
            x_a&#91;:layout.num_condition_audio_rows&#93;,
            x_a&#91;layout.num_condition_audio_rows:&#93; * float(pdd_plan.node_sigmas&#91;&quot;audio&quot;&#93;&#91;0&#93;),
        &#93;,
                             axis=0)

    for step_index in range(num_steps):
        video_t = float(video_scheduler.timesteps&#91;step_index&#93;)
        audio_t = float(audio_scheduler.timesteps&#91;step_index&#93;)
        unique, inverse = build_row_timesteps(
            layout,
            video_timestep=video_t,
            audio_timestep=audio_t,
            condition_video_timestep=max(video_t, MINIMAX_H3_KEYFRAME_NOISE_AUG),
            condition_audio_timestep=1.0,
        )
        step_started = time.perf_counter()
        pdd_forward_kwargs = (&#123;&#125; if pdd_plan is None else &#123;
            &quot;pdd_block&quot;: pdd_plan.block(step_index),
            &quot;pdd_integration_weights&quot;: pdd_plan.integration_weights,
        &#125;)
        record_step = prefix_kv is not None and (step_index == 0 or (prefix_kv_refresh_every &gt; 0
                                                                     and step_index % prefix_kv_refresh_every == 0))
        if prefix_kv is not None and not record_step:
            video_velocity, audio_velocity = dit.forward_target_rows(
                x_v&#91;layout.num_condition_video_rows:&#93;,
                x_a&#91;layout.num_condition_audio_rows:&#93;,
                layout=layout,
                step_timesteps=unique,
                row_timestep_inverse=inverse,
                prefix_kv=prefix_kv,
                step_index=step_index,
                **pdd_forward_kwargs,
            )
            reused_steps.append(step_index)
        else:
            if record_step:
                pdd_forward_kwargs&#91;&quot;record_prefix_kv&quot;&#93; = prefix_kv
                recorded_steps.append(step_index)
            video_velocity, audio_velocity = dit.forward_with_cache(
                x_v,
                x_a,
                text,
                layout=layout,
                step_timesteps=unique,
                row_timestep_inverse=inverse,
                step_index=step_index,
                **pdd_forward_kwargs,
            )
            video_velocity = video_velocity&#91;layout.num_condition_video_rows:&#93;
            audio_velocity = audio_velocity&#91;layout.num_condition_audio_rows:&#93;
        # Only the target rows are denoised: condition rows stay pinned at
        # the timestep the contract trained them at.
        if layout.num_condition_video_rows:
            x_v = mx.concatenate(&#91;
                x_v&#91;:layout.num_condition_video_rows&#93;,
                video_scheduler.step(video_velocity, step_index, x_v&#91;layout.num_condition_video_rows:&#93;)
            &#93;,
                                 axis=0)
        else:
            x_v = video_scheduler.step(video_velocity, step_index, x_v)
        if layout.num_condition_audio_rows:
            x_a = mx.concatenate(&#91;
                x_a&#91;:layout.num_condition_audio_rows&#93;,
                audio_scheduler.step(audio_velocity, step_index, x_a&#91;layout.num_condition_audio_rows:&#93;)
            &#93;,
                                 axis=0)
        else:
            x_a = audio_scheduler.step(audio_velocity, step_index, x_a)
        mx.eval(x_v, x_a)
        step_s = time.perf_counter() - step_started
        dit_forward_s += step_s
        self.last_denoise_step_times.append(step_s)
        logger.info(&quot;H3 denoise step %d/%d (sigma_video=%.4f, sigma_audio=%.4f) in %.2fs&quot;, step_index + 1,
                    num_steps, 1.0 - video_t, 1.0 - audio_t, step_s)
        del video_velocity, audio_velocity
        _cleanup_mlx()
        if inter_step_cooldown_s &gt; 0 and step_index &lt; num_steps - 1:
            logger.info(&quot;Thermal cooldown %.1fs before step %d...&quot;, inter_step_cooldown_s, step_index + 2)
            time.sleep(inter_step_cooldown_s)

    self.last_dit_forward_s = dit_forward_s
    self.last_prefix_kv_stats = &#123;
        &quot;enabled&quot;: prefix_kv is not None,
        &quot;refresh_every&quot;: prefix_kv_refresh_every,
        &quot;storage&quot;: prefix_kv_storage,
        &quot;frozen_rows&quot;: 0 if prefix_kv is None else prefix_kv.rows,
        &quot;cache_gib&quot;: 0.0 if prefix_kv is None else prefix_kv.nbytes / 2**30,
        &quot;max_gib&quot;: prefix_kv_cache_max_gib,
        &quot;recorded_steps&quot;: recorded_steps,
        &quot;reused_steps&quot;: reused_steps,
    &#125;
    prefix_kv = None
    stats = getattr(dit, &quot;last_vsa_stats&quot;, None)
    self.last_vsa_stats = None if stats is None else &#123;
        &quot;enabled&quot;: bool(getattr(dit.vsa_config, &quot;enabled&quot;, False)),
        &quot;configured_sparsity&quot;: stats.configured_sparsity,
        &quot;layer_sparsity&quot;: stats.layer_sparsity,
        &quot;achieved_sparsity&quot;: stats.achieved_sparsity,
        &quot;tile_size&quot;: stats.tile_size,
        &quot;prefix_mode&quot;: stats.prefix_mode,
        &quot;impl&quot;: stats.impl,
        &quot;num_prefix_tiles&quot;: stats.num_prefix_tiles,
        &quot;num_video_tiles&quot;: stats.num_video_tiles,
        &quot;video_keep&quot;: stats.video_keep,
        &quot;dense_fallback_reason&quot;: stats.dense_fallback_reason,
        &quot;attention_calls&quot;: stats.attention_calls,
        &quot;sparse_calls&quot;: stats.sparse_calls,
        &quot;impl_counts&quot;: stats.impl_counts,
        &quot;fallback_reasons&quot;: stats.fallback_reasons,
        &quot;checkpoint_vsa_capable&quot;: bool(getattr(dit, &quot;vsa_capable&quot;, False)),
    &#125;

    video_rows = np.asarray(x_v, dtype=np.float32)&#91;layout.num_condition_video_rows:&#93;
    audio_rows = np.asarray(x_a, dtype=np.float32)&#91;layout.num_condition_audio_rows:&#93;
    if owned_dit:
        del dit
        _cleanup_mlx()
    return video_rows, audio_rows

# -- phase 3a: video decode -------------------------------------------

def decode_video(self,
                 video_rows: np.ndarray,
                 *,
                 height: int,
                 width: int,
                 num_frames: int,
                 tiled: bool = True,
                 vae_tile_height: int = 256,
                 vae_tile_width: int = 256) -&gt; np.ndarray:
    &quot;&quot;&quot;Normalized packed rows -&gt; (T, H, W, 3) uint8 frames.&quot;&quot;&quot;
    _require_positive_vae_tiles(vae_tile_height, vae_tile_width)
    import mlx.core as mx

    geometry = self.resolve_geometry(height, width, num_frames, enforce_duration=False)
    if self.video_decode_backend == &quot;taeh3&quot;:
        from fastvideo.mlx_runtime.minimax_h3_taeh3 import decode_latents_taeh3_mlx

        if self._dit_in_channels != 24:
            raise ValueError(&quot;TAEH3 requires a 24-channel H3 checkpoint.&quot;)
        latents = unpatchify_video_tokens(video_rows, geometry&#91;&quot;latent_frame_count&quot;&#93;, geometry&#91;&quot;latent_height&quot;&#93;,
                                          geometry&#91;&quot;latent_width&quot;&#93;, self._dit_in_channels, self._dit_patch_size)
        pixels = decode_latents_taeh3_mlx(latents,
                                          checkpoint_path=self.taeh3_checkpoint,
                                          dtype=self.vae_dtype,
                                          chunk_size=self.taeh3_chunk_size)
        frames = (pixels&#91;0&#93; * 255.0).astype(np.uint8)
        if frames.shape != (geometry&#91;&quot;num_frames&quot;&#93;, height, width, 3):
            raise RuntimeError(f&quot;TAEH3 produced unexpected frame shape: &#123;frames.shape&#125;&quot;)
        _cleanup_mlx()
        return frames
    if self.resident:
        self.prepare_resident()
        vae = self._resident_components&#91;&quot;video_vae&quot;&#93;
    else:
        vae = self._load_video_vae(include_encoder=False)
    expected_height = height // vae.spatial_compression_ratio
    expected_width = width // vae.spatial_compression_ratio
    if (geometry&#91;&quot;latent_height&quot;&#93;, geometry&#91;&quot;latent_width&quot;&#93;) != (expected_height, expected_width):
        raise RuntimeError(&quot;H3 pipeline/VAE spatial compression mismatch: &quot;
                           f&quot;pipeline=&#123;(geometry&#91;&#x27;latent_height&#x27;&#93;, geometry&#91;&#x27;latent_width&#x27;&#93;)&#125;, &quot;
                           f&quot;VAE=&#123;(expected_height, expected_width)&#125;.&quot;)
    if vae.latent_channels != self._dit_in_channels:
        raise RuntimeError(
            f&quot;H3 DiT/VAE latent-channel mismatch: DiT=&#123;self._dit_in_channels&#125;, VAE=&#123;vae.latent_channels&#125;.&quot;)
    latents = unpatchify_video_tokens(
        video_rows,
        geometry&#91;&quot;latent_frame_count&quot;&#93;,
        geometry&#91;&quot;latent_height&quot;&#93;,
        geometry&#91;&quot;latent_width&quot;&#93;,
        vae.latent_channels,
        self._dit_patch_size,
    )
    z = mx.array(latents)
    z = vae.denormalize_latents(z)
    decoded = vae.decode(z,
                         tiled=tiled,
                         tile_sample_min_height=min(geometry&#91;&quot;height&quot;&#93;, vae_tile_height),
                         tile_sample_min_width=min(geometry&#91;&quot;width&quot;&#93;, vae_tile_width))
    pixels = np.clip(np.asarray(vae.denormalize_pixels(decoded)), 0.0, 1.0)
    del vae, decoded, z
    _cleanup_mlx()
    frames = (pixels&#91;0&#93;.transpose(1, 2, 3, 0) * 255.0).astype(np.uint8)  # (T, H, W, C)
    if frames.shape&#91;0&#93; != geometry&#91;&quot;num_frames&quot;&#93;:
        raise RuntimeError(f&quot;decoded &#123;frames.shape&#91;0&#93;&#125; frames, expected &#123;geometry&#91;&#x27;num_frames&#x27;&#93;&#125;&quot;)
    return frames

# -- phase 3b: audio decode --------------------------------------------

def decode_audio(self, audio_rows: np.ndarray, *, num_frames: int) -&gt; np.ndarray:
    &quot;&quot;&quot;Normalized packed audio rows -&gt; stereo waveform (2, S) fp32 in &#91;-1, 1&#93;.&quot;&quot;&quot;
    import mlx.core as mx

    from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir

    num_audio_latents = audio_latent_num_frames(align_num_frames(num_frames))
    latents = unpack_audio_tokens(audio_rows, num_audio_latents)
    if self.resident:
        self.prepare_resident()
        vae = self._resident_components&#91;&quot;audio_vae&quot;&#93;
    else:
        vae = mlx_h3_audio_vae_from_dir(self.model_root / &quot;audio_vae&quot;, include_encoder=False)
    z = vae.denormalize_latents(mx.array(latents))
    waveform = np.asarray(vae.decode(z))&#91;:, 0, :&#93;  # (B, 1, S) -&gt; (B, S)
    del vae, z
    _cleanup_mlx()
    # Keep audio at least as long as the final video packet. Rounding down
    # by a fractional sample makes ffmpeg&#x27;s ``-shortest`` drop frame 124.
    expected_samples = _audio_sample_count(align_num_frames(num_frames))
    if waveform.shape&#91;-1&#93; &lt; expected_samples:
        waveform = np.pad(waveform, ((0, 0), (0, expected_samples - waveform.shape&#91;-1&#93;)))
    return np.clip(waveform&#91;:, :expected_samples&#93;, -1.0, 1.0)

# -- phase 4: mux --------------------------------------------------------

def mux(self,
        frames: np.ndarray,
        waveform: np.ndarray,
        output_path: str | Path,
        fps: int = MINIMAX_H3_FPS,
        sample_rate: int = 32000) -&gt; Path:
    &quot;&quot;&quot;H.264 video + AAC stereo audio, A/V durations within one frame.&quot;&quot;&quot;
    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_video = output_path.with_suffix(&quot;.tmp.mp4&quot;)
    tmp_audio = output_path.with_suffix(&quot;.tmp.wav&quot;)

    pcm = (np.clip(waveform.T, -1.0, 1.0) * 32767.0).astype(&quot;&lt;i2&quot;)  # (S, 2)
    import wave

    try:
        with wave.open(str(tmp_audio), &quot;wb&quot;) as handle:
            handle.setnchannels(2)
            handle.setsampwidth(2)
            handle.setframerate(sample_rate)
            handle.writeframes(pcm.tobytes())

        height, width = frames.shape&#91;1:3&#93;
        ffmpeg = shutil.which(&quot;ffmpeg&quot;)
        if ffmpeg is None:
            raise RuntimeError(&quot;ffmpeg is required for MP4 muxing.&quot;)
        subprocess.run(
            &#91;
                ffmpeg,
                &quot;-y&quot;,
                &quot;-loglevel&quot;,
                &quot;error&quot;,
                &quot;-f&quot;,
                &quot;rawvideo&quot;,
                &quot;-pix_fmt&quot;,
                &quot;rgb24&quot;,
                &quot;-s&quot;,
                f&quot;&#123;width&#125;x&#123;height&#125;&quot;,
                &quot;-r&quot;,
                str(fps),
                &quot;-i&quot;,
                &quot;-&quot;,
                &quot;-i&quot;,
                str(tmp_audio),
                &quot;-c:v&quot;,
                &quot;libx264&quot;,
                &quot;-preset&quot;,
                &quot;medium&quot;,
                &quot;-crf&quot;,
                &quot;18&quot;,
                &quot;-pix_fmt&quot;,
                &quot;yuv420p&quot;,
                &quot;-c:a&quot;,
                &quot;aac&quot;,
                &quot;-b:a&quot;,
                &quot;192k&quot;,
                &quot;-shortest&quot;,
                &quot;-movflags&quot;,
                &quot;+faststart&quot;,
                str(tmp_video),
            &#93;,
            input=frames.tobytes(),
            check=True,
        )
        tmp_video.replace(output_path)
    finally:
        tmp_audio.unlink(missing_ok=True)
        tmp_video.unlink(missing_ok=True)
    return output_path

def _encode_reference_conditions(
    self,
    references: list&#91;MiniMaxH3PreparedReference&#93;,
    patch_size: tuple&#91;int, int, int&#93;,
    scheduler: MiniMaxH3SchedulerState,
    seed: int,
) -&gt; tuple&#91;np.ndarray, np.ndarray | None&#93;:
    &quot;&quot;&quot;VAE-encode the prepared references into their condition rows.

    The encoders are loaded for this call alone and released afterwards
    rather than kept on the pipeline: only a reference-conditioned request
    needs them, and they double the VAE&#x27;s resident bytes.
    &quot;&quot;&quot;
    from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir
    from fastvideo.mlx_runtime.minimax_h3_ref_conditioning import encode_reference_conditions

    if self.video_decode_backend == &quot;taeh3&quot;:
        raise ValueError(&quot;A reference-conditioned request needs the video VAE encoder; &quot;
                         &quot;use --video-decode-backend h3-vae.&quot;)

    needs_audio = any(reference.has_audio for reference in references)
    video_vae = self._load_video_vae(include_encoder=True, include_decoder=False)
    audio_vae = (mlx_h3_audio_vae_from_dir(self.model_root /
                                           &quot;audio_vae&quot;, include_encoder=True) if needs_audio else None)
    try:
        return encode_reference_conditions(
            video_vae=video_vae,
            audio_vae=audio_vae,
            references=references,
            patch_size=patch_size,
            scheduler=scheduler,
            seed=seed,
        )
    finally:
        del video_vae, audio_vae
        _cleanup_mlx()

# -- end-to-end ----------------------------------------------------------

def generate(
    self,
    prompt: str,
    *,
    output_path: str | Path,
    references: Sequence&#91;MiniMaxH3Reference&#93; | None = None,
    height: int = 480,
    width: int = 832,
    num_frames: int = 124,
    seed: int = 0,
    num_steps: int = 4,
    save_frames: bool = False,
    tiled_video_decode: bool = True,
    vae_tile_height: int = 256,
    vae_tile_width: int = 256,
    fast: bool = False,
    fast_factor: int = 2,
    fast_sharpen: float = 0.6,
    rife_weights_dir: str | Path | None = None,
    fast_spatial: bool = False,
    fast_spatial_scale: int = 2,
    fast_spatial_upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE,
    fast_spatial_sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN,
    vsa: bool | None = None,
    vsa_sparsity: float | None = None,
    vsa_tile_size: int | None = None,
    vsa_ref_policy: str | None = None,
    vsa_ref_keep_rate: float | None = None,
    vsa_prefix_mode: str = &quot;exempt&quot;,
    vsa_dense_first_n_steps: int = 0,
    vsa_dense_layers: tuple&#91;int, ...&#93; = (),
    vsa_impl: str = &quot;auto&quot;,
    inter_step_cooldown_s: float = 0.0,
    reference_max_pixels: int | None = None,
    reuse_prefix_kv: bool = False,
    prefix_kv_refresh_every: int = 0,
    prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB,
    prefix_kv_storage: str = &quot;bf16&quot;,
) -&gt; GenerationResult:
    timings: dict&#91;str, float&#93; = &#123;&#125;
    peaks: dict&#91;str, float&#93; = &#123;&#125;

    if fast_sharpen &lt; 0:
        raise ValueError(f&quot;fast_sharpen must be non-negative, got &#123;fast_sharpen&#125;.&quot;)
    if inter_step_cooldown_s &lt; 0:
        raise ValueError(f&quot;inter_step_cooldown_s must be non-negative, got &#123;inter_step_cooldown_s&#125;.&quot;)
    _require_positive_vae_tiles(vae_tile_height, vae_tile_width)
    _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
    # Unset VSA knobs come from the snapshot&#x27;s FastH3 contract; ``vsa=None``
    # turns VSA on for a VSA contract and ``vsa=False`` opts out.
    vsa_settings = resolve_vsa_settings(
        load_fasth3_inference_contract(self.model_root),
        vsa=vsa,
        sparsity=vsa_sparsity,
        tile_size=vsa_tile_size,
        ref_policy=vsa_ref_policy,
        ref_keep_rate=vsa_ref_keep_rate,
        checkpoint_capable=mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint),
    )
    vsa = vsa_settings.enabled
    vsa_config = MiniMaxH3VSAConfig(
        enabled=vsa,
        sparsity=vsa_settings.sparsity,
        tile_size=vsa_settings.tile_size,
        ref_policy=vsa_settings.ref_policy,
        ref_keep_rate=vsa_settings.ref_keep_rate,
        prefix_mode=vsa_prefix_mode,  # type: ignore&#91;arg-type&#93;
        dense_first_n_steps=vsa_dense_first_n_steps,
        dense_layers=vsa_dense_layers,
        impl=vsa_impl,  # type: ignore&#91;arg-type&#93;
    ) if vsa else MiniMaxH3VSAConfig()
    if vsa_config.enabled and not mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint):
        raise dense_only_vsa_error(self.dit_checkpoint)
    spatial_plan = plan_fast_spatial(
        height,
        width,
        scale=fast_spatial_scale,
        upsample_mode=fast_spatial_upsample_mode,
        sharpen=fast_spatial_sharpen,
    ) if fast_spatial else None
    _preflight_media_dependencies(
        fast=fast,
        fast_sharpen=fast_sharpen,
        rife_weights_dir=rife_weights_dir,
        fast_spatial=fast_spatial,
    )
    if spatial_plan is not None:
        canvas_height, canvas_width = spatial_plan.canvas_height, spatial_plan.canvas_width
    else:
        canvas_height, canvas_width = _model_canvas_size(height, width)
    target_geometry = self.resolve_geometry(canvas_height, canvas_width, num_frames)
    fast_plan = plan_fast_temporal(target_geometry&#91;&quot;num_frames&quot;&#93;, fast_factor) if fast else None
    video_num_frames = fast_plan.source_frames if fast_plan is not None else target_geometry&#91;&quot;num_frames&quot;&#93;
    video_temporal_scale = fast_plan.video_temporal_scale if fast_plan is not None else 1.0
    video_geometry = self.resolve_geometry(
        canvas_height,
        canvas_width,
        video_num_frames,
        enforce_duration=fast_plan is None,
    )
    logger.info(
        &quot;Geometry: output=%dx%dx%d model=%dx%dx%d audio_frames=%d fast=%s fast_spatial=%s&quot;,
        width,
        height,
        target_geometry&#91;&quot;num_frames&quot;&#93;,
        canvas_width,
        canvas_height,
        video_geometry&#91;&quot;num_frames&quot;&#93;,
        target_geometry&#91;&quot;num_frames&quot;&#93;,
        fast_plan,
        spatial_plan,
    )

    if self.video_decode_backend == &quot;taeh3&quot;:
        from fastvideo.mlx_runtime.minimax_h3_taeh3 import ensure_taeh3_checkpoint

        started = time.perf_counter()
        ensure_taeh3_checkpoint(self.taeh3_checkpoint)
        timings&#91;&quot;decoder_prepare_s&quot;&#93; = time.perf_counter() - started
        logger.warning(&quot;TAEH3 is an approximate preview decoder; reconstruction differs from the full H3 VAE.&quot;)

    prepared_references: list&#91;MiniMaxH3PreparedReference&#93; | None = None
    if references is not None:
        prepared_references = self.prepare_references(references,
                                                      video_geometry&#91;&quot;num_frames&quot;&#93;,
                                                      max_pixels=reference_max_pixels)

    _reset_peak_memory()
    started = time.perf_counter()
    if prepared_references is None:
        text_rows, token_tags = self.encode_prompt(prompt)
    else:
        text_rows, token_tags = self.encode_reference_prompt(prompt, prepared_references)
    timings&#91;&quot;condition_s&quot;&#93; = time.perf_counter() - started
    peaks&#91;&quot;condition_gib&quot;&#93; = _peak_memory_gib()
    _cleanup_mlx()

    _reset_peak_memory()
    started = time.perf_counter()
    video_rows, audio_rows = self.denoise(
        text_rows,
        token_tags,
        height=video_geometry&#91;&quot;height&quot;&#93;,
        width=video_geometry&#91;&quot;width&quot;&#93;,
        num_frames=video_geometry&#91;&quot;num_frames&quot;&#93;,
        audio_num_frames=target_geometry&#91;&quot;num_frames&quot;&#93; if fast_plan is not None else None,
        video_temporal_scale=video_temporal_scale,
        seed=seed,
        num_steps=num_steps,
        vsa_config=vsa_config,
        inter_step_cooldown_s=inter_step_cooldown_s,
        references=prepared_references,
        reuse_prefix_kv=reuse_prefix_kv,
        prefix_kv_refresh_every=prefix_kv_refresh_every,
        prefix_kv_cache_max_gib=prefix_kv_cache_max_gib,
        prefix_kv_storage=prefix_kv_storage,
    )
    timings&#91;&quot;denoise_s&quot;&#93; = time.perf_counter() - started
    timings&#91;&quot;dit_forward_s&quot;&#93; = float(getattr(self, &quot;last_dit_forward_s&quot;, 0.0))
    peaks&#91;&quot;denoise_gib&quot;&#93; = _peak_memory_gib()
    del text_rows
    _cleanup_mlx()

    _reset_peak_memory()
    started = time.perf_counter()
    frames = self.decode_video(
        video_rows,
        height=video_geometry&#91;&quot;height&quot;&#93;,
        width=video_geometry&#91;&quot;width&quot;&#93;,
        num_frames=video_geometry&#91;&quot;num_frames&quot;&#93;,
        tiled=tiled_video_decode,
        vae_tile_height=vae_tile_height,
        vae_tile_width=vae_tile_width,
    )
    if spatial_plan is not None:
        frames = _center_crop_frames(frames, spatial_plan.stage1_height, spatial_plan.stage1_width)
    else:
        frames = _center_crop_frames(frames, height, width)
    timings&#91;&quot;video_decode_s&quot;&#93; = time.perf_counter() - started
    peaks&#91;&quot;video_decode_gib&quot;&#93; = _peak_memory_gib()
    _cleanup_mlx()

    if fast_plan is not None:
        from fastvideo.mlx_runtime.rife_interp import interpolate_to_frame_count, load_model

        _reset_peak_memory()
        started = time.perf_counter()
        model = load_model(weights_dir=str(rife_weights_dir) if rife_weights_dir is not None else None)
        try:
            interpolated = interpolate_to_frame_count(
                frames,
                target_geometry&#91;&quot;num_frames&quot;&#93;,
                model=model,
            )
            if spatial_plan is None:
                interpolated = _sharpen_frames(interpolated, fast_sharpen)
            frames = np.stack(interpolated)
            if frames.shape&#91;0&#93; != target_geometry&#91;&quot;num_frames&quot;&#93;:
                raise RuntimeError(
                    f&quot;RIFE produced &#123;frames.shape&#91;0&#93;&#125; frames, expected &#123;target_geometry&#91;&#x27;num_frames&#x27;&#93;&#125;.&quot;)
            del interpolated
            timings&#91;&quot;rife_s&quot;&#93; = time.perf_counter() - started
            peaks&#91;&quot;rife_gib&quot;&#93; = _peak_memory_gib()
        finally:
            load_model.cache_clear()
            del model
            _cleanup_mlx()

    if spatial_plan is not None:
        started = time.perf_counter()
        # One sharpen pass, at full resolution: RIFE and the resample soften
        # for the same reason, so the stronger requested amount is applied
        # once instead of stacking two unsharp masks.
        sharpen = spatial_plan.sharpen if fast_plan is None else max(spatial_plan.sharpen, fast_sharpen)
        frames = np.stack(
            upsample_frames(
                frames,
                width=spatial_plan.target_width,
                height=spatial_plan.target_height,
                mode=spatial_plan.upsample_mode,
                sharpen=sharpen,
            ))
        timings&#91;&quot;spatial_upsample_s&quot;&#93; = time.perf_counter() - started

    _reset_peak_memory()
    started = time.perf_counter()
    waveform = self.decode_audio(audio_rows, num_frames=target_geometry&#91;&quot;num_frames&quot;&#93;)
    timings&#91;&quot;audio_decode_s&quot;&#93; = time.perf_counter() - started
    peaks&#91;&quot;audio_decode_gib&quot;&#93; = _peak_memory_gib()
    _cleanup_mlx()

    started = time.perf_counter()
    video_path = self.mux(frames, waveform, output_path)
    timings&#91;&quot;mux_s&quot;&#93; = time.perf_counter() - started
    timings&#91;&quot;generate_s&quot;&#93; = sum(
        timings.get(key, 0.0) for key in (&quot;decoder_prepare_s&quot;, &quot;condition_s&quot;, &quot;denoise_s&quot;, &quot;video_decode_s&quot;,
                                          &quot;rife_s&quot;, &quot;spatial_upsample_s&quot;, &quot;audio_decode_s&quot;, &quot;mux_s&quot;))

    result = GenerationResult(
        video_path=str(video_path),
        frames=frames if save_frames else None,
        waveform=waveform,
        sample_rate=32000,
        timings=timings,
        peak_memory_gib=peaks,
        video_decode_backend=self.video_decode_backend,
        prefix_kv=&#123;
            **getattr(self, &quot;last_prefix_kv_stats&quot;, &#123;&quot;enabled&quot;: False&#125;),
            &quot;step_times_s&quot;:
            list(getattr(self, &quot;last_denoise_step_times&quot;, &#91;&#93;)),
        &#125;,
        vsa=getattr(self, &quot;last_vsa_stats&quot;, None) or &#123;
            &quot;enabled&quot;: vsa_config.enabled,
            &quot;checkpoint_vsa_capable&quot;: mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint),
        &#125;,
    )
    logger.info(&quot;Generation complete: %s | timings=%s peaks=%s&quot;, video_path, timings, peaks)
    return result</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.resolve_geometry
resolve_geometry(height: int, width: int, num_frames: int, *, enforce_duration: bool = True) -> dict[str, int]
Explicit canvases pass through (positive multiples of 32); the
aspect-ratio resolver only applies when dimensions are omitted.
Source
@staticmethod
def resolve_geometry(
    height: int,
    width: int,
    num_frames: int,
    *,
    enforce_duration: bool = True,
) -> dict[str, int]:
    """Explicit canvases pass through (positive multiples of 32); the
    aspect-ratio resolver only applies when dimensions are omitted."""
    if height <= 0 or width <= 0 or height % 32 or width % 32:
        raise ValueError(f"H3 canvas must be positive multiples of 32, got {height}x{width}.")
    aligned_frames = align_num_frames(num_frames)
    latent_frames = video_latent_num_frames(aligned_frames)
    duration = aligned_frames / MINIMAX_H3_FPS
    if enforce_duration and not MINIMAX_H3_MIN_ALIGNED_FRAMES <= aligned_frames <= MINIMAX_H3_MAX_ALIGNED_FRAMES:
        raise ValueError(f"H3 generates {MINIMAX_H3_MIN_DURATION:g}-{MINIMAX_H3_MAX_DURATION:g} s at "
                         f"{MINIMAX_H3_FPS} fps; {aligned_frames} frames is {duration:.2f} s, outside the "
                         f"accepted {MINIMAX_H3_MIN_ALIGNED_FRAMES}-{MINIMAX_H3_MAX_ALIGNED_FRAMES} frame range.")
    return {
        "height": height,
        "width": width,
        "num_frames": aligned_frames,
        "latent_frame_count": latent_frames,
        "latent_height": height // 16,
        "latent_width": width // 16,
    }
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.prepare_resident
prepare_resident() -> None
Load the encoder, DiT, and both decoders once, before timed requests.
Source
def prepare_resident(self) -> None:
    """Load the encoder, DiT, and both decoders once, before timed requests."""
    if not self.resident or self._resident_components:
        return
    from fastvideo.mlx_runtime.minimax_h3_conditioner import ResidentNVFP4MiniMaxH3TextConditioner
    from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir
    import mlx.core as mx
try:
    conditioner = self._load_conditioner()
    if not isinstance(conditioner, ResidentNVFP4MiniMaxH3TextConditioner):
        conditioner.close()
        raise ValueError(&quot;All-resident generation requires the packed NVFP4 text encoder.&quot;)
    self._resident_components&#91;&quot;conditioner&quot;&#93; = conditioner
    self._resident_components&#91;&quot;dit&quot;&#93; = _load_resident_dit(self.dit_checkpoint)
    self._resident_components&#91;&quot;video_vae&quot;&#93; = self._load_video_vae(include_encoder=False)
    self._resident_components&#91;&quot;audio_vae&quot;&#93; = mlx_h3_audio_vae_from_dir(self.model_root / &quot;audio_vae&quot;,
                                                                       include_encoder=False)
    mx.eval(list(self._resident_components&#91;&quot;audio_vae&quot;&#93;.weights.values()))
    logger.info(&quot;H3 components resident: %.2f GiB active MLX memory&quot;, mx.get_active_memory() / 2**30)
except Exception:
    self.close()
    raise</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.close
close() -> None
Source
def close(self) -> None:
    self._restore_quant_gemm()
    conditioner = self._resident_components.get("conditioner")
    if conditioner is not None:
        conditioner.close()
    self._resident_components.clear()
    _cleanup_mlx()
    previous = getattr(self, "_previous_wired_limit", None)
    if previous is not None:
        import mlx.core as mx
    _restore_metal_wired_limit(mx, previous)
    self._previous_wired_limit = None
previous_memory = getattr(self, &quot;_previous_memory_limit&quot;, None)
if previous_memory is not None:
    import mlx.core as mx

    _restore_metal_memory_limit(mx, previous_memory)
    self._previous_memory_limit = None
previous_cache = getattr(self, &quot;_previous_cache_limit&quot;, None)
if previous_cache is not None:
    import mlx.core as mx

    mx.set_cache_limit(previous_cache)
    self._previous_cache_limit = None</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.encode_prompt
encode_prompt(prompt: str) -> tuple[np.ndarray, np.ndarray]
Returns (hidden states (S, hidden), token tags). Uses the cache or
the streamed conditioner.
Source
def encode_prompt(self, prompt: str) -> tuple[np.ndarray, np.ndarray]:
    """Returns (hidden states (S, hidden), token tags). Uses the cache or
    the streamed conditioner."""
    cache_key = self._prompt_cache_key(prompt)
    if cache_key is not None and cache_key.exists():
        data = np.load(cache_key)
        logger.info("Loaded prompt embeddings from cache %s", cache_key)
        return data["hidden_states"], data["token_tags"]
if self.resident:
    self.prepare_resident()
    conditioner = self._resident_components&#91;&quot;conditioner&quot;&#93;
else:
    conditioner = self._load_conditioner()
hidden, tags = conditioner.encode_prompt(prompt)
if not self.resident:
    conditioner.close()
_cleanup_mlx()
if cache_key is not None:
    cache_key.parent.mkdir(parents=True, exist_ok=True)
    tmp_cache = cache_key.with_name(f&quot;.&#123;cache_key.name&#125;.tmp&quot;)
    try:
        with tmp_cache.open(&quot;wb&quot;) as handle:
            np.savez(handle, hidden_states=hidden, token_tags=tags)
        tmp_cache.replace(cache_key)
    finally:
        tmp_cache.unlink(missing_ok=True)
return hidden, tags</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.load_prompt_cache
load_prompt_cache(path: str | Path) -> tuple[np.ndarray, np.ndarray]
Source
def load_prompt_cache(self, path: str | Path) -> tuple[np.ndarray, np.ndarray]:
    data = np.load(path)
    return data["hidden_states"], data["token_tags"]
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.prepare_references
prepare_references(references: Sequence[MiniMaxH3Reference], num_frames: int, *, max_pixels: int | None = None) -> list[MiniMaxH3PreparedReference]
Validate and decode the ordered reference list.
Decoding happens once, before anything is loaded, so a bad reference
fails before the DiT and the encoders are resident.
Source
def prepare_references(
    self,
    references: Sequence[MiniMaxH3Reference],
    num_frames: int,
    *,
    max_pixels: int | None = None,
) -> list[MiniMaxH3PreparedReference]:
    """Validate and decode the ordered reference list.
Decoding happens once, before anything is loaded, so a bad reference
fails before the DiT and the encoders are resident.
&quot;&quot;&quot;
validated = validate_references(list(references))
target_sample_rate = self.audio_sample_rate()
prepared = &#91;
    prepare_reference(reference, num_frames, target_sample_rate, max_pixels=max_pixels)
    for reference in validated
&#93;
logger.info(&quot;Prepared %d reference(s): %s&quot;, len(prepared),
            &quot;, &quot;.join(reference.media_type for reference in prepared))
return prepared</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.audio_sample_rate
audio_sample_rate() -> int
The audio VAE's sample rate, which reference soundtracks are resampled to.
Source
def audio_sample_rate(self) -> int:
    """The audio VAE's sample rate, which reference soundtracks are resampled to."""
    config_path = self.model_root / "audio_vae" / "config.json"
    if not config_path.is_file():
        return 32000
    return int(json.loads(config_path.read_text(encoding="utf-8")).get("sampling_rate", 32000))
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.encode_reference_prompt
encode_reference_prompt(prompt: str, references: list[MiniMaxH3PreparedReference]) -> tuple[np.ndarray, np.ndarray]
Encode the prompt with its ordered references as Qwen3-VL condition rows.
Source
def encode_reference_prompt(
    self,
    prompt: str,
    references: list[MiniMaxH3PreparedReference],
) -> tuple[np.ndarray, np.ndarray]:
    """Encode the prompt with its ordered references as Qwen3-VL condition rows."""
    from fastvideo.mlx_runtime.minimax_h3_ref_conditioning import encode_reference_prompt, load_vision_limits
    from fastvideo.mlx_runtime.minimax_h3_vision import Qwen3VLVisionConfig
if self.resident:
    self.prepare_resident()
    conditioner = self._resident_components&#91;&quot;conditioner&quot;&#93;
else:
    conditioner = self._load_conditioner()
try:
    config = Qwen3VLVisionConfig.from_config_json(self.conditioner_dir / &quot;config.json&quot;)
    min_pixels, max_pixels = load_vision_limits(self.processor_dir())
    video_min_pixels, video_max_pixels = load_vision_limits(self.processor_dir(), video=True)
    hidden, tags = encode_reference_prompt(
        conditioner,
        prompt,
        references,
        vision_config=config,
        min_pixels=min_pixels,
        max_pixels=max_pixels,
        video_min_pixels=video_min_pixels,
        video_max_pixels=video_max_pixels,
    )
finally:
    if not self.resident:
        conditioner.close()
_cleanup_mlx()
return hidden, tags</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.processor_dir
processor_dir() -> Path
Where the Qwen3-VL processor config lives, defaulting to the tokenizer dir.
Source
def processor_dir(self) -> Path:
    """Where the Qwen3-VL processor config lives, defaulting to the tokenizer dir."""
    for candidate in (self.model_root / "processor", self.tokenizer_dir):
        if (candidate / "preprocessor_config.json").is_file():
            return candidate
    return self.tokenizer_dir
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.has_conditioner_weights
has_conditioner_weights() -> bool
Source
def has_conditioner_weights(self) -> bool:
    marker = self.conditioner_dir / "model.safetensors.index.json"
    single = self.conditioner_dir / "model.safetensors"
    return marker.exists() or single.exists()
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.denoise
denoise(text_rows: np.ndarray, token_tags: np.ndarray, *, height: int, width: int, num_frames: int, audio_num_frames: int | None = None, video_temporal_scale: float = 1.0, seed: int, num_steps: int = 4, dit: Any | None = None, vsa_config: MiniMaxH3VSAConfig | None = None, inter_step_cooldown_s: float = 0.0, references: list[MiniMaxH3PreparedReference] | None = None, condition_video_rows: np.ndarray | None = None, condition_audio_rows: np.ndarray | None = None, reuse_prefix_kv: bool = False, prefix_kv_refresh_every: int = 0, prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB, prefix_kv_storage: str = 'bf16') -> tuple[np.ndarray, np.ndarray]
Denoise joint latents; returns (normalized target video rows, target audio rows).
With references the packed sequence is reference-prefixed and the
encoder rows ride along in every forward, but only the target rows are
stepped and returned: the references are conditioning, not output.
reuse_prefix_kv (Ref2VA only, lossy) records the text/reference
rows' per-block K/V on the first step and runs later steps on the
target rows alone; prefix_kv_refresh_every=R re-records every R
steps (0 never). The cache is refused above prefix_kv_cache_max_gib.
Source
def denoise(
    self,
    text_rows: np.ndarray,
    token_tags: np.ndarray,
    *,
    height: int,
    width: int,
    num_frames: int,
    audio_num_frames: int | None = None,
    video_temporal_scale: float = 1.0,
    seed: int,
    num_steps: int = 4,
    dit: Any | None = None,
    vsa_config: MiniMaxH3VSAConfig | None = None,
    inter_step_cooldown_s: float = 0.0,
    references: list[MiniMaxH3PreparedReference] | None = None,
    condition_video_rows: np.ndarray | None = None,
    condition_audio_rows: np.ndarray | None = None,
    reuse_prefix_kv: bool = False,
    prefix_kv_refresh_every: int = 0,
    prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB,
    prefix_kv_storage: str = "bf16",
) -> tuple[np.ndarray, np.ndarray]:
    """Denoise joint latents; returns (normalized target video rows, target audio rows).
With ``references`` the packed sequence is reference-prefixed and the
encoder rows ride along in every forward, but only the target rows are
stepped and returned: the references are conditioning, not output.

``reuse_prefix_kv`` (Ref2VA only, lossy) records the text/reference
rows&#x27; per-block K/V on the first step and runs later steps on the
target rows alone; ``prefix_kv_refresh_every=R`` re-records every R
steps (0 never). The cache is refused above ``prefix_kv_cache_max_gib``.
&quot;&quot;&quot;
import mlx.core as mx

self.last_prefix_kv_stats: dict&#91;str, object&#93; = &#123;&quot;enabled&quot;: False&#125;
self.last_denoise_step_times = &#91;&#93;
if prefix_kv_refresh_every &lt; 0:
    raise ValueError(f&quot;prefix_kv_refresh_every must be &gt;= 0, got &#123;prefix_kv_refresh_every&#125;.&quot;)
if reuse_prefix_kv and references is None:
    raise ValueError(&quot;Prefix K/V reuse needs Ref2VA references (a reference-prefixed packed layout).&quot;)
validate_prefix_kv_storage(prefix_kv_storage)

geometry = self.resolve_geometry(height, width, num_frames, enforce_duration=audio_num_frames is None)
audio_frames = geometry&#91;&quot;num_frames&quot;&#93; if audio_num_frames is None else align_num_frames(audio_num_frames)

resident_dit = dit is None and self.resident
if resident_dit:
    _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
    self.prepare_resident()
    dit = self._resident_components&#91;&quot;dit&quot;&#93;
owned_dit = dit is None
if owned_dit:
    _validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
    t0 = time.perf_counter()
    dit = load_mlx_h3_checkpoint(self.dit_checkpoint)
    logger.info(&quot;Loaded MLX H3 DiT from %s in %.1fs&quot;, self.dit_checkpoint, time.perf_counter() - t0)

schedule = resolve_h3_denoise_schedule(self.model_root, num_steps)
video_scheduler = schedule.video
audio_scheduler = schedule.audio
union = schedule.adaln_timesteps
num_steps = schedule.num_steps
# A FastH3 PDD export runs one fused head block per trained block; its
# plan supplies the block partition, the per-modality node sigmas
# (already on the schedulers) and the integration weights.
pdd_plan = schedule.pdd_plan
# The keyframe-noise timestep (0.999) is only exercised by FL2VA/Ref2VA
# conditioning rows. A Ref2VA PDD ladder already carries it, so a
# converted DiT&#x27;s baked cache matches; this append is then a no-op.
if references is not None:
    if video_temporal_scale != 1.0:
        # The reference-prefixed builder stretches only the target video
        # rows, which the CUDA packing never does either; refusing beats
        # silently ignoring the request.
        raise ValueError(&quot;Reference-conditioned packing has no temporal-scale stretch; &quot;
                         &quot;run Ref2VA without fast mode.&quot;)
    union = np.unique(np.concatenate(&#91;union, np.array(&#91;MINIMAX_H3_KEYFRAME_NOISE_AUG&#93;,
                                                      dtype=np.float32)&#93;)).astype(np.float32)
    if condition_video_rows is None or condition_audio_rows is None:
        condition_video_rows, condition_audio_rows = self._encode_reference_conditions(
            references, dit.patch_size, video_scheduler, seed)

cache = getattr(dit, &quot;_adaln_cache&quot;, None)
if cache is None or not np.array_equal(cache.timesteps.astype(np.float32), union):
    if cache is not None:
        extra = np.setdiff1d(union, cache.timesteps)
        logger.info(&quot;Recomputing AdaLN cache for %d-step ladder (extra timesteps %s).&quot;, num_steps, extra)
    if _adaln_weights_dropped(dit):
        # An earlier request on this DiT already released the AdaLN projections.
        if not resident_dit:
            raise ValueError(&quot;This H3 DiT dropped its AdaLN weights for another step ladder; &quot;
                             &quot;pass a freshly loaded DiT for a different num_steps.&quot;)
        logger.info(&quot;Reloading the resident H3 DiT for the %d-step ladder.&quot;, num_steps)
        cache = dit = None
        self._resident_components.pop(&quot;dit&quot;, None)
        _cleanup_mlx()
        dit = self._resident_components&#91;&quot;dit&quot;&#93; = _load_resident_dit(self.dit_checkpoint)
    dit.precompute_adaln(union, drop_weights=True)

if resident_dit:
    # The resident DiT keeps the previous request&#x27;s VSA mode unless reset to the default.
    vsa_config = vsa_config or MiniMaxH3VSAConfig()
if vsa_config is not None:
    dit.configure_vsa(vsa_config)
if hasattr(dit, &quot;reset_vsa_stats&quot;):
    dit.reset_vsa_stats()

if references is not None:
    layout = build_ref2va_packed_layout(
        np.asarray(token_tags, dtype=np.int64),
        references,
        geometry&#91;&quot;latent_frame_count&quot;&#93;,
        geometry&#91;&quot;latent_height&quot;&#93;,
        geometry&#91;&quot;latent_width&quot;&#93;,
        audio_latent_num_frames(audio_frames),
        patch_size=dit.patch_size,
    )
else:
    layout = build_packed_layout(
        len(token_tags),
        geometry&#91;&quot;latent_frame_count&quot;&#93;,
        geometry&#91;&quot;latent_height&quot;&#93;,
        geometry&#91;&quot;latent_width&quot;&#93;,
        audio_latent_num_frames(audio_frames),
        patch_size=dit.patch_size,
        text_token_tags=np.asarray(token_tags, dtype=np.int64),
        video_temporal_scale=video_temporal_scale,
    )
if getattr(dit, &quot;vsa_config&quot;, None) is not None and dit.vsa_config.enabled:
    dit.prepare_vsa_geometry(layout)

video_key, audio_key = mx.random.split(mx.random.key(seed))
target_video_rows = int(layout.video_indices.shape&#91;0&#93; - layout.num_condition_video_rows)
target_audio_rows = int(layout.audio_indices.shape&#91;0&#93; - layout.num_condition_audio_rows)
x_v = mx.random.normal((target_video_rows, dit.patch_dim), key=video_key)
x_a = mx.random.normal((target_audio_rows, dit.audio_in_channels), key=audio_key)
if condition_video_rows is not None:
    if condition_video_rows.shape != (layout.num_condition_video_rows, dit.patch_dim):
        raise ValueError(f&quot;Video condition rows &#123;condition_video_rows.shape&#125; do not match the layout&#x27;s &quot;
                         f&quot;&#123;layout.num_condition_video_rows&#125; condition rows of width &#123;dit.patch_dim&#125;.&quot;)
    x_v = mx.concatenate(&#91;mx.array(condition_video_rows), x_v&#93;, axis=0)
elif layout.num_condition_video_rows:
    raise ValueError(&quot;A reference-prefixed layout needs its encoded video condition rows.&quot;)
if condition_audio_rows is not None:
    if condition_audio_rows.shape != (layout.num_condition_audio_rows, dit.audio_in_channels):
        raise ValueError(f&quot;Audio condition rows &#123;condition_audio_rows.shape&#125; do not match the layout&#x27;s &quot;
                         f&quot;&#123;layout.num_condition_audio_rows&#125; condition rows.&quot;)
    x_a = mx.concatenate(&#91;mx.array(condition_audio_rows), x_a&#93;, axis=0)
elif layout.num_condition_audio_rows:
    raise ValueError(&quot;A reference-prefixed layout needs its encoded audio condition rows.&quot;)
text = mx.array(text_rows.astype(np.float32))
dit_forward_s = 0.0
prefix_kv = None
if reuse_prefix_kv:
    frozen_rows = frozen_prefix_rows(layout)
    estimated = estimate_prefix_kv_bytes(len(dit.blocks), frozen_rows, dit.num_heads, dit.head_dim,
                                         prefix_kv_storage)
    check_prefix_kv_budget(estimated, prefix_kv_cache_max_gib, rows=frozen_rows)
    prefix_kv = MiniMaxH3PrefixKVCache(len(dit.blocks), frozen_rows, dit.num_heads, dit.head_dim,
                                       prefix_kv_storage)
    logger.info(&quot;Prefix K/V reuse: %d frozen rows, ~%.2f GiB cache (%s), refresh_every=%d&quot;, frozen_rows,
                estimated / 2**30, prefix_kv_storage, prefix_kv_refresh_every)
recorded_steps: list&#91;int&#93; = &#91;&#93;
reused_steps: list&#91;int&#93; = &#91;&#93;

if pdd_plan is not None:
    # PDD targets start at the first node&#x27;s noise level (0.999 * eps),
    # as the students were trained; conditioning rows stay untouched.
    x_v = mx.concatenate(&#91;
        x_v&#91;:layout.num_condition_video_rows&#93;,
        x_v&#91;layout.num_condition_video_rows:&#93; * float(pdd_plan.node_sigmas&#91;&quot;video&quot;&#93;&#91;0&#93;),
    &#93;,
                         axis=0)
    x_a = mx.concatenate(&#91;
        x_a&#91;:layout.num_condition_audio_rows&#93;,
        x_a&#91;layout.num_condition_audio_rows:&#93; * float(pdd_plan.node_sigmas&#91;&quot;audio&quot;&#93;&#91;0&#93;),
    &#93;,
                         axis=0)

for step_index in range(num_steps):
    video_t = float(video_scheduler.timesteps&#91;step_index&#93;)
    audio_t = float(audio_scheduler.timesteps&#91;step_index&#93;)
    unique, inverse = build_row_timesteps(
        layout,
        video_timestep=video_t,
        audio_timestep=audio_t,
        condition_video_timestep=max(video_t, MINIMAX_H3_KEYFRAME_NOISE_AUG),
        condition_audio_timestep=1.0,
    )
    step_started = time.perf_counter()
    pdd_forward_kwargs = (&#123;&#125; if pdd_plan is None else &#123;
        &quot;pdd_block&quot;: pdd_plan.block(step_index),
        &quot;pdd_integration_weights&quot;: pdd_plan.integration_weights,
    &#125;)
    record_step = prefix_kv is not None and (step_index == 0 or (prefix_kv_refresh_every &gt; 0
                                                                 and step_index % prefix_kv_refresh_every == 0))
    if prefix_kv is not None and not record_step:
        video_velocity, audio_velocity = dit.forward_target_rows(
            x_v&#91;layout.num_condition_video_rows:&#93;,
            x_a&#91;layout.num_condition_audio_rows:&#93;,
            layout=layout,
            step_timesteps=unique,
            row_timestep_inverse=inverse,
            prefix_kv=prefix_kv,
            step_index=step_index,
            **pdd_forward_kwargs,
        )
        reused_steps.append(step_index)
    else:
        if record_step:
            pdd_forward_kwargs&#91;&quot;record_prefix_kv&quot;&#93; = prefix_kv
            recorded_steps.append(step_index)
        video_velocity, audio_velocity = dit.forward_with_cache(
            x_v,
            x_a,
            text,
            layout=layout,
            step_timesteps=unique,
            row_timestep_inverse=inverse,
            step_index=step_index,
            **pdd_forward_kwargs,
        )
        video_velocity = video_velocity&#91;layout.num_condition_video_rows:&#93;
        audio_velocity = audio_velocity&#91;layout.num_condition_audio_rows:&#93;
    # Only the target rows are denoised: condition rows stay pinned at
    # the timestep the contract trained them at.
    if layout.num_condition_video_rows:
        x_v = mx.concatenate(&#91;
            x_v&#91;:layout.num_condition_video_rows&#93;,
            video_scheduler.step(video_velocity, step_index, x_v&#91;layout.num_condition_video_rows:&#93;)
        &#93;,
                             axis=0)
    else:
        x_v = video_scheduler.step(video_velocity, step_index, x_v)
    if layout.num_condition_audio_rows:
        x_a = mx.concatenate(&#91;
            x_a&#91;:layout.num_condition_audio_rows&#93;,
            audio_scheduler.step(audio_velocity, step_index, x_a&#91;layout.num_condition_audio_rows:&#93;)
        &#93;,
                             axis=0)
    else:
        x_a = audio_scheduler.step(audio_velocity, step_index, x_a)
    mx.eval(x_v, x_a)
    step_s = time.perf_counter() - step_started
    dit_forward_s += step_s
    self.last_denoise_step_times.append(step_s)
    logger.info(&quot;H3 denoise step %d/%d (sigma_video=%.4f, sigma_audio=%.4f) in %.2fs&quot;, step_index + 1,
                num_steps, 1.0 - video_t, 1.0 - audio_t, step_s)
    del video_velocity, audio_velocity
    _cleanup_mlx()
    if inter_step_cooldown_s &gt; 0 and step_index &lt; num_steps - 1:
        logger.info(&quot;Thermal cooldown %.1fs before step %d...&quot;, inter_step_cooldown_s, step_index + 2)
        time.sleep(inter_step_cooldown_s)

self.last_dit_forward_s = dit_forward_s
self.last_prefix_kv_stats = &#123;
    &quot;enabled&quot;: prefix_kv is not None,
    &quot;refresh_every&quot;: prefix_kv_refresh_every,
    &quot;storage&quot;: prefix_kv_storage,
    &quot;frozen_rows&quot;: 0 if prefix_kv is None else prefix_kv.rows,
    &quot;cache_gib&quot;: 0.0 if prefix_kv is None else prefix_kv.nbytes / 2**30,
    &quot;max_gib&quot;: prefix_kv_cache_max_gib,
    &quot;recorded_steps&quot;: recorded_steps,
    &quot;reused_steps&quot;: reused_steps,
&#125;
prefix_kv = None
stats = getattr(dit, &quot;last_vsa_stats&quot;, None)
self.last_vsa_stats = None if stats is None else &#123;
    &quot;enabled&quot;: bool(getattr(dit.vsa_config, &quot;enabled&quot;, False)),
    &quot;configured_sparsity&quot;: stats.configured_sparsity,
    &quot;layer_sparsity&quot;: stats.layer_sparsity,
    &quot;achieved_sparsity&quot;: stats.achieved_sparsity,
    &quot;tile_size&quot;: stats.tile_size,
    &quot;prefix_mode&quot;: stats.prefix_mode,
    &quot;impl&quot;: stats.impl,
    &quot;num_prefix_tiles&quot;: stats.num_prefix_tiles,
    &quot;num_video_tiles&quot;: stats.num_video_tiles,
    &quot;video_keep&quot;: stats.video_keep,
    &quot;dense_fallback_reason&quot;: stats.dense_fallback_reason,
    &quot;attention_calls&quot;: stats.attention_calls,
    &quot;sparse_calls&quot;: stats.sparse_calls,
    &quot;impl_counts&quot;: stats.impl_counts,
    &quot;fallback_reasons&quot;: stats.fallback_reasons,
    &quot;checkpoint_vsa_capable&quot;: bool(getattr(dit, &quot;vsa_capable&quot;, False)),
&#125;

video_rows = np.asarray(x_v, dtype=np.float32)&#91;layout.num_condition_video_rows:&#93;
audio_rows = np.asarray(x_a, dtype=np.float32)&#91;layout.num_condition_audio_rows:&#93;
if owned_dit:
    del dit
    _cleanup_mlx()
return video_rows, audio_rows</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.decode_video
decode_video(video_rows: np.ndarray, *, height: int, width: int, num_frames: int, tiled: bool = True, vae_tile_height: int = 256, vae_tile_width: int = 256) -> np.ndarray
Normalized packed rows -> (T, H, W, 3) uint8 frames.
Source
def decode_video(self,
                 video_rows: np.ndarray,
                 *,
                 height: int,
                 width: int,
                 num_frames: int,
                 tiled: bool = True,
                 vae_tile_height: int = 256,
                 vae_tile_width: int = 256) -> np.ndarray:
    """Normalized packed rows -> (T, H, W, 3) uint8 frames."""
    _require_positive_vae_tiles(vae_tile_height, vae_tile_width)
    import mlx.core as mx
geometry = self.resolve_geometry(height, width, num_frames, enforce_duration=False)
if self.video_decode_backend == &quot;taeh3&quot;:
    from fastvideo.mlx_runtime.minimax_h3_taeh3 import decode_latents_taeh3_mlx

    if self._dit_in_channels != 24:
        raise ValueError(&quot;TAEH3 requires a 24-channel H3 checkpoint.&quot;)
    latents = unpatchify_video_tokens(video_rows, geometry&#91;&quot;latent_frame_count&quot;&#93;, geometry&#91;&quot;latent_height&quot;&#93;,
                                      geometry&#91;&quot;latent_width&quot;&#93;, self._dit_in_channels, self._dit_patch_size)
    pixels = decode_latents_taeh3_mlx(latents,
                                      checkpoint_path=self.taeh3_checkpoint,
                                      dtype=self.vae_dtype,
                                      chunk_size=self.taeh3_chunk_size)
    frames = (pixels&#91;0&#93; * 255.0).astype(np.uint8)
    if frames.shape != (geometry&#91;&quot;num_frames&quot;&#93;, height, width, 3):
        raise RuntimeError(f&quot;TAEH3 produced unexpected frame shape: &#123;frames.shape&#125;&quot;)
    _cleanup_mlx()
    return frames
if self.resident:
    self.prepare_resident()
    vae = self._resident_components&#91;&quot;video_vae&quot;&#93;
else:
    vae = self._load_video_vae(include_encoder=False)
expected_height = height // vae.spatial_compression_ratio
expected_width = width // vae.spatial_compression_ratio
if (geometry&#91;&quot;latent_height&quot;&#93;, geometry&#91;&quot;latent_width&quot;&#93;) != (expected_height, expected_width):
    raise RuntimeError(&quot;H3 pipeline/VAE spatial compression mismatch: &quot;
                       f&quot;pipeline=&#123;(geometry&#91;&#x27;latent_height&#x27;&#93;, geometry&#91;&#x27;latent_width&#x27;&#93;)&#125;, &quot;
                       f&quot;VAE=&#123;(expected_height, expected_width)&#125;.&quot;)
if vae.latent_channels != self._dit_in_channels:
    raise RuntimeError(
        f&quot;H3 DiT/VAE latent-channel mismatch: DiT=&#123;self._dit_in_channels&#125;, VAE=&#123;vae.latent_channels&#125;.&quot;)
latents = unpatchify_video_tokens(
    video_rows,
    geometry&#91;&quot;latent_frame_count&quot;&#93;,
    geometry&#91;&quot;latent_height&quot;&#93;,
    geometry&#91;&quot;latent_width&quot;&#93;,
    vae.latent_channels,
    self._dit_patch_size,
)
z = mx.array(latents)
z = vae.denormalize_latents(z)
decoded = vae.decode(z,
                     tiled=tiled,
                     tile_sample_min_height=min(geometry&#91;&quot;height&quot;&#93;, vae_tile_height),
                     tile_sample_min_width=min(geometry&#91;&quot;width&quot;&#93;, vae_tile_width))
pixels = np.clip(np.asarray(vae.denormalize_pixels(decoded)), 0.0, 1.0)
del vae, decoded, z
_cleanup_mlx()
frames = (pixels&#91;0&#93;.transpose(1, 2, 3, 0) * 255.0).astype(np.uint8)  # (T, H, W, C)
if frames.shape&#91;0&#93; != geometry&#91;&quot;num_frames&quot;&#93;:
    raise RuntimeError(f&quot;decoded &#123;frames.shape&#91;0&#93;&#125; frames, expected &#123;geometry&#91;&#x27;num_frames&#x27;&#93;&#125;&quot;)
return frames</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.decode_audio
decode_audio(audio_rows: np.ndarray, *, num_frames: int) -> np.ndarray
Normalized packed audio rows -> stereo waveform (2, S) fp32 in [-1, 1].
Source
def decode_audio(self, audio_rows: np.ndarray, *, num_frames: int) -> np.ndarray:
    """Normalized packed audio rows -> stereo waveform (2, S) fp32 in [-1, 1]."""
    import mlx.core as mx
from fastvideo.mlx_runtime.minimax_h3_audio_vae import mlx_h3_audio_vae_from_dir

num_audio_latents = audio_latent_num_frames(align_num_frames(num_frames))
latents = unpack_audio_tokens(audio_rows, num_audio_latents)
if self.resident:
    self.prepare_resident()
    vae = self._resident_components&#91;&quot;audio_vae&quot;&#93;
else:
    vae = mlx_h3_audio_vae_from_dir(self.model_root / &quot;audio_vae&quot;, include_encoder=False)
z = vae.denormalize_latents(mx.array(latents))
waveform = np.asarray(vae.decode(z))&#91;:, 0, :&#93;  # (B, 1, S) -&gt; (B, S)
del vae, z
_cleanup_mlx()
# Keep audio at least as long as the final video packet. Rounding down
# by a fractional sample makes ffmpeg&#x27;s ``-shortest`` drop frame 124.
expected_samples = _audio_sample_count(align_num_frames(num_frames))
if waveform.shape&#91;-1&#93; &lt; expected_samples:
    waveform = np.pad(waveform, ((0, 0), (0, expected_samples - waveform.shape&#91;-1&#93;)))
return np.clip(waveform&#91;:, :expected_samples&#93;, -1.0, 1.0)</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.mux
mux(frames: np.ndarray, waveform: np.ndarray, output_path: str | Path, fps: int = MINIMAX_H3_FPS, sample_rate: int = 32000) -> Path
H.264 video + AAC stereo audio, A/V durations within one frame.
Source
def mux(self,
        frames: np.ndarray,
        waveform: np.ndarray,
        output_path: str | Path,
        fps: int = MINIMAX_H3_FPS,
        sample_rate: int = 32000) -> Path:
    """H.264 video + AAC stereo audio, A/V durations within one frame."""
    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_video = output_path.with_suffix(".tmp.mp4")
    tmp_audio = output_path.with_suffix(".tmp.wav")
pcm = (np.clip(waveform.T, -1.0, 1.0) * 32767.0).astype(&quot;&lt;i2&quot;)  # (S, 2)
import wave

try:
    with wave.open(str(tmp_audio), &quot;wb&quot;) as handle:
        handle.setnchannels(2)
        handle.setsampwidth(2)
        handle.setframerate(sample_rate)
        handle.writeframes(pcm.tobytes())

    height, width = frames.shape&#91;1:3&#93;
    ffmpeg = shutil.which(&quot;ffmpeg&quot;)
    if ffmpeg is None:
        raise RuntimeError(&quot;ffmpeg is required for MP4 muxing.&quot;)
    subprocess.run(
        &#91;
            ffmpeg,
            &quot;-y&quot;,
            &quot;-loglevel&quot;,
            &quot;error&quot;,
            &quot;-f&quot;,
            &quot;rawvideo&quot;,
            &quot;-pix_fmt&quot;,
            &quot;rgb24&quot;,
            &quot;-s&quot;,
            f&quot;&#123;width&#125;x&#123;height&#125;&quot;,
            &quot;-r&quot;,
            str(fps),
            &quot;-i&quot;,
            &quot;-&quot;,
            &quot;-i&quot;,
            str(tmp_audio),
            &quot;-c:v&quot;,
            &quot;libx264&quot;,
            &quot;-preset&quot;,
            &quot;medium&quot;,
            &quot;-crf&quot;,
            &quot;18&quot;,
            &quot;-pix_fmt&quot;,
            &quot;yuv420p&quot;,
            &quot;-c:a&quot;,
            &quot;aac&quot;,
            &quot;-b:a&quot;,
            &quot;192k&quot;,
            &quot;-shortest&quot;,
            &quot;-movflags&quot;,
            &quot;+faststart&quot;,
            str(tmp_video),
        &#93;,
        input=frames.tobytes(),
        check=True,
    )
    tmp_video.replace(output_path)
finally:
    tmp_audio.unlink(missing_ok=True)
    tmp_video.unlink(missing_ok=True)
return output_path</code></pre></details>
fastvideo.mlx_runtime.minimax_h3_pipeline.MiniMaxH3MLXPipeline.generate
generate(prompt: str, *, output_path: str | Path, references: Sequence[MiniMaxH3Reference] | None = None, height: int = 480, width: int = 832, num_frames: int = 124, seed: int = 0, num_steps: int = 4, save_frames: bool = False, tiled_video_decode: bool = True, vae_tile_height: int = 256, vae_tile_width: int = 256, fast: bool = False, fast_factor: int = 2, fast_sharpen: float = 0.6, rife_weights_dir: str | Path | None = None, fast_spatial: bool = False, fast_spatial_scale: int = 2, fast_spatial_upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE, fast_spatial_sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN, vsa: bool | None = None, vsa_sparsity: float | None = None, vsa_tile_size: int | None = None, vsa_ref_policy: str | None = None, vsa_ref_keep_rate: float | None = None, vsa_prefix_mode: str = 'exempt', vsa_dense_first_n_steps: int = 0, vsa_dense_layers: tuple[int, ...] = (), vsa_impl: str = 'auto', inter_step_cooldown_s: float = 0.0, reference_max_pixels: int | None = None, reuse_prefix_kv: bool = False, prefix_kv_refresh_every: int = 0, prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB, prefix_kv_storage: str = 'bf16') -> GenerationResult
Source
def generate(
    self,
    prompt: str,
    *,
    output_path: str | Path,
    references: Sequence[MiniMaxH3Reference] | None = None,
    height: int = 480,
    width: int = 832,
    num_frames: int = 124,
    seed: int = 0,
    num_steps: int = 4,
    save_frames: bool = False,
    tiled_video_decode: bool = True,
    vae_tile_height: int = 256,
    vae_tile_width: int = 256,
    fast: bool = False,
    fast_factor: int = 2,
    fast_sharpen: float = 0.6,
    rife_weights_dir: str | Path | None = None,
    fast_spatial: bool = False,
    fast_spatial_scale: int = 2,
    fast_spatial_upsample_mode: str = DEFAULT_PIXEL_UPSAMPLE_MODE,
    fast_spatial_sharpen: float = DEFAULT_FAST_SPATIAL_SHARPEN,
    vsa: bool | None = None,
    vsa_sparsity: float | None = None,
    vsa_tile_size: int | None = None,
    vsa_ref_policy: str | None = None,
    vsa_ref_keep_rate: float | None = None,
    vsa_prefix_mode: str = "exempt",
    vsa_dense_first_n_steps: int = 0,
    vsa_dense_layers: tuple[int, ...] = (),
    vsa_impl: str = "auto",
    inter_step_cooldown_s: float = 0.0,
    reference_max_pixels: int | None = None,
    reuse_prefix_kv: bool = False,
    prefix_kv_refresh_every: int = 0,
    prefix_kv_cache_max_gib: float = DEFAULT_PREFIX_KV_MAX_GIB,
    prefix_kv_storage: str = "bf16",
) -> GenerationResult:
    timings: dict[str, float] = {}
    peaks: dict[str, float] = {}
if fast_sharpen &lt; 0:
    raise ValueError(f&quot;fast_sharpen must be non-negative, got &#123;fast_sharpen&#125;.&quot;)
if inter_step_cooldown_s &lt; 0:
    raise ValueError(f&quot;inter_step_cooldown_s must be non-negative, got &#123;inter_step_cooldown_s&#125;.&quot;)
_require_positive_vae_tiles(vae_tile_height, vae_tile_width)
_validate_checkpoint_step_ladder(self.dit_checkpoint, num_steps, model_root=self.model_root)
# Unset VSA knobs come from the snapshot&#x27;s FastH3 contract; ``vsa=None``
# turns VSA on for a VSA contract and ``vsa=False`` opts out.
vsa_settings = resolve_vsa_settings(
    load_fasth3_inference_contract(self.model_root),
    vsa=vsa,
    sparsity=vsa_sparsity,
    tile_size=vsa_tile_size,
    ref_policy=vsa_ref_policy,
    ref_keep_rate=vsa_ref_keep_rate,
    checkpoint_capable=mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint),
)
vsa = vsa_settings.enabled
vsa_config = MiniMaxH3VSAConfig(
    enabled=vsa,
    sparsity=vsa_settings.sparsity,
    tile_size=vsa_settings.tile_size,
    ref_policy=vsa_settings.ref_policy,
    ref_keep_rate=vsa_settings.ref_keep_rate,
    prefix_mode=vsa_prefix_mode,  # type: ignore&#91;arg-type&#93;
    dense_first_n_steps=vsa_dense_first_n_steps,
    dense_layers=vsa_dense_layers,
    impl=vsa_impl,  # type: ignore&#91;arg-type&#93;
) if vsa else MiniMaxH3VSAConfig()
if vsa_config.enabled and not mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint):
    raise dense_only_vsa_error(self.dit_checkpoint)
spatial_plan = plan_fast_spatial(
    height,
    width,
    scale=fast_spatial_scale,
    upsample_mode=fast_spatial_upsample_mode,
    sharpen=fast_spatial_sharpen,
) if fast_spatial else None
_preflight_media_dependencies(
    fast=fast,
    fast_sharpen=fast_sharpen,
    rife_weights_dir=rife_weights_dir,
    fast_spatial=fast_spatial,
)
if spatial_plan is not None:
    canvas_height, canvas_width = spatial_plan.canvas_height, spatial_plan.canvas_width
else:
    canvas_height, canvas_width = _model_canvas_size(height, width)
target_geometry = self.resolve_geometry(canvas_height, canvas_width, num_frames)
fast_plan = plan_fast_temporal(target_geometry&#91;&quot;num_frames&quot;&#93;, fast_factor) if fast else None
video_num_frames = fast_plan.source_frames if fast_plan is not None else target_geometry&#91;&quot;num_frames&quot;&#93;
video_temporal_scale = fast_plan.video_temporal_scale if fast_plan is not None else 1.0
video_geometry = self.resolve_geometry(
    canvas_height,
    canvas_width,
    video_num_frames,
    enforce_duration=fast_plan is None,
)
logger.info(
    &quot;Geometry: output=%dx%dx%d model=%dx%dx%d audio_frames=%d fast=%s fast_spatial=%s&quot;,
    width,
    height,
    target_geometry&#91;&quot;num_frames&quot;&#93;,
    canvas_width,
    canvas_height,
    video_geometry&#91;&quot;num_frames&quot;&#93;,
    target_geometry&#91;&quot;num_frames&quot;&#93;,
    fast_plan,
    spatial_plan,
)

if self.video_decode_backend == &quot;taeh3&quot;:
    from fastvideo.mlx_runtime.minimax_h3_taeh3 import ensure_taeh3_checkpoint

    started = time.perf_counter()
    ensure_taeh3_checkpoint(self.taeh3_checkpoint)
    timings&#91;&quot;decoder_prepare_s&quot;&#93; = time.perf_counter() - started
    logger.warning(&quot;TAEH3 is an approximate preview decoder; reconstruction differs from the full H3 VAE.&quot;)

prepared_references: list&#91;MiniMaxH3PreparedReference&#93; | None = None
if references is not None:
    prepared_references = self.prepare_references(references,
                                                  video_geometry&#91;&quot;num_frames&quot;&#93;,
                                                  max_pixels=reference_max_pixels)

_reset_peak_memory()
started = time.perf_counter()
if prepared_references is None:
    text_rows, token_tags = self.encode_prompt(prompt)
else:
    text_rows, token_tags = self.encode_reference_prompt(prompt, prepared_references)
timings&#91;&quot;condition_s&quot;&#93; = time.perf_counter() - started
peaks&#91;&quot;condition_gib&quot;&#93; = _peak_memory_gib()
_cleanup_mlx()

_reset_peak_memory()
started = time.perf_counter()
video_rows, audio_rows = self.denoise(
    text_rows,
    token_tags,
    height=video_geometry&#91;&quot;height&quot;&#93;,
    width=video_geometry&#91;&quot;width&quot;&#93;,
    num_frames=video_geometry&#91;&quot;num_frames&quot;&#93;,
    audio_num_frames=target_geometry&#91;&quot;num_frames&quot;&#93; if fast_plan is not None else None,
    video_temporal_scale=video_temporal_scale,
    seed=seed,
    num_steps=num_steps,
    vsa_config=vsa_config,
    inter_step_cooldown_s=inter_step_cooldown_s,
    references=prepared_references,
    reuse_prefix_kv=reuse_prefix_kv,
    prefix_kv_refresh_every=prefix_kv_refresh_every,
    prefix_kv_cache_max_gib=prefix_kv_cache_max_gib,
    prefix_kv_storage=prefix_kv_storage,
)
timings&#91;&quot;denoise_s&quot;&#93; = time.perf_counter() - started
timings&#91;&quot;dit_forward_s&quot;&#93; = float(getattr(self, &quot;last_dit_forward_s&quot;, 0.0))
peaks&#91;&quot;denoise_gib&quot;&#93; = _peak_memory_gib()
del text_rows
_cleanup_mlx()

_reset_peak_memory()
started = time.perf_counter()
frames = self.decode_video(
    video_rows,
    height=video_geometry&#91;&quot;height&quot;&#93;,
    width=video_geometry&#91;&quot;width&quot;&#93;,
    num_frames=video_geometry&#91;&quot;num_frames&quot;&#93;,
    tiled=tiled_video_decode,
    vae_tile_height=vae_tile_height,
    vae_tile_width=vae_tile_width,
)
if spatial_plan is not None:
    frames = _center_crop_frames(frames, spatial_plan.stage1_height, spatial_plan.stage1_width)
else:
    frames = _center_crop_frames(frames, height, width)
timings&#91;&quot;video_decode_s&quot;&#93; = time.perf_counter() - started
peaks&#91;&quot;video_decode_gib&quot;&#93; = _peak_memory_gib()
_cleanup_mlx()

if fast_plan is not None:
    from fastvideo.mlx_runtime.rife_interp import interpolate_to_frame_count, load_model

    _reset_peak_memory()
    started = time.perf_counter()
    model = load_model(weights_dir=str(rife_weights_dir) if rife_weights_dir is not None else None)
    try:
        interpolated = interpolate_to_frame_count(
            frames,
            target_geometry&#91;&quot;num_frames&quot;&#93;,
            model=model,
        )
        if spatial_plan is None:
            interpolated = _sharpen_frames(interpolated, fast_sharpen)
        frames = np.stack(interpolated)
        if frames.shape&#91;0&#93; != target_geometry&#91;&quot;num_frames&quot;&#93;:
            raise RuntimeError(
                f&quot;RIFE produced &#123;frames.shape&#91;0&#93;&#125; frames, expected &#123;target_geometry&#91;&#x27;num_frames&#x27;&#93;&#125;.&quot;)
        del interpolated
        timings&#91;&quot;rife_s&quot;&#93; = time.perf_counter() - started
        peaks&#91;&quot;rife_gib&quot;&#93; = _peak_memory_gib()
    finally:
        load_model.cache_clear()
        del model
        _cleanup_mlx()

if spatial_plan is not None:
    started = time.perf_counter()
    # One sharpen pass, at full resolution: RIFE and the resample soften
    # for the same reason, so the stronger requested amount is applied
    # once instead of stacking two unsharp masks.
    sharpen = spatial_plan.sharpen if fast_plan is None else max(spatial_plan.sharpen, fast_sharpen)
    frames = np.stack(
        upsample_frames(
            frames,
            width=spatial_plan.target_width,
            height=spatial_plan.target_height,
            mode=spatial_plan.upsample_mode,
            sharpen=sharpen,
        ))
    timings&#91;&quot;spatial_upsample_s&quot;&#93; = time.perf_counter() - started

_reset_peak_memory()
started = time.perf_counter()
waveform = self.decode_audio(audio_rows, num_frames=target_geometry&#91;&quot;num_frames&quot;&#93;)
timings&#91;&quot;audio_decode_s&quot;&#93; = time.perf_counter() - started
peaks&#91;&quot;audio_decode_gib&quot;&#93; = _peak_memory_gib()
_cleanup_mlx()

started = time.perf_counter()
video_path = self.mux(frames, waveform, output_path)
timings&#91;&quot;mux_s&quot;&#93; = time.perf_counter() - started
timings&#91;&quot;generate_s&quot;&#93; = sum(
    timings.get(key, 0.0) for key in (&quot;decoder_prepare_s&quot;, &quot;condition_s&quot;, &quot;denoise_s&quot;, &quot;video_decode_s&quot;,
                                      &quot;rife_s&quot;, &quot;spatial_upsample_s&quot;, &quot;audio_decode_s&quot;, &quot;mux_s&quot;))

result = GenerationResult(
    video_path=str(video_path),
    frames=frames if save_frames else None,
    waveform=waveform,
    sample_rate=32000,
    timings=timings,
    peak_memory_gib=peaks,
    video_decode_backend=self.video_decode_backend,
    prefix_kv=&#123;
        **getattr(self, &quot;last_prefix_kv_stats&quot;, &#123;&quot;enabled&quot;: False&#125;),
        &quot;step_times_s&quot;:
        list(getattr(self, &quot;last_denoise_step_times&quot;, &#91;&#93;)),
    &#125;,
    vsa=getattr(self, &quot;last_vsa_stats&quot;, None) or &#123;
        &quot;enabled&quot;: vsa_config.enabled,
        &quot;checkpoint_vsa_capable&quot;: mlx_h3_checkpoint_vsa_capable(self.dit_checkpoint),
    &#125;,
)
logger.info(&quot;Generation complete: %s | timings=%s peaks=%s&quot;, video_path, timings, peaks)
return result</code></pre></details>