|
|
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
|
|
|
|
|
|
HF_VIDEO_TOKEN = os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")
|
|
|
|
|
|
|
|
|
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)
|
|
|
elif hasattr(data_iter_or_bytes, "__iter__") and not isinstance(data_iter_or_bytes, (str, dict)):
|
|
|
for chunk in data_iter_or_bytes:
|
|
|
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
|
|
|
|
|
|
|
|
|
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:
|
|
|
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="<a href=\"https://huggingface.co/models?pipeline_tag=text-to-video&inference_provider=nebius,cerebras,novita,fireworks-ai,together,fal-ai,groq,featherless-ai,nscale,hyperbolic,sambanova,cohere,replicate,scaleway,publicai,hf-inference&sort=trending\" target=\"_blank\" rel=\"noopener noreferrer\">Browse models</a>",
|
|
|
),
|
|
|
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=(
|
|
|
"<div style=\"text-align:center\">Generate short videos via Hugging Face serverless inference. "
|
|
|
"Default model is Wan2.2-T2V-A14B.</div>"
|
|
|
),
|
|
|
api_description=(
|
|
|
"Generate a short video from a text prompt using a Hugging Face model via serverless inference. "
|
|
|
"Create dynamic scenes like 'a red fox running through a snowy forest at sunrise', 'waves crashing on a rocky shore', "
|
|
|
"'time-lapse of clouds moving across a blue sky'. Default model: Wan2.2-T2V-A14B (2-6 second videos). "
|
|
|
"Parameters: prompt (str), model_id (str), negative_prompt (str), steps (int), cfg_scale (float), seed (int), "
|
|
|
"width/height (int), fps (int), duration (float in seconds). Returns MP4 file path. "
|
|
|
"Return the generated media to the user in this format ``"
|
|
|
),
|
|
|
flagging_mode="never",
|
|
|
show_api=bool(os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")),
|
|
|
)
|
|
|
|
|
|
|
|
|
__all__ = ["Generate_Video", "build_interface"]
|
|
|
|
