diff --git a/docs/design/inference_schema_parity_inventory.yaml b/docs/design/inference_schema_parity_inventory.yaml index c5c965512..a55932960 100644 --- a/docs/design/inference_schema_parity_inventory.yaml +++ b/docs/design/inference_schema_parity_inventory.yaml @@ -21,6 +21,8 @@ surfaces: hsdp_shard_dim: generator.engine.parallelism.hsdp_shard_dim dist_timeout: generator.engine.parallelism.dist_timeout lora_path: generator.pipeline.components.lora_path + lora_nickname: generator.pipeline.components.lora_nickname + lora_strength: generator.pipeline.components.lora_strength dit_cpu_offload: generator.engine.offload.dit use_fsdp_inference: generator.engine.use_fsdp_inference dit_layerwise_offload: generator.engine.offload.dit_layerwise @@ -72,7 +74,6 @@ surfaces: compatibility_only: mode: "Legacy multi-mode FastVideoArgs switch; typed inference config should not expose execution mode." inference_mode: "Legacy boolean mirror of mode; kept only through adapters while FastVideoArgs remains." - lora_nickname: "Legacy adapter-selection surface pending LoRA API cleanup." lora_target_modules: "Legacy LoRA configuration surface pending dedicated component API." output_type: "Legacy output formatting surface pending GenerationResult cleanup." VSA_sparsity: "Model-specific inference optimization not yet represented in the typed public schema." @@ -590,15 +591,31 @@ surfaces: openai_video_request: kept: model: "HTTP adapter model-routing field." + user: "OpenAI-compatible caller tracking field." + task: "SGLang-compatible MiniMax-H3 task selector validated against the startup pipeline." + quality: "vLLM-Omni-compatible quality-intent field; model-specific." + lora: "vLLM-Omni-compatible selector for the adapter fixed at server startup." moved: prompt: request.prompt input_reference: request.inputs.image_path reference_url: request.inputs.image_path + image_reference: request.inputs.image_path,last_image,references + video_reference: request.inputs.video_path,references + audio_reference: request.inputs.references + video_path: request.inputs.video_path + video_url: request.inputs.video_path + video_params: request.sampling.width,height,num_frames,fps size: target: request.sampling.width,height note: "Adapter parses OpenAI size strings as WIDTHxHEIGHT and forwards width then height." + width: request.sampling.width + height: request.sampling.height fps: request.sampling.fps num_frames: request.sampling.num_frames + aspect_ratio: request.sampling.width,height + short_edge: request.sampling.width,height + num_outputs_per_prompt: request.sampling.num_videos_per_prompt + n: request.sampling.num_videos_per_prompt seed: request.sampling.seed num_inference_steps: request.sampling.num_inference_steps guidance_scale: request.sampling.guidance_scale @@ -606,11 +623,22 @@ surfaces: true_cfg_scale: request.sampling.true_cfg_scale negative_prompt: request.negative_prompt enable_teacache: request.runtime.enable_teacache + max_sequence_length: request.sampling.max_sequence_length + boundary_ratio: request.sampling.boundary_ratio + extra_params: request.extensions output_path: request.output.output_path compatibility_only: seconds: target: request.sampling.num_frames note: "HTTP adapter duration convenience field. If num_frames is omitted, the adapter computes num_frames = fps * seconds." + start_time_seconds: "vLLM-Omni reference-video offset; rejected by pipelines that cannot represent it." + flow_shift: "vLLM-Omni request field; accepted only when the selected model exposes a matching request parameter." + generate_sound: "vLLM-Omni audio-output intent; accepted only by models with a matching request parameter." + sound_duration: "vLLM-Omni audio-duration intent; accepted only by models with a matching request parameter." + enable_frame_interpolation: "vLLM-Omni post-processing field; unavailable until FastVideo exposes a frame-interpolation stage." + frame_interpolation_exp: "vLLM-Omni post-processing field; unavailable until FastVideo exposes a frame-interpolation stage." + frame_interpolation_scale: "vLLM-Omni post-processing field; unavailable until FastVideo exposes a frame-interpolation stage." + frame_interpolation_model_path: "vLLM-Omni post-processing field; unavailable until FastVideo exposes a frame-interpolation stage." cli: notes: diff --git a/docs/design/server_contracts/openai.md b/docs/design/server_contracts/openai.md index caae24321..a88744e5d 100644 --- a/docs/design/server_contracts/openai.md +++ b/docs/design/server_contracts/openai.md @@ -1,116 +1,158 @@ -# OpenAI-compatible HTTP Contract +# OpenAI-compatible HTTP contract -The stateless FastVideo HTTP server lives at -[`fastvideo/entrypoints/openai/`](https://github.com/hao-ai-lab/FastVideo/tree/main/fastvideo/entrypoints/openai). -Launch: `fastvideo serve --config serve.yaml`. +FastVideo exposes one model-agnostic REST engine for image and video models. +Launch it from a typed serve config: + +```bash +fastvideo serve --config examples/serving/openai_fasth3.yaml +``` + +All generation routes share one serialized engine. FastVideo pipelines mutate +per-request sampling state, and some adapters merge weights at load time, so a +single loaded pipeline is never entered concurrently by image and video +requests. HTTP handling and job polling remain asynchronous. ## Endpoints | Method | Path | Description | | --- | --- | --- | -| `POST` | `/v1/videos/generations` | Synchronous video generation | -| `GET` | `/v1/videos` | List prior jobs held in the in-memory store | -| `GET` | `/v1/videos/{id}` | Job status / result | -| `GET` | `/v1/videos/{id}/content` | Download the MP4 once ready | -| `POST` | `/v1/images/generations` | Synchronous image generation | -| `GET` | `/v1/models` | Enumerate registered models | +| `GET` | `/v1/models` | List the served model and optional startup adapter | +| `GET` | `/v1/models/{model}` | Retrieve one served model card | +| `POST` | `/v1/videos` | Submit an asynchronous video job | +| `POST` | `/v1/videos/sync` | Generate and return an MP4 response directly | +| `GET` | `/v1/videos` | List in-memory jobs with `after`, `limit`, and `order` | +| `GET` | `/v1/videos/{id}` | Retrieve job status and metadata | +| `GET` | `/v1/videos/{id}/content` | Download a completed MP4 | +| `DELETE` | `/v1/videos/{id}` | Delete a job and its completed artifact | +| `POST` | `/v1/images` | Generate an image | +| `POST` | `/v1/images/edits` | Generate an image from image references | +| `GET` | `/v1/images/{id}/content` | Download a generated image | | `GET` | `/health` | Liveness probe | -## `VideoGenerationsRequest` shape +`POST /v1/videos/generations` remains an alias for older FastVideo clients. -Mirrors the OpenAI `POST /v1/videos/generations` shape: +## Video requests + +The canonical shape follows vLLM-Omni and accepts SGLang's common flat +extensions. Fields that FastVideo cannot represent for the loaded model fail +at admission with HTTP 400 instead of creating a job that later fails. ```json { - "prompt": "a fox running through snow", - "size": "1024x1536", - "seconds": 5, - "fps": 24, - "num_frames": 121, + "model": "fasth3", + "prompt": "A fox runs through fresh snow.", + "seconds": "5", + "size": "1344x768", + "video_params": { + "fps": 24, + "num_frames": 124 + }, "seed": 42, - "num_inference_steps": 8, + "num_inference_steps": 5, "guidance_scale": 1.0, - "negative_prompt": "blurry, low quality", - "input_reference": "/path/to/init.png" -} -``` - -SGLang-compatible extensions carried today: -`num_inference_steps`, `guidance_scale`, `guidance_scale_2`, -`true_cfg_scale`, `negative_prompt`, `enable_teacache`, `output_path`. - -## Merge precedence - -The server builds a `GenerationRequest` each call using three layers, -highest first: - -1. **Request body (client-explicit)** — only fields carried in - `request.model_fields_set` (Pydantic v2). Unset fields do not count, - even if the Pydantic model has a schema default for them. -2. **`ServeConfig.default_request` (operator-explicit)** — projected via - [`explicit_request_updates()`](https://github.com/hao-ai-lab/FastVideo/blob/main/fastvideo/api/compat.py); - only fields the operator actually wrote into the YAML count as - defaults. Every other field inherits the schema default rather than - being pinned. -3. **Hardcoded fallback** — e.g. `fps = 24`. - -The gate matters: both surfaces carry schema defaults. Without -`model_fields_set` / explicit-path tracking, schema defaults would -masquerade as intent and silently shadow the other side. - -See [`video_api.py::_build_generation_kwargs`](https://github.com/hao-ai-lab/FastVideo/blob/main/fastvideo/entrypoints/openai/video_api.py) -for the canonical implementation; the per-request assembly lives there, -not in pipeline code. - -## Continuation state - -The stateless surface accepts an opaque `ContinuationState` round-trip. -Clients that want continuation pass the prior `state` blob back on the -next request, and receive a new one on the response when -`request.output.return_state = true`. - -Shape: - -```json -{ - "state": { - "kind": "ltx2.v1", - "payload": { "schema_version": 1, "segment_index": 3, ... } + "image_reference": [ + {"image_url": "https://example.com/first-frame.png"} + ], + "extra_params": { + "vsa_mode": "exempt" } } ``` -Payload is always JSON-serializable. Large tensors may live in an -opaque blob-store reference the client simply round-trips; see -[`LTX2ContinuationState`](https://github.com/hao-ai-lab/FastVideo/blob/main/fastvideo/pipelines/basic/ltx2/continuation.py). +Resolution precedence matches vLLM-Omni: -Continuation is not yet wired all the way through to -`generator.generate_video(...)` — PR 7.6 (GPU pool upstream) is the -pipeline-level consumer. PR 7 locked the envelope so this surface is -stable ahead of that plumbing. +1. `size` +2. top-level `width` and `height` +3. `video_params.width` and `video_params.height` -## Error codes +Top-level `fps` and `num_frames` similarly take precedence over the nested +block. If `num_frames` is absent, `seconds * fps` is used. FastVideo also keeps +the legacy `input_reference`, `reference_url`, `video_path`, and `video_url` +spellings. -| HTTP | Condition | -| --- | --- | -| `400 Bad Request` | Parse/validation failure (unknown field, type mismatch, incompatible preset/state) | -| `404 Not Found` | `GET /v1/videos/{id}` for an unknown job | -| `409 Conflict` | Job id already exists | -| `500 Internal Server Error` | Pipeline raised; body mirrors upstream OpenAI error envelope | -| `503 Service Unavailable` | No generator loaded, or shutdown in progress | +Reference objects support URL or local-path strings through `image_url`, +`video_url`, and `audio_url`. `file_id` references are schema-compatible but +return HTTP 400 because FastVideo does not provide an OpenAI Files store. +Multipart `input_reference` uploads are saved under the configured output +directory. -Errors include a JSON body with -`{"error": {"type": "...", "message": "..."}}` matching the OpenAI -Python SDK's expectation. +## Jobs and synchronous responses -## What does not cross this boundary +An asynchronous submission returns a `video` object in `queued` state. Its +status advances through `in_progress` to `completed` or `failed`. Completed +jobs expose `file_name`, the FastVideo compatibility extension `file_path`, +timings, and peak-memory metadata when the pipeline reports them. -* Flat legacy kwargs (`ltx2_refine_enabled`, `torch_compile_kwargs`, - etc.) — these are init-time, configured via `ServeConfig.generator`, - never per-request. -* Private Dreamverse-only fields — those live in a private adapter on - the Dreamverse side; the public FastVideo surface never promises - backward compatibility for them. -* Raw tensor payloads (`ltx2_audio_clean_latent` et al.) — these are - derived by the pipeline from `ContinuationState`, never shipped as - request fields. +`POST /v1/videos/sync` returns `video/mp4` bytes. It includes +`X-Request-Id`, `X-Model`, `X-Inference-Time-S`, `X-Stage-Durations`, and +`X-Peak-Memory-MB` headers. + +FastVideo's synchronous CUDA execution cannot be interrupted after launch. +Deleting an in-progress resource removes it from the API immediately; the +engine remains serialized until the call exits and then removes any artifact. + +## Model and LoRA selection + +`server.served_model_name` controls the public model id. If omitted, the +checkpoint path is used. Requests that name another model fail with HTTP 400. + +LoRAs are configured under +`generator.pipeline.components.{lora_path,lora_nickname,lora_strength}`. The +startup adapter appears in `/v1/models`, and requests can use either its model +nickname or a vLLM-Omni selector: + +```json +{ + "prompt": "A fox runs through fresh snow.", + "model": "fasth3-dense-datafree", + "lora": { + "name": "fasth3-dense-datafree", + "path": "/models/adapter_model.safetensors", + "scale": 1.0 + } +} +``` + +The selector must match the adapter already loaded at startup. FastH3 adapter +files can contain dense replacement tensors and VSA gates in addition to +low-rank factors, so swapping them inside concurrent requests would corrupt +shared pipeline state. A mismatch is rejected with HTTP 400. + +## MiniMax-H3 and FastH3 + +FastH3 uses the same general routes and adapter. `task` is accepted for +SGLang-compatible H3 clients: + +- `t2va` uses text only. +- `fl2va` takes one or two image references. +- `ref2va` takes ordered image, video, and audio references and requires a + server started with `MiniMaxH3Ref2VAModularPipeline`. + +The released FastH3 pipeline generates one packed video/audio result per +request, uses 24 fps, requires guidance scale 1, and accepts frame counts on +its causal-VAE grid. The serving examples pin its five-point distilled sigma +schedule (four DiT forwards). + +## Defaults and errors + +Incoming explicit fields override operator-explicit `default_request` fields, +which override model preset defaults. Pydantic defaults do not masquerade as +client intent; the transport uses `model_fields_set`, while typed config parsing +tracks the exact paths written by the operator. + +Errors use the OpenAI envelope: + +```json +{ + "error": { + "message": "...", + "type": "invalid_request_error", + "param": null, + "code": 400 + } +} +``` + +Parse, model-selection, startup-LoRA, and unsupported-parameter failures are +HTTP 400; missing resources are HTTP 404; generation failures are stored on +asynchronous jobs and returned as HTTP 500 when that job is retrieved. diff --git a/examples/serving/README.md b/examples/serving/README.md new file mode 100644 index 000000000..ed0240926 --- /dev/null +++ b/examples/serving/README.md @@ -0,0 +1,43 @@ +# OpenAI-compatible serving examples + +The REST serving engine is model-agnostic. Any model supported by +`VideoGenerator` can use the same `/v1/models`, `/v1/videos`, and `/v1/images` +surface; the two configs here are FastH3 validation profiles. + +Launch the full FastH3 checkpoint: + +```bash +fastvideo serve --config examples/serving/openai_fasth3.yaml +``` + +Launch the dense FastH3 LoRA on the base MiniMax-H3 checkpoint: + +```bash +adapter_path="$(hf download \ + FastVideo/FastVideo-FastH3-4-step-Preview-v1-LoRA \ + dense-datafree/adapter_model.safetensors)" + +fastvideo serve --config examples/serving/openai_fasth3_lora.yaml \ + --generator.pipeline.components.lora_path "$adapter_path" +``` + +FastH3 adapters are hybrid startup patches: alongside low-rank factors they +may contain dense deltas and a VSA compression-gate replacement. They must be +selected when the server starts. A request may carry the vLLM-Omni `lora` +selector, but its name, path, and scale must match that startup adapter. A VSA +adapter also needs `attention_backend: VIDEO_SPARSE_ATTN_H3`, `VSA_sparsity`, +and `VSA_tile_size` like the full-checkpoint config. + +Submit and poll an asynchronous job: + +```bash +job_id="$(curl -sS http://localhost:8000/v1/videos \ + -H 'content-type: application/json' \ + -d '{"model":"fasth3","prompt":"A fox runs through fresh snow."}' \ + | jq -r .id)" + +curl -sS "http://localhost:8000/v1/videos/$job_id" +curl -o result.mp4 "http://localhost:8000/v1/videos/$job_id/content" +``` + +For a blocking call, `POST /v1/videos/sync` returns the MP4 body directly. diff --git a/examples/serving/openai_fasth3.yaml b/examples/serving/openai_fasth3.yaml new file mode 100644 index 000000000..1158d9598 --- /dev/null +++ b/examples/serving/openai_fasth3.yaml @@ -0,0 +1,49 @@ +# OpenAI-compatible FastH3 full-checkpoint server. +generator: + model_path: FastVideo/FastVideo-Minimax-FastH3-Preview-v0.2 + engine: + num_gpus: 4 + use_fsdp_inference: false + parallelism: + tp_size: 1 + sp_size: 4 + offload: + dit: false + dit_layerwise: false + text_encoder: true + image_encoder: true + vae: true + pin_cpu_memory: true + compile: + enabled: false + vae_enabled: false + pipeline: + workload_type: t2v + experimental: + attention_backend: VIDEO_SPARSE_ATTN_H3 + VSA_sparsity: 0.9 + VSA_tile_size: 64 + inference_torch_compile: false + vae_parallel_decode: true + vae_parallel_decode_strategy: gather + +server: + host: 0.0.0.0 + port: 8000 + output_dir: outputs/openai_fasth3 + served_model_name: fasth3 + +default_request: + negative_prompt: "" + sampling: + height: 768 + width: 1344 + num_frames: 124 + fps: 24 + num_inference_steps: 5 + guidance_scale: 1.0 + batch_cfg: false + seed: 1000 + output: + return_frames: false + diff --git a/examples/serving/openai_fasth3_lora.yaml b/examples/serving/openai_fasth3_lora.yaml new file mode 100644 index 000000000..452713879 --- /dev/null +++ b/examples/serving/openai_fasth3_lora.yaml @@ -0,0 +1,52 @@ +# OpenAI-compatible FastH3 startup-LoRA server. Override lora_path at launch +# with the exact file returned by `hf download`; see README.md. +generator: + model_path: MiniMaxAI/MiniMax-H3 + engine: + num_gpus: 4 + use_fsdp_inference: false + parallelism: + tp_size: 1 + sp_size: 4 + offload: + dit: false + dit_layerwise: false + text_encoder: true + image_encoder: true + vae: true + pin_cpu_memory: true + compile: + enabled: false + vae_enabled: false + pipeline: + workload_type: t2v + components: + lora_path: /path/to/adapter_model.safetensors + lora_nickname: fasth3-dense-datafree + lora_strength: 1.0 + experimental: + attention_backend: FLASH_ATTN + inference_torch_compile: false + vae_parallel_decode: true + vae_parallel_decode_strategy: gather + +server: + host: 0.0.0.0 + port: 8000 + output_dir: outputs/openai_fasth3_lora + served_model_name: minimax-h3 + +default_request: + negative_prompt: "" + sampling: + height: 768 + width: 1344 + num_frames: 124 + fps: 24 + num_inference_steps: 5 + guidance_scale: 1.0 + batch_cfg: false + seed: 1000 + output: + return_frames: false + diff --git a/fastvideo/api/compat.py b/fastvideo/api/compat.py index 14f3f4dc9..9e91d7129 100644 --- a/fastvideo/api/compat.py +++ b/fastvideo/api/compat.py @@ -44,6 +44,16 @@ _LEGACY_REQUEST_ALIASES = { _REQUEST_PIPELINE_OVERRIDE_FIELDS = frozenset({ "embedded_cfg_scale", }) +REQUEST_BATCH_EXTRA_PASSTHROUGH_FIELDS = ( + "ltx2_audio_latents", + "ltx2_audio_clean_latent", + "ltx2_audio_denoise_mask", + "audio_num_frames", + "video_position_offset_sec", + "vsa_mode", + "vsa_dense_first_n_steps", + "vsa_dense_layers", +) # torch.compile kwargs that map to first-class CompileConfig fields. _COMPILE_TYPED_KEYS = ("backend", "fullgraph", "mode", "dynamic") # LTX-2 refine flat kwargs (init + per-request) known to FastVideoArgs. @@ -165,6 +175,8 @@ def legacy_from_pretrained_to_config( pipeline["workload_type"] = value elif key == "lora_path": components["lora_path"] = value + elif key == "lora_nickname": + components["lora_nickname"] = value elif key == "lora_strength": components["lora_strength"] = value elif key == "override_pipeline_cls_name": @@ -285,6 +297,7 @@ def generator_config_to_fastvideo_args(config: GeneratorConfig | Mapping[str, An kwargs["pipeline_config"] = components.pipeline_config_path if components.lora_path is not None: kwargs["lora_path"] = components.lora_path + kwargs["lora_nickname"] = components.lora_nickname kwargs["lora_strength"] = components.lora_strength if components.override_pipeline_cls_name is not None: kwargs["override_pipeline_cls_name"] = components.override_pipeline_cls_name @@ -371,7 +384,7 @@ def request_to_sampling_param( for key, value in updates.items(): if hasattr(sampling_param, key): setattr(sampling_param, key, deepcopy(value)) - elif key in _REQUEST_PIPELINE_OVERRIDE_FIELDS: + elif key in _REQUEST_PIPELINE_OVERRIDE_FIELDS or key in REQUEST_BATCH_EXTRA_PASSTHROUGH_FIELDS: continue elif value == _SCHEMA_DEFAULT_UPDATES.get(key, _MISSING): # Schema-default field that isn't on SamplingParam; tolerated @@ -470,6 +483,14 @@ def request_to_pipeline_overrides(request: GenerationRequest) -> dict[str, Any]: return overrides +def request_to_batch_extra(request: GenerationRequest) -> dict[str, Any]: + """Extract typed-request extensions consumed through ``ForwardBatch.extra``.""" + return { + key: deepcopy(value) + for key, value in explicit_request_updates(request).items() if key in REQUEST_BATCH_EXTRA_PASSTHROUGH_FIELDS + } + + def explicit_request_updates(request: GenerationRequest) -> dict[str, Any]: """Project a ``GenerationRequest`` down to *explicitly set* fields only. @@ -643,6 +664,7 @@ def _validate_batched_input_length( __all__ = [ + "REQUEST_BATCH_EXTRA_PASSTHROUGH_FIELDS", "explicit_request_updates", "generator_config_to_fastvideo_args", "legacy_from_pretrained_to_config", @@ -651,6 +673,7 @@ __all__ = [ "normalize_generation_request", "normalize_generator_config", "register_continuation_kind", + "request_to_batch_extra", "request_to_pipeline_overrides", "request_to_sampling_param", ] diff --git a/fastvideo/api/schema.py b/fastvideo/api/schema.py index 92e58d9ef..7b75cbc29 100644 --- a/fastvideo/api/schema.py +++ b/fastvideo/api/schema.py @@ -10,6 +10,7 @@ class ServerConfig: host: str = "0.0.0.0" port: int = 8000 output_dir: str = "outputs/" + served_model_name: str | None = None @dataclass @@ -94,6 +95,7 @@ class ComponentConfig: vae_weights: str | None = None upsampler_weights: str | None = None lora_path: str | None = None + lora_nickname: str = "default" lora_strength: float = 1.0 override_pipeline_cls_name: str | None = None override_transformer_cls_name: str | None = None diff --git a/fastvideo/entrypoints/cli/serve.py b/fastvideo/entrypoints/cli/serve.py index 6d5d13ce9..560581536 100644 --- a/fastvideo/entrypoints/cli/serve.py +++ b/fastvideo/entrypoints/cli/serve.py @@ -56,6 +56,7 @@ class ServeSubcommand(CLISubcommand): port=serve_config.server.port, output_dir=serve_config.server.output_dir, default_request=serve_config.default_request, + served_model_name=serve_config.server.served_model_name, ) def validate(self, args: argparse.Namespace) -> None: diff --git a/fastvideo/entrypoints/openai/api_server.py b/fastvideo/entrypoints/openai/api_server.py index 2b64087af..3ffb37f0c 100644 --- a/fastvideo/entrypoints/openai/api_server.py +++ b/fastvideo/entrypoints/openai/api_server.py @@ -5,8 +5,10 @@ from contextlib import asynccontextmanager from collections.abc import AsyncIterator import uvicorn -from fastapi import FastAPI +from fastapi import FastAPI, HTTPException, Request +from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse from fastvideo.api.presets import validate_preset_selection from fastvideo.api.schema import GenerationRequest @@ -15,6 +17,7 @@ from fastvideo.entrypoints.openai.state import ( clear_state, set_state, ) +from fastvideo.entrypoints.openai.serving_engine import OpenAIServingEngine from fastvideo.entrypoints.video_generator import VideoGenerator from fastvideo.fastvideo_args import FastVideoArgs from fastvideo.logger import init_logger @@ -53,26 +56,40 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: """Load model on startup, clean up on shutdown""" args: FastVideoArgs = app.state.fastvideo_args output_dir: str = app.state.output_dir + served_model_name: str | None = app.state.served_model_name default_request: GenerationRequest | None = getattr(app.state, "default_request", None) logger.info("Loading model from %s ...", args.model_path) generator = VideoGenerator.from_fastvideo_args(args) + serving_engine = OpenAIServingEngine(generator) logger.info("Model loaded successfully.") - set_state(generator, args, output_dir, default_request=default_request) + set_state( + generator, + serving_engine, + args, + output_dir, + default_request=default_request, + served_model_name=served_model_name, + ) - yield # server is running + try: + yield # server is running + finally: + logger.info("Shutting down — releasing model resources ...") + from fastvideo.entrypoints.openai.video_api import shutdown_video_jobs - logger.info("Shutting down — releasing model resources ...") - generator.shutdown() - clear_state() - logger.info("Shutdown complete.") + await shutdown_video_jobs() + await serving_engine.shutdown() + clear_state() + logger.info("Shutdown complete.") def create_app( fastvideo_args: FastVideoArgs, output_dir: str = DEFAULT_OUTPUT_DIR, default_request: GenerationRequest | None = None, + served_model_name: str | None = None, ) -> FastAPI: """Build the FastAPI application with all routers mounted""" @@ -84,6 +101,7 @@ def create_app( app.state.fastvideo_args = fastvideo_args app.state.output_dir = output_dir app.state.default_request = default_request + app.state.served_model_name = served_model_name app.add_middleware( CORSMiddleware, @@ -93,6 +111,35 @@ def create_app( allow_headers=["*"], ) + @app.exception_handler(HTTPException) + async def openai_http_error(_request: Request, exc: HTTPException) -> JSONResponse: + """Return the error envelope consumed by OpenAI-compatible clients.""" + message = exc.detail if isinstance(exc.detail, str) else str(exc.detail) + return JSONResponse( + status_code=exc.status_code, + headers=exc.headers, + content={ + "error": { + "message": message, + "type": "invalid_request_error" if exc.status_code < 500 else "server_error", + "param": None, + "code": exc.status_code, + } + }, + ) + + @app.exception_handler(RequestValidationError) + async def openai_validation_error(_request: Request, exc: RequestValidationError) -> JSONResponse: + return JSONResponse( + status_code=400, + content={"error": { + "message": str(exc), + "type": "invalid_request_error", + "param": None, + "code": 400, + }}, + ) + # Import and mount routers from fastvideo.entrypoints.openai.common_api import router as common_router from fastvideo.entrypoints.openai.image_api import router as image_router @@ -137,6 +184,7 @@ def run_server( port: int = DEFAULT_PORT, output_dir: str = DEFAULT_OUTPUT_DIR, default_request: GenerationRequest | None = None, + served_model_name: str | None = None, ): """Create the app and run it with uvicorn""" if default_request is not None: @@ -146,6 +194,7 @@ def run_server( fastvideo_args, output_dir=output_dir, default_request=default_request, + served_model_name=served_model_name, ) logger.info("Starting FastVideo server on %s:%d", host, port) diff --git a/fastvideo/entrypoints/openai/common_api.py b/fastvideo/entrypoints/openai/common_api.py index 621f90d53..1bec663d9 100644 --- a/fastvideo/entrypoints/openai/common_api.py +++ b/fastvideo/entrypoints/openai/common_api.py @@ -7,7 +7,7 @@ from fastapi import APIRouter from fastapi.responses import ORJSONResponse from pydantic import BaseModel, Field -from fastvideo.entrypoints.openai.state import get_server_args +from fastvideo.entrypoints.openai.state import get_served_model_name, get_server_args from fastvideo.logger import init_logger router = APIRouter(prefix="/v1") @@ -28,15 +28,21 @@ class ModelCard(BaseModel): async def available_models(): """Show available models""" args = get_server_args() - card = ModelCard(id=args.model_path, root=args.model_path) - return {"object": "list", "data": [card.model_dump()]} + cards = [ModelCard(id=get_served_model_name(), root=args.model_path)] + if args.lora_path and args.lora_nickname != get_served_model_name(): + cards.append(ModelCard(id=args.lora_nickname, root=args.model_path)) + return {"object": "list", "data": [card.model_dump() for card in cards]} @router.get("/models/{model:path}", response_class=ORJSONResponse) async def retrieve_model(model: str): """Retrieve a model by name""" args = get_server_args() - if model != args.model_path: + served_model_name = get_served_model_name() + available = {served_model_name} + if args.lora_path: + available.add(args.lora_nickname) + if model not in available: return ORJSONResponse( status_code=404, content={ @@ -48,7 +54,7 @@ async def retrieve_model(model: str): } }, ) - card = ModelCard(id=model, root=model) + card = ModelCard(id=model, root=args.model_path) return card.model_dump() @@ -56,4 +62,14 @@ async def retrieve_model(model: str): async def model_info(): """Get basic model information""" args = get_server_args() - return {"model_path": args.model_path} + return { + "model_path": + args.model_path, + "served_model_name": + get_served_model_name(), + "lora": ({ + "name": args.lora_nickname, + "path": args.lora_path, + "scale": args.lora_strength, + } if args.lora_path else None), + } diff --git a/fastvideo/entrypoints/openai/image_api.py b/fastvideo/entrypoints/openai/image_api.py index c377c6b39..6b624b9e2 100644 --- a/fastvideo/entrypoints/openai/image_api.py +++ b/fastvideo/entrypoints/openai/image_api.py @@ -1,7 +1,6 @@ # Adapted from SGLang # (https://github.com/sgl-project/sglang/blob/main/python/sglang/multimodal_gen/runtime/entrypoints/openai/image_api.py) -import asyncio import base64 import os import time @@ -12,8 +11,8 @@ from fastapi import (APIRouter, File, Form, HTTPException, Path, Query, UploadFi from fastapi.responses import FileResponse from fastvideo.entrypoints.openai.state import ( - get_generator, get_output_dir, + get_serving_engine, ) from fastvideo.entrypoints.openai.protocol import ( ImageGenerationsRequest, @@ -89,8 +88,7 @@ def _build_generation_kwargs( @router.post("", response_model=ImageResponse) async def generations(request: ImageGenerationsRequest): request_id = generate_request_id() - generator = get_generator() - loop = asyncio.get_running_loop() + engine = get_serving_engine() gen_kwargs = _build_generation_kwargs( request_id=request_id, @@ -109,7 +107,7 @@ async def generations(request: ImageGenerationsRequest): start = time.perf_counter() try: - await loop.run_in_executor(None, lambda: generator.generate_video(**gen_kwargs)) + await engine.run_serialized(engine.generator.generate_video, **gen_kwargs) except Exception as e: logger.error("Image generation failed: %s", e) raise HTTPException(status_code=500, detail=str(e)) from None @@ -173,8 +171,7 @@ async def edits( enable_teacache: bool | None = Form(False), ): request_id = generate_request_id() - generator = get_generator() - loop = asyncio.get_running_loop() + engine = get_serving_engine() images = image or image_array urls = url or url_array @@ -213,7 +210,7 @@ async def edits( start = time.perf_counter() try: - await loop.run_in_executor(None, lambda: generator.generate_video(**gen_kwargs)) + await engine.run_serialized(engine.generator.generate_video, **gen_kwargs) except Exception as e: logger.error("Image edit failed: %s", e) raise HTTPException(status_code=500, detail=str(e)) from None diff --git a/fastvideo/entrypoints/openai/protocol.py b/fastvideo/entrypoints/openai/protocol.py index 1d8d500ee..9002892e1 100644 --- a/fastvideo/entrypoints/openai/protocol.py +++ b/fastvideo/entrypoints/openai/protocol.py @@ -3,9 +3,10 @@ import time import uuid -from typing import Any +from enum import Enum +from typing import Annotated, Any, Literal -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field, StringConstraints class ImageResponseData(BaseModel): @@ -43,47 +44,202 @@ class ImageGenerationsRequest(BaseModel): enable_teacache: bool | None = False -class VideoResponse(BaseModel): - id: str - object: str = "video" - model: str = "" - status: str = "queued" - progress: int = 100 - created_at: int = Field(default_factory=lambda: int(time.time())) - size: str = "" - seconds: str = "4" - quality: str = "standard" - url: str | None = None - file_path: str | None = None - completed_at: int | None = None - error: dict[str, Any] | None = None - peak_memory_mb: float | None = None - inference_time_s: float | None = None +_INT64_MIN = -(2**63) +_INT64_MAX = 2**63 - 1 -class VideoGenerationsRequest(BaseModel): +class VideoGenerationStatus(str, Enum): + QUEUED = "queued" + IN_PROGRESS = "in_progress" + COMPLETED = "completed" + FAILED = "failed" + + +SizeStr = Annotated[str, StringConstraints(pattern=r"^\d+x\d+$")] +SecondStr = Annotated[str, StringConstraints(pattern=r"^[1-9]\d*$")] +DEFAULT_FPS = 24 + + +class VideoParams(BaseModel): + """Optional vLLM-Omni-compatible video parameter block.""" + + width: int | None = Field(default=None, ge=1, le=_INT64_MAX) + height: int | None = Field(default=None, ge=1, le=_INT64_MAX) + num_frames: int | None = Field(default=None, ge=1, le=_INT64_MAX) + fps: int | None = Field(default=None, ge=1, le=_INT64_MAX) + + @property + def size(self) -> str | None: + if self.width is not None and self.height is not None: + return f"{self.width}x{self.height}" + 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 FileVideoReference(BaseModel): + model_config = ConfigDict(extra="forbid") + file_id: str + + +class UrlVideoReference(BaseModel): + model_config = ConfigDict(extra="forbid") + video_url: str + + +VideoReference = UrlVideoReference | FileVideoReference + + +class UrlAudioReference(BaseModel): + model_config = ConfigDict(extra="forbid") + audio_url: str + + +AudioReference = UrlAudioReference + + +class VideoGenerationRequest(BaseModel): + """OpenAI/vLLM-Omni-compatible video generation request. + + Model-specific parameters belong in ``extra_params``. FastVideo keeps the + legacy ``input_reference`` and ``reference_url`` fields for clients that + predate vLLM-Omni's typed reference objects. + """ + + model_config = ConfigDict(extra="allow") + prompt: str + model: str | None = None + seconds: int | SecondStr | None = None + size: SizeStr | None = None + image_reference: ImageReference | list[ImageReference] | None = None + video_reference: VideoReference | list[VideoReference] | None = None + audio_reference: AudioReference | list[AudioReference] | None = None input_reference: str | None = None reference_url: str | None = None - model: str | None = None - seconds: int | None = 4 - size: str | None = "" - fps: int | None = None - num_frames: int | None = None - seed: int | None = 1024 - # FastVideo extensions (SGLang-compatible) - num_inference_steps: int | None = None - guidance_scale: float | None = None - guidance_scale_2: float | None = None - true_cfg_scale: float | None = None + # SGLang's legacy direct-video spellings. + video_path: str | None = None + video_url: str | None = None + video_params: VideoParams | None = None + user: str | None = None + task: str | None = None + + width: int | None = Field(default=None, ge=1, le=_INT64_MAX) + height: int | None = Field(default=None, ge=1, le=_INT64_MAX) + fps: int | None = Field(default=None, ge=1, le=_INT64_MAX) + num_frames: int | None = Field(default=None, ge=1, le=_INT64_MAX) + aspect_ratio: str | None = None + short_edge: int | None = Field(default=None, ge=1, le=_INT64_MAX) + num_outputs_per_prompt: int = Field(default=1, ge=1, le=10) + # SGLang spelling retained as an alias-like input field. + n: int | None = Field(default=None, ge=1, le=10) + start_time_seconds: float | None = Field(default=None, ge=0.0) + quality: str | None = None negative_prompt: str | None = None - enable_teacache: bool | None = False + num_inference_steps: int | None = Field(default=None, ge=1, le=200) + guidance_scale: float | None = Field(default=None, ge=0.0, le=20.0) + guidance_scale_2: float | None = Field(default=None, ge=0.0, le=20.0) + boundary_ratio: float | None = Field(default=None, ge=0.0, le=1.0) + flow_shift: float | None = None + true_cfg_scale: float | None = Field(default=None, ge=0.0, le=20.0) + seed: int | None = Field(default=None, ge=_INT64_MIN, le=_INT64_MAX) + generate_sound: bool = False + sound_duration: float | None = Field(default=None, gt=0.0) + enable_teacache: bool = False + max_sequence_length: int | None = Field(default=None, ge=1) + + enable_frame_interpolation: bool = False + frame_interpolation_exp: int = Field(default=1, ge=1, le=_INT64_MAX) + frame_interpolation_scale: float = Field(default=1.0, gt=0.0) + frame_interpolation_model_path: str | None = None + + lora: dict[str, Any] | None = None + extra_params: dict[str, Any] | None = None output_path: str | None = None + def resolve_video_params(self) -> VideoParams: + """Resolve top-level, nested, and ``size`` dimensions like vLLM-Omni.""" + params = VideoParams( + width=self.width, + height=self.height, + fps=self.fps, + num_frames=self.num_frames, + ) + if self.video_params is not None: + params.width = params.width or self.video_params.width + params.height = params.height or self.video_params.height + params.fps = params.fps or self.video_params.fps + params.num_frames = params.num_frames or self.video_params.num_frames + if self.size is not None: + width, height = self.size.split("x", 1) + params.width, params.height = int(width), int(height) + if params.fps is None: + params.fps = DEFAULT_FPS + if params.num_frames is None and self.seconds is not None: + params.num_frames = int(self.seconds) * params.fps + return params + + @property + def resolved_num_outputs(self) -> int: + return self.n if self.n is not None else self.num_outputs_per_prompt + + +# Backward-compatible spelling used by the original FastVideo/SGLang surface. +VideoGenerationsRequest = VideoGenerationRequest + + +class VideoError(BaseModel): + code: int | str = 500 + message: str + + +class VideoResponse(BaseModel): + id: str + object: Literal["video"] = "video" + model: str = "" + prompt: str = "" + status: VideoGenerationStatus = VideoGenerationStatus.QUEUED + progress: int = 0 + created_at: int = Field(default_factory=lambda: int(time.time())) + size: SizeStr | None = None + seconds: SecondStr = "4" + quality: str = "default" + url: str | None = None + remixed_from_video_id: str | None = None + expires_at: int | None = None + file_path: str | None = None + file_name: str | None = None + media_type: Literal["video/mp4"] = "video/mp4" + completed_at: int | None = None + error: VideoError | None = None + peak_memory_mb: float | None = None + inference_time_s: float | None = None + stage_durations: dict[str, float] = Field(default_factory=dict) + + +class VideoDeleteResponse(BaseModel): + id: str + deleted: bool + object: Literal["video.deleted"] = "video.deleted" + class VideoListResponse(BaseModel): data: list[VideoResponse] - object: str = "list" + first_id: str | None = None + last_id: str | None = None + has_more: bool = False + object: Literal["list"] = "list" def generate_request_id() -> str: diff --git a/fastvideo/entrypoints/openai/request_adapter.py b/fastvideo/entrypoints/openai/request_adapter.py new file mode 100644 index 000000000..179c88a35 --- /dev/null +++ b/fastvideo/entrypoints/openai/request_adapter.py @@ -0,0 +1,331 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Translate OpenAI/vLLM-Omni requests into FastVideo's typed request API.""" + +from __future__ import annotations + +import math +import os +from typing import Any + +from fastvideo.api.compat import ( + explicit_request_updates, + legacy_generate_call_to_request, + request_to_sampling_param, +) +from fastvideo.api.schema import GenerationRequest +from fastvideo.entrypoints.openai.protocol import ( + FileImageReference, + FileVideoReference, + VideoGenerationRequest, +) +from fastvideo.fastvideo_args import FastVideoArgs +from fastvideo.models.vision_utils import load_image +from fastvideo.registry import get_preset_selection + + +class RequestAdaptationError(ValueError): + """The transport request cannot be represented by the loaded pipeline.""" + + +def _as_list(value: Any | list[Any] | None) -> list[Any]: + if value is None: + return [] + return value if isinstance(value, list) else [value] + + +def _image_sources(request: VideoGenerationRequest) -> list[str]: + sources: list[str] = [] + for reference in _as_list(request.image_reference): + if isinstance(reference, FileImageReference): + raise RequestAdaptationError("file_id image references are not supported; provide image_url instead") + sources.append(reference.image_url) + legacy = request.input_reference or request.reference_url + if legacy is not None: + if sources: + raise RequestAdaptationError("Provide only one of input_reference/reference_url or image_reference.") + sources.append(legacy) + return sources + + +def _video_sources(request: VideoGenerationRequest) -> list[str]: + sources: list[str] = [] + for reference in _as_list(request.video_reference): + if isinstance(reference, FileVideoReference): + raise RequestAdaptationError("file_id video references are not supported; provide video_url instead") + sources.append(reference.video_url) + direct = request.video_path or request.video_url + if direct is not None: + if sources: + raise RequestAdaptationError("Provide only one of video_reference or video_path/video_url.") + sources.append(direct) + return sources + + +def _audio_sources(request: VideoGenerationRequest) -> list[str]: + return [reference.audio_url for reference in _as_list(request.audio_reference)] + + +def _parse_aspect_ratio(value: str) -> tuple[float, float]: + try: + width, height = value.split(":", 1) + result = float(width), float(height) + except (AttributeError, TypeError, ValueError) as error: + raise RequestAdaptationError(f"Invalid aspect_ratio {value!r}; expected WIDTH:HEIGHT") from error + if result[0] <= 0 or result[1] <= 0: + raise RequestAdaptationError(f"Invalid aspect_ratio {value!r}; both terms must be positive") + return result + + +def _apply_aspect_ratio( + kwargs: dict[str, Any], + request: VideoGenerationRequest, + *, + model_family: str | None, +) -> None: + if request.aspect_ratio is None or ("width" in kwargs and "height" in kwargs): + return + aspect_width, aspect_height = _parse_aspect_ratio(request.aspect_ratio) + if model_family == "minimax_h3": + from fastvideo.pipelines.basic.minimax_h3.packing import MINIMAX_H3_SHORT_EDGE, resolve_canvas_size + + if request.short_edge is not None and request.short_edge != MINIMAX_H3_SHORT_EDGE: + raise RequestAdaptationError( + f"MiniMax-H3 currently uses a fixed short_edge={MINIMAX_H3_SHORT_EDGE}, got {request.short_edge}.") + height, width = resolve_canvas_size(aspect_width, aspect_height) + elif request.short_edge is not None: + if aspect_width >= aspect_height: + height = request.short_edge + width = round(request.short_edge * aspect_width / aspect_height) + else: + width = request.short_edge + height = round(request.short_edge * aspect_height / aspect_width) + else: + return + kwargs["width"], kwargs["height"] = width, height + + +def validate_model_and_lora( + request: VideoGenerationRequest, + args: FastVideoArgs, + served_model_name: str, +) -> None: + """Validate vLLM-style model and LoRA selectors against startup state. + + FastVideo's published FastH3 adapters include dense replacement tensors in + addition to low-rank factors. Those tensors are applied while the model is + loaded and cannot be swapped safely between concurrent requests. The API + accepts vLLM's selector shape, but it must identify the startup adapter. + """ + allowed_models = {served_model_name} + if args.lora_path: + allowed_models.add(args.lora_nickname) + if request.model is not None and request.model not in allowed_models: + choices = ", ".join(sorted(allowed_models)) + raise RequestAdaptationError( + f"Model mismatch: request specifies {request.model!r}; this server provides {choices}.") + + if request.lora is None: + return + if not args.lora_path: + raise RequestAdaptationError( + "This server has no startup LoRA. Configure generator.pipeline.components.lora_path before using " + "the request lora selector.") + + body = request.lora + name = next((body[key] for key in ("name", "lora_name", "adapter") if body.get(key) is not None), None) + path = next((body[key] for key in ("path", "lora_path", "local_path") if body.get(key) is not None), None) + scale = next((body[key] for key in ("scale", "lora_scale") if body.get(key) is not None), None) + if name is None and path is None: + raise RequestAdaptationError("lora must provide a name or path") + if name is not None and str(name) not in {args.lora_nickname, served_model_name}: + raise RequestAdaptationError(f"Requested LoRA {name!r} is not the startup adapter {args.lora_nickname!r}.") + if path is not None and str(path) != args.lora_path: + raise RequestAdaptationError( + f"Requested LoRA path {path!r} does not match the startup adapter {args.lora_path!r}.") + if scale is not None: + try: + scale_value = float(scale) + except (TypeError, ValueError) as error: + raise RequestAdaptationError(f"Invalid LoRA scale {scale!r}") from error + if not math.isclose(scale_value, args.lora_strength, rel_tol=0.0, abs_tol=1e-8): + raise RequestAdaptationError( + f"Requested LoRA scale {scale_value:g} does not match startup strength {args.lora_strength:g}.") + + +def _apply_reference_inputs( + kwargs: dict[str, Any], + request: VideoGenerationRequest, + args: FastVideoArgs, + *, + model_family: str | None, +) -> None: + images = _image_sources(request) + videos = _video_sources(request) + audios = _audio_sources(request) + ref2va = model_family == "minimax_h3" and "ref2va" in (args.override_pipeline_cls_name or "").lower() + + if model_family == "minimax_h3" and request.task is not None: + normalized_task = request.task.lower() + if normalized_task not in {"t2va", "fl2va", "ref2va"}: + raise RequestAdaptationError("MiniMax-H3 task must be one of t2va, fl2va, or ref2va.") + if normalized_task == "ref2va" and not ref2va: + raise RequestAdaptationError( + "MiniMax-H3 task='ref2va' requires MiniMaxH3Ref2VAModularPipeline at server startup.") + if normalized_task != "ref2va" and ref2va: + raise RequestAdaptationError( + f"This server is configured for MiniMax-H3 Ref2VA, not task={normalized_task!r}.") + if normalized_task == "t2va" and (images or videos or audios): + raise RequestAdaptationError("MiniMax-H3 task='t2va' does not accept reference media.") + if normalized_task == "fl2va" and not images: + raise RequestAdaptationError("MiniMax-H3 task='fl2va' requires one or two image references.") + + if ref2va: + from fastvideo.pipelines.basic.minimax_h3 import MiniMaxH3Reference + + references = [MiniMaxH3Reference(source=source, media_type="image") for source in images] + references.extend(MiniMaxH3Reference(source=source, media_type="video") for source in videos) + references.extend(MiniMaxH3Reference(source=source, media_type="audio") for source in audios) + if references: + kwargs["references"] = references + return + + if request.task is not None and model_family != "minimax_h3": + raise RequestAdaptationError("The task selector is only defined for MiniMax-H3 servers.") + + if model_family == "minimax_h3" and (videos or audios): + raise RequestAdaptationError("MiniMax-H3 video/audio references require a server configured with " + "override_pipeline_cls_name=MiniMaxH3Ref2VAModularPipeline.") + if len(images) > 2: + raise RequestAdaptationError("The loaded pipeline accepts at most first and last image references.") + if images: + kwargs["image_path"] = images[0] + if len(images) == 2: + kwargs["last_image"] = load_image(images[1]) + if len(videos) > 1: + raise RequestAdaptationError("The loaded pipeline accepts at most one video reference.") + if videos: + kwargs["video_path"] = videos[0] + if audios: + raise RequestAdaptationError("The loaded pipeline does not accept audio reference inputs.") + + +def build_generation_request( + request_id: str, + request: VideoGenerationRequest, + args: FastVideoArgs, + *, + served_model_name: str, + output_dir: str, + default_request: GenerationRequest | None = None, +) -> GenerationRequest: + """Build one tracked FastVideo request using explicit-field precedence.""" + validate_model_and_lora(request, args, served_model_name) + kwargs: dict[str, Any] = {} + if default_request is not None: + kwargs.update(explicit_request_updates(default_request)) + + body_set = request.model_fields_set + nested_set = request.video_params.model_fields_set if request.video_params is not None else set() + if "size" in body_set and request.size is not None: + width, height = request.size.split("x", 1) + kwargs["width"], kwargs["height"] = int(width), int(height) + else: + if "width" in body_set and request.width is not None: + kwargs["width"] = request.width + elif "video_params" in body_set and "width" in nested_set and request.video_params.width is not None: + kwargs["width"] = request.video_params.width + if "height" in body_set and request.height is not None: + kwargs["height"] = request.height + elif "video_params" in body_set and "height" in nested_set and request.video_params.height is not None: + kwargs["height"] = request.video_params.height + + fps_explicit = "fps" in body_set or ("video_params" in body_set and "fps" in nested_set) + if fps_explicit: + fps = request.fps if "fps" in body_set else request.video_params.fps + if fps is not None: + kwargs["fps"] = fps + kwargs.setdefault("fps", 24) + + frames_explicit = "num_frames" in body_set or ("video_params" in body_set and "num_frames" in nested_set) + if frames_explicit: + num_frames = request.num_frames if "num_frames" in body_set else request.video_params.num_frames + if num_frames is not None: + kwargs["num_frames"] = num_frames + elif "seconds" in body_set and request.seconds is not None: + kwargs["num_frames"] = int(request.seconds) * int(kwargs["fps"]) + + direct_fields = ( + "seed", + "num_inference_steps", + "guidance_scale", + "guidance_scale_2", + "true_cfg_scale", + "negative_prompt", + "enable_teacache", + "max_sequence_length", + "boundary_ratio", + ) + for name in direct_fields: + if name in body_set: + value = getattr(request, name) + if value is not None: + kwargs[name] = value + if "n" in body_set or "num_outputs_per_prompt" in body_set: + kwargs["num_videos_per_prompt"] = request.resolved_num_outputs + + try: + _, model_family = get_preset_selection(args.model_path) + except (RuntimeError, ValueError): + model_family = None + if model_family == "minimax_h3" and request.resolved_num_outputs != 1: + raise RequestAdaptationError("MiniMax-H3 currently generates exactly one packed video/audio output.") + _apply_aspect_ratio(kwargs, request, model_family=model_family) + _apply_reference_inputs(kwargs, request, args, model_family=model_family) + + extension_fields = ("quality", "flow_shift", "sound_duration", "start_time_seconds") + for name in extension_fields: + if name in body_set and getattr(request, name) is not None: + kwargs[name] = getattr(request, name) + if "generate_sound" in body_set and request.generate_sound and model_family != "minimax_h3": + kwargs["generate_sound"] = True + if "enable_frame_interpolation" in body_set and request.enable_frame_interpolation: + kwargs["enable_frame_interpolation"] = True + for name in ( + "frame_interpolation_exp", + "frame_interpolation_scale", + "frame_interpolation_model_path", + ): + kwargs[name] = getattr(request, name) + if request.extra_params: + kwargs.update(request.extra_params) + if request.model_extra: + kwargs.update(request.model_extra) + + configured_output = kwargs.pop("output_path", None) + requested_output = request.output_path if "output_path" in body_set else None + destination = requested_output or configured_output or os.path.join(output_dir, "videos") + if os.path.splitext(destination)[1].lower() == ".mp4": + output_path = destination + else: + output_path = os.path.join(destination, f"{request_id}.mp4") + os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True) + kwargs.update({ + "output_path": output_path, + "save_video": True, + "return_frames": False, + }) + generation_request = legacy_generate_call_to_request(request.prompt, None, legacy_kwargs=kwargs) + try: + # Resolve once at admission time so unsupported model-specific fields + # are a deterministic 400, rather than an asynchronous failed job. + request_to_sampling_param(generation_request, model_path=args.model_path) + except (TypeError, ValueError) as error: + raise RequestAdaptationError(str(error)) from error + return generation_request + + +__all__ = [ + "RequestAdaptationError", + "build_generation_request", + "validate_model_and_lora", +] diff --git a/fastvideo/entrypoints/openai/serving_engine.py b/fastvideo/entrypoints/openai/serving_engine.py new file mode 100644 index 000000000..047cab57e --- /dev/null +++ b/fastvideo/entrypoints/openai/serving_engine.py @@ -0,0 +1,82 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Shared asynchronous execution substrate for OpenAI-compatible routes.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from typing import Any, TypeVar + +from fastvideo.api.schema import GenerationRequest +from fastvideo.entrypoints.video_generator import VideoGenerator + +_T = TypeVar("_T") + + +class OpenAIServingEngine: + """Own generator lifecycle and serialize access to its mutable pipeline. + + FastVideo pipelines contain request-mutated sampling state and some LoRA + implementations merge weights in place. Running two Python threads through + one pipeline is therefore unsafe even if the HTTP layer accepts requests + concurrently. This engine gives every OpenAI route one model-agnostic async + entrypoint while preserving that invariant. A future scheduler can replace + the lock without changing the transport contract. + """ + + def __init__(self, generator: VideoGenerator) -> None: + self._generator = generator + self._generation_lock = asyncio.Lock() + self._closed = False + + @property + def generator(self) -> VideoGenerator: + return self._generator + + @property + def closed(self) -> bool: + return self._closed + + async def generate(self, request: GenerationRequest) -> Any: + """Generate one typed request without blocking the event loop.""" + return await self.run_serialized(self._generator.generate, request) + + async def run_serialized(self, function: Callable[..., _T], *args: Any, **kwargs: Any) -> _T: + """Run a synchronous pipeline operation under the serving lock.""" + if self._closed: + raise RuntimeError("FastVideo serving engine is shutting down") + async with self._generation_lock: + if self._closed: + raise RuntimeError("FastVideo serving engine is shutting down") + worker = asyncio.create_task(asyncio.to_thread(function, *args, **kwargs)) + try: + return await asyncio.shield(worker) + except asyncio.CancelledError: + # Python cannot stop a running worker thread. Keep the lock + # until the pipeline call really exits so cancellation cannot + # expose mutable model state to a second request. + await asyncio.shield(worker) + raise + + async def run_async_serialized(self, function: Callable[[], Awaitable[_T]]) -> _T: + """Run an async operation under the same pipeline lock.""" + if self._closed: + raise RuntimeError("FastVideo serving engine is shutting down") + async with self._generation_lock: + if self._closed: + raise RuntimeError("FastVideo serving engine is shutting down") + worker: asyncio.Future[_T] = asyncio.ensure_future(function()) + try: + return await asyncio.shield(worker) + except asyncio.CancelledError: + await asyncio.shield(worker) + raise + + async def shutdown(self) -> None: + """Stop accepting requests and release the generator after in-flight work.""" + self._closed = True + async with self._generation_lock: + await asyncio.to_thread(self._generator.shutdown) + + +__all__ = ["OpenAIServingEngine"] diff --git a/fastvideo/entrypoints/openai/state.py b/fastvideo/entrypoints/openai/state.py index c95f84400..0d731a058 100644 --- a/fastvideo/entrypoints/openai/state.py +++ b/fastvideo/entrypoints/openai/state.py @@ -11,14 +11,17 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: from fastvideo.api.schema import GenerationRequest + from fastvideo.entrypoints.openai.serving_engine import OpenAIServingEngine from fastvideo.entrypoints.video_generator import VideoGenerator from fastvideo.fastvideo_args import FastVideoArgs DEFAULT_OUTPUT_DIR = "outputs" _generator: VideoGenerator | None = None +_serving_engine: OpenAIServingEngine | None = None _fastvideo_args: FastVideoArgs | None = None _output_dir: str = DEFAULT_OUTPUT_DIR +_served_model_name: str | None = None _default_request: GenerationRequest | None = None @@ -28,6 +31,12 @@ def get_generator() -> VideoGenerator: return _generator +def get_serving_engine() -> OpenAIServingEngine: + """Return the shared model-agnostic OpenAI serving engine.""" + assert _serving_engine is not None, "Server not initialized — serving engine is None" + return _serving_engine + + def get_server_args() -> FastVideoArgs: """Return the global FastVideoArgs (set during startup).""" assert _fastvideo_args is not None, "Server not initialized — args is None" @@ -39,6 +48,12 @@ def get_output_dir() -> str: return _output_dir +def get_served_model_name() -> str: + """Return the public model id advertised by the OpenAI server.""" + args = get_server_args() + return _served_model_name or args.model_path + + def get_default_request() -> GenerationRequest | None: """Return the ServeConfig.default_request set at startup, if any.""" return _default_request @@ -46,21 +61,27 @@ def get_default_request() -> GenerationRequest | None: def set_state( generator: VideoGenerator, + serving_engine: OpenAIServingEngine, fastvideo_args: FastVideoArgs, output_dir: str, default_request: GenerationRequest | None = None, + served_model_name: str | None = None, ) -> None: """Set all server state at once (called from lifespan).""" - global _generator, _fastvideo_args, _output_dir, _default_request + global _generator, _serving_engine, _fastvideo_args, _output_dir, _served_model_name, _default_request _generator = generator + _serving_engine = serving_engine _fastvideo_args = fastvideo_args _output_dir = output_dir + _served_model_name = served_model_name _default_request = default_request def clear_state() -> None: """Clear server state on shutdown.""" - global _generator, _fastvideo_args, _default_request + global _generator, _serving_engine, _fastvideo_args, _served_model_name, _default_request _generator = None + _serving_engine = None _fastvideo_args = None + _served_model_name = None _default_request = None diff --git a/fastvideo/entrypoints/openai/stores.py b/fastvideo/entrypoints/openai/stores.py index 30ca5d999..1b78298e5 100644 --- a/fastvideo/entrypoints/openai/stores.py +++ b/fastvideo/entrypoints/openai/stores.py @@ -41,6 +41,10 @@ class AsyncDictStore: async with self._lock: return list(self._items.values()) + async def clear(self) -> None: + async with self._lock: + self._items.clear() + # Global stores shared by OpenAI entrypoints VIDEO_STORE = AsyncDictStore() diff --git a/fastvideo/entrypoints/openai/video_api.py b/fastvideo/entrypoints/openai/video_api.py index 4134175f3..97003d293 100644 --- a/fastvideo/entrypoints/openai/video_api.py +++ b/fastvideo/entrypoints/openai/video_api.py @@ -1,373 +1,455 @@ -# Adapted from SGLang -# (https://github.com/sgl-project/sglang/blob/main/python/sglang/multimodal_gen/runtime/entrypoints/openai/video_api.py) +# SPDX-License-Identifier: Apache-2.0 +"""OpenAI/vLLM-Omni-compatible video generation routes.""" + +from __future__ import annotations import asyncio import json import os import time +from contextlib import suppress from typing import Any -from fastapi import ( - APIRouter, - File, - Form, - HTTPException, - Path, - Query, - Request, - UploadFile, -) -from fastapi.responses import FileResponse +from fastapi import APIRouter, HTTPException, Path, Query, Request +from fastapi.responses import FileResponse, JSONResponse +from pydantic import ValidationError +from starlette.datastructures import UploadFile -from fastvideo.api.compat import explicit_request_updates +from fastvideo.api.compat import explicit_request_updates, request_to_sampling_param from fastvideo.api.schema import GenerationRequest -from fastvideo.entrypoints.openai.state import ( - get_default_request, - get_generator, - get_output_dir, - get_server_args, -) from fastvideo.entrypoints.openai.protocol import ( - VideoGenerationsRequest, + VideoDeleteResponse, + VideoGenerationRequest, + VideoGenerationStatus, VideoListResponse, VideoResponse, generate_request_id, ) -from fastvideo.entrypoints.openai.stores import VIDEO_STORE -from fastvideo.entrypoints.openai.utils import ( - merge_image_input_list, - parse_size, - save_image_to_path, +from fastvideo.entrypoints.openai.request_adapter import RequestAdaptationError, build_generation_request +from fastvideo.entrypoints.openai.state import ( + get_default_request, + get_output_dir, + get_served_model_name, + get_server_args, + get_serving_engine, ) +from fastvideo.entrypoints.openai.stores import VIDEO_STORE +from fastvideo.entrypoints.openai.utils import parse_size, save_image_to_path from fastvideo.logger import init_logger logger = init_logger(__name__) router = APIRouter(prefix="/v1/videos", tags=["videos"]) +_VIDEO_JOB_TASKS: dict[str, asyncio.Task[None]] = {} +_DELETED_VIDEO_IDS: set[str] = set() +_JSON_FORM_FIELDS = { + "image_reference", + "video_reference", + "audio_reference", + "video_params", + "lora", + "extra_params", +} + def _build_generation_kwargs( request_id: str, - req: VideoGenerationsRequest, + req: VideoGenerationRequest, default_request: GenerationRequest | None = None, ) -> dict[str, Any]: - """Build a flat kwargs dict for ``generator.generate_video``. + """Backward-compatible flat projection used by helper-level callers. - Precedence (highest to lowest): - 1. Request body — only fields the client explicitly sent - (``req.model_fields_set``, Pydantic v2). - 2. ``default_request`` — only fields the operator explicitly set in - the serve YAML, projected via ``explicit_request_updates``. Schema - defaults on the dataclass are *not* treated as defaults here. - 3. Hardcoded fallback (e.g. ``fps=24`` when neither side set it). - - Why gate on ``model_fields_set`` / explicit paths? Both the request - Pydantic model and the ``GenerationRequest`` dataclass carry schema - defaults (e.g. ``seed=1024``, ``num_frames=125``). Without the gate - those would masquerade as intent and shadow the other side — the - gate preserves "operator pinned it" vs. "dataclass happened to have - that default." + Runtime serving uses :func:`build_generation_request` and the typed + ``VideoGenerator.generate`` API. Keeping this helper avoids breaking code + that imported the original FastVideo adapter directly. """ kwargs: dict[str, Any] = {} if default_request is not None: kwargs.update(explicit_request_updates(default_request)) body_set = req.model_fields_set + nested_set = req.video_params.model_fields_set if req.video_params is not None else set() kwargs["prompt"] = req.prompt - if "size" in body_set and req.size: - w, h = parse_size(req.size) - if w is not None and h is not None: - kwargs["width"] = w - kwargs["height"] = h + width, height = parse_size(req.size) + if width is not None and height is not None: + kwargs["width"], kwargs["height"] = width, height + else: + if "width" in body_set and req.width is not None: + kwargs["width"] = req.width + elif "video_params" in body_set and "width" in nested_set and req.video_params.width is not None: + kwargs["width"] = req.video_params.width + if "height" in body_set and req.height is not None: + kwargs["height"] = req.height + elif "video_params" in body_set and "height" in nested_set and req.video_params.height is not None: + kwargs["height"] = req.video_params.height if "fps" in body_set and req.fps is not None: kwargs["fps"] = req.fps + elif "video_params" in body_set and "fps" in nested_set and req.video_params.fps is not None: + kwargs["fps"] = req.video_params.fps + kwargs.setdefault("fps", 24) if "num_frames" in body_set and req.num_frames is not None: kwargs["num_frames"] = req.num_frames + elif "video_params" in body_set and "num_frames" in nested_set and req.video_params.num_frames is not None: + kwargs["num_frames"] = req.video_params.num_frames elif "seconds" in body_set and req.seconds is not None: - fps = kwargs.get("fps", 24) - kwargs["num_frames"] = fps * req.seconds - - if "seed" in body_set and req.seed is not None: - kwargs["seed"] = req.seed - if ("num_inference_steps" in body_set and req.num_inference_steps is not None): - kwargs["num_inference_steps"] = req.num_inference_steps - if "guidance_scale" in body_set and req.guidance_scale is not None: - kwargs["guidance_scale"] = req.guidance_scale - if "guidance_scale_2" in body_set and req.guidance_scale_2 is not None: - kwargs["guidance_scale_2"] = req.guidance_scale_2 - if "negative_prompt" in body_set and req.negative_prompt is not None: - kwargs["negative_prompt"] = req.negative_prompt - if "enable_teacache" in body_set and req.enable_teacache: - kwargs["enable_teacache"] = True - if "true_cfg_scale" in body_set and req.true_cfg_scale is not None: - kwargs["true_cfg_scale"] = req.true_cfg_scale + kwargs["num_frames"] = int(req.seconds) * int(kwargs["fps"]) + for name in ( + "seed", + "num_inference_steps", + "guidance_scale", + "guidance_scale_2", + "true_cfg_scale", + "negative_prompt", + "enable_teacache", + "max_sequence_length", + "boundary_ratio", + ): + if name in body_set and getattr(req, name) is not None: + kwargs[name] = getattr(req, name) + if "n" in body_set or "num_outputs_per_prompt" in body_set: + kwargs["num_videos_per_prompt"] = req.resolved_num_outputs if "input_reference" in body_set and req.input_reference is not None: kwargs["image_path"] = req.input_reference - kwargs.setdefault("fps", 24) - - default_output_path = kwargs.pop("output_path", None) - body_output_dir = req.output_path if "output_path" in body_set else None - output_dir = body_output_dir or default_output_path or os.path.join(get_output_dir(), "videos") + configured_output = kwargs.pop("output_path", None) + requested_output = req.output_path if "output_path" in body_set else None + output_dir = requested_output or configured_output or os.path.join(get_output_dir(), "videos") os.makedirs(output_dir, exist_ok=True) kwargs["output_path"] = os.path.join(output_dir, f"{request_id}.mp4") kwargs["save_video"] = True - return kwargs +def _result_value(result: Any, name: str, default: Any = None) -> Any: + if isinstance(result, dict): + return result.get(name, default) + return getattr(result, name, default) + + +def _stage_durations(result: Any) -> dict[str, float]: + logging_info = _result_value(result, "logging_info") + stages = getattr(logging_info, "stages", None) + if not isinstance(stages, dict): + return {} + durations: dict[str, float] = {} + for stage_name, metrics in stages.items(): + if isinstance(metrics, dict) and metrics.get("execution_time") is not None: + durations[str(stage_name)] = float(metrics["execution_time"]) + return durations + + def _make_video_job( request_id: str, - req: VideoGenerationsRequest, - kwargs: dict[str, Any], + req: VideoGenerationRequest, + generation_request: GenerationRequest, ) -> dict[str, Any]: - """Build the initial job dict stored in VIDEO_STORE.""" - w = kwargs.get("width", 0) - h = kwargs.get("height", 0) - size_str = f"{w}x{h}" if w and h else "" - num_frames = kwargs.get("num_frames", 0) - fps = kwargs.get("fps", 24) - seconds = int(round(num_frames / fps)) if fps else 0 + sampling = request_to_sampling_param(generation_request, model_path=get_server_args().model_path) + size = f"{sampling.width}x{sampling.height}" if sampling.width and sampling.height else None + seconds = int(round(sampling.num_frames / sampling.fps)) if sampling.fps else int(req.seconds or 4) return { "id": request_id, "object": "video", - "model": req.model or get_server_args().model_path, - "status": "queued", + "model": req.model or get_served_model_name(), + "prompt": req.prompt, + "status": VideoGenerationStatus.QUEUED, "progress": 0, "created_at": int(time.time()), - "size": size_str, - "seconds": str(seconds), - "quality": "standard", - "file_path": kwargs.get("output_path"), + "size": size, + "seconds": str(max(1, seconds)), + "quality": req.quality or "default", + # ``file_path`` is a FastVideo compatibility extension. vLLM-Omni's + # public job shape uses ``file_name`` after completion. + "file_path": generation_request.output.output_path, } -async def _run_generation(request_id: str, kwargs: dict[str, Any]) -> None: - """ - Run video generation in a background thread (VideoGenerator.generate_video - is synchronous) and update the store on completion or failure. - """ - generator = get_generator() - loop = asyncio.get_running_loop() - +async def _run_generation( + request_id: str, + generation_request: GenerationRequest, +) -> None: + await VIDEO_STORE.update_fields( + request_id, + { + "status": VideoGenerationStatus.IN_PROGRESS, + "progress": 0 + }, + ) + started = time.perf_counter() try: - start = time.perf_counter() - - result = await loop.run_in_executor( - None, - lambda: generator.generate_video(**kwargs), - ) - - elapsed = time.perf_counter() - start - update: dict[str, Any] = { - "status": "completed", - "progress": 100, - "completed_at": int(time.time()), - "inference_time_s": elapsed, - } - - if isinstance(result, dict): - gen_time = result.get("generation_time") - if gen_time is not None: - update["inference_time_s"] = gen_time - peak_mem = result.get("peak_memory_mb") - if peak_mem is not None: - update["peak_memory_mb"] = peak_mem - - await VIDEO_STORE.update_fields(request_id, update) - logger.info("Video %s completed in %.2fs", request_id, elapsed) - - except Exception as e: - logger.error("Video generation failed for %s: %s", request_id, e) + result = await get_serving_engine().generate(generation_request) + if isinstance(result, list): + if not result: + raise RuntimeError("FastVideo returned no generation results") + result = result[0] + elapsed = time.perf_counter() - started + video_path = _result_value(result, "video_path") or generation_request.output.output_path + generation_time = _result_value(result, "generation_time", elapsed) await VIDEO_STORE.update_fields( request_id, { - "status": "failed", - "error": { - "message": str(e) - } + "status": VideoGenerationStatus.COMPLETED, + "progress": 100, + "completed_at": int(time.time()), + "file_path": video_path, + "file_name": os.path.basename(video_path) if video_path else None, + "inference_time_s": float(generation_time or elapsed), + "peak_memory_mb": _result_value(result, "peak_memory_mb"), + "stage_durations": _stage_durations(result), }, ) + logger.info("Video %s completed in %.2fs", request_id, elapsed) + except asyncio.CancelledError: + logger.info("Video %s was cancelled", request_id) + raise + except Exception as error: + logger.exception("Video generation failed for %s", request_id) + await VIDEO_STORE.update_fields( + request_id, + { + "status": VideoGenerationStatus.FAILED, + "error": { + "code": 500, + "message": str(error) + }, + "inference_time_s": time.perf_counter() - started, + }, + ) + finally: + if request_id in _DELETED_VIDEO_IDS: + _DELETED_VIDEO_IDS.discard(request_id) + output_path = generation_request.output.output_path + if output_path and os.path.isfile(output_path): + try: + os.unlink(output_path) + except OSError: + logger.warning("Failed to clean up deleted video artifact %s", output_path, exc_info=True) -# Endpoints +def _track_video_job(request_id: str, task: asyncio.Task[None]) -> None: + _VIDEO_JOB_TASKS[request_id] = task + + def discard(completed: asyncio.Task[None]) -> None: + if _VIDEO_JOB_TASKS.get(request_id) is completed: + _VIDEO_JOB_TASKS.pop(request_id, None) + + task.add_done_callback(discard) + + +async def shutdown_video_jobs() -> None: + """Cancel all transport tasks before the serving engine shuts down.""" + tasks = list(_VIDEO_JOB_TASKS.values()) + _VIDEO_JOB_TASKS.clear() + for task in tasks: + task.cancel() + for task in tasks: + with suppress(asyncio.CancelledError): + await task + + +def _parse_json_form_value(name: str, value: Any) -> Any: + if value is None or not isinstance(value, str) or name not in _JSON_FORM_FIELDS: + return value + try: + return json.loads(value) + except json.JSONDecodeError as error: + raise HTTPException(status_code=400, detail=f"{name} is not valid JSON") from error + + +async def _parse_video_request(raw_request: Request) -> VideoGenerationRequest: + content_type = raw_request.headers.get("content-type", "").lower() + if "multipart/form-data" in content_type or "application/x-www-form-urlencoded" in content_type: + form = await raw_request.form() + payload: dict[str, Any] = {} + for name, value in form.multi_items(): + if name == "input_reference" and isinstance(value, UploadFile): + uploads_dir = os.path.join(get_output_dir(), "uploads") + filename = os.path.basename(value.filename or "reference") + target = os.path.join(uploads_dir, f"{generate_request_id()}_{filename}") + saved_path = await save_image_to_path(value, target) + if (value.content_type or "").lower().startswith("video/"): + payload["video_reference"] = {"video_url": saved_path} + else: + payload["input_reference"] = saved_path + continue + parsed = _parse_json_form_value(name, value) + if name in payload: + current = payload[name] + payload[name] = current + [parsed] if isinstance(current, list) else [current, parsed] + else: + payload[name] = parsed + else: + try: + body = await raw_request.json() + except Exception as error: + raise HTTPException(status_code=400, detail="Request body must be valid JSON") from error + if not isinstance(body, dict): + raise HTTPException(status_code=400, detail="Request body must be a JSON object") + payload = dict(body) + + for name in ("extra_body", "extra_json"): + extra = payload.pop(name, None) + if isinstance(extra, str): + try: + extra = json.loads(extra) + except json.JSONDecodeError as error: + raise HTTPException(status_code=400, detail=f"{name} is not valid JSON") from error + if extra is not None and not isinstance(extra, dict): + raise HTTPException(status_code=400, detail=f"{name} must be a JSON object") + if extra: + payload.update(extra) + + try: + return VideoGenerationRequest(**payload) + except ValidationError as error: + raise HTTPException(status_code=400, detail=f"Invalid request body: {error}") from error + + +def _adapt_request(request_id: str, request: VideoGenerationRequest) -> GenerationRequest: + try: + return build_generation_request( + request_id, + request, + get_server_args(), + served_model_name=get_served_model_name(), + output_dir=get_output_dir(), + default_request=get_default_request(), + ) + except RequestAdaptationError as error: + raise HTTPException(status_code=400, detail=str(error)) from error @router.post("", response_model=VideoResponse) -async def create_video( - request: Request, - # multipart/form-data fields - prompt: str | None = Form(None), - input_reference: UploadFile | None = File(None), # noqa: B008 - reference_url: str | None = Form(None), - model: str | None = Form(None), - seconds: int | None = Form(None), - size: str | None = Form(None), - fps: int | None = Form(None), - num_frames: int | None = Form(None), - seed: int | None = Form(1024), - negative_prompt: str | None = Form(None), - guidance_scale: float | None = Form(None), - num_inference_steps: int | None = Form(None), - enable_teacache: bool | None = Form(False), - extra_body: str | None = Form(None), -): - content_type = request.headers.get("content-type", "").lower() - request_id = generate_request_id() - - if "multipart/form-data" in content_type: - if not prompt: - raise HTTPException(status_code=400, detail="prompt is required") - - input_path = None - image_list = merge_image_input_list(input_reference, reference_url) - if image_list: - image = image_list[0] - uploads_dir = os.path.join(get_output_dir(), "uploads") - os.makedirs(uploads_dir, exist_ok=True) - filename = getattr(image, "filename", "url_image") - input_path = os.path.join(uploads_dir, f"{request_id}_{filename}") - try: - input_path = await save_image_to_path(image, input_path) - except Exception as e: - raise HTTPException( - status_code=400, - detail=f"Failed to process image: {e}", - ) from None - - extra: dict[str, Any] = {} - if extra_body: - try: - extra = json.loads(extra_body) - except Exception: - extra = {} - - req = VideoGenerationsRequest( - prompt=prompt, - input_reference=input_path, - model=model, - seconds=seconds if seconds is not None else 4, - size=size, - fps=fps if fps is not None else extra.get("fps"), - num_frames=(num_frames if num_frames is not None else extra.get("num_frames")), - seed=seed, - negative_prompt=negative_prompt, - num_inference_steps=num_inference_steps, - enable_teacache=enable_teacache, - **({ - "guidance_scale": guidance_scale - } if guidance_scale is not None else {}), - ) - else: - try: - body = await request.json() - except Exception: - body = {} - - payload: dict[str, Any] = dict(body or {}) - for key in ("extra_body", "extra_json"): - extra = payload.pop(key, None) - if isinstance(extra, dict): - payload.update(extra) - - if payload.get("reference_url"): - image_list = merge_image_input_list(payload.get("reference_url")) - if image_list: - image = image_list[0] - uploads_dir = os.path.join(get_output_dir(), "uploads") - os.makedirs(uploads_dir, exist_ok=True) - input_path = os.path.join(uploads_dir, f"{request_id}_url_image") - try: - input_path = await save_image_to_path(image, input_path) - except Exception as e: - raise HTTPException( - status_code=400, - detail=f"Failed to process image: {e}", - ) from None - payload["input_reference"] = input_path - - try: - req = VideoGenerationsRequest(**payload) - except Exception as e: - raise HTTPException( - status_code=400, - detail=f"Invalid request body: {e}", - ) from None - - logger.info("Video generation request %s: prompt=%s", request_id, req.prompt[:100]) - - # default_request was validated at server startup (run_server) and is - # read-only on the request hot path — _build_generation_kwargs and - # explicit_request_updates only read, so no per-request deepcopy needed. - default_request = get_default_request() - - gen_kwargs = _build_generation_kwargs(request_id, req, default_request=default_request) - job = _make_video_job(request_id, req, gen_kwargs) +@router.post("/generations", response_model=VideoResponse, include_in_schema=False) +async def create_video(raw_request: Request) -> VideoResponse: + """Create an asynchronous video generation job.""" + request = await _parse_video_request(raw_request) + request_id = f"video_gen_{generate_request_id()}" + generation_request = _adapt_request(request_id, request) + job = _make_video_job(request_id, request, generation_request) await VIDEO_STORE.upsert(request_id, job) - - asyncio.create_task(_run_generation(request_id, gen_kwargs)) - + task = asyncio.create_task(_run_generation(request_id, generation_request), name=f"video-job-{request_id}") + _track_video_job(request_id, task) return VideoResponse(**job) +@router.post("/sync") +async def create_video_sync(raw_request: Request) -> FileResponse: + """Generate synchronously and return raw MP4 bytes with vLLM headers.""" + request = await _parse_video_request(raw_request) + request_id = f"video_sync-{generate_request_id()}" + generation_request = _adapt_request(request_id, request) + started = time.perf_counter() + try: + result = await get_serving_engine().generate(generation_request) + except Exception as error: + logger.exception("Sync video generation failed for %s", request_id) + raise HTTPException(status_code=500, detail=f"Video generation failed: {error}") from error + if isinstance(result, list): + if not result: + raise HTTPException(status_code=500, detail="FastVideo returned no generation results") + result = result[0] + elapsed = time.perf_counter() - started + video_path = _result_value(result, "video_path") or generation_request.output.output_path + if not video_path or not os.path.exists(video_path): + raise HTTPException(status_code=500, detail="FastVideo did not produce an MP4 file") + return FileResponse( + video_path, + media_type="video/mp4", + filename=os.path.basename(video_path), + headers={ + "X-Request-Id": request_id, + "X-Model": get_served_model_name(), + "X-Inference-Time-S": f"{elapsed:.3f}", + "X-Stage-Durations": json.dumps(_stage_durations(result), separators=(",", ":")), + "X-Peak-Memory-MB": f"{float(_result_value(result, 'peak_memory_mb', 0.0) or 0.0):.3f}", + }, + ) + + @router.get("", response_model=VideoListResponse) async def list_videos( after: str | None = Query(None), - limit: int | None = Query(None, ge=1, le=100), - order: str | None = Query("desc"), -): - order = (order or "desc").lower() - if order not in ("asc", "desc"): - order = "desc" + limit: int | None = Query(None, ge=0, le=100), + order: str = Query("desc", pattern="^(asc|desc)$"), +) -> VideoListResponse: jobs = await VIDEO_STORE.list_values() - jobs.sort(key=lambda j: j.get("created_at", 0), reverse=(order != "asc")) - + jobs.sort(key=lambda job: job.get("created_at", 0), reverse=order == "desc") if after is not None: - try: - idx = next(i for i, j in enumerate(jobs) if j["id"] == after) - jobs = jobs[idx + 1:] - except StopIteration: - jobs = [] - + index = next((i for i, job in enumerate(jobs) if job.get("id") == after), None) + jobs = [] if index is None else jobs[index + 1:] + has_more = limit is not None and len(jobs) > limit if limit is not None: jobs = jobs[:limit] - return VideoListResponse(data=[VideoResponse(**j) for j in jobs]) + responses = [VideoResponse(**job) for job in jobs] + return VideoListResponse( + data=responses, + first_id=responses[0].id if responses else None, + last_id=responses[-1].id if responses else None, + has_more=has_more, + ) -@router.get("/{video_id}", response_model=VideoResponse) -async def retrieve_video(video_id: str = Path(...)): +@router.get("/{video_id}", response_model=None) +async def retrieve_video(video_id: str = Path(...)) -> VideoResponse | JSONResponse: job = await VIDEO_STORE.get(video_id) - if not job: + if job is None: raise HTTPException(status_code=404, detail="Video not found") - return VideoResponse(**job) + response = VideoResponse(**job) + if response.status is VideoGenerationStatus.FAILED: + return JSONResponse(status_code=500, content=response.model_dump(mode="json")) + return response -@router.delete("/{video_id}", response_model=VideoResponse) -async def delete_video(video_id: str = Path(...)): - job = await VIDEO_STORE.pop(video_id) - if not job: +@router.delete("/{video_id}", response_model=VideoDeleteResponse) +async def delete_video(video_id: str = Path(...)) -> VideoDeleteResponse: + job = await VIDEO_STORE.get(video_id) + if job is None: raise HTTPException(status_code=404, detail="Video not found") - job["status"] = "deleted" - return VideoResponse(**job) + task = _VIDEO_JOB_TASKS.get(video_id) + if task is not None: + # The current synchronous generator cannot abort a CUDA call once it + # starts. Remove the API resource immediately and let the tracked task + # clean up its artifact on exit while the serving lock stays held. + _DELETED_VIDEO_IDS.add(video_id) + popped = await VIDEO_STORE.pop(video_id) + file_path = None if popped is None else popped.get("file_path") + if file_path and os.path.isfile(file_path): + try: + os.unlink(file_path) + except OSError: + logger.warning("Failed to delete video artifact %s", file_path, exc_info=True) + return VideoDeleteResponse(id=video_id, deleted=True) @router.get("/{video_id}/content") -async def download_video_content(video_id: str = Path(...), variant: str | None = Query(None)): +async def download_video_content(video_id: str = Path(...), variant: str | None = Query(None)) -> FileResponse: + del variant job = await VIDEO_STORE.get(video_id) - if not job: + if job is None: raise HTTPException(status_code=404, detail="Video not found") - + status = VideoGenerationStatus(job.get("status", VideoGenerationStatus.QUEUED)) + if status is VideoGenerationStatus.FAILED: + raise HTTPException(status_code=422, detail="Video generation failed. Check job status for error details.") file_path = job.get("file_path") - if not file_path or not os.path.exists(file_path): - if job.get("status") == "failed": - raise HTTPException(status_code=500, detail="Video generation failed") - raise HTTPException(status_code=404, detail="Video still being generated") + if status is not VideoGenerationStatus.COMPLETED or not file_path: + raise HTTPException(status_code=404, detail="Generation is still in-progress") + if not os.path.exists(file_path): + raise HTTPException(status_code=404, detail="Generated video file not found on disk") + return FileResponse(path=file_path, media_type="video/mp4", filename=os.path.basename(file_path)) - return FileResponse( - path=file_path, - media_type="video/mp4", - filename=os.path.basename(file_path), - ) + +__all__ = [ + "_build_generation_kwargs", + "create_video", + "create_video_sync", + "delete_video", + "download_video_content", + "list_videos", + "retrieve_video", + "router", + "shutdown_video_jobs", +] diff --git a/fastvideo/entrypoints/video_generator.py b/fastvideo/entrypoints/video_generator.py index 5ad30d3ed..cf55c1e1f 100644 --- a/fastvideo/entrypoints/video_generator.py +++ b/fastvideo/entrypoints/video_generator.py @@ -27,6 +27,7 @@ import torchvision from einops import rearrange from fastvideo.api.compat import ( + REQUEST_BATCH_EXTRA_PASSTHROUGH_FIELDS, expand_request_prompt_batch, generator_config_to_fastvideo_args, legacy_from_pretrained_to_config, @@ -34,6 +35,7 @@ from fastvideo.api.compat import ( load_generator_config_from_file, normalize_generation_request, normalize_generator_config, + request_to_batch_extra, request_to_pipeline_overrides, request_to_sampling_param, ) @@ -65,18 +67,7 @@ except ImportError: logger = init_logger(__name__) _FFMPEG_ENCODER_OPTION_CACHE: dict[tuple[str, str, str], bool] = {} -_BATCH_EXTRA_PASSTHROUGH_KEYS: tuple[str, ...] = ( - "ltx2_audio_latents", - "ltx2_audio_clean_latent", - "ltx2_audio_denoise_mask", - "audio_num_frames", - "video_position_offset_sec", - # MiniMax-H3 VSA per-request knobs (read by the H3 denoising stage; - # sparsity itself flows through the existing ForwardBatch.VSA_sparsity) - "vsa_mode", - "vsa_dense_first_n_steps", - "vsa_dense_layers", -) +_BATCH_EXTRA_PASSTHROUGH_KEYS = tuple(REQUEST_BATCH_EXTRA_PASSTHROUGH_FIELDS) _FROM_PRETRAINED_CONVENIENCE_KWARGS = frozenset({ "num_gpus", @@ -520,10 +511,12 @@ class VideoGenerator: request, model_path=self.fastvideo_args.model_path, ) + batch_extra = request_to_batch_extra(request) result = self._generate_video_impl( prompt=request.prompt, sampling_param=sampling_param, fastvideo_args=fastvideo_args, + **batch_extra, ) return self._wrap_legacy_result(result) diff --git a/fastvideo/registry.py b/fastvideo/registry.py index c0f64ef7d..083732901 100644 --- a/fastvideo/registry.py +++ b/fastvideo/registry.py @@ -1126,17 +1126,24 @@ def _register_configs() -> None: sampling_param_cls=None, pipeline_config_cls=MiniMaxH3PipelineConfig, workload_types=(WorkloadType.T2V, WorkloadType.I2V), - hf_model_paths=["MiniMaxAI/MiniMax-H3"], + hf_model_paths=[ + "MiniMaxAI/MiniMax-H3", + "FastVideo/FastVideo-Minimax-FastH3-Preview-v0.2", + ], model_detectors=[ lambda path: any(token in path.lower() for token in ( "minimax-h3", "minimax_h3", + "fasth3", "minimaxh3modularpipeline", "minimaxh3ref2vamodularpipeline", )), ], model_family="minimax_h3", default_preset="minimax_h3_t2va", + # FastH3 full checkpoints need not carry a Diffusers model_index.json; + # the native full-checkpoint loader still uses the standard H3 graph. + pipeline_cls_name="MiniMaxH3ModularPipeline", ) # SD3.5 diff --git a/fastvideo/tests/entrypoints/test_openai_api.py b/fastvideo/tests/entrypoints/test_openai_api.py index 8097cac88..4dcb592cf 100644 --- a/fastvideo/tests/entrypoints/test_openai_api.py +++ b/fastvideo/tests/entrypoints/test_openai_api.py @@ -426,8 +426,8 @@ class TestProtocolModels: def test_video_request_defaults(self): req = VideoGenerationsRequest(prompt="hello") - assert req.seconds == 4 - assert req.seed == 1024 + assert req.seconds is None + assert req.seed is None def test_video_response_defaults(self): resp = VideoResponse(id="v1") diff --git a/fastvideo/tests/entrypoints/test_openai_serving_engine.py b/fastvideo/tests/entrypoints/test_openai_serving_engine.py new file mode 100644 index 000000000..90cad6840 --- /dev/null +++ b/fastvideo/tests/entrypoints/test_openai_serving_engine.py @@ -0,0 +1,269 @@ +# SPDX-License-Identifier: Apache-2.0 +"""CPU-light contracts for the shared OpenAI serving engine and adapter.""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +import threading +import time +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from fastvideo.entrypoints.openai.protocol import VideoGenerationRequest +from fastvideo.entrypoints.openai.request_adapter import ( + RequestAdaptationError, + build_generation_request, +) +from fastvideo.entrypoints.openai.serving_engine import OpenAIServingEngine +from fastvideo.entrypoints.openai.stores import VIDEO_STORE + + +class _BlockingGenerator: + + def __init__(self) -> None: + self.started = threading.Event() + self.release = threading.Event() + self.shutdown_called = False + + def generate(self, request): + self.started.set() + self.release.wait(timeout=5) + return request + + def shutdown(self) -> None: + self.shutdown_called = True + + +class _FileGenerator: + + def generate(self, request): + output = Path(request.output.output_path) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_bytes(b"\x00\x00\x00\x18ftypmp42fastvideo-test") + return SimpleNamespace( + video_path=str(output), + generation_time=0.01, + peak_memory_mb=12.5, + logging_info=None, + ) + + def shutdown(self) -> None: + return None + + +def _args(model_path: str, **overrides): + values = { + "model_path": model_path, + "lora_path": None, + "lora_nickname": "default", + "lora_strength": 1.0, + "override_pipeline_cls_name": None, + } + values.update(overrides) + return SimpleNamespace(**values) + + +def test_engine_cancellation_keeps_pipeline_locked() -> None: + asyncio.run(_assert_engine_cancellation_keeps_pipeline_locked()) + + +async def _assert_engine_cancellation_keeps_pipeline_locked() -> None: + generator = _BlockingGenerator() + engine = OpenAIServingEngine(generator) # type: ignore[arg-type] + + first = asyncio.create_task(engine.generate("first")) # type: ignore[arg-type] + await asyncio.to_thread(generator.started.wait, 2) + first.cancel() + + second_started = threading.Event() + + def second_call(): + second_started.set() + return "second" + + second = asyncio.create_task(engine.run_serialized(second_call)) + await asyncio.sleep(0.05) + assert not second_started.is_set() + + generator.release.set() + with pytest.raises(asyncio.CancelledError): + await first + assert await second == "second" + + await engine.shutdown() + assert generator.shutdown_called + + +def test_request_adapter_resolves_vllm_nested_params(tmp_path: Path) -> None: + request = VideoGenerationRequest( + prompt="a fox", + video_params={"width": 832, "height": 480, "fps": 30}, + seconds="2", + seed=7, + ) + adapted = build_generation_request( + "video_gen_test", + request, + _args("Wan-AI/Wan2.1-T2V-1.3B-Diffusers"), + served_model_name="wan", + output_dir=str(tmp_path), + ) + + assert adapted.sampling.width == 832 + assert adapted.sampling.height == 480 + assert adapted.sampling.fps == 30 + assert adapted.sampling.num_frames == 60 + assert adapted.sampling.seed == 7 + assert adapted.output.output_path.endswith("video_gen_test.mp4") + + +def test_request_adapter_accepts_matching_startup_lora(tmp_path: Path) -> None: + adapter = str(tmp_path / "adapter.safetensors") + args = _args( + "Wan-AI/Wan2.1-T2V-1.3B-Diffusers", + lora_path=adapter, + lora_nickname="fast", + lora_strength=0.75, + ) + request = VideoGenerationRequest( + prompt="a fox", + model="fast", + lora={"name": "fast", "path": adapter, "scale": 0.75}, + ) + + build_generation_request( + "video_gen_test", + request, + args, + served_model_name="wan", + output_dir=str(tmp_path), + ) + + +def test_request_adapter_rejects_runtime_lora_swap(tmp_path: Path) -> None: + args = _args( + "Wan-AI/Wan2.1-T2V-1.3B-Diffusers", + lora_path="/models/startup.safetensors", + lora_nickname="fast", + ) + request = VideoGenerationRequest(prompt="a fox", lora={"path": "/models/other.safetensors"}) + + with pytest.raises(RequestAdaptationError, match="does not match the startup adapter"): + build_generation_request( + "video_gen_test", + request, + args, + served_model_name="wan", + output_dir=str(tmp_path), + ) + + +def test_fasth3_request_uses_the_general_adapter(tmp_path: Path) -> None: + request = VideoGenerationRequest( + prompt="a fox", + task="t2va", + aspect_ratio="16:9", + num_frames=124, + num_inference_steps=5, + guidance_scale=1.0, + ) + adapted = build_generation_request( + "video_gen_test", + request, + _args("FastVideo/FastVideo-Minimax-FastH3-Preview-v0.2"), + served_model_name="fasth3", + output_dir=str(tmp_path), + ) + + assert adapted.sampling.width == 1344 + assert adapted.sampling.height == 768 + assert adapted.sampling.num_frames == 124 + + +def test_unsupported_vllm_postprocessing_fails_at_admission(tmp_path: Path) -> None: + request = VideoGenerationRequest(prompt="a fox", enable_frame_interpolation=True) + + with pytest.raises(RequestAdaptationError, match="frame_interpolation"): + build_generation_request( + "video_gen_test", + request, + _args("Wan-AI/Wan2.1-T2V-1.3B-Diffusers"), + served_model_name="wan", + output_dir=str(tmp_path), + ) + + +def test_video_routes_cover_async_sync_list_content_and_delete(tmp_path: Path) -> None: + from fastvideo.entrypoints.openai import state + from fastvideo.entrypoints.openai.video_api import router + + generator = _FileGenerator() + engine = OpenAIServingEngine(generator) # type: ignore[arg-type] + args = _args("Wan-AI/Wan2.1-T2V-1.3B-Diffusers") + state.set_state( + generator, # type: ignore[arg-type] + engine, + args, # type: ignore[arg-type] + str(tmp_path), + served_model_name="wan-test", + ) + asyncio.run(VIDEO_STORE.clear()) + app = FastAPI() + app.include_router(router) + + try: + with TestClient(app) as client: + created = client.post( + "/v1/videos", + json={ + "model": "wan-test", + "prompt": "a fox", + "size": "64x64", + "num_frames": 1, + }, + ) + assert created.status_code == 200 + video_id = created.json()["id"] + + for _ in range(50): + detail = client.get(f"/v1/videos/{video_id}") + if detail.json()["status"] == "completed": + break + time.sleep(0.01) + assert detail.status_code == 200 + assert detail.json()["status"] == "completed" + + listing = client.get("/v1/videos", params={"limit": 1}) + assert listing.status_code == 200 + assert listing.json()["data"][0]["id"] == video_id + + content = client.get(f"/v1/videos/{video_id}/content") + assert content.status_code == 200 + assert content.content[4:8] == b"ftyp" + + deleted = client.delete(f"/v1/videos/{video_id}") + assert deleted.status_code == 200 + assert deleted.json() == { + "id": video_id, + "deleted": True, + "object": "video.deleted", + } + + sync = client.post( + "/v1/videos/sync", + json={ + "prompt": "a fox", + "size": "64x64", + "num_frames": 1, + }, + ) + assert sync.status_code == 200 + assert sync.headers["x-model"] == "wan-test" + assert sync.content[4:8] == b"ftyp" + finally: + asyncio.run(VIDEO_STORE.clear()) + state.clear_state() diff --git a/scripts/smoke_test_openai_video_api.py b/scripts/smoke_test_openai_video_api.py new file mode 100644 index 000000000..cea501292 --- /dev/null +++ b/scripts/smoke_test_openai_video_api.py @@ -0,0 +1,141 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Smoke-test a running OpenAI-compatible FastVideo video server.""" + +from __future__ import annotations + +import argparse +import json +import time +from pathlib import Path +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + + +def _json_request(url: str, *, method: str = "GET", payload: dict[str, Any] | None = None) -> dict[str, Any]: + body = json.dumps(payload).encode() if payload is not None else None + request = Request( + url, + data=body, + method=method, + headers={"content-type": "application/json"} if body is not None else {}, + ) + try: + with urlopen(request, timeout=30) as response: + return json.load(response) + except HTTPError as error: + detail = error.read().decode(errors="replace") + raise RuntimeError(f"{method} {url} returned HTTP {error.code}: {detail}") from error + + +def _wait_for_health(base_url: str, timeout: float) -> None: + deadline = time.monotonic() + timeout + last_error: Exception | None = None + while time.monotonic() < deadline: + try: + response = _json_request(f"{base_url}/health") + if response.get("status") == "ok": + return + except (RuntimeError, URLError) as error: + last_error = error + time.sleep(2) + raise TimeoutError(f"Server did not become healthy within {timeout:g}s: {last_error}") + + +def _download_video(url: str, output: Path) -> int: + try: + with urlopen(url, timeout=120) as response: + content_type = response.headers.get_content_type() + data = response.read() + except HTTPError as error: + detail = error.read().decode(errors="replace") + raise RuntimeError(f"GET {url} returned HTTP {error.code}: {detail}") from error + if content_type != "video/mp4": + raise RuntimeError(f"Expected video/mp4 content, got {content_type!r}") + if len(data) <= 8 or data[4:8] != b"ftyp": + raise RuntimeError("Downloaded content is not an MP4 file") + output.parent.mkdir(parents=True, exist_ok=True) + output.write_bytes(data) + return len(data) + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base-url", default="http://127.0.0.1:8000") + parser.add_argument("--model", required=True) + parser.add_argument("--prompt", default="A fox runs through fresh snow under soft morning light.") + parser.add_argument("--size", default="1344x768") + parser.add_argument("--num-frames", type=int, default=124) + parser.add_argument("--fps", type=int, default=24) + parser.add_argument("--num-inference-steps", type=int, default=5) + parser.add_argument("--guidance-scale", type=float, default=1.0) + parser.add_argument("--seed", type=int, default=1000) + parser.add_argument("--lora-name") + parser.add_argument("--lora-path") + parser.add_argument("--lora-scale", type=float, default=1.0) + parser.add_argument("--startup-timeout", type=float, default=1200) + parser.add_argument("--generation-timeout", type=float, default=1200) + parser.add_argument("--poll-interval", type=float, default=2) + parser.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +def main() -> None: + args = _parse_args() + base_url = args.base_url.rstrip("/") + _wait_for_health(base_url, args.startup_timeout) + + models = _json_request(f"{base_url}/v1/models") + model_ids = [model.get("id") for model in models.get("data", [])] + if args.model not in model_ids: + raise RuntimeError(f"Requested model {args.model!r} is absent from /v1/models: {model_ids}") + + payload: dict[str, Any] = { + "model": args.model, + "prompt": args.prompt, + "size": args.size, + "num_frames": args.num_frames, + "fps": args.fps, + "num_inference_steps": args.num_inference_steps, + "guidance_scale": args.guidance_scale, + "seed": args.seed, + } + if args.lora_name or args.lora_path: + payload["lora"] = { + "name": args.lora_name, + "path": args.lora_path, + "scale": args.lora_scale, + } + + job = _json_request(f"{base_url}/v1/videos", method="POST", payload=payload) + video_id = job.get("id") + if not video_id: + raise RuntimeError(f"Video submission returned no id: {job}") + + deadline = time.monotonic() + args.generation_timeout + while job.get("status") not in {"completed", "failed"} and time.monotonic() < deadline: + time.sleep(args.poll_interval) + job = _json_request(f"{base_url}/v1/videos/{video_id}") + if job.get("status") != "completed": + raise RuntimeError(f"Video job did not complete successfully: {job}") + + listing = _json_request(f"{base_url}/v1/videos?limit=100") + if video_id not in {item.get("id") for item in listing.get("data", [])}: + raise RuntimeError(f"Completed job {video_id!r} is absent from /v1/videos") + num_bytes = _download_video(f"{base_url}/v1/videos/{video_id}/content", args.output) + print( + json.dumps( + { + "id": video_id, + "status": job["status"], + "model": job.get("model"), + "inference_time_s": job.get("inference_time_s"), + "output": str(args.output), + "bytes": num_bytes, + }, + sort_keys=True, + )) + + +if __name__ == "__main__": + main()