Spaces:

FastVideo
/

FastH3-Omni

Running on Zero

Tile-64 attention and a persistent kernel cache

798a242 verified 2 days ago

18.6 kB

"""FastH3-Omni · reference-to-video (+audio) on ZeroGPU.
Ordered image, video and audio references plus a prompt become a short video
with a matching 32 kHz soundtrack in eight fused forwards of the FastVideo
FastH3 OmniRef student. A plain prompt is wrapped into the model's six-section
H3 format; "Upsample prompt" rewrites it with the MiniMax-H3 prompt rewriter
instead. See fasth3_engine.py for the ZeroGPU + FastVideo integration notes.
"""
import contextvars
import os
import tempfile
from functools import cache
from pathlib import Path

import gradio as gr
from PIL import Image

try:
    import spaces
    IS_ZERO = True
except ImportError:  # local / CI
    IS_ZERO = False

    class _NoSpaces:
        @staticmethod
        def GPU(*a, **k):
            if a and callable(a[0]):
                return a[0]
            return lambda f: f

    spaces = _NoSpaces()  # type: ignore

from fasth3_engine import FastH3OmniRefEngine, MODEL_REPO, prefetch

REWRITER_SPACE = os.getenv("FASTH3_REWRITER", "hugging-apps/minimax-h3-prompt-rewriter")

CANVASES = {
    "1344x768 · 16:9 recommended": (768, 1344),
    "832x480 · 16:9 fast": (480, 832),
    "1280x704 · 16:9": (704, 1280),
    "960x544 · 16:9": (544, 960),
    "768x1344 · 9:16 recommended": (1344, 768),
    "480x832 · 9:16 fast": (832, 480),
    "768x768 · 1:1": (768, 768),
    "1024x1024 · 1:1": (1024, 1024),
    "1024x768 · 4:3": (768, 1024),
    "768x1024 · 3:4": (1024, 768),
    "1536x672 · 21:9": (672, 1536),
}
DEFAULT_CANVAS = "1344x768 · 16:9 recommended"
PORTRAIT_CANVAS = "768x1344 · 9:16 recommended"
FPS, FRAMES_PER_CHUNK, LATENTS_PER_CHUNK = 24, 17, 5
MIN_FRAMES, MAX_FRAMES = 124, 362
MIN_DURATION, MAX_UI_DURATION = 5, 14
MIN_REFERENCE_VIDEO, MAX_REFERENCE_VIDEO = 2.0, 15.0
MAX_IMAGE_SLOTS, OPEN_IMAGE_SLOTS = 9, 2
REWRITER_IMAGE_SLOTS = 4
DEFAULT_PIXEL_FRAMES = 1344 * 768 * 124
GPU_SECONDS_WARM = int(os.getenv("FASTH3_GPU_SECONDS", "170"))
GPU_SECONDS_COLD = int(os.getenv("FASTH3_GPU_SECONDS_COLD", "360"))
GPU_SECONDS_PER_EXTRA_IMAGE, GPU_SECONDS_PER_VIDEO = 110, 160

APP_DIR = Path(__file__).resolve().parent
EXAMPLES_DIR = APP_DIR / "examples"
LIMITS = {"image": 9, "video": 3, "audio": 3}
REWRITER_LIMITS = {"image": 4, "video": 1, "audio": 1}

EXAMPLES = [
    {
        "title": "Beach promenade at golden hour",
        "image": "woman.jpg",
        "desc": "a young woman with a dark curly afro and gold hoop earrings, wearing a white t-shirt",
        "scene": "walks along a sunlit beach promenade at golden hour, smiling and glancing toward the camera, "
        "palm trees and the ocean behind her.",
        "sound": "ocean waves, a gentle breeze and distant seagulls",
        "music": "none",
    },
    {
        "title": "Dog bounding through snow",
        "image": "dog.jpg",
        "desc": "a pale cream, fluffy long-haired dog with floppy ears and a dark nose",
        "scene": "bounds across a snowy field in bright winter sunlight, kicking up powder snow, then shakes itself off.",
        "sound": "crunching snow, happy panting and a cold wind",
        "music": "none",
    },
    {
        "title": "Birthday cake with candles",
        "image": "cake.jpg",
        "desc": "a whole chocolate layer cake with chocolate frosting, pastel sprinkles and chocolate shards on top",
        "scene": "sits at the center of a festive birthday table in a warm, dim room as lit candles flicker and "
        "colorful confetti drifts down around it.",
        "sound": "soft party chatter and a faint crackle",
        "music": "cheerful, upbeat birthday music",
    },
]


def composer_examples():
    rows = []
    for example in EXAMPLES:
        path = str(EXAMPLES_DIR / example["image"])
        rows.append({
            "title": example["title"],
            "refs": [{"kind": "image", "path": path, "url": f"/gradio_api/file={path}", "name": example["image"],
                      "desc": example["desc"]}],
            "builder": {"scene": example["scene"], "sound": example["sound"], "music": example["music"],
                        "style": "realistic photographic"},
        })
    return rows


