from __future__ import annotations import os import random import tempfile from typing import Annotated import gradio as gr from huggingface_hub import InferenceClient from app import _log_call_end, _log_call_start, _truncate_for_log from ._docstrings import autodoc HF_VIDEO_TOKEN = os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN") # Single source of truth for the LLM-facing tool description TOOL_SUMMARY = ( "Generate a short MP4 video from a text prompt via Hugging Face serverless inference; " "control model, steps, guidance, seed, size, fps, and duration; returns a temporary MP4 file path. " "Return the generated media to the user in this format `![Alt text](URL)`." ) def _write_video_tmp(data_iter_or_bytes: object, suffix: str = ".mp4") -> str: fd, fname = tempfile.mkstemp(suffix=suffix) try: with os.fdopen(fd, "wb") as file: if isinstance(data_iter_or_bytes, (bytes, bytearray)): file.write(data_iter_or_bytes) elif hasattr(data_iter_or_bytes, "read"): file.write(data_iter_or_bytes.read()) elif hasattr(data_iter_or_bytes, "content"): file.write(data_iter_or_bytes.content) # type: ignore[attr-defined] elif hasattr(data_iter_or_bytes, "__iter__") and not isinstance(data_iter_or_bytes, (str, dict)): for chunk in data_iter_or_bytes: # type: ignore[assignment] if chunk: file.write(chunk) else: raise gr.Error("Unsupported video data type returned by provider.") except Exception: try: os.remove(fname) except Exception: pass raise return fname @autodoc( summary=TOOL_SUMMARY, ) def Generate_Video( prompt: Annotated[str, "Text description of the video to generate (e.g., 'a red fox running through a snowy forest at sunrise')."], model_id: Annotated[str, "Hugging Face model id in the form 'creator/model-name'. Defaults to Wan-AI/Wan2.2-T2V-A14B."] = "Wan-AI/Wan2.2-T2V-A14B", negative_prompt: Annotated[str, "What should NOT appear in the video."] = "", steps: Annotated[int, "Number of denoising steps (1–100). Higher can improve quality but is slower."] = 25, cfg_scale: Annotated[float, "Guidance scale (1–20). Higher = follow the prompt more closely, lower = more creative."] = 3.5, seed: Annotated[int, "Random seed for reproducibility. Use -1 for a random seed per call."] = -1, width: Annotated[int, "Output width in pixels (multiples of 8 recommended)."] = 768, height: Annotated[int, "Output height in pixels (multiples of 8 recommended)."] = 768, fps: Annotated[int, "Frames per second of the output video (e.g., 24)."] = 24, duration: Annotated[float, "Target duration in seconds (provider/model dependent, commonly 2–6s)."] = 4.0, ) -> str: _log_call_start( "Generate_Video", prompt=_truncate_for_log(prompt, 160), model_id=model_id, steps=steps, cfg_scale=cfg_scale, fps=fps, duration=duration, size=f"{width}x{height}", ) if not prompt or not prompt.strip(): _log_call_end("Generate_Video", "error=empty prompt") raise gr.Error("Please provide a non-empty prompt.") providers = ["auto", "replicate", "fal-ai"] last_error: Exception | None = None parameters = { "negative_prompt": negative_prompt or None, "num_inference_steps": steps, "guidance_scale": cfg_scale, "seed": seed if seed != -1 else random.randint(1, 1_000_000_000), "width": width, "height": height, "fps": fps, "duration": duration, } for provider in providers: try: client = InferenceClient(api_key=HF_VIDEO_TOKEN, provider=provider) if hasattr(client, "text_to_video"): num_frames = int(duration * fps) if duration and fps else None extra_body = {} if width: extra_body["width"] = width if height: extra_body["height"] = height if fps: extra_body["fps"] = fps if duration: extra_body["duration"] = duration result = client.text_to_video( prompt=prompt, model=model_id, guidance_scale=cfg_scale, negative_prompt=[negative_prompt] if negative_prompt else None, num_frames=num_frames, num_inference_steps=steps, seed=parameters["seed"], extra_body=extra_body if extra_body else None, ) else: result = client.post( model=model_id, json={"inputs": prompt, "parameters": {k: v for k, v in parameters.items() if v is not None}}, ) path = _write_video_tmp(result, suffix=".mp4") try: size = os.path.getsize(path) except Exception: size = -1 _log_call_end("Generate_Video", f"provider={provider} path={os.path.basename(path)} bytes={size}") return path except Exception as exc: # pylint: disable=broad-except last_error = exc continue msg = str(last_error) if last_error else "Unknown error" lowered = msg.lower() if "404" in msg: raise gr.Error(f"Model not found or unavailable: {model_id}. Check the id and HF token access.") if "503" in msg: raise gr.Error("The model is warming up. Please try again shortly.") if "401" in msg or "403" in msg: raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.") if ("api_key" in lowered) or ("hf auth login" in lowered) or ("unauthorized" in lowered) or ("forbidden" in lowered): raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.") _log_call_end("Generate_Video", f"error={_truncate_for_log(msg, 200)}") raise gr.Error(f"Video generation failed: {msg}") def build_interface() -> gr.Interface: return gr.Interface( fn=Generate_Video, inputs=[ gr.Textbox(label="Prompt", placeholder="Enter a prompt for the video", lines=2), gr.Textbox( label="Model", value="Wan-AI/Wan2.2-T2V-A14B", placeholder="creator/model-name", max_lines=1, info="Browse models", ), gr.Textbox(label="Negative Prompt", value="", lines=2), gr.Slider(minimum=1, maximum=100, value=25, step=1, label="Steps"), gr.Slider(minimum=1.0, maximum=20.0, value=3.5, step=0.1, label="CFG Scale"), gr.Slider(minimum=-1, maximum=1_000_000_000, value=-1, step=1, label="Seed (-1 = random)"), gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Width"), gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Height"), gr.Slider(minimum=4, maximum=60, value=24, step=1, label="FPS"), gr.Slider(minimum=1.0, maximum=10.0, value=4.0, step=0.5, label="Duration (s)"), ], outputs=gr.Video(label="Generated Video", show_download_button=True, format="mp4"), title="Generate Video", description=( "
Generate short videos via Hugging Face serverless inference. " "Default model is Wan2.2-T2V-A14B.
" ), api_description=TOOL_SUMMARY, flagging_mode="never", show_api=bool(os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")), ) __all__ = ["Generate_Video", "build_interface"]