File size: 4,732 Bytes
2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e afceeed 2557c6e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
from typing import Dict, Any
import os
from pathlib import Path
import time
from datetime import datetime
import argparse
from hyvideo.utils.file_utils import save_videos_grid
from hyvideo.inference import HunyuanVideoSampler
from hyvideo.config import parse_args
from hyvideo.constants import NEGATIVE_PROMPT
def get_default_args():
"""Create default arguments instead of parsing from command line"""
parser = argparse.ArgumentParser()
# Add all the arguments that were in the original parser
parser.add_argument("--model", type=str, default="HYVideo-T/2")
parser.add_argument("--model-resolution", type=str, default="720p", choices=["540p", "720p"])
parser.add_argument("--latent-channels", type=int, default=4)
parser.add_argument("--precision", type=str, default="bf16", choices=["bf16", "fp32", "fp16"])
parser.add_argument("--batch-size", type=int, default=1)
parser.add_argument("--infer-steps", type=int, default=50)
parser.add_argument("--model-base", type=str, default=None)
parser.add_argument("--save-path", type=str, default="outputs")
parser.add_argument("--video-length", type=int, default=129) # 5 seconds
# Parse with empty args list to avoid reading sys.argv
args = parser.parse_args([])
return args
class EndpointHandler:
def __init__(self, path: str = ""):
"""Initialize the handler with model path and default config."""
# Use default args instead of parsing from command line
self.args = get_default_args()
self.args.model_base = path # Use the provided model path
# Initialize model
models_root_path = Path(path)
if not models_root_path.exists():
raise ValueError(f"`models_root` not exists: {models_root_path}")
self.model = HunyuanVideoSampler.from_pretrained(models_root_path, args=self.args)
def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""Process a single request
Args:
data: Dictionary containing:
- inputs (str): The prompt text
- resolution (str, optional): Video resolution like "1280x720"
- video_length (int, optional): Number of frames
- num_inference_steps (int, optional): Number of inference steps
- seed (int, optional): Random seed (-1 for random)
- guidance_scale (float, optional): Guidance scale value
- flow_shift (float, optional): Flow shift value
- embedded_guidance_scale (float, optional): Embedded guidance scale
Returns:
Dictionary containing the generated video as base64 string
"""
# Get inputs from request data
prompt = data.pop("inputs", None)
if prompt is None:
raise ValueError("No prompt provided in the 'inputs' field")
# Parse resolution
resolution = data.pop("resolution", "1280x720")
width, height = map(int, resolution.split("x"))
# Get other parameters with defaults
video_length = int(data.pop("video_length", 129))
seed = data.pop("seed", -1)
seed = None if seed == -1 else int(seed)
num_inference_steps = int(data.pop("num_inference_steps", 50))
guidance_scale = float(data.pop("guidance_scale", 1.0))
flow_shift = float(data.pop("flow_shift", 7.0))
embedded_guidance_scale = float(data.pop("embedded_guidance_scale", 6.0))
# Run inference
outputs = self.model.predict(
prompt=prompt,
height=height,
width=width,
video_length=video_length,
seed=seed,
negative_prompt="",
infer_steps=num_inference_steps,
guidance_scale=guidance_scale,
num_videos_per_prompt=1,
flow_shift=flow_shift,
batch_size=1,
embedded_guidance_scale=embedded_guidance_scale
)
# Get the video tensor
samples = outputs['samples']
sample = samples[0].unsqueeze(0)
# Save to temporary file
temp_path = "/tmp/temp_video.mp4"
save_videos_grid(sample, temp_path, fps=24)
# Read video file and convert to base64
with open(temp_path, "rb") as f:
video_bytes = f.read()
import base64
video_base64 = base64.b64encode(video_bytes).decode()
# Cleanup
os.remove(temp_path)
return {
"video_base64": video_base64,
"seed": outputs['seeds'][0],
"prompt": outputs['prompts'][0]
}
|