def snap_frames(seconds: float) -> int:
    frames = max(1, round(float(seconds) * FPS))
    while frames % FRAMES_PER_CHUNK != LATENTS_PER_CHUNK:
        frames += 1
    return frames


def reference_labels(references):
    names = {"image": "Picture", "video": "Video", "audio": "Audio"}
    counts, labels = {}, []
    for kind, _ in references:
        counts[kind] = counts.get(kind, 0) + 1
        labels.append((kind, f"<{names[kind]} {counts[kind]}>"))
    return labels


def to_h3_prompt(prompt: str, references) -> str:
    """Wrap a plain prompt into the six-section H3 format the checkpoint was trained on."""
    prompt = (prompt or "").strip()
    if "subject_definitions:" in prompt:
        return prompt
    labels = reference_labels(references)
    visual = " and ".join(label for kind, label in labels if kind != "audio") or "the references"
    audio = [label for kind, label in labels if kind == "audio"]
    sound = (f"the voice and sound of {' and '.join(audio)}, in a natural acoustic space matching the scene"
             if audio else "natural ambient sound that matches the scene")
    return (
        f"subject_definitions:\n<Subject 1> is the subject shown in {visual}.\n\n"
        f"summary:\n{prompt}\n\n"
        f"retention_analysis:\n<Subject 1>: fully_preserved - the reference subject's identity, "
        f"appearance and details are kept.\n\n"
        f"detailed_description:\nThe target video is in a realistic photographic style.\n"
        f"[Shot 1] {prompt}\n\n"
        f"overall_soundscape:\n{sound}.\n\n"
        f"non_diegetic_music:\nnone."
    )


# --- engine: built once at startup, on ZeroGPU's emulated CUDA ------------
# The weights land on "cuda" in the main process; ZeroGPU packs them and maps
# them onto the real card at the start of every GPU call, so no call rebuilds.
from fasth3_engine import fetch_kernel_cache, zerogpu_main_process_shims

prefetch()
KERNELS_CACHED = fetch_kernel_cache()
if IS_ZERO:
    zerogpu_main_process_shims()
ENGINE = FastH3OmniRefEngine()
ENGINE.initialize()
_COMPLETED_GPU_CALLS = 0


def probe(path: str):
    import av

    def seconds(stream, container):
        if stream.duration is not None and stream.time_base is not None:
            return float(stream.duration * stream.time_base)
        return None if container.duration is None else container.duration / av.time_base

    with av.open(path) as container:
        video = seconds(container.streams.video[0], container) if container.streams.video else None
        audio = seconds(container.streams.audio[0], container) if container.streams.audio else None
    return video, audio


def _allowed(path: str) -> bool:
    from gradio.utils import get_upload_folder

    real = os.path.realpath(path)
    roots = [os.path.realpath(get_upload_folder()), os.path.realpath(EXAMPLES_DIR)]
    return any(real == root or real.startswith(root + os.sep) for root in roots) and os.path.isfile(real)


def references_from(composer: dict):
    refs = (composer or {}).get("refs") or []
    if any(ref.get("uploading") for ref in refs):
        raise gr.Error("A reference is still uploading. Give it a second and try again.")
    for ref in refs:
        if ref.get("error"):
            raise gr.Error(ref["error"])
    references = []
    for ref in refs:
        kind, path = ref.get("kind"), ref.get("path")
        if kind not in LIMITS or not path or not _allowed(path):
            raise gr.Error("One of the references could not be read. Remove it and add it again.")
        references.append((kind, _as_rgb_png(path, len(references)) if kind == "image" else path))
    for kind, limit in LIMITS.items():
        if sum(k == kind for k, _ in references) > limit:
            raise gr.Error(f"At most {limit} {kind} references.")
    return references


def _as_rgb_png(path: str, index: int) -> str:
    tmp = tempfile.NamedTemporaryFile(suffix=f"_ref{index + 1}.png", delete=False)
    Image.open(path).convert("RGB").save(tmp.name)
    return tmp.name


def soundtrack_seconds(references):
    carried = []
    for kind, path in references:
        if kind == "image":
            continue
        _, audio_seconds = probe(path)
        if audio_seconds is not None:
            carried.append(audio_seconds)
    return carried


def check(prompt: str, references) -> None:
    if not references:
        raise gr.Error("Add at least one image or video reference.")
    if not any(kind != "audio" for kind, _ in references):
        raise gr.Error("Audio needs an image or a video alongside it.")
    if not prompt or not prompt.strip():
        raise gr.Error("Describe the video: fill in the builder or write a prompt.")
    if "<removed>" in prompt:
        raise gr.Error("The prompt still refers to a removed reference (<removed>). Edit it out first.")
    for kind, path in references:
        if kind != "video":
            continue
        video_seconds, _ = probe(path)
        if video_seconds is None:
            raise gr.Error("A reference video has no video stream. Add it as audio instead.")
        if not MIN_REFERENCE_VIDEO <= video_seconds <= MAX_REFERENCE_VIDEO:
            raise gr.Error(
                f"A reference video is {video_seconds:.1f} s. Use clips between "
                f"{MIN_REFERENCE_VIDEO:g} and {MAX_REFERENCE_VIDEO:g} seconds."
            )


