diff --git a/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/api_client.py b/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/api_client.py index 2c936d33f..8600fe393 100644 --- a/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/api_client.py +++ b/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/api_client.py @@ -6,6 +6,7 @@ The image generation part is derived from dougbtv/comfyui-vllm-omni by Doug (@do Original source at https://github.com/dougbtv/comfyui-vllm-omni, distributed under the MIT License. """ +import asyncio import json from typing import Any @@ -18,8 +19,8 @@ from .format import ( audio_to_base64, base64_to_audio, base64_to_image_tensor, - base64_to_video, bytes_to_audio, + bytes_to_video, image_tensor_to_base64, image_tensor_to_png_bytes, video_to_base64, @@ -31,10 +32,43 @@ from .types import AudioFormat logger = get_logger(__name__) +async def url_json(session: aiohttp.ClientSession, url: str, verb: str = "get", **kwargs) -> dict[str, Any]: + try: + async with getattr(session, verb)(url, **kwargs) as response: + if not response.ok: + error_text = await response.text() + raise (ValueError if response.status < 500 else RuntimeError)( + f"vLLM-Omni API returned status {response.status}: {error_text}" + ) + try: + return await response.json() + except aiohttp.ContentTypeError as e: + raise RuntimeError(f"Invalid JSON response from vLLM-Omni: {e}") + except aiohttp.ClientError as e: + raise RuntimeError(f"Network error connecting to vLLM-Omni at {url}: {e}") + + +async def url_bytes(session: aiohttp.ClientSession, url: str, verb: str = "get", **kwargs) -> bytes: + try: + async with getattr(session, verb)(url, **kwargs) as response: + if not response.ok: + error_text = await response.text() + raise (ValueError if response.status < 500 else RuntimeError)( + f"vLLM-Omni API returned status {response.status}: {error_text}" + ) + return await response.read() + except aiohttp.ClientError as e: + raise RuntimeError(f"Network error connecting to vLLM-Omni at {url}: {e}") + + class VLLMOmniClient: - def __init__(self, base_url: str, timeout: float | None = None): + def __init__( + self, base_url: str, timeout: float | None = None, poll_interval: float = 5.0, max_poll_duration: float = 60 * 5 + ): self.base_url = base_url self.timeout = aiohttp.ClientTimeout(total=timeout) + self.poll_interval = poll_interval + self.max_poll_duration = max_poll_duration async def generate_image( self, @@ -262,33 +296,41 @@ class VLLMOmniClient: content_type="image/png", ) - url = self.base_url + "/videos" async with aiohttp.ClientSession(timeout=self.timeout) as session: + # Start the video generation job + url = f"{self.base_url}/videos" + data = await url_json(session, url, "post", data=form) + if (job_id := data.get("id", None)) is None: + raise RuntimeError("API response missing job 'id' field - expected OpenAI compliant format") + if (job_status := data.get("status", None)) is None: + raise RuntimeError("API response missing job 'status' field - expected OpenAI compliant format") + + # Poll for video generation job completion + deadline = asyncio.get_running_loop().time() + self.max_poll_duration + url = f"{self.base_url}/videos/{job_id}" + while job_status not in {"completed", "failed"}: + await asyncio.sleep(self.poll_interval) + + data = await url_json(session, url) + if (job_status := data.get("status", None)) is None: + raise RuntimeError("API response missing job 'status' field - expected OpenAI compliant format") + if asyncio.get_running_loop().time() >= deadline: + raise RuntimeError(f"Timed out waiting for video job {job_id} to complete") + + if job_status == "failed": + raise RuntimeError(f"Video job failed: {data}") + + # Retrieve completed content + video_bytes = await url_bytes(session, f"{url}/content") + + # Decode video and make a best effort at cleaning up server resources try: - async with session.post(url, data=form) as response: - if not response.ok: - error_text = await response.text() - raise (ValueError if response.status < 500 else RuntimeError)( - f"vLLM-Omni API returned status {response.status}: {error_text}" - ) - - try: - data = await response.json() - except aiohttp.ContentTypeError as e: - raise RuntimeError(f"Invalid JSON response from vLLM-Omni: {e}") - except aiohttp.ClientError as e: - raise RuntimeError(f"Network error connecting to vLLM-Omni at {url}: {e}") - - if "data" not in data: - raise RuntimeError("API response missing 'data' field - expected OpenAI DALL-E format") - if not data["data"]: - raise RuntimeError("API returned empty data array") - try: - base64_str = data["data"][0]["b64_json"] - except (KeyError, IndexError): - raise RuntimeError("API response missing 'b64_json' field in first data item") - - return base64_to_video(base64_str) + return bytes_to_video(video_bytes) + finally: + try: + await url_json(session, url, "delete") + except Exception as exc: + logger.warning("Failed to clean up video job %s: %s", job_id, exc) async def generate_understanding_chat_completion( self, diff --git a/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/format.py b/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/format.py index 642c83c8d..42d396f06 100644 --- a/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/format.py +++ b/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/format.py @@ -143,15 +143,7 @@ def video_to_base64(video: VideoInput, filename: str = "video.mp4") -> str: return f"data:{mime_type};base64,{base64_str}" -def base64_to_video(base64_str: str) -> VideoInput: - if base64_str.startswith("data:video"): - _, base64_str = base64_str.split(",", 1) - - try: - video_bytes = base64.b64decode(base64_str) - except Exception as e: - raise ValueError(f"Invalid base64 string: {e}") - +def bytes_to_video(video_bytes: bytes) -> VideoInput: video_buffer = BytesIO(video_bytes) try: @@ -208,6 +200,18 @@ def base64_to_video(base64_str: str) -> VideoInput: return InputImpl.VideoFromComponents(components) +def base64_to_video(base64_str: str) -> VideoInput: + if base64_str.startswith("data:video"): + _, base64_str = base64_str.split(",", 1) + + try: + video_bytes = base64.b64decode(base64_str) + except Exception as e: + raise ValueError(f"Invalid base64 string: {e}") + + return bytes_to_video(video_bytes) + + def audio_to_bytes(audio: AudioInput, filename: str = "audio.mp3", quality: str = "128k") -> BytesIO: waveform = audio["waveform"][0] # Shape: (C, T) sample_rate = audio["sample_rate"] diff --git a/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/validators.py b/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/validators.py index c5ac4bdf4..f607f2cb8 100644 --- a/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/validators.py +++ b/apps/ComfyUI-vLLM-Omni/comfyui_vllm_omni/utils/validators.py @@ -70,13 +70,17 @@ def add_sampling_parameters_to_stage( stages = pipeline_spec["stages"] if isinstance(sampling_param_list, dict): + sampling_param_list = sampling_param_list.__class__(sampling_param_list) sampling_param_list.update(params_to_add) elif sampling_param_list is None: sampling_param_list = params_to_add.copy() else: for i, stage in enumerate(stages): if stage == stage_type: - sampling_param_list[i].update(params_to_add) + stage_param = sampling_param_list[i] + stage_param = stage_param.__class__(stage_param) + stage_param.update(params_to_add) + sampling_param_list[i] = stage_param return sampling_param_list diff --git a/docs/user_guide/examples/online_serving/image_to_video.md b/docs/user_guide/examples/online_serving/image_to_video.md index dcbb10067..a2b30320b 100644 --- a/docs/user_guide/examples/online_serving/image_to_video.md +++ b/docs/user_guide/examples/online_serving/image_to_video.md @@ -29,6 +29,35 @@ The script allows overriding: - `CACHE_BACKEND` (default: `none`) - `ENABLE_CACHE_DIT_SUMMARY` (default: `0`) +## Async Job Behavior + +`POST /v1/videos` is asynchronous. It creates a video job and immediately +returns metadata like the job ID and initial `queued` status. To get the final +artifact, poll the job status and then download the completed file from the +content endpoint. + +The main endpoints are: +- `POST /v1/videos`: create a video generation job +- `GET /v1/videos/{video_id}`: retrieve the current job status and metadata +- `GET /v1/videos`: list stored video jobs +- `GET /v1/videos/{video_id}/content`: download the generated video file +- `DELETE /v1/videos/{video_id}`: delete the job and any stored output + +## Storage + +Generated video files are stored on local disk by the async video API. +Local file storage behavior can be controlled via the following environment variables: + +- `VLLM_OMNI_STORAGE_PATH`: directory used for generated files (default: `/tmp/storage`) +- `VLLM_OMNI_STORAGE_MAX_CONCURRENCY`: max concurrent save/delete operations (default: `4`) + +Example: + +```bash +export VLLM_OMNI_STORAGE_PATH=/var/tmp/vllm-omni-videos +export VLLM_OMNI_STORAGE_MAX_CONCURRENCY=8 +``` + ## API Calls ### Method 1: Using curl @@ -38,7 +67,7 @@ The script allows overriding: bash run_curl_image_to_video.sh # Or execute directly (OpenAI-style multipart) -curl -X POST http://localhost:8091/v1/videos \ +create_response=$(curl -s http://localhost:8091/v1/videos \ -H "Accept: application/json" \ -F "prompt=A bear playing with yarn, smooth motion" \ -F "negative_prompt=low quality, blurry, static" \ @@ -52,7 +81,23 @@ curl -X POST http://localhost:8091/v1/videos \ -F "guidance_scale_2=1.0" \ -F "boundary_ratio=0.875" \ -F "flow_shift=12.0" \ - -F "seed=42" | jq -r '.data[0].b64_json' | base64 -d > wan22_i2v_output.mp4 + -F "seed=42") + +video_id=$(echo "$create_response" | jq -r '.id') +while true; do + status=$(curl -s "http://localhost:8091/v1/videos/${video_id}" | jq -r '.status') + if [ "$status" = "completed" ]; then + break + fi + if [ "$status" = "failed" ]; then + echo "Video generation failed" + exit 1 + fi + sleep 2 +done + +curl -s "http://localhost:8091/v1/videos/${video_id}" | jq . +curl -L "http://localhost:8091/v1/videos/${video_id}/content" -o wan22_i2v_output.mp4 ``` ## Request Format @@ -66,6 +111,18 @@ curl -X POST http://localhost:8091/v1/videos \ -F "input_reference=@/path/to/qwen-bear.png" ``` +### Alternative JSON-Safe Reference Input + +Use `image_reference` when you want to pass a URL or JSON-safe image reference +instead of uploading a file. Do not send `input_reference` and +`image_reference` together. + +```bash +curl -X POST http://localhost:8091/v1/videos \ + -F "prompt=A bear playing with yarn, smooth motion" \ + -F 'image_reference={"image_url":"https://example.com/qwen-bear.png"}' +``` + ### Generation with Parameters ```bash @@ -85,6 +142,63 @@ curl -X POST http://localhost:8091/v1/videos \ -F "seed=42" ``` +## Create Response Format + +`POST /v1/videos` returns a job record. + +```json +{ + "id": "video_gen_123", + "object": "video", + "status": "queued", + "model": "Wan-AI/Wan2.2-I2V-A14B-Diffusers", + "prompt": "A bear playing with yarn, smooth motion", + "created_at": 1234567890 +} +``` + +## Retrieve, List, Download, and Delete + +### Retrieve a job + +```bash +curl -s http://localhost:8091/v1/videos/${video_id} | jq . +``` + +### List jobs + +```bash +curl -s http://localhost:8091/v1/videos | jq . +``` + +### Download the completed video + +```bash +curl -L http://localhost:8091/v1/videos/${video_id}/content -o wan22_i2v_output.mp4 +``` + +### Delete a job and its stored file + +```bash +curl -X DELETE http://localhost:8091/v1/videos/${video_id} | jq . +``` + +## Poll Until Complete + +```bash +while true; do + status=$(curl -s http://localhost:8091/v1/videos/${video_id} | jq -r '.status') + if [ "$status" = "completed" ]; then + break + fi + if [ "$status" = "failed" ]; then + echo "Video generation failed" + exit 1 + fi + sleep 2 +done +``` + ## Example materials ??? abstract "run_curl_image_to_video.sh" diff --git a/docs/user_guide/examples/online_serving/text_to_video.md b/docs/user_guide/examples/online_serving/text_to_video.md index a632c49a7..309edb840 100644 --- a/docs/user_guide/examples/online_serving/text_to_video.md +++ b/docs/user_guide/examples/online_serving/text_to_video.md @@ -29,6 +29,34 @@ The script allows overriding: - `CACHE_BACKEND` (default: `none`) - `ENABLE_CACHE_DIT_SUMMARY` (default: `0`) +## Async Job Behavior + +`POST /v1/videos` is asynchronous. It creates a video job and immediately +returns metadata like the job ID and initial `queued` status. To get the final +artifact, poll the job status and then download the completed file from the +content endpoint. + +The main endpoints are: +- `POST /v1/videos`: create a video generation job +- `GET /v1/videos/{video_id}`: retrieve the current job status and metadata +- `GET /v1/videos`: list stored video jobs +- `GET /v1/videos/{video_id}/content`: download the generated video file +- `DELETE /v1/videos/{video_id}`: delete the job and any stored output + +## Storage + +Generated video files are stored on local disk by the async video API. + +- `VLLM_OMNI_STORAGE_PATH`: directory used for generated files (default: `/tmp/storage`) +- `VLLM_OMNI_STORAGE_MAX_CONCURRENCY`: max concurrent save/delete operations (default: `4`) + +Example: + +```bash +export VLLM_OMNI_STORAGE_PATH=/var/tmp/vllm-omni-videos +export VLLM_OMNI_STORAGE_MAX_CONCURRENCY=8 +``` + ## API Calls ### Method 1: Using curl @@ -38,7 +66,7 @@ The script allows overriding: bash run_curl_text_to_video.sh # Or execute directly (OpenAI-style multipart) -curl -s http://localhost:8091/v1/videos \ +create_response=$(curl -s http://localhost:8091/v1/videos \ -H "Accept: application/json" \ -F "prompt=Two anthropomorphic cats in comfy boxing gear and bright gloves fight intensely on a spotlighted stage." \ -F "width=832" \ @@ -50,7 +78,24 @@ curl -s http://localhost:8091/v1/videos \ -F "guidance_scale=4.0" \ -F "guidance_scale_2=4.0" \ -F "boundary_ratio=0.875" \ - -F "seed=42" | jq -r '.data[0].b64_json' | base64 -d > wan22_output.mp4 + -F "flow_shift=5.0" \ + -F "seed=42") + +video_id=$(echo "$create_response" | jq -r '.id') +while true; do + status=$(curl -s "http://localhost:8091/v1/videos/${video_id}" | jq -r '.status') + if [ "$status" = "completed" ]; then + break + fi + if [ "$status" = "failed" ]; then + echo "Video generation failed" + exit 1 + fi + sleep 2 +done + +curl -s "http://localhost:8091/v1/videos/${video_id}" | jq . +curl -L "http://localhost:8091/v1/videos/${video_id}/content" -o wan22_output.mp4 ``` ## Request Format @@ -85,8 +130,9 @@ curl -X POST http://localhost:8091/v1/videos \ | Parameter | Type | Default | Description | | --------------------- | ------ | ------- | ------------------------------------------------ | | `prompt` | str | - | Text description of the desired video | +| `seconds` | str | None | Clip duration in seconds | +| `size` | str | None | Output size in `WIDTHxHEIGHT` format | | `negative_prompt` | str | None | Negative prompt | -| `n` | int | 1 | Number of videos to generate | | `width` | int | None | Video width in pixels | | `height` | int | None | Video height in pixels | | `num_frames` | int | None | Number of frames to generate | @@ -98,24 +144,62 @@ curl -X POST http://localhost:8091/v1/videos \ | `flow_shift` | float | None | Scheduler flow shift (Wan2.2) | | `seed` | int | None | Random seed (reproducible) | | `lora` | object | None | LoRA configuration | -| `extra_body` | object | None | Model-specific extra parameters | -## Response Format +## Create Response Format + +`POST /v1/videos` returns a job record. ```json { - "created": 1234567890, - "data": [ - { "b64_json": "" } - ] + "id": "video_gen_123", + "object": "video", + "status": "queued", + "model": "Wan-AI/Wan2.2-T2V-A14B-Diffusers", + "prompt": "A cinematic view of a futuristic city at sunset", + "created_at": 1234567890 } ``` -## Extract Video +## Retrieve, List, Download, and Delete + +### Retrieve a job ```bash -# Extract base64 from response and decode to video -cat response.json | jq -r '.data[0].b64_json' | base64 -d > wan22_output.mp4 +curl -s http://localhost:8091/v1/videos/${video_id} | jq . +``` + +### List jobs + +```bash +curl -s http://localhost:8091/v1/videos | jq . +``` + +### Download the completed video + +```bash +curl -L http://localhost:8091/v1/videos/${video_id}/content -o wan22_output.mp4 +``` + +### Delete a job and its stored file + +```bash +curl -X DELETE http://localhost:8091/v1/videos/${video_id} | jq . +``` + +## Poll Until Complete + +```bash +while true; do + status=$(curl -s http://localhost:8091/v1/videos/${video_id} | jq -r '.status') + if [ "$status" = "completed" ]; then + break + fi + if [ "$status" = "failed" ]; then + echo "Video generation failed" + exit 1 + fi + sleep 2 +done ``` ## Example materials diff --git a/examples/online_serving/image_to_video/README.md b/examples/online_serving/image_to_video/README.md index 7e67a15e4..fea99efa6 100644 --- a/examples/online_serving/image_to_video/README.md +++ b/examples/online_serving/image_to_video/README.md @@ -26,6 +26,35 @@ The script allows overriding: - `CACHE_BACKEND` (default: `none`) - `ENABLE_CACHE_DIT_SUMMARY` (default: `0`) +## Async Job Behavior + +`POST /v1/videos` is asynchronous. It creates a video job and immediately +returns metadata like the job ID and initial `queued` status. To get the final +artifact, poll the job status and then download the completed file from the +content endpoint. + +The main endpoints are: +- `POST /v1/videos`: create a video generation job +- `GET /v1/videos/{video_id}`: retrieve the current job status and metadata +- `GET /v1/videos`: list stored video jobs +- `GET /v1/videos/{video_id}/content`: download the generated video file +- `DELETE /v1/videos/{video_id}`: delete the job and any stored output + +## Storage + +Generated video files are stored on local disk by the async video API. +Local file storage behavior can be controlled via the following environment variables: + +- `VLLM_OMNI_STORAGE_PATH`: directory used for generated files (default: `/tmp/storage`) +- `VLLM_OMNI_STORAGE_MAX_CONCURRENCY`: max concurrent save/delete operations (default: `4`) + +Example: + +```bash +export VLLM_OMNI_STORAGE_PATH=/var/tmp/vllm-omni-videos +export VLLM_OMNI_STORAGE_MAX_CONCURRENCY=8 +``` + ## API Calls ### Method 1: Using curl @@ -35,7 +64,7 @@ The script allows overriding: bash run_curl_image_to_video.sh # Or execute directly (OpenAI-style multipart) -curl -X POST http://localhost:8091/v1/videos \ +create_response=$(curl -s http://localhost:8091/v1/videos \ -H "Accept: application/json" \ -F "prompt=A bear playing with yarn, smooth motion" \ -F "negative_prompt=low quality, blurry, static" \ @@ -49,7 +78,23 @@ curl -X POST http://localhost:8091/v1/videos \ -F "guidance_scale_2=1.0" \ -F "boundary_ratio=0.875" \ -F "flow_shift=12.0" \ - -F "seed=42" | jq -r '.data[0].b64_json' | base64 -d > wan22_i2v_output.mp4 + -F "seed=42") + +video_id=$(echo "$create_response" | jq -r '.id') +while true; do + status=$(curl -s "http://localhost:8091/v1/videos/${video_id}" | jq -r '.status') + if [ "$status" = "completed" ]; then + break + fi + if [ "$status" = "failed" ]; then + echo "Video generation failed" + exit 1 + fi + sleep 2 +done + +curl -s "http://localhost:8091/v1/videos/${video_id}" | jq . +curl -L "http://localhost:8091/v1/videos/${video_id}/content" -o wan22_i2v_output.mp4 ``` ## Request Format @@ -63,6 +108,18 @@ curl -X POST http://localhost:8091/v1/videos \ -F "input_reference=@/path/to/qwen-bear.png" ``` +### Alternative JSON-Safe Reference Input + +Use `image_reference` when you want to pass a URL or JSON-safe image reference +instead of uploading a file. Do not send `input_reference` and +`image_reference` together. + +```bash +curl -X POST http://localhost:8091/v1/videos \ + -F "prompt=A bear playing with yarn, smooth motion" \ + -F 'image_reference={"image_url":"https://example.com/qwen-bear.png"}' +``` + ### Generation with Parameters ```bash @@ -81,3 +138,60 @@ curl -X POST http://localhost:8091/v1/videos \ -F "flow_shift=12.0" \ -F "seed=42" ``` + +## Create Response Format + +`POST /v1/videos` returns a job record, not inline base64 video data. + +```json +{ + "id": "video_gen_123", + "object": "video", + "status": "queued", + "model": "Wan-AI/Wan2.2-I2V-A14B-Diffusers", + "prompt": "A bear playing with yarn, smooth motion", + "created_at": 1234567890 +} +``` + +## Retrieve, List, Download, and Delete + +### Retrieve a job + +```bash +curl -s http://localhost:8091/v1/videos/${video_id} | jq . +``` + +### List jobs + +```bash +curl -s http://localhost:8091/v1/videos | jq . +``` + +### Download the completed video + +```bash +curl -L http://localhost:8091/v1/videos/${video_id}/content -o wan22_i2v_output.mp4 +``` + +### Delete a job and its stored file + +```bash +curl -X DELETE http://localhost:8091/v1/videos/${video_id} | jq . +``` + +## Poll Until Complete + +```bash +while true; do + status=$(curl -s http://localhost:8091/v1/videos/${video_id} | jq -r '.status') + if [ "$status" = "completed" ]; then + break + fi + if [ "$status" = "failed" ]; then + echo "Video generation failed" + exit 1 + fi + sleep 2 +done +``` diff --git a/examples/online_serving/image_to_video/run_curl_image_to_video.sh b/examples/online_serving/image_to_video/run_curl_image_to_video.sh index 5e6faf477..f4c1496a6 100644 --- a/examples/online_serving/image_to_video/run_curl_image_to_video.sh +++ b/examples/online_serving/image_to_video/run_curl_image_to_video.sh @@ -1,33 +1,75 @@ #!/bin/bash -# Wan2.2 image-to-video curl example (OpenAI-style multipart) +# Wan2.2 image-to-video curl example using the async video job API. + +set -euo pipefail INPUT_IMAGE="${INPUT_IMAGE:-../../offline_inference/image_to_video/qwen-bear.png}" +BASE_URL="${BASE_URL:-http://localhost:8099}" OUTPUT_PATH="${OUTPUT_PATH:-wan22_i2v_output.mp4}" NEGATIVE_PROMPT="${NEGATIVE_PROMPT:-}" +POLL_INTERVAL="${POLL_INTERVAL:-2}" if [ ! -f "$INPUT_IMAGE" ]; then echo "Input image not found: $INPUT_IMAGE" exit 1 fi -NEGATIVE_PROMPT_FLAG="" -if [ -n "$NEGATIVE_PROMPT" ]; then - NEGATIVE_PROMPT_FLAG="-F negative_prompt=${NEGATIVE_PROMPT}" +create_cmd=( + curl -sS -X POST "${BASE_URL}/v1/videos" + -H "Accept: application/json" + -F "prompt=A bear playing with yarn, smooth motion" + -F "input_reference=@${INPUT_IMAGE}" + -F "seconds=2" + -F "size=832x480" + -F "fps=16" + -F "num_inference_steps=40" + -F "guidance_scale=1.0" + -F "guidance_scale_2=1.0" + -F "boundary_ratio=0.875" + -F "flow_shift=12.0" + -F "seed=42" +) + +if [ -n "${NEGATIVE_PROMPT}" ]; then + create_cmd+=(-F "negative_prompt=${NEGATIVE_PROMPT}") fi -curl -X POST http://localhost:8099/v1/videos \ - -H "Accept: application/json" \ - -F "prompt=A bear playing with yarn, smooth motion" \ - $NEGATIVE_PROMPT_FLAG \ - -F "input_reference=@${INPUT_IMAGE}" \ - -F "seconds=2" \ - -F "size=832x480" \ - -F "fps=16" \ - -F "num_inference_steps=40" \ - -F "guidance_scale=1.0" \ - -F "guidance_scale_2=1.0" \ - -F "boundary_ratio=0.875" \ - -F "flow_shift=12.0" \ - -F "seed=42" | jq -r '.data[0].b64_json' | base64 -d > "${OUTPUT_PATH}" +create_response="$("${create_cmd[@]}")" +video_id="$(echo "${create_response}" | jq -r '.id')" +if [ -z "${video_id}" ] || [ "${video_id}" = "null" ]; then + echo "Failed to create video job:" + echo "${create_response}" | jq . + exit 1 +fi +echo "Created video job ${video_id}" +echo "${create_response}" | jq . + +while true; do + status_response="$(curl -sS "${BASE_URL}/v1/videos/${video_id}")" + status="$(echo "${status_response}" | jq -r '.status')" + + case "${status}" in + queued|in_progress) + echo "Video job ${video_id} status: ${status}" + sleep "${POLL_INTERVAL}" + ;; + completed) + echo "${status_response}" | jq . + break + ;; + failed) + echo "Video generation failed:" + echo "${status_response}" | jq . + exit 1 + ;; + *) + echo "Unexpected status response:" + echo "${status_response}" | jq . + exit 1 + ;; + esac +done + +curl -sS -L "${BASE_URL}/v1/videos/${video_id}/content" -o "${OUTPUT_PATH}" echo "Saved video to ${OUTPUT_PATH}" diff --git a/examples/online_serving/text_to_video/README.md b/examples/online_serving/text_to_video/README.md index a8c69e87b..92afb1fc9 100644 --- a/examples/online_serving/text_to_video/README.md +++ b/examples/online_serving/text_to_video/README.md @@ -26,6 +26,35 @@ The script allows overriding: - `CACHE_BACKEND` (default: `none`) - `ENABLE_CACHE_DIT_SUMMARY` (default: `0`) +## Async Job Behavior + +`POST /v1/videos` is asynchronous. It creates a video job and immediately +returns metadata like the job ID and initial `queued` status. To get the final +artifact, poll the job status and then download the completed file from the +content endpoint. + +The main endpoints are: +- `POST /v1/videos`: create a video generation job +- `GET /v1/videos/{video_id}`: retrieve the current job status and metadata +- `GET /v1/videos`: list stored video jobs +- `GET /v1/videos/{video_id}/content`: download the generated video file +- `DELETE /v1/videos/{video_id}`: delete the job and any stored output + +## Storage + +Generated video files are stored on local disk by the async video API. +Local file storage behavior can be controlled via the following environment variables: + +- `VLLM_OMNI_STORAGE_PATH`: directory used for generated files (default: `/tmp/storage`) +- `VLLM_OMNI_STORAGE_MAX_CONCURRENCY`: max concurrent save/delete operations (default: `4`) + +Example: + +```bash +export VLLM_OMNI_STORAGE_PATH=/var/tmp/vllm-omni-videos +export VLLM_OMNI_STORAGE_MAX_CONCURRENCY=8 +``` + ## API Calls ### Method 1: Using curl @@ -35,7 +64,7 @@ The script allows overriding: bash run_curl_text_to_video.sh # Or execute directly (OpenAI-style multipart) -curl -s http://localhost:8091/v1/videos \ +create_response=$(curl -s http://localhost:8091/v1/videos \ -H "Accept: application/json" \ -F "prompt=Two anthropomorphic cats in comfy boxing gear and bright gloves fight intensely on a spotlighted stage." \ -F "width=832" \ @@ -47,7 +76,24 @@ curl -s http://localhost:8091/v1/videos \ -F "guidance_scale=4.0" \ -F "guidance_scale_2=4.0" \ -F "boundary_ratio=0.875" \ - -F "seed=42" | jq -r '.data[0].b64_json' | base64 -d > wan22_output.mp4 + -F "flow_shift=5.0" \ + -F "seed=42") + +video_id=$(echo "$create_response" | jq -r '.id') +while true; do + status=$(curl -s "http://localhost:8091/v1/videos/${video_id}" | jq -r '.status') + if [ "$status" = "completed" ]; then + break + fi + if [ "$status" = "failed" ]; then + echo "Video generation failed" + exit 1 + fi + sleep 2 +done + +curl -s "http://localhost:8091/v1/videos/${video_id}" | jq . +curl -L "http://localhost:8091/v1/videos/${video_id}/content" -o wan22_output.mp4 ``` ## Request Format @@ -82,8 +128,9 @@ curl -X POST http://localhost:8091/v1/videos \ | Parameter | Type | Default | Description | | --------------------- | ------ | ------- | ------------------------------------------------ | | `prompt` | str | - | Text description of the desired video | +| `seconds` | str | None | Clip duration in seconds | +| `size` | str | None | Output size in `WIDTHxHEIGHT` format | | `negative_prompt` | str | None | Negative prompt | -| `n` | int | 1 | Number of videos to generate | | `width` | int | None | Video width in pixels | | `height` | int | None | Video height in pixels | | `num_frames` | int | None | Number of frames to generate | @@ -95,22 +142,60 @@ curl -X POST http://localhost:8091/v1/videos \ | `flow_shift` | float | None | Scheduler flow shift (Wan2.2) | | `seed` | int | None | Random seed (reproducible) | | `lora` | object | None | LoRA configuration | -| `extra_body` | object | None | Model-specific extra parameters | -## Response Format +## Create Response Format + +`POST /v1/videos` returns a job record, not inline base64 video data. ```json { - "created": 1234567890, - "data": [ - { "b64_json": "" } - ] + "id": "video_gen_123", + "object": "video", + "status": "queued", + "model": "Wan-AI/Wan2.2-T2V-A14B-Diffusers", + "prompt": "A cinematic view of a futuristic city at sunset", + "created_at": 1234567890 } ``` -## Extract Video +## Retrieve, List, Download, and Delete + +### Retrieve a job ```bash -# Extract base64 from response and decode to video -cat response.json | jq -r '.data[0].b64_json' | base64 -d > wan22_output.mp4 +curl -s http://localhost:8091/v1/videos/${video_id} | jq . +``` + +### List jobs + +```bash +curl -s http://localhost:8091/v1/videos | jq . +``` + +### Download the completed video + +```bash +curl -L http://localhost:8091/v1/videos/${video_id}/content -o wan22_output.mp4 +``` + +### Delete a job and its stored file + +```bash +curl -X DELETE http://localhost:8091/v1/videos/${video_id} | jq . +``` + +## Poll Until Complete + +```bash +while true; do + status=$(curl -s http://localhost:8091/v1/videos/${video_id} | jq -r '.status') + if [ "$status" = "completed" ]; then + break + fi + if [ "$status" = "failed" ]; then + echo "Video generation failed" + exit 1 + fi + sleep 2 +done ``` diff --git a/examples/online_serving/text_to_video/run_curl_text_to_video.sh b/examples/online_serving/text_to_video/run_curl_text_to_video.sh index 43889c398..a95fa53f8 100644 --- a/examples/online_serving/text_to_video/run_curl_text_to_video.sh +++ b/examples/online_serving/text_to_video/run_curl_text_to_video.sh @@ -1,20 +1,63 @@ #!/bin/bash -# Wan2.2 text-to-video curl example +# Wan2.2 text-to-video curl example using the async video job API. -OUTPUT_PATH="wan22_output.mp4" +set -euo pipefail -curl -X POST http://localhost:8098/v1/videos \ - -H "Accept: application/json" \ - -F "prompt=Two anthropomorphic cats in comfy boxing gear and bright gloves fight intensely on a spotlighted stage." \ - -F "seconds=2" \ - -F "size=832x480" \ - -F "negative_prompt=色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走" \ - -F "fps=16" \ - -F "num_inference_steps=40" \ - -F "guidance_scale=4.0" \ - -F "guidance_scale_2=4.0" \ - -F "boundary_ratio=0.875" \ - -F "flow_shift=5.0" \ - -F "seed=42" | jq -r '.data[0].b64_json' | base64 -d > "${OUTPUT_PATH}" +BASE_URL="${BASE_URL:-http://localhost:8098}" +OUTPUT_PATH="${OUTPUT_PATH:-wan22_output.mp4}" +POLL_INTERVAL="${POLL_INTERVAL:-2}" +create_response=$( + curl -sS -X POST "${BASE_URL}/v1/videos" \ + -H "Accept: application/json" \ + -F "prompt=Two anthropomorphic cats in comfy boxing gear and bright gloves fight intensely on a spotlighted stage." \ + -F "seconds=2" \ + -F "size=832x480" \ + -F "negative_prompt=色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走" \ + -F "fps=16" \ + -F "num_inference_steps=40" \ + -F "guidance_scale=4.0" \ + -F "guidance_scale_2=4.0" \ + -F "boundary_ratio=0.875" \ + -F "flow_shift=5.0" \ + -F "seed=42" +) + +video_id="$(echo "${create_response}" | jq -r '.id')" +if [ -z "${video_id}" ] || [ "${video_id}" = "null" ]; then + echo "Failed to create video job:" + echo "${create_response}" | jq . + exit 1 +fi + +echo "Created video job ${video_id}" +echo "${create_response}" | jq . + +while true; do + status_response="$(curl -sS "${BASE_URL}/v1/videos/${video_id}")" + status="$(echo "${status_response}" | jq -r '.status')" + + case "${status}" in + queued|in_progress) + echo "Video job ${video_id} status: ${status}" + sleep "${POLL_INTERVAL}" + ;; + completed) + echo "${status_response}" | jq . + break + ;; + failed) + echo "Video generation failed:" + echo "${status_response}" | jq . + exit 1 + ;; + *) + echo "Unexpected status response:" + echo "${status_response}" | jq . + exit 1 + ;; + esac +done + +curl -sS -L "${BASE_URL}/v1/videos/${video_id}/content" -o "${OUTPUT_PATH}" echo "Saved video to ${OUTPUT_PATH}" diff --git a/tests/comfyui/test_comfyui_integration.py b/tests/comfyui/test_comfyui_integration.py index d28818ab0..24f16d5e9 100644 --- a/tests/comfyui/test_comfyui_integration.py +++ b/tests/comfyui/test_comfyui_integration.py @@ -82,6 +82,15 @@ DIFFUSION_SINGLE_SAMPLING_PARAMS = DiffusionSamplingParams( } ) +DIFFUSION_VIDEO_SINGLE_SAMPLING_PARAMS = DiffusionSamplingParams( + { + "num_inference_steps": 30, + "guidance_scale": 6.0, + "true_cfg_scale": 1.5, + } +) + + AR_LIST_SAMPLING_PARAMS = [ AutoregressionSamplingParams( { @@ -310,8 +319,8 @@ def _build_mock_outputs(outputs: Iterable[OmniRequestOutput], sampling_case: Sam ) elif sampling_case.kind is SamplingKind.VIDEO_DIFFUSION_SINGLE: assert len(received_sampling_params_list) == 1 - expected = DIFFUSION_SINGLE_SAMPLING_PARAMS.copy() - expected["num_outputs_per_prompt"] = expected.pop("n") # convert from n to num_outputs_per_prompt + expected = DIFFUSION_VIDEO_SINGLE_SAMPLING_PARAMS.copy() + # expected["num_outputs_per_prompt"] = expected.pop("n") # convert from n to num_outputs_per_prompt _assert_sampling_param_values( received_sampling_params_list[0], { @@ -690,7 +699,7 @@ async def test_tts_nodes(api_server: str, node_cls, call_kwargs: dict, sampling_ pytest.param( SamplingCase( kind=SamplingKind.VIDEO_DIFFUSION_SINGLE, - sampling_params=DIFFUSION_SINGLE_SAMPLING_PARAMS, + sampling_params=DIFFUSION_VIDEO_SINGLE_SAMPLING_PARAMS, lora=LORA_PARAMS, ), id="single-diffusion-sampling-params", diff --git a/tests/e2e/online_serving/test_video_generation_api.py b/tests/e2e/online_serving/test_video_generation_api.py new file mode 100644 index 000000000..0711a1048 --- /dev/null +++ b/tests/e2e/online_serving/test_video_generation_api.py @@ -0,0 +1,260 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +""" +E2E online serving tests for the async video generation API. + +These tests exercise the real `/v1/videos` job lifecycle against a running +diffusion model, including create, retrieve, list, download, and delete flows. +""" + +import os +import time +import uuid +from pathlib import Path +from typing import Any + +import pytest +import requests + +from tests.conftest import OmniServer +from tests.utils import hardware_test + +os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn" +os.environ["VLLM_TEST_CLEAN_GPU_MEMORY"] = "0" + +MODEL = "Wan-AI/Wan2.2-T2V-A14B-Diffusers" +VIDEO_POLL_INTERVAL_S = 2.0 +VIDEO_TIMEOUT_S = 900.0 + + +@pytest.fixture(scope="module") +def video_server(): + """Start a real video generation server for async video API tests.""" + with OmniServer( + MODEL, + [ + "--num-gpus", + "1", + "--boundary-ratio", + "0.875", + "--flow-shift", + "5.0", + "--disable-log-stats", + ], + ) as server: + yield server + + +def _video_api_url(server: OmniServer, suffix: str = "") -> str: + return f"http://{server.host}:{server.port}/v1/videos{suffix}" + + +def _multipart_fields(payload: dict[str, Any]) -> list[tuple[str, tuple[None, str]]]: + return [(key, (None, str(value))) for key, value in payload.items() if value is not None] + + +def _create_video_job( + server: OmniServer, + *, + prompt: str | None = None, + **overrides: Any, +) -> requests.Response: + payload: dict[str, Any] = { + "prompt": prompt or f"video api e2e {uuid.uuid4().hex[:8]}", + "width": 640, + "height": 480, + "num_frames": 5, + "fps": 8, + "num_inference_steps": 2, + "guidance_scale": 1.0, + "guidance_scale_2": 1.0, + "boundary_ratio": 0.875, + "flow_shift": 5.0, + "seed": 42, + } + payload.update(overrides) + return requests.post( + _video_api_url(server), + files=_multipart_fields(payload), + timeout=VIDEO_TIMEOUT_S, + ) + + +def _retrieve_video_job(server: OmniServer, video_id: str) -> requests.Response: + return requests.get(_video_api_url(server, f"/{video_id}"), timeout=VIDEO_TIMEOUT_S) + + +def _wait_for_video_status(server: OmniServer, video_id: str, expected_status: str) -> dict[str, Any]: + deadline = time.time() + VIDEO_TIMEOUT_S + last_payload: dict[str, Any] | None = None + + while time.time() < deadline: + response = _retrieve_video_job(server, video_id) + assert response.status_code == 200, response.text + last_payload = response.json() + status = last_payload["status"] + if status == expected_status: + return last_payload + if status == "failed": + raise AssertionError(f"Video job {video_id} failed unexpectedly: {last_payload}") + time.sleep(VIDEO_POLL_INTERVAL_S) + + raise AssertionError( + f"Timed out waiting for video job {video_id} to reach status={expected_status}. Last payload: {last_payload}" + ) + + +def _wait_for_video_missing(server: OmniServer, video_id: str) -> None: + deadline = time.time() + 30.0 + last_status: int | None = None + + while time.time() < deadline: + response = _retrieve_video_job(server, video_id) + last_status = response.status_code + if response.status_code == 404: + return + time.sleep(0.5) + + raise AssertionError(f"Timed out waiting for video job {video_id} to disappear. Last status={last_status}") + + +def _delete_video_job(server: OmniServer, video_id: str) -> requests.Response: + return requests.delete(_video_api_url(server, f"/{video_id}"), timeout=VIDEO_TIMEOUT_S) + + +def _delete_video_job_with_retry(server: OmniServer, video_id: str) -> requests.Response: + deadline = time.time() + 30.0 + last_response: requests.Response | None = None + + while time.time() < deadline: + response = _delete_video_job(server, video_id) + last_response = response + if response.status_code != 409: + return response + time.sleep(1.0) + + raise AssertionError( + f"Timed out waiting to delete video job {video_id}. " + f"Last response: {None if last_response is None else last_response.text}" + ) + + +def _best_effort_delete(server: OmniServer, video_id: str) -> None: + try: + response = _delete_video_job_with_retry(server, video_id) + if response.status_code not in (200, 404): + print(f"Cleanup delete for {video_id} returned {response.status_code}: {response.text}") + except Exception as exc: + print(f"Cleanup delete for {video_id} failed: {exc}") + + +def _assert_mp4_payload(content: bytes) -> None: + assert len(content) > 32, f"Downloaded video payload is unexpectedly small: {len(content)} bytes" + assert content[4:8] == b"ftyp", "Downloaded payload does not look like an MP4 file." + + +@pytest.mark.core_model +@pytest.mark.diffusion +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}) +def test_create_list_retrieve_download_video(video_server: OmniServer, tmp_path: Path) -> None: + video_id: str | None = None + try: + create_response = _create_video_job(video_server, prompt="A small paper airplane gliding over a lake at dawn.") + assert create_response.status_code == 200, create_response.text + + created = create_response.json() + video_id = created["id"] + assert created["object"] == "video" + assert created["status"] == "queued" + assert created["model"] == MODEL + + completed = _wait_for_video_status(video_server, video_id, "completed") + assert completed["file_name"] is not None + assert completed["completed_at"] is not None + assert completed["progress"] == 100 + + retrieve_response = _retrieve_video_job(video_server, video_id) + assert retrieve_response.status_code == 200, retrieve_response.text + retrieved = retrieve_response.json() + assert retrieved["id"] == video_id + assert retrieved["status"] == "completed" + + list_response = requests.get(_video_api_url(video_server), timeout=VIDEO_TIMEOUT_S) + assert list_response.status_code == 200, list_response.text + listed = list_response.json() + assert listed["object"] == "list" + assert any(item["id"] == video_id for item in listed["data"]) + + download_response = requests.get( + _video_api_url(video_server, f"/{video_id}/content"), + timeout=VIDEO_TIMEOUT_S, + ) + assert download_response.status_code == 200, download_response.text + assert download_response.headers["content-type"].startswith("video/mp4") + assert completed["file_name"] in download_response.headers.get("content-disposition", "") + _assert_mp4_payload(download_response.content) + + output_path = tmp_path / completed["file_name"] + output_path.write_bytes(download_response.content) + assert output_path.stat().st_size == len(download_response.content) + finally: + if video_id is not None: + _best_effort_delete(video_server, video_id) + + +@pytest.mark.core_model +@pytest.mark.diffusion +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}) +def test_delete_video_while_queued_or_in_progress(video_server: OmniServer) -> None: + create_response = _create_video_job( + video_server, + prompt="A toy boat sailing through fog.", + num_frames=13, + num_inference_steps=4, + ) + assert create_response.status_code == 200, create_response.text + + created = create_response.json() + video_id = created["id"] + assert created["status"] == "queued" + + # Delete immediately so the job is still queued or just entering execution. + delete_response = _delete_video_job_with_retry(video_server, video_id) + assert delete_response.status_code == 200, delete_response.text + deleted = delete_response.json() + assert deleted["id"] == video_id + assert deleted["deleted"] is True + assert deleted["object"] == "video.deleted" + + _wait_for_video_missing(video_server, video_id) + + +@pytest.mark.core_model +@pytest.mark.diffusion +@hardware_test(res={"cuda": "H100", "rocm": "MI325"}) +def test_delete_video_after_processing(video_server: OmniServer) -> None: + create_response = _create_video_job( + video_server, + prompt="A lantern drifting above a quiet forest.", + seed=43, + ) + assert create_response.status_code == 200, create_response.text + + video_id = create_response.json()["id"] + completed = _wait_for_video_status(video_server, video_id, "completed") + assert completed["file_name"] is not None + + delete_response = _delete_video_job_with_retry(video_server, video_id) + assert delete_response.status_code == 200, delete_response.text + deleted = delete_response.json() + assert deleted["id"] == video_id + assert deleted["deleted"] is True + assert deleted["object"] == "video.deleted" + + _wait_for_video_missing(video_server, video_id) + + download_response = requests.get( + _video_api_url(video_server, f"/{video_id}/content"), + timeout=VIDEO_TIMEOUT_S, + ) + assert download_response.status_code == 404, download_response.text diff --git a/tests/entrypoints/openai_api/test_video_server.py b/tests/entrypoints/openai_api/test_video_server.py index 307780309..0f905d9e0 100644 --- a/tests/entrypoints/openai_api/test_video_server.py +++ b/tests/entrypoints/openai_api/test_video_server.py @@ -4,19 +4,31 @@ Unit tests for OpenAI-compatible video generation endpoints. """ +import asyncio +import base64 import io +import json +import os +import threading +import time from types import SimpleNamespace import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from PIL import Image -from pydantic import ValidationError from pytest_mock import MockerFixture +from vllm_omni.entrypoints.openai import api_server from vllm_omni.entrypoints.openai.api_server import router -from vllm_omni.entrypoints.openai.protocol.videos import VideoGenerationRequest, VideoResponseFormat +from vllm_omni.entrypoints.openai.protocol.videos import ( + VideoGenerationRequest, + VideoGenerationStatus, + VideoResponse, +) from vllm_omni.entrypoints.openai.serving_video import OmniOpenAIServingVideo +from vllm_omni.entrypoints.openai.storage import LocalStorageManager +from vllm_omni.entrypoints.openai.stores import AsyncDictStore, TaskRegistry pytestmark = [pytest.mark.core_model, pytest.mark.cpu] @@ -44,6 +56,38 @@ class FakeAsyncOmni: yield MockVideoResult(videos) +class BlockingVideoHandler: + def __init__(self): + self.model_name = "Wan-AI/Wan2.2-T2V-A14B-Diffusers" + self.stage_configs = None + self.started = threading.Event() + self.cancelled = threading.Event() + + def set_stage_configs_if_missing(self, stage_configs): + if self.stage_configs is None: + self.stage_configs = stage_configs + + async def generate_videos(self, request, reference_id, *, reference_image=None): + self.started.set() + try: + await asyncio.Future() + except asyncio.CancelledError: + self.cancelled.set() + raise + + +@pytest.fixture(autouse=True) +def isolated_video_backends(tmp_path, monkeypatch): + """Use isolated in-memory metadata and local storage for each test.""" + store: AsyncDictStore[VideoResponse] = AsyncDictStore() + tasks = TaskRegistry() + storage = LocalStorageManager(storage_path=str(tmp_path / "storage")) + monkeypatch.setattr(api_server, "VIDEO_STORE", store) + monkeypatch.setattr(api_server, "VIDEO_TASKS", tasks) + monkeypatch.setattr(api_server, "STORAGE_MANAGER", storage) + return store, tasks, storage + + @pytest.fixture def test_client(): app = FastAPI() @@ -52,7 +96,8 @@ def test_client(): diffusion_engine=FakeAsyncOmni(), model_name="Wan-AI/Wan2.2-T2V-A14B-Diffusers", ) - return TestClient(app) + with TestClient(app) as client: + yield client def _make_test_image_bytes(size=(64, 64)) -> bytes: @@ -62,6 +107,34 @@ def _make_test_image_bytes(size=(64, 64)) -> bytes: return buf.getvalue() +def _make_test_image_data_url(size=(64, 64)) -> str: + image_bytes = _make_test_image_bytes(size) + encoded = base64.b64encode(image_bytes).decode("utf-8") + return f"data:image/png;base64,{encoded}" + + +def _wait_for_status(client: TestClient, video_id: str, status: str, timeout_s: float = 2.0): + deadline = time.time() + timeout_s + last_payload = None + while time.time() < deadline: + response = client.get(f"/v1/videos/{video_id}") + assert response.status_code == 200 + last_payload = response.json() + if last_payload["status"] == status: + return last_payload + time.sleep(0.02) + raise AssertionError(f"Timed out waiting for status={status}. Last payload: {last_payload}") + + +def _wait_until(predicate, timeout_s: float = 2.0, interval_s: float = 0.02): + deadline = time.time() + timeout_s + while time.time() < deadline: + if predicate(): + return + time.sleep(interval_s) + raise AssertionError("Timed out waiting for condition") + + def test_t2v_video_generation_form(test_client, mocker: MockerFixture): fps_values = [] @@ -80,24 +153,22 @@ def test_t2v_video_generation_form(test_client, mocker: MockerFixture): "size": "640x360", "seconds": "2", "fps": "12", - "n": "2", }, ) assert response.status_code == 200 - data = response.json() - assert "data" in data and len(data["data"]) == 2 - assert all(item["b64_json"] == "Zg==" for item in data["data"]) + video_id = response.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) engine = test_client.app.state.openai_serving_video._engine_client captured = engine.captured_sampling_params_list[0] - assert captured.num_outputs_per_prompt == 2 + assert captured.num_outputs_per_prompt == 1 assert captured.width == 640 assert captured.height == 360 assert captured.num_frames == 24 assert captured.fps == 12 assert captured.frame_rate == 12.0 - assert fps_values == [12, 12] + assert fps_values == [12] def test_i2v_video_generation_form(test_client, mocker: MockerFixture): @@ -114,9 +185,8 @@ def test_i2v_video_generation_form(test_client, mocker: MockerFixture): ) assert response.status_code == 200 - data = response.json() - assert "data" in data and len(data["data"]) == 1 - assert data["data"][0]["b64_json"] == "Zg==" + video_id = response.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) engine = test_client.app.state.openai_serving_video._engine_client prompt = engine.captured_prompt @@ -145,6 +215,8 @@ def test_i2v_video_generation_resizes_input_to_requested_dimensions(test_client, ) assert response.status_code == 200 + video_id = response.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) engine = test_client.app.state.openai_serving_video._engine_client prompt = engine.captured_prompt @@ -153,6 +225,30 @@ def test_i2v_video_generation_resizes_input_to_requested_dimensions(test_client, assert input_image.size == (96, 64) +def test_i2v_video_generation_with_image_reference_form(test_client, mocker: MockerFixture): + mocker.patch( + "vllm_omni.entrypoints.openai.serving_video.encode_video_base64", + return_value="Zg==", + ) + response = test_client.post( + "/v1/videos", + data={ + "prompt": "A fox running through snow.", + "image_reference": json.dumps({"image_url": _make_test_image_data_url((40, 24))}), + }, + ) + + assert response.status_code == 200 + video_id = response.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) + + engine = test_client.app.state.openai_serving_video._engine_client + prompt = engine.captured_prompt + input_image = prompt["multi_modal_data"]["image"] + assert isinstance(input_image, Image.Image) + assert input_image.size == (40, 24) + + def test_seconds_defaults_fps_and_frames(test_client, mocker: MockerFixture): fps_values = [] @@ -173,10 +269,13 @@ def test_seconds_defaults_fps_and_frames(test_client, mocker: MockerFixture): ) assert response.status_code == 200 + video_id = response.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) engine = test_client.app.state.openai_serving_video._engine_client captured = engine.captured_sampling_params_list[0] assert captured.num_frames == 72 assert captured.fps == 24 + assert captured.frame_rate == 24.0 assert fps_values == [24] @@ -194,16 +293,50 @@ def test_size_param_sets_width_height(test_client, mocker: MockerFixture): ) assert response.status_code == 200 + video_id = response.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) engine = test_client.app.state.openai_serving_video._engine_client captured = engine.captured_sampling_params_list[0] assert captured.width == 320 assert captured.height == 240 +def test_sampling_params_pass_through(test_client, mocker: MockerFixture): + mocker.patch( + "vllm_omni.entrypoints.openai.serving_video.encode_video_base64", + return_value="Zg==", + ) + response = test_client.post( + "/v1/videos", + data={ + "prompt": "param pass", + "num_inference_steps": "30", + "guidance_scale": "6.5", + "guidance_scale_2": "8.0", + "true_cfg_scale": "4.0", + "boundary_ratio": "0.7", + "flow_shift": "0.25", + }, + ) + + assert response.status_code == 200 + video_id = response.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) + engine = test_client.app.state.openai_serving_video._engine_client + captured = engine.captured_sampling_params_list[0] + assert captured.num_inference_steps == 30 + assert captured.guidance_scale == 6.5 + assert captured.guidance_scale_2 == 8.0 + assert captured.true_cfg_scale == 4.0 + assert captured.boundary_ratio == 0.7 + assert captured.extra_args["flow_shift"] == 0.25 + + def test_audio_sample_rate_comes_from_model_config(test_client, mocker: MockerFixture): audio_sample_rates = [] def _fake_encode(video, fps, audio=None, audio_sample_rate=None): + del video, fps, audio audio_sample_rates.append(audio_sample_rate) return "Zg==" @@ -233,38 +366,11 @@ def test_audio_sample_rate_comes_from_model_config(test_client, mocker: MockerFi ) assert response.status_code == 200 + video_id = response.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) assert audio_sample_rates == [16000] -def test_sampling_params_pass_through(test_client, mocker: MockerFixture): - mocker.patch( - "vllm_omni.entrypoints.openai.serving_video.encode_video_base64", - return_value="Zg==", - ) - response = test_client.post( - "/v1/videos", - data={ - "prompt": "param pass", - "num_inference_steps": "30", - "guidance_scale": "6.5", - "guidance_scale_2": "8.0", - "true_cfg_scale": "4.0", - "boundary_ratio": "0.7", - "flow_shift": "0.25", - }, - ) - - assert response.status_code == 200 - engine = test_client.app.state.openai_serving_video._engine_client - captured = engine.captured_sampling_params_list[0] - assert captured.num_inference_steps == 30 - assert captured.guidance_scale == 6.5 - assert captured.guidance_scale_2 == 8.0 - assert captured.true_cfg_scale == 4.0 - assert captured.boundary_ratio == 0.7 - assert captured.extra_args["flow_shift"] == 0.25 - - def test_missing_handler_returns_503(): app = FastAPI() app.include_router(router) @@ -287,29 +393,29 @@ def test_missing_prompt_returns_422(test_client): assert response.status_code == 422 -def test_invalid_size_format_raises_validation_error(test_client): - with pytest.raises(ValidationError): - test_client.post( - "/v1/videos", - data={"prompt": "bad size", "size": "invalid"}, - ) - - -def test_invalid_size_parse_returns_500(test_client): +def test_invalid_size_parse_returns_422(test_client): response = test_client.post( "/v1/videos", data={"prompt": "bad size", "size": "640x"}, ) - assert response.status_code == 500 - assert "video generation failed" in response.json()["detail"].lower() + assert response.status_code == 422 + body = response.json() + assert body["detail"][0]["loc"] == ["body", "size"] + assert body["detail"][0]["type"] == "string_pattern_mismatch" + assert body["detail"][0]["input"] == "640x" -def test_invalid_response_format_raises_validation_error(test_client): - with pytest.raises(ValidationError): - test_client.post( - "/v1/videos", - data={"prompt": "bad format", "response_format": "url"}, - ) +def test_rejects_input_reference_and_image_reference_together(test_client): + response = test_client.post( + "/v1/videos", + data={ + "prompt": "bad refs", + "image_reference": '{"image_url": "https://example.com/cat.png"}', + }, + files={"input_reference": ("input.png", _make_test_image_bytes(), "image/png")}, + ) + assert response.status_code == 400 + assert "either input_reference or image_reference" in response.json()["detail"].lower() def test_invalid_seconds_returns_422(test_client): @@ -320,20 +426,6 @@ def test_invalid_seconds_returns_422(test_client): assert response.status_code == 422 -def test_invalid_n_raises_validation_error(test_client): - with pytest.raises(ValidationError): - test_client.post( - "/v1/videos", - data={"prompt": "bad n", "n": "0"}, - ) - - with pytest.raises(ValidationError): - test_client.post( - "/v1/videos", - data={"prompt": "bad n", "n": "5"}, - ) - - def test_negative_prompt_and_seed_pass_through(test_client, mocker: MockerFixture): mocker.patch( "vllm_omni.entrypoints.openai.serving_video.encode_video_base64", @@ -349,6 +441,8 @@ def test_negative_prompt_and_seed_pass_through(test_client, mocker: MockerFixtur ) assert response.status_code == 200 + video_id = response.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) engine = test_client.app.state.openai_serving_video._engine_client captured_prompt = engine.captured_prompt captured_params = engine.captured_sampling_params_list[0] @@ -364,19 +458,38 @@ def test_invalid_lora_returns_400(test_client): "lora": '{"name": "bad-lora"}', }, ) + assert response.status_code == 200 + video_id = response.json()["id"] + failed = _wait_for_status(test_client, video_id, VideoGenerationStatus.FAILED.value) + assert failed["error"]["code"] == "HTTPException" + assert "lora object" in failed["error"]["message"].lower() + + +def test_unsupported_image_reference_file_id_returns_400(test_client): + response = test_client.post( + "/v1/videos", + data={ + "prompt": "unsupported ref", + "image_reference": '{"file_id": "file-123"}', + }, + ) assert response.status_code == 400 - assert "lora" in response.json()["detail"].lower() + assert response.json()["detail"] == "Invalid image_reference: file_id is not supported yet." + + +def test_invalid_uploaded_input_reference_returns_400(test_client): + response = test_client.post( + "/v1/videos", + data={"prompt": "bad upload"}, + files={"input_reference": ("input.png", b"not-an-image", "image/png")}, + ) + assert response.status_code == 400 + assert response.json()["detail"] == "Invalid input_reference: provided content is not a valid image." def test_video_request_validation(): req = VideoGenerationRequest(prompt="test") assert req.prompt == "test" - assert req.n == 1 - assert req.response_format == VideoResponseFormat.B64_JSON - - with pytest.raises(ValueError): - VideoGenerationRequest(prompt="test", response_format="url") - with pytest.raises(ValueError): VideoGenerationRequest(prompt="test", size="invalid") @@ -384,4 +497,144 @@ def test_video_request_validation(): VideoGenerationRequest(prompt="test", seconds="abc") with pytest.raises(ValueError): - VideoGenerationRequest(prompt="test", n=0) + VideoGenerationRequest(prompt="test", image_reference={"file_id": "file-1", "image_url": "https://example.com"}) + + +def test_list_videos_supports_order_after_and_limit(test_client, mocker: MockerFixture): + mocker.patch( + "vllm_omni.entrypoints.openai.serving_video.encode_video_base64", + return_value="Zg==", + ) + ids = [] + for i in range(3): + create_resp = test_client.post("/v1/videos", data={"prompt": f"video-{i}"}) + assert create_resp.status_code == 200 + video_id = create_resp.json()["id"] + _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) + ids.append(video_id) + + asyncio.run(api_server.VIDEO_STORE.update_fields(ids[0], {"created_at": 100})) + asyncio.run(api_server.VIDEO_STORE.update_fields(ids[1], {"created_at": 200})) + asyncio.run(api_server.VIDEO_STORE.update_fields(ids[2], {"created_at": 300})) + + asc_resp = test_client.get("/v1/videos", params={"order": "asc"}) + assert asc_resp.status_code == 200 + asc_body = asc_resp.json() + asc_ids = [item["id"] for item in asc_body["data"]] + assert asc_ids == [ids[0], ids[1], ids[2]] + assert asc_body["object"] == "list" + assert asc_body["first_id"] == ids[0] + assert asc_body["last_id"] == ids[2] + assert asc_body["has_more"] is False + + desc_resp = test_client.get("/v1/videos", params={"order": "desc", "limit": 2}) + assert desc_resp.status_code == 200 + desc_body = desc_resp.json() + desc_ids = [item["id"] for item in desc_body["data"]] + assert desc_ids == [ids[2], ids[1]] + assert desc_body["object"] == "list" + assert desc_body["first_id"] == ids[2] + assert desc_body["last_id"] == ids[1] + assert desc_body["has_more"] is True + + after_resp = test_client.get("/v1/videos", params={"order": "asc", "after": ids[0]}) + assert after_resp.status_code == 200 + after_body = after_resp.json() + after_ids = [item["id"] for item in after_body["data"]] + assert after_ids == [ids[1], ids[2]] + assert after_body["object"] == "list" + assert after_body["first_id"] == ids[1] + assert after_body["last_id"] == ids[2] + assert after_body["has_more"] is False + + zero_limit_resp = test_client.get("/v1/videos", params={"order": "asc", "limit": 0}) + assert zero_limit_resp.status_code == 200 + zero_limit_body = zero_limit_resp.json() + assert zero_limit_body["data"] == [] + assert zero_limit_body["object"] == "list" + assert zero_limit_body["first_id"] is None + assert zero_limit_body["last_id"] is None + assert zero_limit_body["has_more"] is True + + zero_limit_after_resp = test_client.get( + "/v1/videos", + params={"order": "asc", "after": ids[2], "limit": 0}, + ) + assert zero_limit_after_resp.status_code == 200 + zero_limit_after_body = zero_limit_after_resp.json() + assert zero_limit_after_body["data"] == [] + assert zero_limit_after_body["object"] == "list" + assert zero_limit_after_body["first_id"] is None + assert zero_limit_after_body["last_id"] is None + assert zero_limit_after_body["has_more"] is False + + +def test_delete_completed_job_removes_file_and_metadata(test_client, mocker: MockerFixture): + mocker.patch( + "vllm_omni.entrypoints.openai.serving_video.encode_video_base64", + return_value="Zg==", + ) + create_resp = test_client.post("/v1/videos", data={"prompt": "Delete this video"}) + assert create_resp.status_code == 200 + video_id = create_resp.json()["id"] + + final = _wait_for_status(test_client, video_id, VideoGenerationStatus.COMPLETED.value) + file_name = final["file_name"] + assert file_name is not None + file_path = os.path.join(api_server.STORAGE_MANAGER.storage_path, file_name) + assert os.path.exists(file_path) + + delete_resp = test_client.delete(f"/v1/videos/{video_id}") + assert delete_resp.status_code == 200 + assert delete_resp.json()["id"] == video_id + assert delete_resp.json()["deleted"] is True + assert delete_resp.json()["object"] == "video.deleted" + assert not os.path.exists(file_path) + + +def test_delete_in_progress_job_cancels_task_and_removes_metadata(test_client): + handler = BlockingVideoHandler() + test_client.app.state.openai_serving_video = handler + + create_resp = test_client.post("/v1/videos", data={"prompt": "Cancel this video"}) + assert create_resp.status_code == 200 + video_id = create_resp.json()["id"] + + assert handler.started.wait(timeout=2.0) + + delete_resp = test_client.delete(f"/v1/videos/{video_id}") + assert delete_resp.status_code == 200 + assert delete_resp.json()["id"] == video_id + assert delete_resp.json()["deleted"] is True + assert delete_resp.json()["object"] == "video.deleted" + + assert handler.cancelled.wait(timeout=2.0) + _wait_until(lambda: asyncio.run(api_server.VIDEO_TASKS.get(video_id)) is None) + assert asyncio.run(api_server.VIDEO_STORE.get(video_id)) is None + + retrieve_resp = test_client.get(f"/v1/videos/{video_id}") + assert retrieve_resp.status_code == 404 + + +def test_video_response_file_extension_is_robust(): + response = VideoResponse(model="test-model", prompt="Make something beautiful") + assert response.file_extension == "mp4" + + with_params = VideoResponse.model_construct( + model="test-model", + media_type="video/mp4; charset=binary", + ) + assert with_params.file_extension == "mp4" + + webm = VideoResponse.model_construct( + model="test-model", + media_type="video/webm", + ) + assert webm.file_extension == "webm" + + with pytest.raises(ValueError): + unknown = VideoResponse.model_construct( + model="test-model", + media_type="application/x-custom-video", + ) + _ = unknown.file_extension diff --git a/vllm_omni/benchmarks/metrics/metrics.py b/vllm_omni/benchmarks/metrics/metrics.py index b68418e17..a2acc7d75 100644 --- a/vllm_omni/benchmarks/metrics/metrics.py +++ b/vllm_omni/benchmarks/metrics/metrics.py @@ -228,8 +228,8 @@ def calculate_metrics( good_completed += 1 if completed == 0: - warnings.formatwarning = ( - lambda msg, category, filename, lineno, line=None: f"{filename}:{lineno}: {category.__name__}: {msg}\n" + warnings.formatwarning = lambda msg, category, filename, lineno, line=None: ( + f"{filename}:{lineno}: {category.__name__}: {msg}\n" ) warnings.warn( "All requests failed. This is likely due to a misconfiguration on the benchmark arguments.", diff --git a/vllm_omni/diffusion/envs.py b/vllm_omni/diffusion/envs.py index dd566b059..a71dc2e8e 100644 --- a/vllm_omni/diffusion/envs.py +++ b/vllm_omni/diffusion/envs.py @@ -20,7 +20,7 @@ environment_variables: dict[str, Callable[[], Any]] = { # used in distributed environment to determine the master address "MASTER_ADDR": lambda: os.getenv("MASTER_ADDR", ""), # used in distributed environment to manually set the communication port - "MASTER_PORT": lambda: (int(os.getenv("MASTER_PORT", "0")) if "MASTER_PORT" in os.environ else None), + "MASTER_PORT": lambda: int(os.getenv("MASTER_PORT", "0")) if "MASTER_PORT" in os.environ else None, # path to cudatoolkit home directory, under which should be bin, include, # and lib directories. "CUDA_HOME": lambda: os.environ.get("CUDA_HOME", None), diff --git a/vllm_omni/entrypoints/openai/api_server.py b/vllm_omni/entrypoints/openai/api_server.py index 1507ce8c3..f73cf3c8d 100644 --- a/vllm_omni/entrypoints/openai/api_server.py +++ b/vllm_omni/entrypoints/openai/api_server.py @@ -1,5 +1,6 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project +import asyncio import base64 import io import json @@ -15,12 +16,12 @@ from argparse import Namespace from collections.abc import AsyncIterator from contextlib import asynccontextmanager from http import HTTPStatus -from typing import Annotated, Any, cast +from typing import Annotated, Any, Literal, cast import httpx import vllm.envs as envs -from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile, WebSocket -from fastapi.responses import JSONResponse, StreamingResponse +from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Request, UploadFile, WebSocket +from fastapi.responses import FileResponse, JSONResponse, StreamingResponse from PIL import Image from starlette.datastructures import State from starlette.routing import Route @@ -82,6 +83,7 @@ from vllm.tool_parsers import ToolParserManager from vllm.utils.system_utils import decorate_logs from vllm_omni.entrypoints.async_omni import AsyncOmni +from vllm_omni.entrypoints.openai.errors import InvalidInputReferenceError from vllm_omni.entrypoints.openai.image_api_utils import ( encode_image_base64, parse_size, @@ -93,13 +95,22 @@ from vllm_omni.entrypoints.openai.protocol.images import ( ImageGenerationResponse, ) from vllm_omni.entrypoints.openai.protocol.videos import ( + SecondStr, + SizeStr, + VideoDeleteResponse, + VideoError, VideoGenerationRequest, - VideoGenerationResponse, + VideoGenerationStatus, + VideoListResponse, + VideoResponse, ) from vllm_omni.entrypoints.openai.serving_chat import OmniOpenAIServingChat from vllm_omni.entrypoints.openai.serving_speech import OmniOpenAIServingSpeech from vllm_omni.entrypoints.openai.serving_speech_stream import OmniStreamingSpeechHandler -from vllm_omni.entrypoints.openai.serving_video import OmniOpenAIServingVideo +from vllm_omni.entrypoints.openai.serving_video import OmniOpenAIServingVideo, ReferenceImage +from vllm_omni.entrypoints.openai.storage import STORAGE_MANAGER +from vllm_omni.entrypoints.openai.stores import VIDEO_STORE, VIDEO_TASKS +from vllm_omni.entrypoints.openai.video_api_utils import decode_input_reference from vllm_omni.inputs.data import OmniDiffusionSamplingParams, OmniSamplingParams, OmniTextPrompt from vllm_omni.lora.request import LoRARequest from vllm_omni.lora.utils import stable_lora_int_id @@ -827,6 +838,20 @@ _remove_route_from_router(router, "/v1/audio/speech", {"POST"}) @with_cancellation @load_aware_call async def create_speech(request: OpenAICreateSpeechRequest, raw_request: Request): + """Generate speech audio from text using the loaded TTS model. + + Args: + request: Speech synthesis request in OpenAI-compatible format. + raw_request: Raw FastAPI request for accessing app state. + + Returns: + The generated audio response, or an OpenAI-style error payload when + the request cannot be fulfilled. + + Raises: + HTTPException: If the server does not support speech generation or the + synthesis request fails unexpectedly. + """ handler = Omnispeech(raw_request) if handler is None: base_server = getattr(raw_request.app.state, "openai_serving_tokenization", None) @@ -858,7 +883,16 @@ async def create_speech(request: OpenAICreateSpeechRequest, raw_request: Request }, ) async def list_voices(raw_request: Request): - """List available TTS voices/speakers from the loaded model.""" + """List available TTS voices exposed by the loaded speech model. + + Args: + raw_request: Raw FastAPI request for accessing app state. + + Returns: + A JSON payload containing the sorted set of supported speaker names, or + an OpenAI-style error response when the current server configuration + does not support the Speech API. + """ handler = Omnispeech(raw_request) if handler is None: return base(raw_request).create_error_response(message="The model does not support Speech API") @@ -1401,7 +1435,7 @@ async def _generate_with_async_omni( return result -def _update_if_not_none(object: any, key: str, val: any) -> None: +def _update_if_not_none(object: Any, key: str, val: Any) -> None: if val is not None: setattr(object, key, val) @@ -1507,7 +1541,7 @@ def _encode_image_base64_with_compression( def apply_stage_default_sampling_params( default_params_json: str | None, - sampling_params: any, + sampling_params: Any, stage_key: str, ) -> None: """ @@ -1527,29 +1561,16 @@ def apply_stage_default_sampling_params( setattr(sampling_params, param_name, param_value) -async def _run_video_generation( - request: VideoGenerationRequest, - raw_request: Request, - *, - input_reference_bytes: bytes | None = None, -) -> VideoGenerationResponse: - handler = Omnivideo(raw_request) - if handler is None: - raise HTTPException( - status_code=HTTPStatus.SERVICE_UNAVAILABLE.value, - detail="Video generation handler not initialized.", - ) - logger.info("Video generation handler: %s", type(handler).__name__) - try: - return await handler.generate_videos(request, raw_request, input_reference_bytes=input_reference_bytes) - except HTTPException: - raise - except Exception as e: - logger.exception("Video generation failed: %s", e) - raise HTTPException( - status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, - detail=f"Video generation failed: {str(e)}", - ) +def _resolve_video_runtime_context(raw_request: Request) -> tuple[str | None, list[Any] | None]: + app_model_name = None + serving_models = getattr(raw_request.app.state, "openai_serving_models", None) + if serving_models and getattr(serving_models, "base_model_paths", None): + base_paths = serving_models.base_model_paths + if base_paths: + app_model_name = base_paths[0].name + + app_stage_configs = getattr(raw_request.app.state, "stage_configs", None) + return app_model_name, app_stage_configs def _parse_form_json(value: str | None) -> Any: @@ -1564,10 +1585,97 @@ def _parse_form_json(value: str | None) -> Any: ) from exc +def video_response_from_request(model_name: str, req: VideoGenerationRequest) -> VideoResponse: + resp = VideoResponse( + model=model_name, + status=VideoGenerationStatus.QUEUED, + size=req.size, + prompt=req.prompt, + ) + resp.seconds = str(req.seconds or resp.seconds) + return resp + + +async def decode_and_save_video_output(output: Any, file_name: str) -> str: + if not output.b64_json: + raise RuntimeError(f"Video output for {file_name} did not include b64_json content.") + + try: + video_bytes = base64.b64decode(output.b64_json) + except Exception as decode_exc: + raise RuntimeError(f"Failed to decode generated video payload for {file_name}") from decode_exc + + return await STORAGE_MANAGER.save(video_bytes, file_name) + + +def _cleanup_video(video_id: str, output_path: str | None): + try: + if output_path is not None: + os.remove(output_path) + except OSError: + logger.warning("Failed to cleanup partial video file '%s' for id=%s", output_path, video_id) + + +async def _run_video_generation_job( + handler: OmniOpenAIServingVideo, + request: VideoGenerationRequest, + video_id: str, + reference_image: ReferenceImage | None = None, +) -> None: + job = await VIDEO_STORE.get(video_id) + if job is None: + logger.warning("Video job %s missing before generation task started; skipping", video_id) + return + + await VIDEO_STORE.update_fields(video_id, {"status": VideoGenerationStatus.IN_PROGRESS}) + started_at = time.perf_counter() + output_path = None + try: + response = await handler.generate_videos(request, video_id, reference_image=reference_image) + if not response.data: + raise RuntimeError("Video generation completed but returned no outputs.") + + if (video_count := len(response.data)) > 1: + logger.warning("Video request %s generated %s outputs but we only expected one.", video_id, video_count) + + file_name = f"{video_id}.{job.file_extension}" + output_path = await decode_and_save_video_output(response.data[0], file_name) + logger.info("Video request %s persisted %s output file.", video_id, output_path) + + await VIDEO_STORE.update_fields( + video_id, + { + "status": VideoGenerationStatus.COMPLETED, + "progress": 100, + "file_name": file_name, + "completed_at": int(time.time()), + "inference_time_s": time.perf_counter() - started_at, + }, + ) + except Exception as exc: + logger.exception("Video generation failed for id=%s", video_id) + + _cleanup_video(video_id, output_path) + # TODO: It would be better to have a finite collection of errors to return rather than the exception name + await VIDEO_STORE.update_fields( + video_id, + { + "status": VideoGenerationStatus.FAILED, + "completed_at": int(time.time()), + "error": VideoError(code=type(exc).__name__, message=str(exc)), + "inference_time_s": time.perf_counter() - started_at, + }, + ) + except asyncio.CancelledError: + _cleanup_video(video_id, output_path) + await VIDEO_STORE.pop(video_id) + raise + + @router.post( "/v1/videos", responses={ - HTTPStatus.OK.value: {"model": VideoGenerationResponse}, + HTTPStatus.OK.value: {"model": VideoResponse}, HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse}, HTTPStatus.SERVICE_UNAVAILABLE.value: {"model": ErrorResponse}, HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse}, @@ -1577,11 +1685,10 @@ async def create_video( raw_request: Request, prompt: str = Form(...), input_reference: UploadFile | None = File(default=None), + image_reference: str | None = Form(default=None), model: str | None = Form(default=None), - n: int | None = Form(default=None), - seconds: int | None = Form(default=None), - size: str | None = Form(default=None), - response_format: str | None = Form(default=None), + seconds: SecondStr | None = Form(default=None), + size: SizeStr | None = Form(default=None), user: str | None = Form(default=None), width: int | None = Form(default=None), height: int | None = Form(default=None), @@ -1596,19 +1703,60 @@ async def create_video( seed: int | None = Form(default=None), negative_prompt: str | None = Form(default=None), lora: str | None = Form(default=None), -) -> VideoGenerationResponse: - """OpenAI-style video create endpoint (multipart form-data).""" +) -> VideoResponse: + """Create an asynchronous video generation job. + + This OpenAI-style endpoint accepts multipart form-data, validates the + request payload, persists a queued job record, and starts generation in the + background. The response contains metadata for polling job status rather + than the generated video bytes. + + Args: + raw_request: Raw FastAPI request for accessing app state. + prompt: Text prompt describing the requested video. + input_reference: Optional uploaded reference image file. + image_reference: Optional JSON-encoded reference image descriptor. + model: Optional model name supplied by the client. + seconds: Optional target duration string accepted by the video API. + size: Optional output size string such as ``1280x720``. + user: Optional user identifier forwarded in the stored request. + width: Optional explicit output width override. + height: Optional explicit output height override. + num_frames: Optional explicit frame count override. + fps: Optional explicit frame rate override. + num_inference_steps: Optional inference step override. + guidance_scale: Optional primary guidance scale override. + guidance_scale_2: Optional secondary guidance scale override. + boundary_ratio: Optional boundary ratio override. + flow_shift: Optional flow shift override. + true_cfg_scale: Optional true CFG scale override. + seed: Optional random seed override. + negative_prompt: Optional negative prompt. + lora: Optional JSON-encoded per-request LoRA configuration. + + Returns: + A queued ``VideoResponse`` that includes the generated job identifier + and initial metadata for later retrieval. + + Raises: + HTTPException: If the request is invalid, the video handler is + unavailable, or job initialization fails. + """ input_reference_bytes = await input_reference.read() if input_reference is not None else None + parsed_image_reference = _parse_form_json(image_reference) + if parsed_image_reference is not None and input_reference_bytes is not None: + raise HTTPException( + status_code=HTTPStatus.BAD_REQUEST.value, + detail="Provide either input_reference or image_reference, not both.", + ) request_data: dict[str, Any] = { "prompt": prompt, "model": model, "seconds": seconds, "size": size, + "image_reference": parsed_image_reference, "user": user, - "response_format": response_format, - "input_reference": None, - "n": n, "width": width, "height": height, "num_frames": num_frames, @@ -1623,6 +1771,183 @@ async def create_video( "negative_prompt": negative_prompt, "lora": _parse_form_json(lora), } + request_data = {k: v for k, v in request_data.items() if v is not None} request = VideoGenerationRequest(**request_data) - return await _run_video_generation(request, raw_request, input_reference_bytes=input_reference_bytes) + + handler = Omnivideo(raw_request) + if handler is None: + raise HTTPException( + status_code=HTTPStatus.SERVICE_UNAVAILABLE.value, + detail="Video generation handler not initialized.", + ) + logger.info("Video generation handler: %s", type(handler).__name__) + try: + app_model_name, app_stage_configs = _resolve_video_runtime_context(raw_request) + effective_model_name = handler.model_name or app_model_name or request.model or "unknown" + if request.model is not None and effective_model_name is not None and request.model != effective_model_name: + logger.warning( + "Model mismatch: request specifies '%s' but server is running '%s'. Using server model.", + request.model, + effective_model_name, + ) + handler.set_stage_configs_if_missing(app_stage_configs) + except HTTPException: + raise + except Exception as e: + logger.exception("Video generation failed: %s", e) + raise HTTPException( + status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, + detail=f"Video generation failed: {str(e)}", + ) + ref = video_response_from_request(effective_model_name, request) + + try: + image_data = await decode_input_reference(request.image_reference, input_reference_bytes) + except InvalidInputReferenceError as exc: + raise HTTPException(400, detail=str(exc) or "Invalid input reference.") from exc + + reference_image = ReferenceImage(data=image_data) if image_data is not None else image_data + await VIDEO_STORE.upsert(ref.id, ref) + task = asyncio.create_task(_run_video_generation_job(handler, request, ref.id, reference_image)) + await VIDEO_TASKS.upsert(ref.id, task) + return ref + + +@router.get("/v1/videos", response_model=VideoListResponse) +async def list_videos( + after: str | None = None, + limit: int | None = Query(None, ge=0, le=100), + order: Annotated[Literal["asc", "desc"], Query()] = "desc", +): + """List stored video generation jobs. + + Args: + after: Optional cursor indicating the last seen video ID. + limit: Optional maximum number of jobs to return. + order: Sort order for the returned jobs by creation time. + + Returns: + A ``VideoListResponse`` containing paginated job metadata and cursor + information. + """ + jobs = await VIDEO_STORE.list_values() + jobs.sort(key=lambda j: j.created_at, reverse=order == "desc") + + if after is not None: + idx = next((i for i, job in enumerate(jobs) if job.id == after), None) + jobs = [] if idx is None else jobs[idx + 1 :] + + has_more = False + if limit is not None: + has_more = len(jobs) > limit + jobs = jobs[:limit] + + first_id, last_id = None, None + if len(jobs) > 0: + first_id = jobs[0].id + last_id = jobs[-1].id + + return VideoListResponse(data=jobs, has_more=has_more, first_id=first_id, last_id=last_id) + + +@router.get("/v1/videos/{video_id}") +async def retrieve_video(video_id: str) -> VideoResponse: + """Retrieve metadata for a previously created video job. + + Args: + video_id: Identifier returned by ``POST /v1/videos``. + + Returns: + The stored ``VideoResponse`` for the requested job. + + Raises: + HTTPException: If the video job does not exist. + """ + job = await VIDEO_STORE.get(video_id) + if job is None: + raise HTTPException(status_code=404, detail="Video not found") + return job + + +@router.delete("/v1/videos/{video_id}") +async def delete_video(video_id: str) -> VideoDeleteResponse: + """Delete a stored video job and any generated output. + + If the job is still queued or running, this endpoint first attempts to + cancel the in-flight generation task before removing the stored metadata. + + Args: + video_id: Identifier of the video job to delete. + + Returns: + A ``VideoDeleteResponse`` confirming the job was removed. + + Raises: + HTTPException: If the video job does not exist, cancellation is still + in progress, or output is not yet ready for a completed job. + """ + job = await VIDEO_STORE.get(video_id) + if job is None: + raise HTTPException(status_code=404, detail="Video not found") + + if job.status in (VideoGenerationStatus.QUEUED, VideoGenerationStatus.IN_PROGRESS): + task = await VIDEO_TASKS.get(video_id) + if task is not None: + task.cancel() + try: + await asyncio.wait_for(task, timeout=2.0) + except asyncio.TimeoutError: + raise HTTPException(status_code=409, detail="Cancellation in progress. Please try again later.") + except asyncio.CancelledError: + pass + + await VIDEO_STORE.pop(video_id) + return VideoDeleteResponse(id=job.id, deleted=True) + elif job.status is VideoGenerationStatus.FAILED: + if job.file_name is not None: + try: + await STORAGE_MANAGER.delete(job.file_name) + except Exception: + logger.warning("Failed to delete stored artifact for failed video job %s", video_id, exc_info=True) + + await VIDEO_STORE.pop(video_id) + return VideoDeleteResponse(id=job.id, deleted=True) + + if job.file_name is None: + raise HTTPException(status_code=409, detail="Video output not yet available. Please try again later.") + + await STORAGE_MANAGER.delete(job.file_name) + await VIDEO_STORE.pop(video_id) + return VideoDeleteResponse(id=job.id, deleted=True) + + +@router.get("/v1/videos/{video_id}/content") +async def download_video(video_id: str) -> FileResponse: + """Download the generated file for a completed video job. + + Args: + video_id: Identifier of the video job whose output should be returned. + + Returns: + A ``FileResponse`` streaming the generated video file from local + storage. + + Raises: + HTTPException: If the job does not exist, is still in progress, or the + generated file is missing from disk. + """ + job = await VIDEO_STORE.get(video_id) + if job is None: + raise HTTPException(status_code=404, detail="Video not found") + + if job.status == VideoGenerationStatus.FAILED: + raise HTTPException(status_code=422, detail="Video generation failed. Check job status for error details.") + if not job.file_name: + raise HTTPException(status_code=404, detail="Generation is still in-progress") + + full_path = STORAGE_MANAGER.get_full_file_path(job.file_name) + if not os.path.exists(full_path): + raise HTTPException(status_code=404, detail="Generated video file not found on disk") + + return FileResponse(path=full_path, media_type=job.media_type, filename=job.file_name) diff --git a/vllm_omni/entrypoints/openai/errors.py b/vllm_omni/entrypoints/openai/errors.py new file mode 100644 index 000000000..fda9b6436 --- /dev/null +++ b/vllm_omni/entrypoints/openai/errors.py @@ -0,0 +1,3 @@ +class InvalidInputReferenceError(ValueError): + def __init__(self, message: str = "Invalid input reference.") -> None: + super().__init__(message) diff --git a/vllm_omni/entrypoints/openai/protocol/__init__.py b/vllm_omni/entrypoints/openai/protocol/__init__.py index 3788ee2af..c73203cc4 100644 --- a/vllm_omni/entrypoints/openai/protocol/__init__.py +++ b/vllm_omni/entrypoints/openai/protocol/__init__.py @@ -12,7 +12,6 @@ from vllm_omni.entrypoints.openai.protocol.videos import ( VideoData, VideoGenerationRequest, VideoGenerationResponse, - VideoResponseFormat, ) __all__ = [ @@ -23,6 +22,5 @@ __all__ = [ "VideoData", "VideoGenerationRequest", "VideoGenerationResponse", - "VideoResponseFormat", "OmniChatCompletionStreamResponse", ] diff --git a/vllm_omni/entrypoints/openai/protocol/videos.py b/vllm_omni/entrypoints/openai/protocol/videos.py index 49fa43189..1da6bdb08 100644 --- a/vllm_omni/entrypoints/openai/protocol/videos.py +++ b/vllm_omni/entrypoints/openai/protocol/videos.py @@ -8,17 +8,40 @@ mirrors the OpenAI Images API shape, with vllm-omni extensions for diffusion video models (e.g., Wan2.2). """ +import mimetypes +import time +import uuid from enum import Enum -from typing import Any +from functools import lru_cache +from typing import Annotated, Any, Literal -from pydantic import BaseModel, Field, field_validator +from pydantic import BaseModel, ConfigDict, Field, StringConstraints + +from vllm_omni.entrypoints.openai.image_api_utils import parse_size -class VideoResponseFormat(str, Enum): - """Video response format.""" +class VideoGenerationStatus(str, Enum): + QUEUED = "queued" + IN_PROGRESS = "in_progress" + COMPLETED = "completed" + FAILED = "failed" - B64_JSON = "b64_json" - URL = "url" # Not implemented in PoC + +SizeStr = Annotated[str, StringConstraints(pattern=r"^\d+x\d+$")] +SecondStr = Annotated[str, StringConstraints(pattern=r"^[1-9]\d*$")] +DEFAULT_FPS = 24 + + +@lru_cache +def file_extension(media_type: str): + media_type = str(media_type).split(";", 1)[0].strip().lower() + ext = mimetypes.guess_extension(media_type, strict=False) + + if ext is None: + raise ValueError(f"No recognized file extension for media_type {media_type}") + + # Keep naming stable for unknown/unsupported MIME types. + return ext.lstrip(".") class VideoParams(BaseModel): @@ -29,6 +52,26 @@ class VideoParams(BaseModel): num_frames: int | None = Field(default=None, ge=1, description="Number of frames") fps: int | None = Field(default=None, ge=1, description="Frames per second for output video") + @property + def size(self) -> str | None: + if self.width and self.height: + return f"{self.width}x{self.height}" + else: + return None + + +class FileImageReference(BaseModel): + model_config = ConfigDict(extra="forbid") + file_id: str + + +class UrlImageReference(BaseModel): + model_config = ConfigDict(extra="forbid") + image_url: str + + +ImageReference = UrlImageReference | FileImageReference + class VideoGenerationRequest(BaseModel): """ @@ -37,42 +80,37 @@ class VideoGenerationRequest(BaseModel): Follows the OpenAI Images API conventions with extensions for video. """ - prompt: str = Field(..., description="Text description of the desired video(s)") - # OpenAI standard fields model: str | None = Field( default=None, description="Model to use (optional, uses server's configured model if omitted)", ) - n: int = Field(default=1, ge=1, le=4, description="Number of videos to generate") - seconds: int | str | None = Field( + prompt: str = Field(..., description="Text description of the desired video(s)") + seconds: SecondStr | None = Field( default=None, description="Clip duration in seconds (OpenAI-compatible, e.g., 4, 8, 12)", ) - size: str | None = Field( + size: SizeStr | None = Field( default=None, description="Video dimensions in WIDTHxHEIGHT format (e.g., '1280x720')", ) - response_format: VideoResponseFormat = Field( - default=VideoResponseFormat.B64_JSON, description="Format of the returned video" + + image_reference: ImageReference | None = Field( + default=None, + description="Optional JSON-safe image reference that guides generation. Provide either image_url or file_id.", ) - user: str | None = Field(default=None, description="User identifier for tracking") - stream: bool | None = Field(default=False, description="Whether to stream outputs (not yet supported)") # Video params block for extensibility video_params: VideoParams | None = Field(default=None, description="Optional video-specific parameters") - # Image-to-video input (OpenAI-style name) - input_reference: str | None = Field( - default=None, - description="Image input for I2V; provided via multipart form-data as a file upload.", - ) + # User specific tracking field + user: str | None = Field(default=None, description="User identifier for tracking") # Video-specific fields (top-level for OpenAI-style compatibility) width: int | None = Field(default=None, ge=1, description="Video width in pixels") height: int | None = Field(default=None, ge=1, description="Video height in pixels") - num_frames: int | None = Field(default=None, ge=1, description="Number of frames to generate") fps: int | None = Field(default=None, ge=1, description="Frames per second for output video") + num_frames: int | None = Field(default=None, ge=1, description="Number of frames to generate") # vllm-omni extensions for diffusion control negative_prompt: str | None = Field(default=None, description="Text describing what to avoid in the video") @@ -123,34 +161,25 @@ class VideoGenerationRequest(BaseModel): ), ) - @field_validator("size") - @classmethod - def validate_size(cls, v): - if v is None: - return None - if not isinstance(v, str) or "x" not in v: - raise ValueError("size must be in format 'WIDTHxHEIGHT' (e.g., '1280x720')") - return v + def resolve_video_params(self) -> VideoParams: + vp = VideoParams(width=self.width, height=self.height, fps=self.fps, num_frames=self.num_frames) - @field_validator("response_format") - @classmethod - def validate_response_format(cls, v): - if v is not None and v != VideoResponseFormat.B64_JSON: - raise ValueError(f"Only 'b64_json' response format is supported, got: {v}") - return v + if self.video_params is not None: + vp.width = vp.width or self.video_params.width + vp.height = vp.height or self.video_params.height + vp.fps = vp.fps or self.video_params.fps + vp.num_frames = vp.num_frames or self.video_params.num_frames - @field_validator("seconds") - @classmethod - def validate_seconds(cls, v): - if v is None: - return None - if isinstance(v, str): - if not v.isdigit(): - raise ValueError("seconds must be an integer or numeric string") - return int(v) - if isinstance(v, int): - return v - raise ValueError("seconds must be an integer or numeric string") + if self.size: + vp.width, vp.height = parse_size(self.size) + + if vp.fps is None: + vp.fps = DEFAULT_FPS + + if vp.num_frames is None and self.seconds is not None: + vp.num_frames = int(self.seconds) * int(vp.fps) + + return vp class VideoData(BaseModel): @@ -166,3 +195,74 @@ class VideoGenerationResponse(BaseModel): created: int = Field(..., description="Unix timestamp of when the generation completed") data: list[VideoData] = Field(..., description="Array of generated videos") + + +class VideoError(BaseModel): + code: str = Field(..., description="A machine-readable error code that was returned.") + message: str = Field(..., description="A human-readable description of the error that was returned.") + + +class VideoResponse(BaseModel): + """Stored metadata for an async video generation job.""" + + # OpenAI standard fields + model: str = Field(..., description="Model name used for video generation.") + prompt: str = Field(..., description="The prompt that was used to generate the video.") + id: str = Field( + default_factory=lambda: f"video_gen_{uuid.uuid4().hex}", description="Unique id for a video request" + ) + object: Literal["video"] = Field(default="video", description="Object type identifier.") + status: VideoGenerationStatus = Field( + default=VideoGenerationStatus.QUEUED, description="Current lifecycle status of the video job." + ) + size: SizeStr | None = Field(default=None, description="Requested output size in WIDTHxHEIGHT format.") + progress: int = Field(default=0, description="Best-effort progress indicator from 0 to 100.") + seconds: SecondStr = Field(default="4", description="Requested clip length in seconds.") + quality: str = Field(default="default", description="Requested quality level for generation.") + completed_at: int | None = Field( + default=None, description="Unix timestamp (seconds) for when the job completed, if finished." + ) + created_at: int = Field( + default_factory=lambda: int(time.time()), + description="Unix timestamp (seconds) for when the job was created.", + ) + remixed_from_video_id: str | None = Field( + default=None, + description="Optional source video id for remix/edit flows.", + ) + error: VideoError | None = Field( + default=None, + description="Error payload that explains why generation failed, if applicable.", + ) + + # vLLM specific fields + media_type: Literal["video/mp4"] = Field(default="video/mp4", description="MIME type of the generated artifact.") + + expires_at: int | None = Field(default=None, description="Unix timestamp when this record is considered expired.") + file_name: str | None = Field( + default=None, + description="Filename of the saved output video files for this job.", + ) + inference_time_s: float | None = Field(default=None, description="End-to-end inference time in seconds.") + + @property + def file_extension(self) -> str: + return file_extension(self.media_type) + + +class VideoDeleteResponse(BaseModel): + id: str = Field(description="Identifier of the deleted video.") + deleted: bool = Field(description="Indicates that the video resource was deleted.") + object: Literal["video.deleted"] = Field( + default="video.deleted", description="The object type that signals the deletion response." + ) + + +class VideoListResponse(BaseModel): + """Paginated-style wrapper for listing stored video jobs.""" + + first_id: str | None = Field(..., description="The ID of the first item in the list.") + last_id: str | None = Field(..., description="The ID of the last item in the list.") + has_more: bool = Field(..., description="Whether there are more items available.") + data: list[VideoResponse] = Field(..., description="Array of video job records.") + object: Literal["list"] = Field(default="list", description="Object type identifier for list responses.") diff --git a/vllm_omni/entrypoints/openai/serving_video.py b/vllm_omni/entrypoints/openai/serving_video.py index b663eb52f..6a16e7c70 100644 --- a/vllm_omni/entrypoints/openai/serving_video.py +++ b/vllm_omni/entrypoints/openai/serving_video.py @@ -4,23 +4,22 @@ from __future__ import annotations import time -import uuid +from dataclasses import dataclass from http import HTTPStatus from typing import Any, cast -from fastapi import HTTPException, Request +from fastapi import HTTPException from PIL import Image from vllm.engine.protocol import EngineClient from vllm.logger import init_logger from vllm_omni.entrypoints.async_omni import AsyncOmni -from vllm_omni.entrypoints.openai.image_api_utils import parse_size from vllm_omni.entrypoints.openai.protocol.videos import ( VideoData, VideoGenerationRequest, VideoGenerationResponse, ) -from vllm_omni.entrypoints.openai.video_api_utils import decode_input_reference, encode_video_base64 +from vllm_omni.entrypoints.openai.video_api_utils import encode_video_base64 from vllm_omni.inputs.data import OmniDiffusionSamplingParams, OmniSamplingParams, OmniTextPrompt from vllm_omni.lora.request import LoRARequest from vllm_omni.lora.utils import stable_lora_int_id @@ -28,6 +27,13 @@ from vllm_omni.lora.utils import stable_lora_int_id logger = init_logger(__name__) +@dataclass +class ReferenceImage: + """Reference class for tracking additional metadata if needed""" + + data: Image.Image + + class OmniOpenAIServingVideo: """OpenAI-style video generation handler for omni diffusion models.""" @@ -41,6 +47,18 @@ class OmniOpenAIServingVideo: self._model_name = model_name self._stage_configs = stage_configs + @property + def model_name(self) -> str | None: + return self._model_name + + @property + def stage_configs(self) -> list[Any] | None: + return self._stage_configs + + def set_stage_configs_if_missing(self, stage_configs: list[Any] | None) -> None: + if self._stage_configs is None and stage_configs is not None: + self._stage_configs = stage_configs + @classmethod def for_diffusion( cls, @@ -57,57 +75,32 @@ class OmniOpenAIServingVideo: async def generate_videos( self, request: VideoGenerationRequest, - raw_request: Request | None = None, + reference_id: str, *, - input_reference_bytes: bytes | None = None, + reference_image: ReferenceImage | None = None, ) -> VideoGenerationResponse: - if request.stream: - raise HTTPException( - status_code=HTTPStatus.BAD_REQUEST.value, - detail="Streaming video generation is not supported yet.", - ) - - request_id = f"video_gen_{uuid.uuid4().hex}" - model_name = self._resolve_model_name(raw_request) - - if request.model is not None and model_name is not None and request.model != model_name: - logger.warning( - "Model mismatch: request specifies '%s' but server is running '%s'. Using server model.", - request.model, - model_name, - ) - - prompt: OmniTextPrompt = {"prompt": request.prompt} + prompt: OmniTextPrompt = OmniTextPrompt(prompt=request.prompt) if request.negative_prompt is not None: prompt["negative_prompt"] = request.negative_prompt - input_image = None - try: - input_image = decode_input_reference(request.input_reference, input_reference_bytes) - except ValueError as exc: - raise HTTPException( - status_code=HTTPStatus.BAD_REQUEST.value, - detail=str(exc), - ) from exc + gen_params = OmniDiffusionSamplingParams() - gen_params = OmniDiffusionSamplingParams(num_outputs_per_prompt=request.n) - - width, height, num_frames, fps = self._resolve_video_params(request) - if input_image is not None and width is not None and height is not None: - target_size = (width, height) + input_image = None if reference_image is None else reference_image.data + vp = request.resolve_video_params() + if input_image is not None and vp.width is not None and vp.height is not None: + target_size = (vp.width, vp.height) if input_image.size != target_size: input_image = input_image.resize(target_size, Image.Resampling.LANCZOS) if input_image is not None: prompt["multi_modal_data"] = {"image": input_image} - - if width is not None and height is not None: - gen_params.width = width - gen_params.height = height - if num_frames is not None: - gen_params.num_frames = num_frames - if fps is not None: - gen_params.fps = fps - gen_params.frame_rate = float(fps) + if vp.width is not None and vp.height is not None: + gen_params.width = vp.width + gen_params.height = vp.height + if vp.num_frames is not None: + gen_params.num_frames = vp.num_frames + if vp.fps is not None: + gen_params.fps = vp.fps + gen_params.frame_rate = float(vp.fps) if request.num_inference_steps is not None: gen_params.num_inference_steps = request.num_inference_steps @@ -140,12 +133,12 @@ class OmniOpenAIServingVideo: gen_params.seed, ) - result = await self._run_generation(prompt, gen_params, request_id, raw_request) + result = await self._run_generation(prompt, gen_params, reference_id) _t_encode_start = time.perf_counter() videos = self._extract_video_outputs(result) audios = self._extract_audio_outputs(result, expected_count=len(videos)) - output_fps = fps or 24 audio_sample_rate = self._resolve_audio_sample_rate(result) + output_fps = vp.fps or 24 video_data = [ VideoData( @@ -166,37 +159,6 @@ class OmniOpenAIServingVideo: logger.info("Video response encoding (MP4+base64): %.2f ms", _t_encode_ms) return VideoGenerationResponse(created=int(time.time()), data=video_data) - def _resolve_model_name(self, raw_request: Request | None) -> str | None: - if self._model_name: - return self._model_name - if raw_request is None: - return None - serving_models = getattr(raw_request.app.state, "openai_serving_models", None) - if serving_models and getattr(serving_models, "base_model_paths", None): - base_paths = serving_models.base_model_paths - if base_paths: - return base_paths[0].name - return None - - @staticmethod - def _resolve_video_params(request: VideoGenerationRequest) -> tuple[int | None, int | None, int | None, int | None]: - width = request.width or (request.video_params.width if request.video_params else None) - height = request.height or (request.video_params.height if request.video_params else None) - num_frames = request.num_frames or (request.video_params.num_frames if request.video_params else None) - fps = request.fps or (request.video_params.fps if request.video_params else None) - seconds = request.seconds - - if request.size: - width, height = parse_size(request.size) - - if fps is None: - fps = 24 # Default FPS if not specified - - if num_frames is None and seconds is not None: - num_frames = int(seconds) * int(fps) - - return width, height, num_frames, fps - @staticmethod def _apply_lora(lora_body: Any, gen_params: OmniDiffusionSamplingParams) -> None: if lora_body is None: @@ -237,22 +199,15 @@ class OmniOpenAIServingVideo: prompt: OmniTextPrompt, gen_params: OmniDiffusionSamplingParams, request_id: str, - raw_request: Request | None, ) -> Any: has_stage_list = hasattr(self._engine_client, "stage_list") logger.info( "Video generation routing: stage_configs=%s, has_stage_list=%s, engine_type=%s", - "present" - if (self._stage_configs or (getattr(raw_request.app.state, "stage_configs", None) if raw_request else None)) - else "missing", + "present" if self._stage_configs else "missing", has_stage_list, type(self._engine_client).__name__, ) - stage_configs = ( - self._stage_configs - or (getattr(raw_request.app.state, "stage_configs", None) if raw_request else None) - or getattr(self._engine_client, "stage_configs", None) - ) + stage_configs = self._stage_configs or getattr(self._engine_client, "stage_configs", None) if not stage_configs: if not hasattr(self._engine_client, "stage_list"): diff --git a/vllm_omni/entrypoints/openai/storage.py b/vllm_omni/entrypoints/openai/storage.py new file mode 100644 index 000000000..52aef8948 --- /dev/null +++ b/vllm_omni/entrypoints/openai/storage.py @@ -0,0 +1,70 @@ +import asyncio +import os +from abc import ABC, abstractmethod +from tempfile import NamedTemporaryFile + + +class StorageBaseManager(ABC): + @abstractmethod + async def save(self, *args, **kwargs) -> str: + pass + + @abstractmethod + async def delete(self, *args, **kwargs) -> bool: + pass + + +class LocalStorageManager(StorageBaseManager): + def __init__(self, storage_path: str | None = None): + if storage_path is None: + storage_path = os.getenv("VLLM_OMNI_STORAGE_PATH", "/tmp/storage") + self.storage_path = storage_path + os.makedirs(self.storage_path, exist_ok=True) + + max_concurrency = int(os.getenv("VLLM_OMNI_STORAGE_MAX_CONCURRENCY", "4")) + self._io_semaphore = asyncio.Semaphore(max(1, max_concurrency)) + + def _save_sync(self, data: bytes, file_name: str) -> str: + filename = self.get_full_file_path(file_name) + tmp_name: str | None = None + try: + with NamedTemporaryFile("wb", dir=self.storage_path, delete=False) as f: + tmp_name = f.name + f.write(data) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_name, filename) + return filename + except Exception: + if tmp_name is not None: + try: + os.remove(tmp_name) + except OSError: + pass + raise + + async def save(self, data: bytes, file_name: str) -> str: + async with self._io_semaphore: + return await asyncio.to_thread(self._save_sync, data, file_name) + + def _delete_sync(self, file_name: str) -> bool: + try: + os.remove(self.get_full_file_path(file_name)) + except FileNotFoundError: + return False + return True + + async def delete(self, file_name: str) -> bool: + async with self._io_semaphore: + return await asyncio.to_thread(self._delete_sync, file_name) + + def exists(self, file_name: str) -> bool: + return os.path.exists(self.get_full_file_path(file_name)) + + def get_full_file_path(self, file_name: str) -> str: + return os.path.join(self.storage_path, file_name) + + +# Should implement TTL management on local file storage +# and recovery on restart in the *STORE's +STORAGE_MANAGER = LocalStorageManager() diff --git a/vllm_omni/entrypoints/openai/stores.py b/vllm_omni/entrypoints/openai/stores.py new file mode 100644 index 000000000..98f8696af --- /dev/null +++ b/vllm_omni/entrypoints/openai/stores.py @@ -0,0 +1,74 @@ +import asyncio +from typing import Any, Generic, TypeVar + +from pydantic import BaseModel + +from vllm_omni.entrypoints.openai.protocol.videos import VideoResponse + +T = TypeVar("T", bound=BaseModel) +U = TypeVar("U") + + +class TaskRegistry: + def __init__(self) -> None: + self._tasks: dict[str, asyncio.Task[None]] = {} + self._lock = asyncio.Lock() + + async def get(self, key: str) -> asyncio.Task[None] | None: + async with self._lock: + return self._tasks.get(key) + + async def pop(self, key: str) -> asyncio.Task[None] | None: + async with self._lock: + return self._tasks.pop(key, None) + + async def upsert(self, key: str, task: asyncio.Task[None]) -> None: + def _cleanup(_: asyncio.Task[None]) -> None: + asyncio.create_task(self.pop(key)) + + task.add_done_callback(_cleanup) + + async with self._lock: + self._tasks[key] = task + + +class AsyncDictStore(Generic[T]): + """A small async-safe in-memory key-value store for dict items. + + This encapsulates the usual pattern of a module-level dict guarded by + an asyncio.Lock and provides simple CRUD methods that are safe to call + concurrently from FastAPI request handlers and background tasks. + """ + + def __init__(self) -> None: + self._items: dict[str, T] = {} + self._lock = asyncio.Lock() + + async def upsert(self, key: str, value: T) -> None: + async with self._lock: + self._items[key] = value + + async def update_fields(self, key: str, updates: dict[str, Any]) -> T | None: + async with self._lock: + item: T | None = self._items.get(key) + if item is None: + return None + new_item = item.model_copy(update=updates) + self._items[key] = new_item + return new_item + + async def get(self, key: str) -> T | None: + async with self._lock: + return self._items.get(key) + + async def pop(self, key: str) -> T | None: + async with self._lock: + return self._items.pop(key, None) + + async def list_values(self) -> list[T]: + async with self._lock: + return list(self._items.values()) + + +VIDEO_STORE: AsyncDictStore[VideoResponse] = AsyncDictStore() +VIDEO_TASKS: TaskRegistry = TaskRegistry() diff --git a/vllm_omni/entrypoints/openai/video_api_utils.py b/vllm_omni/entrypoints/openai/video_api_utils.py index 64672ef6f..2ed1fd3de 100644 --- a/vllm_omni/entrypoints/openai/video_api_utils.py +++ b/vllm_omni/entrypoints/openai/video_api_utils.py @@ -7,32 +7,82 @@ Shared helper utilities for OpenAI-compatible video generation API. from __future__ import annotations import base64 +import binascii import os import tempfile from io import BytesIO from typing import Any +import httpx import numpy as np import torch -from PIL import Image +from PIL import Image, UnidentifiedImageError + +from vllm_omni.entrypoints.openai.errors import InvalidInputReferenceError +from vllm_omni.entrypoints.openai.protocol.videos import ( + FileImageReference, + ImageReference, + UrlImageReference, +) -def decode_input_reference(input_reference: str | None, input_reference_bytes: bytes | None) -> Image.Image | None: - """Decode image input from multipart bytes or base64/data URL.""" - if input_reference and input_reference_bytes: - raise ValueError("Provide input_reference either as file upload or base64, not both.") - if input_reference_bytes: - return Image.open(BytesIO(input_reference_bytes)).convert("RGB") +def _decode_image_bytes(image_bytes: bytes, *, source: str) -> Image.Image: + try: + return Image.open(BytesIO(image_bytes)).convert("RGB") + except (UnidentifiedImageError, OSError, ValueError) as exc: + raise InvalidInputReferenceError(f"Invalid {source}: provided content is not a valid image.") from exc + + +def _decode_base64_image(input_reference: str, *, source: str) -> Image.Image: if input_reference: if input_reference.startswith("data:image"): _, b64_data = input_reference.split(",", 1) else: b64_data = input_reference + try: image_bytes = base64.b64decode(b64_data) - except Exception as exc: # pragma: no cover - malformed base64 - raise ValueError("Invalid base64 input_reference.") from exc - return Image.open(BytesIO(image_bytes)).convert("RGB") + except (binascii.Error, ValueError) as exc: # pragma: no cover - malformed base64 + raise InvalidInputReferenceError(f"Invalid {source}: image data is not valid base64.") from exc + return _decode_image_bytes(image_bytes, source=source) + raise InvalidInputReferenceError(f"Invalid {source}: image data is empty.") + + +async def decode_image_url(image_url: str) -> Image.Image: + if image_url.startswith("data:image"): + return _decode_base64_image(image_url, source="image_reference.image_url") + + if image_url.startswith(("http://", "https://")): + async with httpx.AsyncClient(timeout=60) as client: + try: + response = await client.get(image_url) + response.raise_for_status() + except httpx.HTTPError as exc: + raise InvalidInputReferenceError( + "Invalid image_reference.image_url: failed to download image." + ) from exc + return _decode_image_bytes(response.content, source="image_reference.image_url") + + raise InvalidInputReferenceError("Invalid image_reference.image_url: must be an http(s) URL or data URL.") + + +async def decode_input_reference( + image_reference: ImageReference | None, + input_reference_bytes: bytes | None, +) -> Image.Image | None: + """Decode image input from multipart bytes, base64/data URL, or image_reference.""" + + if input_reference_bytes is not None and image_reference is not None: + raise InvalidInputReferenceError("Provide either input_reference or image_reference, not both.") + + if isinstance(input_reference_bytes, bytes): + return _decode_image_bytes(input_reference_bytes, source="input_reference") + + if isinstance(image_reference, UrlImageReference): + return await decode_image_url(image_reference.image_url) + elif isinstance(image_reference, FileImageReference): + raise InvalidInputReferenceError("Invalid image_reference: file_id is not supported yet.") + return None