@cache
def rewriter():
    from gradio_client import Client

    return Client(REWRITER_SPACE, token=False)


def aspect_label(height: int, width: int) -> str:
    ratio = width / height
    options = {"21:9": 21 / 9, "16:9": 16 / 9, "4:3": 4 / 3, "1:1": 1.0, "3:4": 3 / 4, "9:16": 9 / 16}
    return min(options, key=lambda key: abs(options[key] - ratio))


def resolve_frames(references, match, duration) -> int:
    carried = soundtrack_seconds(references)
    if match and len(carried) == 1 and MIN_FRAMES <= snap_frames(carried[0]) <= MAX_FRAMES:
        return snap_frames(carried[0])
    return min(MAX_FRAMES, max(MIN_FRAMES, snap_frames(duration)))


def upsample(composer, canvas, match, duration):
    from gradio_client import handle_file

    composer = dict(composer or {})
    try:
        references = references_from(composer)
        if not any(kind != "audio" for kind, _ in references):
            raise gr.Error("Add an image or a video reference first.")
        for kind, limit in REWRITER_LIMITS.items():
            if sum(k == kind for k, _ in references) > limit:
                raise gr.Error(
                    f"The rewriter takes at most {REWRITER_LIMITS['image']} images, 1 video and 1 audio clip."
                )
        prompt = (composer.get("upsample_source") or "").strip()
        if not prompt:
            raise gr.Error("Describe the video first, then upsample it.")
        missing = [label for _, label in reference_labels(references) if label not in prompt]
        if missing:
            prompt = f"Keep the subject shown in {' and '.join(missing)}. {prompt}"
        height, width = CANVASES.get(canvas, CANVASES[DEFAULT_CANVAS])
        num_frames = resolve_frames(references, match, duration)
        images = [path for kind, path in references if kind == "image"]
        images += [None] * (REWRITER_LIMITS["image"] - len(images))
        video = next((path for kind, path in references if kind == "video"), None)
        audio = next((path for kind, path in references if kind == "audio"), None)
        rewritten, status = rewriter().predict(
            task="Ref2AV",
            prompt=prompt,
            duration=round(num_frames / FPS, 2),
            resolution=aspect_label(height, width),
            image_1=handle_file(images[0]) if images[0] else None,
            image_2=handle_file(images[1]) if images[1] else None,
            image_3=handle_file(images[2]) if images[2] else None,
            image_4=handle_file(images[3]) if images[3] else None,
            ref_video=handle_file(video) if video else None,
            ref_audio=handle_file(audio) if audio else None,
            api_name="/rewrite",
        )
        rewritten = (rewritten or "").strip()
        if not rewritten:
            raise gr.Error(f"The rewriter returned nothing: {status}")
    except gr.Error as error:
        gr.Warning(str(error.message))
        rewritten = ""
    except Exception as error:
        gr.Warning(f"The prompt rewriter failed: {type(error).__name__}: {error}")
        rewritten = ""
    composer["upsampled"] = rewritten
    return composer


def _gpu_duration(prompt, references, height, width, num_frames, seed, *args, **kwargs):
    cold = _COMPLETED_GPU_CALLS == 0 and not KERNELS_CACHED
    base = GPU_SECONDS_COLD if cold else GPU_SECONDS_WARM
    seconds = base * max(1.0, (height * width * num_frames) / DEFAULT_PIXEL_FRAMES)
    images = sum(kind == "image" for kind, _ in references)
    seconds += GPU_SECONDS_PER_EXTRA_IMAGE * max(0, images - 1)
    seconds += GPU_SECONDS_PER_VIDEO * sum(kind == "video" for kind, _ in references)
    return int(min(900, seconds))


@spaces.GPU(duration=_gpu_duration, size="xlarge")
def _generate(prompt, references, height, width, num_frames, seed, progress=gr.Progress()):
    progress(0.01, desc="Starting on the GPU")
    context = contextvars.copy_context()

    def report(fraction, desc):
        context.copy().run(progress, fraction, desc=desc)

    return ENGINE.generate(prompt, references, height, width, num_frames, seed, on_progress=report)


def generate(composer, canvas=DEFAULT_CANVAS, match=True, duration=5, seed=1234, progress=gr.Progress()):
    references = references_from(composer)
    prompt = ((composer or {}).get("prompt") or "").strip()
    check(prompt, references)
    height, width = CANVASES.get(canvas, CANVASES[DEFAULT_CANVAS])
    num_frames = resolve_frames(references, match, duration)
    seed = int(seed) if seed is not None else 0
    global _COMPLETED_GPU_CALLS
    progress(0, desc="Waiting for a GPU")
    path = _generate(to_h3_prompt(prompt, references), references, height, width, num_frames, seed)
    _COMPLETED_GPU_CALLS += 1
    return path


# --- UI ---------------------------------------------------------------------
HEADER = """
<div class="fh-header">
  <h1>FastH3-<span class="fh-accent">Omni</span></h1>
  <p>FastH3-Omni is FastVideo's 8-step reference-to-video student that turns reference images, videos and
  audio plus a prompt into a short clip with a matching soundtrack. Drop in references, then describe the video
  with the builder or write the prompt yourself, tagging each reference by name.</p>
  <div class="fh-pills">
    <a href="https://huggingface.co/FastVideo/FastVideo-FastH3-Omni-8-Step-V1" target="_blank">Model</a>
    <a href="https://huggingface.co/collections/FastVideo/fasth3-omni-6ad501acfccdebd3c3b7445b" target="_blank">Collection</a>
    <a href="https://github.com/hao-ai-lab/FastVideo" target="_blank">FastVideo</a>
    <a href="https://huggingface.co/MiniMaxAI/MiniMax-H3" target="_blank">MiniMax-H3</a>
  </div>
</div>
"""

FASTVIDEO_BLUE = gr.themes.Color(
    c50="#eef3ff", c100="#dbe5ff", c200="#bccfff", c300="#8eaeff", c400="#5f8cff",
    c500="#346cff", c600="#1f55ec", c700="#1843c4", c800="#17389c", c900="#16317c", c950="#0e1d4a",
    name="fastvideo",
)

CSS = """
.main.fillable { max-width: 1250px !important; }
.fh-header { padding: 6px 2px 12px; }
.fh-header h1 { font-size: 1.6rem; margin: 0 0 6px; font-weight: 700; }
.fh-header h1 .fh-accent { color: #346cff; }
.fh-header p { color: var(--body-text-color-subdued); margin: 0 0 12px; max-width: 76ch; }
.fh-pills { display: flex; gap: 8px; flex-wrap: wrap; }
.fh-pills a { font-size: .8rem; padding: 3px 11px; border-radius: 999px;
  background: var(--button-secondary-background-fill); color: var(--body-text-color);
  text-decoration: none; border: 1px solid var(--border-color-primary); }
.fh-pills a:hover { border-color: #346cff; color: #346cff; }
#gen-btn { background: #346cff; color: #fff; border: none; }
#gen-btn:hover { background: #1f55ec; }
"""

THEME = gr.themes.Base(
    primary_hue=FASTVIDEO_BLUE,
    secondary_hue=FASTVIDEO_BLUE,
    neutral_hue="zinc",
    font=[gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"],
).set(
    body_background_fill_dark="#000000",
    block_background_fill_dark="#0b0b0f",
    background_fill_secondary_dark="#0b0b0f",
    color_accent="#346cff",
    slider_color="#346cff",
    checkbox_background_color_selected="#346cff",
    checkbox_background_color_selected_dark="#346cff",
)

COMPOSER_HTML = (APP_DIR / "composer.html").read_text()
COMPOSER_CSS = (APP_DIR / "composer.css").read_text()
COMPOSER_JS = (APP_DIR / "composer.js").read_text()

with gr.Blocks(title="FastH3-Omni") as demo:
    gr.HTML(HEADER)

    with gr.Row():
        with gr.Column(scale=1):
            composer = gr.HTML(
                value={"examples": composer_examples()},
                html_template=COMPOSER_HTML,
                css_template=COMPOSER_CSS,
                js_on_load=COMPOSER_JS,
                apply_default_css=False,
            )
            run = gr.Button("Generate", variant="primary", size="lg", elem_id="gen-btn")
            with gr.Accordion("Advanced options", open=False):
                canvas = gr.Dropdown(label="Canvas", choices=list(CANVASES), value=DEFAULT_CANVAS)
                match = gr.Checkbox(
                    label="Match the reference soundtrack",
                    info="When exactly one reference carries a soundtrack, the clip takes its length.",
                    value=True,
                )
                duration = gr.Slider(
                    label="Duration (s)", minimum=MIN_DURATION, maximum=MAX_UI_DURATION, step=1, value=5
                )
                seed = gr.Number(label="Seed", value=1234, precision=0)
                gr.Markdown("8 fused steps and guidance 1.0 are fixed by the model's contract.")

        with gr.Column(scale=1):
            result = gr.Video(label="Video + soundtrack", autoplay=True)

    composer.edit(upsample, [composer, canvas, match, duration], [composer], api_name=False, show_progress="hidden")
    run.click(generate, [composer, canvas, match, duration, seed], [result], api_name="generate")


if __name__ == "__main__":
    demo.queue(max_size=12).launch(
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True,
        theme=THEME,
        css=CSS,
        ssr_mode=False,
        allowed_paths=[str(EXAMPLES_DIR)],
    )

	"""FastH3-Omni · reference-to-video (+audio) on ZeroGPU.

	Ordered image, video and audio references plus a prompt become a short video
	with a matching 32 kHz soundtrack in eight fused forwards of the FastVideo
	FastH3 OmniRef student. A plain prompt is wrapped into the model's six-section
	H3 format; "Upsample prompt" rewrites it with the MiniMax-H3 prompt rewriter
	instead. See fasth3_engine.py for the ZeroGPU + FastVideo integration notes.
	"""
	import contextvars
	import os
	import tempfile
	from functools import cache
	from pathlib import Path

	import gradio as gr
	from PIL import Image

	try:
	import spaces
	IS_ZERO = True
	except ImportError: # local / CI
	IS_ZERO = False

	class _NoSpaces:
	@staticmethod
	def GPU(a, *k):
	if a and callable(a[0]):
	return a[0]
	return lambda f: f

	spaces = _NoSpaces() # type: ignore

	from fasth3_engine import FastH3OmniRefEngine, MODEL_REPO, prefetch

	REWRITER_SPACE = os.getenv("FASTH3_REWRITER", "hugging-apps/minimax-h3-prompt-rewriter")

	CANVASES = {
	"1344x768 · 16:9 recommended": (768, 1344),
	"832x480 · 16:9 fast": (480, 832),
	"1280x704 · 16:9": (704, 1280),
	"960x544 · 16:9": (544, 960),
	"768x1344 · 9:16 recommended": (1344, 768),
	"480x832 · 9:16 fast": (832, 480),
	"768x768 · 1:1": (768, 768),
	"1024x1024 · 1:1": (1024, 1024),
	"1024x768 · 4:3": (768, 1024),
	"768x1024 · 3:4": (1024, 768),
	"1536x672 · 21:9": (672, 1536),
	}
	DEFAULT_CANVAS = "1344x768 · 16:9 recommended"
	PORTRAIT_CANVAS = "768x1344 · 9:16 recommended"
	FPS, FRAMES_PER_CHUNK, LATENTS_PER_CHUNK = 24, 17, 5
	MIN_FRAMES, MAX_FRAMES = 124, 362
	MIN_DURATION, MAX_UI_DURATION = 5, 14
	MIN_REFERENCE_VIDEO, MAX_REFERENCE_VIDEO = 2.0, 15.0
	MAX_IMAGE_SLOTS, OPEN_IMAGE_SLOTS = 9, 2
	REWRITER_IMAGE_SLOTS = 4
	DEFAULT_PIXEL_FRAMES = 1344 * 768 * 124
	GPU_SECONDS_WARM = int(os.getenv("FASTH3_GPU_SECONDS", "170"))
	GPU_SECONDS_COLD = int(os.getenv("FASTH3_GPU_SECONDS_COLD", "360"))
	GPU_SECONDS_PER_EXTRA_IMAGE, GPU_SECONDS_PER_VIDEO = 110, 160

	APP_DIR = Path(__file__).resolve().parent
	EXAMPLES_DIR = APP_DIR / "examples"
	LIMITS = {"image": 9, "video": 3, "audio": 3}
	REWRITER_LIMITS = {"image": 4, "video": 1, "audio": 1}

	EXAMPLES = [
	{
	"title": "Beach promenade at golden hour",
	"image": "woman.jpg",
	"desc": "a young woman with a dark curly afro and gold hoop earrings, wearing a white t-shirt",
	"scene": "walks along a sunlit beach promenade at golden hour, smiling and glancing toward the camera, "
	"palm trees and the ocean behind her.",
	"sound": "ocean waves, a gentle breeze and distant seagulls",
	"music": "none",
	},
	{
	"title": "Dog bounding through snow",
	"image": "dog.jpg",
	"desc": "a pale cream, fluffy long-haired dog with floppy ears and a dark nose",
	"scene": "bounds across a snowy field in bright winter sunlight, kicking up powder snow, then shakes itself off.",
	"sound": "crunching snow, happy panting and a cold wind",
	"music": "none",
	},
	{
	"title": "Birthday cake with candles",
	"image": "cake.jpg",
	"desc": "a whole chocolate layer cake with chocolate frosting, pastel sprinkles and chocolate shards on top",
	"scene": "sits at the center of a festive birthday table in a warm, dim room as lit candles flicker and "
	"colorful confetti drifts down around it.",
	"sound": "soft party chatter and a faint crackle",
	"music": "cheerful, upbeat birthday music",
	},
	]


	def composer_examples():
	rows = []
	for example in EXAMPLES:
	path = str(EXAMPLES_DIR / example["image"])
	rows.append({
	"title": example["title"],
	"refs": [{"kind": "image", "path": path, "url": f"/gradio_api/file={path}", "name": example["image"],
	"desc": example["desc"]}],
	"builder": {"scene": example["scene"], "sound": example["sound"], "music": example["music"],
	"style": "realistic photographic"},
	})
	return rows


	def snap_frames(seconds: float) -> int:
	frames = max(1, round(float(seconds) * FPS))
	while frames % FRAMES_PER_CHUNK != LATENTS_PER_CHUNK:
	frames += 1
	return frames


	def reference_labels(references):
	names = {"image": "Picture", "video": "Video", "audio": "Audio"}
	counts, labels = {}, []
	for kind, _ in references:
	counts[kind] = counts.get(kind, 0) + 1
	labels.append((kind, f"<{names[kind]} {counts[kind]}>"))
	return labels


	def to_h3_prompt(prompt: str, references) -> str:
	"""Wrap a plain prompt into the six-section H3 format the checkpoint was trained on."""
	prompt = (prompt or "").strip()
	if "subject_definitions:" in prompt:
	return prompt
	labels = reference_labels(references)
	visual = " and ".join(label for kind, label in labels if kind != "audio") or "the references"
	audio = [label for kind, label in labels if kind == "audio"]
	sound = (f"the voice and sound of {' and '.join(audio)}, in a natural acoustic space matching the scene"
	if audio else "natural ambient sound that matches the scene")
	return (
	f"subject_definitions:\n<Subject 1> is the subject shown in {visual}.\n\n"
	f"summary:\n{prompt}\n\n"
	f"retention_analysis:\n<Subject 1>: fully_preserved - the reference subject's identity, "
	f"appearance and details are kept.\n\n"
	f"detailed_description:\nThe target video is in a realistic photographic style.\n"
	f"[Shot 1] {prompt}\n\n"
	f"overall_soundscape:\n{sound}.\n\n"
	f"non_diegetic_music:\nnone."
	)


	# --- engine: built once at startup, on ZeroGPU's emulated CUDA ------------
	# The weights land on "cuda" in the main process; ZeroGPU packs them and maps
	# them onto the real card at the start of every GPU call, so no call rebuilds.
	from fasth3_engine import fetch_kernel_cache, zerogpu_main_process_shims

	prefetch()
	KERNELS_CACHED = fetch_kernel_cache()
	if IS_ZERO:
	zerogpu_main_process_shims()
	ENGINE = FastH3OmniRefEngine()
	ENGINE.initialize()
	_COMPLETED_GPU_CALLS = 0


	def probe(path: str):
	import av

	def seconds(stream, container):
	if stream.duration is not None and stream.time_base is not None:
	return float(stream.duration * stream.time_base)
	return None if container.duration is None else container.duration / av.time_base

	with av.open(path) as container:
	video = seconds(container.streams.video[0], container) if container.streams.video else None
	audio = seconds(container.streams.audio[0], container) if container.streams.audio else None
	return video, audio


	def _allowed(path: str) -> bool:
	from gradio.utils import get_upload_folder

	real = os.path.realpath(path)
	roots = [os.path.realpath(get_upload_folder()), os.path.realpath(EXAMPLES_DIR)]
	return any(real == root or real.startswith(root + os.sep) for root in roots) and os.path.isfile(real)


	def references_from(composer: dict):
	refs = (composer or {}).get("refs") or []
	if any(ref.get("uploading") for ref in refs):
	raise gr.Error("A reference is still uploading. Give it a second and try again.")
	for ref in refs:
	if ref.get("error"):
	raise gr.Error(ref["error"])
	references = []
	for ref in refs:
	kind, path = ref.get("kind"), ref.get("path")
	if kind not in LIMITS or not path or not _allowed(path):
	raise gr.Error("One of the references could not be read. Remove it and add it again.")
	references.append((kind, _as_rgb_png(path, len(references)) if kind == "image" else path))
	for kind, limit in LIMITS.items():
	if sum(k == kind for k, _ in references) > limit:
	raise gr.Error(f"At most {limit} {kind} references.")
	return references


	def _as_rgb_png(path: str, index: int) -> str:
	tmp = tempfile.NamedTemporaryFile(suffix=f"_ref{index + 1}.png", delete=False)
	Image.open(path).convert("RGB").save(tmp.name)
	return tmp.name


	def soundtrack_seconds(references):
	carried = []
	for kind, path in references:
	if kind == "image":
	continue
	_, audio_seconds = probe(path)
	if audio_seconds is not None:
	carried.append(audio_seconds)
	return carried


	def check(prompt: str, references) -> None:
	if not references:
	raise gr.Error("Add at least one image or video reference.")
	if not any(kind != "audio" for kind, _ in references):
	raise gr.Error("Audio needs an image or a video alongside it.")
	if not prompt or not prompt.strip():
	raise gr.Error("Describe the video: fill in the builder or write a prompt.")
	if "<removed>" in prompt:
	raise gr.Error("The prompt still refers to a removed reference (<removed>). Edit it out first.")
	for kind, path in references:
	if kind != "video":
	continue
	video_seconds, _ = probe(path)
	if video_seconds is None:
	raise gr.Error("A reference video has no video stream. Add it as audio instead.")
	if not MIN_REFERENCE_VIDEO <= video_seconds <= MAX_REFERENCE_VIDEO:
	raise gr.Error(
	f"A reference video is {video_seconds:.1f} s. Use clips between "
	f"{MIN_REFERENCE_VIDEO:g} and {MAX_REFERENCE_VIDEO:g} seconds."
	)


	@cache
	def rewriter():
	from gradio_client import Client

	return Client(REWRITER_SPACE, token=False)


	def aspect_label(height: int, width: int) -> str:
	ratio = width / height
	options = {"21:9": 21 / 9, "16:9": 16 / 9, "4:3": 4 / 3, "1:1": 1.0, "3:4": 3 / 4, "9:16": 9 / 16}
	return min(options, key=lambda key: abs(options[key] - ratio))


	def resolve_frames(references, match, duration) -> int:
	carried = soundtrack_seconds(references)
	if match and len(carried) == 1 and MIN_FRAMES <= snap_frames(carried[0]) <= MAX_FRAMES:
	return snap_frames(carried[0])
	return min(MAX_FRAMES, max(MIN_FRAMES, snap_frames(duration)))


	def upsample(composer, canvas, match, duration):
	from gradio_client import handle_file

	composer = dict(composer or {})
	try:
	references = references_from(composer)
	if not any(kind != "audio" for kind, _ in references):
	raise gr.Error("Add an image or a video reference first.")
	for kind, limit in REWRITER_LIMITS.items():
	if sum(k == kind for k, _ in references) > limit:
	raise gr.Error(
	f"The rewriter takes at most {REWRITER_LIMITS['image']} images, 1 video and 1 audio clip."
	)
	prompt = (composer.get("upsample_source") or "").strip()
	if not prompt:
	raise gr.Error("Describe the video first, then upsample it.")
	missing = [label for _, label in reference_labels(references) if label not in prompt]
	if missing:
	prompt = f"Keep the subject shown in {' and '.join(missing)}. {prompt}"
	height, width = CANVASES.get(canvas, CANVASES[DEFAULT_CANVAS])
	num_frames = resolve_frames(references, match, duration)
	images = [path for kind, path in references if kind == "image"]
	images += [None] * (REWRITER_LIMITS["image"] - len(images))
	video = next((path for kind, path in references if kind == "video"), None)
	audio = next((path for kind, path in references if kind == "audio"), None)
	rewritten, status = rewriter().predict(
	task="Ref2AV",
	prompt=prompt,
	duration=round(num_frames / FPS, 2),
	resolution=aspect_label(height, width),
	image_1=handle_file(images[0]) if images[0] else None,
	image_2=handle_file(images[1]) if images[1] else None,
	image_3=handle_file(images[2]) if images[2] else None,
	image_4=handle_file(images[3]) if images[3] else None,
	ref_video=handle_file(video) if video else None,
	ref_audio=handle_file(audio) if audio else None,
	api_name="/rewrite",
	)
	rewritten = (rewritten or "").strip()
	if not rewritten:
	raise gr.Error(f"The rewriter returned nothing: {status}")
	except gr.Error as error:
	gr.Warning(str(error.message))
	rewritten = ""
	except Exception as error:
	gr.Warning(f"The prompt rewriter failed: {type(error).__name__}: {error}")
	rewritten = ""
	composer["upsampled"] = rewritten
	return composer


	def _gpu_duration(prompt, references, height, width, num_frames, seed, args, *kwargs):
	cold = _COMPLETED_GPU_CALLS == 0 and not KERNELS_CACHED
	base = GPU_SECONDS_COLD if cold else GPU_SECONDS_WARM
	seconds = base * max(1.0, (height * width * num_frames) / DEFAULT_PIXEL_FRAMES)
	images = sum(kind == "image" for kind, _ in references)
	seconds += GPU_SECONDS_PER_EXTRA_IMAGE * max(0, images - 1)
	seconds += GPU_SECONDS_PER_VIDEO * sum(kind == "video" for kind, _ in references)
	return int(min(900, seconds))


	@spaces.GPU(duration=_gpu_duration, size="xlarge")
	def _generate(prompt, references, height, width, num_frames, seed, progress=gr.Progress()):
	progress(0.01, desc="Starting on the GPU")
	context = contextvars.copy_context()

	def report(fraction, desc):
	context.copy().run(progress, fraction, desc=desc)

	return ENGINE.generate(prompt, references, height, width, num_frames, seed, on_progress=report)


	def generate(composer, canvas=DEFAULT_CANVAS, match=True, duration=5, seed=1234, progress=gr.Progress()):
	references = references_from(composer)
	prompt = ((composer or {}).get("prompt") or "").strip()
	check(prompt, references)
	height, width = CANVASES.get(canvas, CANVASES[DEFAULT_CANVAS])
	num_frames = resolve_frames(references, match, duration)
	seed = int(seed) if seed is not None else 0
	global _COMPLETED_GPU_CALLS
	progress(0, desc="Waiting for a GPU")
	path = _generate(to_h3_prompt(prompt, references), references, height, width, num_frames, seed)
	_COMPLETED_GPU_CALLS += 1
	return path


	# --- UI ---------------------------------------------------------------------
	HEADER = """
	<div class="fh-header">
	<h1>FastH3-<span class="fh-accent">Omni</span></h1>
	<p>FastH3-Omni is FastVideo's 8-step reference-to-video student that turns reference images, videos and
	audio plus a prompt into a short clip with a matching soundtrack. Drop in references, then describe the video
	with the builder or write the prompt yourself, tagging each reference by name.</p>
	<div class="fh-pills">
	<a href="https://huggingface.co/FastVideo/FastVideo-FastH3-Omni-8-Step-V1" target="_blank">Model</a>
	<a href="https://huggingface.co/collections/FastVideo/fasth3-omni-6ad501acfccdebd3c3b7445b" target="_blank">Collection</a>
	<a href="https://github.com/hao-ai-lab/FastVideo" target="_blank">FastVideo</a>
	<a href="https://huggingface.co/MiniMaxAI/MiniMax-H3" target="_blank">MiniMax-H3</a>
	</div>
	</div>
	"""

	FASTVIDEO_BLUE = gr.themes.Color(
	c50="#eef3ff", c100="#dbe5ff", c200="#bccfff", c300="#8eaeff", c400="#5f8cff",
	c500="#346cff", c600="#1f55ec", c700="#1843c4", c800="#17389c", c900="#16317c", c950="#0e1d4a",
	name="fastvideo",
	)

	CSS = """
	.main.fillable { max-width: 1250px !important; }
	.fh-header { padding: 6px 2px 12px; }
	.fh-header h1 { font-size: 1.6rem; margin: 0 0 6px; font-weight: 700; }
	.fh-header h1 .fh-accent { color: #346cff; }
	.fh-header p { color: var(--body-text-color-subdued); margin: 0 0 12px; max-width: 76ch; }
	.fh-pills { display: flex; gap: 8px; flex-wrap: wrap; }
	.fh-pills a { font-size: .8rem; padding: 3px 11px; border-radius: 999px;
	background: var(--button-secondary-background-fill); color: var(--body-text-color);
	text-decoration: none; border: 1px solid var(--border-color-primary); }
	.fh-pills a:hover { border-color: #346cff; color: #346cff; }
	#gen-btn { background: #346cff; color: #fff; border: none; }
	#gen-btn:hover { background: #1f55ec; }
	"""

	THEME = gr.themes.Base(
	primary_hue=FASTVIDEO_BLUE,
	secondary_hue=FASTVIDEO_BLUE,
	neutral_hue="zinc",
	font=[gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"],
	).set(
	body_background_fill_dark="#000000",
	block_background_fill_dark="#0b0b0f",
	background_fill_secondary_dark="#0b0b0f",
	color_accent="#346cff",
	slider_color="#346cff",
	checkbox_background_color_selected="#346cff",
	checkbox_background_color_selected_dark="#346cff",
	)

	COMPOSER_HTML = (APP_DIR / "composer.html").read_text()
	COMPOSER_CSS = (APP_DIR / "composer.css").read_text()
	COMPOSER_JS = (APP_DIR / "composer.js").read_text()

	with gr.Blocks(title="FastH3-Omni") as demo:
	gr.HTML(HEADER)

	with gr.Row():
	with gr.Column(scale=1):
	composer = gr.HTML(
	value={"examples": composer_examples()},
	html_template=COMPOSER_HTML,
	css_template=COMPOSER_CSS,
	js_on_load=COMPOSER_JS,
	apply_default_css=False,
	)
	run = gr.Button("Generate", variant="primary", size="lg", elem_id="gen-btn")
	with gr.Accordion("Advanced options", open=False):
	canvas = gr.Dropdown(label="Canvas", choices=list(CANVASES), value=DEFAULT_CANVAS)
	match = gr.Checkbox(
	label="Match the reference soundtrack",
	info="When exactly one reference carries a soundtrack, the clip takes its length.",
	value=True,
	)
	duration = gr.Slider(
	label="Duration (s)", minimum=MIN_DURATION, maximum=MAX_UI_DURATION, step=1, value=5
	)
	seed = gr.Number(label="Seed", value=1234, precision=0)
	gr.Markdown("8 fused steps and guidance 1.0 are fixed by the model's contract.")

	with gr.Column(scale=1):
	result = gr.Video(label="Video + soundtrack", autoplay=True)

	composer.edit(upsample, [composer, canvas, match, duration], [composer], api_name=False, show_progress="hidden")
	run.click(generate, [composer, canvas, match, duration, seed], [result], api_name="generate")


	if __name__ == "__main__":
	demo.queue(max_size=12).launch(
	server_name="0.0.0.0",
	server_port=7860,
	show_error=True,
	theme=THEME,
	css=CSS,
	ssr_mode=False,
	allowed_paths=[str(EXAMPLES_DIR)],
	)