diff --git a/cli/api-video.py b/cli/api-video.py new file mode 100644 index 000000000..adc354a0d --- /dev/null +++ b/cli/api-video.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python +# python cli/api-video.py --prompt "a paper boat drifting down a rain gutter" --frames 17 --steps 8 --output /tmp/video.mp4 +import os +import time +import base64 +import logging +import argparse +import threading +import requests +import urllib3 + +sd_url = os.environ.get('SDAPI_URL', "http://127.0.0.1:7860") +sd_username = os.environ.get('SDAPI_USR', None) +sd_password = os.environ.get('SDAPI_PWD', None) +options = { + "send_video": True, + "send_thumbnail": False, +} + +logging.basicConfig(level = logging.INFO, format = '%(asctime)s %(levelname)s: %(message)s') +log = logging.getLogger(__name__) +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + +def auth(): + if sd_username is not None and sd_password is not None: + return requests.auth.HTTPBasicAuth(sd_username, sd_password) + return None + + +def get(endpoint: str, params: dict | None = None, timeout: int = 60): + req = requests.get(f'{sd_url}{endpoint}', params=params, timeout=timeout, verify=False, auth=auth()) + if req.status_code != 200: + return { 'error': req.status_code, 'reason': req.reason, 'url': req.url } + return req.json() + + +def post(endpoint: str, dct: dict | None = None, timeout: int = 3600): + req = requests.post(f'{sd_url}{endpoint}', json=dct, timeout=timeout, verify=False, auth=auth()) + if req.status_code != 200: + res = { 'error': req.status_code, 'reason': req.reason, 'url': req.url } + try: + res['detail'] = req.json().get('detail', None) + except Exception: + pass + return res + return req.json() + + +def encode(f: str): + with open(f, 'rb') as file: + return base64.b64encode(file.read()).decode() + + +def list_models(): + data = get('/sdapi/v1/video/models') + if isinstance(data, dict) and 'error' in data: + log.error(f'video models: {data}') + return + for item in data: + loaded = ' loaded=true' if item.get('loaded') else '' + log.info(f'engine="{item["engine"]}" model="{item["name"]}" mode={item["mode"]}{loaded}') + log.info(f'video models: {len(data)}') + + +def watch_progress(stop_event: threading.Event): + while not stop_event.is_set(): + status = get('/sdapi/v1/progress', params={ 'skip_current_image': True }) + if 'error' not in status: + state = status.get('state') or {} + log.info(f'progress={status.get("progress", 0):.2f} eta={status.get("eta_relative", 0):.1f} step={state.get("sampling_step", 0)}/{state.get("sampling_steps", 0)} info="{status.get("textinfo") or ""}"') + stop_event.wait(5) + + +def save_output(data: dict, output: str): + if data.get('video'): + with open(output, 'wb') as f: + f.write(base64.b64decode(data['video'])) + log.info(f'video saved: filename={output}') + elif data.get('still') and data.get('frames'): + with open(output, 'wb') as f: + f.write(base64.b64decode(data['frames'][0])) + log.info(f'still saved: filename={output}') + elif data.get('video_path'): + req = requests.get(f'{sd_url}/sdapi/v1/video/file', params={ 'file': data['video_path'] }, timeout=300, verify=False, auth=auth()) + if req.status_code == 200: + with open(output, 'wb') as f: + f.write(req.content) + log.info(f'video fetched: filename={output} size={len(req.content)}') + else: + log.error(f'video fetch failed: code={req.status_code} reason={req.reason}') + else: + log.warning('no video output received') + + +def generate(args): # pylint: disable=redefined-outer-name + t0 = time.time() + if args.engine: + options['engine'] = args.engine + if args.model: + options['model'] = args.model + options['prompt'] = args.prompt + options['negative_prompt'] = args.negative + options['width'] = int(args.width) + options['height'] = int(args.height) + options['frames'] = int(args.frames) + options['steps'] = int(args.steps) + options['seed'] = int(args.seed) + options['sampler_name'] = args.sampler + options['mp4_fps'] = int(args.fps) + options['mp4_interpolate'] = int(args.interpolate) + options['audio'] = bool(args.audio) + if args.init: + options['init_image'] = encode(args.init) + if args.last: + options['last_image'] = encode(args.last) + stop_event = threading.Event() + if args.progress: + threading.Thread(target=watch_progress, args=(stop_event,), daemon=True).start() + data = post('/sdapi/v1/video', options, timeout=int(args.timeout)) + stop_event.set() + t1 = time.time() + if 'error' in data: + log.error(f'generate failed: {data}') + return + log.info(f'video received: frames={data.get("frames_count")} fps={data.get("fps")} duration={data.get("duration")} audio={data.get("has_audio")} still={data.get("still")} path={data.get("video_path")} time={t1-t0:.2f}') + if args.output: + save_output(data, args.output) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description = 'api-video') + parser.add_argument('--list', action='store_true', help='list video engines and models') + parser.add_argument('--engine', required=False, default=None, help='video engine; omit with --model to use the loaded checkpoint') + parser.add_argument('--model', required=False, default=None, help='video model name within the engine') + parser.add_argument('--prompt', required=False, default='', help='prompt text') + parser.add_argument('--negative', required=False, default='', help='negative prompt text') + parser.add_argument('--width', required=False, default=832, help='video width') + parser.add_argument('--height', required=False, default=480, help='video height') + parser.add_argument('--frames', required=False, default=17, help='number of frames; 1 for a still image') + parser.add_argument('--steps', required=False, default=20, help='number of steps') + parser.add_argument('--seed', required=False, default=-1, help='initial seed') + parser.add_argument('--sampler', required=False, default='Default', help='sampler name') + parser.add_argument('--fps', required=False, default=24, help='frames per second') + parser.add_argument('--interpolate', required=False, default=0, help='rife interpolation passes') + parser.add_argument('--audio', action=argparse.BooleanOptionalAction, default=True, help='generate audio on supported models') + parser.add_argument('--init', required=False, default=None, help='init image file') + parser.add_argument('--last', required=False, default=None, help='last frame image file') + parser.add_argument('--output', required=False, default=None, help='output video file') + parser.add_argument('--progress', action='store_true', help='poll and log progress during generation') + parser.add_argument('--timeout', required=False, default=3600, help='request timeout in seconds') + args = parser.parse_args() + log.info(f'api-video: {args}') + if args.list: + list_models() + else: + generate(args) diff --git a/modules/api/api.py b/modules/api/api.py index 97a5404f7..70b98019e 100644 --- a/modules/api/api.py +++ b/modules/api/api.py @@ -6,7 +6,7 @@ from fastapi.security import HTTPBasic, HTTPBasicCredentials from fastapi.exceptions import HTTPException from modules import errors, shared, paths from modules.logger import log -from modules.api import models, endpoints, script, helpers, server, generate, process, control, docs, gpu +from modules.api import models, endpoints, script, helpers, server, generate, process, control, video, docs, gpu errors.install() @@ -33,6 +33,7 @@ class Api: self.generate = generate.APIGenerate(queue_lock) self.process = process.APIProcess(queue_lock) self.control = control.APIControl(queue_lock) + self.video = video.APIVideo(queue_lock) # compatibility api self.text2imgapi = self.generate.post_text2img self.img2imgapi = self.generate.post_img2img @@ -68,6 +69,7 @@ class Api: self.add_api_route("/sdapi/v1/txt2img", self.generate.post_text2img, methods=["POST"], response_model=models.ResTxt2Img, tags=["Generation"]) self.add_api_route("/sdapi/v1/img2img", self.generate.post_img2img, methods=["POST"], response_model=models.ResImg2Img, tags=["Generation"]) self.add_api_route("/sdapi/v1/control", self.control.post_control, methods=["POST"], response_model=control.ResControl, tags=["Generation"]) + self.add_api_route("/sdapi/v1/video", self.video.post_video, methods=["POST"], response_model=video.ResVideo, tags=["Generation"]) self.add_api_route("/sdapi/v1/process", self.process.extras_single_image_api, methods=["POST"], response_model=models.ResProcessImage, tags=["Processing"]) self.add_api_route("/sdapi/v1/extra-single-image", self.process.extras_single_image_api, methods=["POST"], response_model=models.ResProcessImage, tags=["Processing"]) self.add_api_route("/sdapi/v1/process-batch", self.process.extras_batch_images_api, methods=["POST"], response_model=models.ResProcessBatch, tags=["Processing"]) @@ -104,9 +106,11 @@ class Api: self.add_api_route("/sdapi/v1/extra-network-detail", endpoints.get_extra_network_detail, methods=["GET"], response_model=models.ItemExtraNetworkFull, tags=["Enumerators"]) self.add_api_route("/sdapi/v1/extra-network-details", endpoints.get_extra_network_details, methods=["GET"], response_model=models.ResExtraNetworkDetails, tags=["Enumerators"]) self.add_api_route("/sdapi/v1/unets", endpoints.get_unets, methods=["GET"], response_model=list[models.ItemUNet], tags=["Enumerators"]) + self.add_api_route("/sdapi/v1/video/models", self.video.get_video_models, methods=["GET"], response_model=list[video.ItemVideoModel], tags=["Enumerators"]) # functional api self.add_api_route("/sdapi/v1/file", endpoints.get_file, methods=["GET"], tags=["Functional"]) + self.add_api_route("/sdapi/v1/video/file", self.video.get_video_file, methods=["GET"], tags=["Functional"]) self.add_api_route("/sdapi/v1/delete-image", endpoints.get_deleteimage, methods=["DELETE"], tags=["Functional"]) self.add_api_route("/sdapi/v1/delete-file", endpoints.get_deletefile, methods=["DELETE"], tags=["Functional"]) self.add_api_route("/sdapi/v1/png-info", endpoints.get_pnginfo, methods=["GET"], response_model=models.ResImageInfo, tags=["Functional"]) diff --git a/modules/api/helpers.py b/modules/api/helpers.py index 487499a80..f5087e5e8 100644 --- a/modules/api/helpers.py +++ b/modules/api/helpers.py @@ -1,4 +1,5 @@ import io +import os import base64 from PIL import Image, PngImagePlugin import piexif @@ -90,6 +91,24 @@ def encode_pil_to_base64(image): return b64 +MAX_B64_BYTES = 256 * 1024 * 1024 # base64 expands ~4/3 and the response is built in memory; larger artifacts are fetched by path instead + + +def encode_file_to_base64(fn: str, max_bytes: int = MAX_B64_BYTES) -> str | None: + try: + if fn is None or not os.path.isfile(fn): + return None + size = os.path.getsize(fn) + if size > max_bytes: + log.warning(f'API cannot encode file: fn="{fn}" size={size} max={max_bytes}') + return None + with open(fn, 'rb') as f: + return base64.b64encode(f.read()).decode('ascii') + except Exception as e: + log.warning(f'API cannot encode file: fn="{fn}" {e}') + return None + + def upscaler_to_index(name: str): try: return [x.name.lower() for x in shared.sd_upscalers].index(name.lower()) diff --git a/modules/api/mime.py b/modules/api/mime.py index 7b244f152..824af43be 100644 --- a/modules/api/mime.py +++ b/modules/api/mime.py @@ -10,3 +10,6 @@ def register(): mimetypes.add_type('image/webp', '.webp') mimetypes.add_type('image/jxl', '.jxl') mimetypes.add_type('font/ttf', '.ttf') + mimetypes.add_type('video/mp4', '.mp4') + mimetypes.add_type('video/webm', '.webm') + mimetypes.add_type('video/x-matroska', '.mkv') diff --git a/modules/api/validate.py b/modules/api/validate.py index 9738c0efb..7de27254a 100644 --- a/modules/api/validate.py +++ b/modules/api/validate.py @@ -12,6 +12,7 @@ request_cost = { "/sdapi/v1/txt2img": 5, "/sdapi/v1/img2img": 5, "/sdapi/v1/control": 5, + "/sdapi/v1/video": 5, } log_cost = { "/.well-known/appspecific/com.chrome.devtools.json": -1, diff --git a/modules/api/video.py b/modules/api/video.py new file mode 100644 index 000000000..414951900 --- /dev/null +++ b/modules/api/video.py @@ -0,0 +1,265 @@ +from types import SimpleNamespace +from threading import Lock +from pydantic import BaseModel, Field # pylint: disable=no-name-in-module +from fastapi.exceptions import HTTPException +from modules import errors, shared, scripts_manager, ui +from modules.api import script, helpers +from modules.paths import resolve_output_path +from modules.video_models import models_def, video_load, video_run + + +errors.install() + + +class ReqVideo(BaseModel): + engine: str | None = Field(default=None, title="Engine", description="Video engine family; omit together with model to use the currently loaded checkpoint") + model: str | None = Field(default=None, title="Model", description="Video model name within the engine; see GET /sdapi/v1/video/models") + prompt: str = Field(default="", title="Prompt", description="Text prompt") + negative_prompt: str = Field(default="", title="Negative prompt", description="Negative text prompt") + styles: list[str] = Field(default=[], title="Styles", description="Prompt style names to apply") + width: int = Field(default=832, ge=64, le=4096, title="Width", description="Output width; snapped to the model canvas multiple") + height: int = Field(default=480, ge=64, le=4096, title="Height", description="Output height; snapped to the model canvas multiple") + frames: int = Field(default=17, ge=1, le=1024, title="Frames", description="Number of frames; 1 produces a single still image on workflow models") + steps: int = Field(default=50, ge=1, le=200, title="Steps", description="Number of inference steps") + sampler_name: str = Field(default="Default", title="Sampler", description="Sampler name; Default keeps the model scheduler") + sampler_shift: float = Field(default=-1.0, title="Sampler shift", description="Scheduler flow shift; -1 keeps the model default") + dynamic_shift: bool = Field(default=False, title="Dynamic shift", description="Enable dynamic scheduler shifting") + seed: int = Field(default=-1, title="Seed", description="Generation seed; -1 for random") + guidance_scale: float = Field(default=-1.0, title="Guidance scale", description="CFG scale; -1 keeps the model default") + guidance_true: float = Field(default=-1.0, title="True guidance", description="True CFG scale; -1 keeps the model default") + init_image: str | None = Field(default=None, title="Init image", description="Base64, data URI, or upload reference for the first-frame image") + init_strength: float = Field(default=0.8, ge=0.0, le=1.0, title="Init strength", description="Denoising strength for the init image") + last_image: str | None = Field(default=None, title="Last image", description="Base64, data URI, or upload reference for the last-frame image") + vae_type: str = Field(default="Default", title="VAE type", description="Decode variant: Default, Tiny, Remote, or Upscale") + vae_tile_frames: int = Field(default=16, ge=1, le=64, title="VAE tile frames", description="Frames per VAE decode tile") + audio: bool = Field(default=True, title="Audio", description="Generate audio on models that support it") + mp4_fps: int = Field(default=24, ge=1, le=60, title="FPS", description="Frames per second of the saved video") + mp4_interpolate: int = Field(default=0, ge=0, le=10, title="Interpolation", description="RIFE interpolation passes between frames") + mp4_codec: str = Field(default="libx264", title="Codec", description="Video codec; none skips video encoding") + mp4_ext: str = Field(default="mp4", title="Container", description="Container extension; the muxer is inferred from it") + mp4_opt: str = Field(default="crf:16", title="Codec options", description="Encoder options as comma-separated key:value pairs") + mp4_video: bool = Field(default=True, title="Save video", description="Write the video container to disk") + mp4_frames: bool = Field(default=False, title="Save frames", description="Write individual frame images to disk") + mp4_sf: bool = Field(default=False, title="Save safetensors", description="Write raw frames as a safetensors file") + mp4_thumb: bool = Field(default=True, title="Save thumbnail", description="Write a thumbnail image next to the video") + override_settings: dict = Field(default={}, title="Override settings", description="Setting overrides applied for this generation only") + script_args: list = Field(default=[], title="Script args", description="Positional arguments for a selectable script") + alwayson_scripts: dict = Field(default={}, title="Always-on scripts", description="Per-script argument overrides, keyed by script name") + send_video: bool = Field(default=True, title="Send video", description="Return the video base64-encoded in the response") + send_frames: bool = Field(default=False, title="Send frames", description="Return every frame base64-encoded in the response") + send_thumbnail: bool = Field(default=True, title="Send thumbnail", description="Return the thumbnail base64-encoded in the response") + extra: dict | None = Field(default={}, exclude=True, title="Extra", description="Extra attributes set on the processing object") + + +class ResVideo(BaseModel): + video: str | None = Field(default=None, title="Video", description="Base64-encoded video file; empty when not requested, above the size cap, or in still mode") + video_path: str | None = Field(default=None, title="Video path", description="Server path of the saved video; fetch via GET /sdapi/v1/video/file") + thumbnail: str | None = Field(default=None, title="Thumbnail", description="Base64-encoded thumbnail image") + thumbnail_path: str | None = Field(default=None, title="Thumbnail path", description="Server path of the saved thumbnail") + frames: list[str] = Field(default=[], title="Frames", description="Base64-encoded frames; always populated in still mode") + frames_count: int = Field(default=0, title="Frame count", description="Number of frames written, after interpolation") + fps: float = Field(default=0.0, title="FPS", description="Effective frames per second of the saved video") + duration: float = Field(default=0.0, title="Duration", description="Video duration in seconds") + has_audio: bool = Field(default=False, title="Has audio", description="Whether the video carries an audio track") + still: bool = Field(default=False, title="Still", description="Single-frame result; the product is in frames and no video was written") + params: dict = Field(default={}, title="Parameters", description="Echo of the request parameters used for generation") + info: str = Field(default="", title="Info", description="Generation info string with seed, sampler, and pipeline details") + + +class ItemVideoModel(BaseModel): + engine: str = Field(title="Engine", description="Video engine family") + name: str = Field(title="Name", description="Model name; pass together with engine to select it") + repo: str = Field(default="", title="Repo", description="Model repository or path") + url: str = Field(default="", title="URL", description="Model information page") + mode: str = Field(title="Mode", description="Input mode: workflow, t2v, i2v, flf2v, vace, or animate") + workflow: str | None = Field(default=None, title="Workflow", description="Modular workflow name when the model dispatches on inputs") + base: bool = Field(default=False, title="Base", description="Also listed in the base checkpoint dropdown") + loaded: bool = Field(default=False, title="Loaded", description="Currently loaded through the video registry") + + +def model_mode(m: models_def.Model) -> str: + # mirrors the dispatch order in video_run.run: workflow models route on inputs, the rest on name markers + if m.workflow is not None: + return 'workflow' + if 'T2V' in m.name: + return 't2v' + if 'I2V' in m.name: + return 'i2v' + if 'FLF2V' in m.name: + return 'flf2v' + if 'VACE' in m.name: + return 'vace' + if 'Animate' in m.name: + return 'animate' + return 't2v' + + +class APIVideo: + def __init__(self, queue_lock: Lock): + self.queue_lock = queue_lock + self.default_script_arg_video = [] + + def prepare_scripts(self, p_stub, req: ReqVideo): + script_runner = scripts_manager.scripts_video + if not script_runner.scripts: + script_runner.initialize_scripts(is_img2img=False, is_control=False, is_video=True) + ui.create_ui(None) + if not self.default_script_arg_video: + self.default_script_arg_video = script.init_default_script_args(script_runner) + script_args = script.init_script_args(p_stub, req, self.default_script_arg_video, None, None, script_runner) + return script_runner, script_args + + def sanitize_b64(self, req: ReqVideo): + def sanitize_str(args: list): + for idx in range(0, len(args)): + if isinstance(args[idx], str) and len(args[idx]) >= 1000: + args[idx] = f"" + for name in ('init_image', 'last_image'): + val = getattr(req, name, None) + if isinstance(val, str) and len(val) >= 1000: + setattr(req, name, f"") + if req.script_args: + sanitize_str(req.script_args) + if req.alwayson_scripts: + for script_obj in req.alwayson_scripts.values(): + if script_obj and "args" in script_obj and script_obj["args"]: + sanitize_str(script_obj["args"]) + + def post_video(self, req: ReqVideo): + """Generate a video, or a single still frame, using a video model. + + Omit `engine` and `model` to drive the currently loaded checkpoint when it is + video-capable; this covers models loaded from local folders that have no registry + entry. Pass both names to select a registry model, which is loaded on demand; + `GET /sdapi/v1/video/models` enumerates the valid pairs. + + `frames` of 1 on a workflow model produces a single still image returned in `frames`. + Disk outputs are controlled by `mp4_video`, `mp4_frames`, `mp4_sf`, and `mp4_thumb`; + response payloads are controlled independently by `send_video`, `send_frames`, and + `send_thumbnail`. Artifacts above the base64 size cap return `video` empty with + `video_path` set; fetch those via `GET /sdapi/v1/video/file`. + + `init_image` and `last_image` accept base64 data, data URIs, or upload references. + Progress is reported on `GET /sdapi/v1/progress`; `POST /sdapi/v1/interrupt` cancels. + Switching checkpoints via `override_settings` is not supported here; use + `POST /sdapi/v1/checkpoint` before generating. + """ + try: + selected, needs_load = video_run.resolve_model(req.engine, req.model) + except video_run.VideoError as e: + raise HTTPException(status_code=e.code, detail=str(e)) from e + sampler_name = helpers.validate_sampler_name(req.sampler_name) + init_image = helpers.decode_base64_to_image(req.init_image) if req.init_image else None + last_image = helpers.decode_base64_to_image(req.last_image) if req.last_image else None + overrides = dict(req.override_settings or {}) + for key in ('sd_model_checkpoint', 'sd_model_refiner'): + if key in overrides: + raise HTTPException(status_code=400, detail=f"{key} override is not supported here: switch models via POST /sdapi/v1/checkpoint before generating") + p_stub = SimpleNamespace(per_script_args={}) + script_runner, script_args = self.prepare_scripts(p_stub, req) + extra = getattr(req, 'extra', {}) or {} + + with self.queue_lock: + jobid = shared.state.begin('API-VID', api=True) + try: + res = video_run.run( + selected, + prompt=req.prompt, + negative=req.negative_prompt, + styles=req.styles, + width=req.width, + height=req.height, + frames=req.frames, + steps=req.steps, + sampler_name=sampler_name, + sampler_shift=req.sampler_shift, + dynamic_shift=req.dynamic_shift, + seed=req.seed, + guidance_scale=req.guidance_scale, + guidance_true=req.guidance_true, + init_image=init_image, + init_strength=req.init_strength, + last_image=last_image, + vae_type=req.vae_type, + vae_tile_frames=req.vae_tile_frames, + audio=req.audio, + mp4_fps=req.mp4_fps, + mp4_interpolate=req.mp4_interpolate, + mp4_codec=req.mp4_codec, + mp4_ext=req.mp4_ext, + mp4_opt=req.mp4_opt, + mp4_video=req.mp4_video, + mp4_frames=req.mp4_frames, + mp4_sf=req.mp4_sf, + mp4_thumb=req.mp4_thumb, + override_settings=overrides, + engine=req.engine, + scripts=script_runner, + script_args=script_args, + per_script_args=p_stub.per_script_args, + extra_p=extra, + needs_load=needs_load, + ) + except video_run.VideoError as e: + raise HTTPException(status_code=e.code, detail=str(e)) from e + finally: + shared.state.end(jobid, api=False) + + send_frames = req.send_frames or res.still # a still request has no other product to return + b64_frames = list(map(helpers.encode_pil_to_base64, res.images)) if send_frames else [] + video_b64 = helpers.encode_file_to_base64(res.video_path) if req.send_video and res.video_path else None + thumb_b64 = helpers.encode_file_to_base64(res.thumb_path) if req.send_thumbnail and res.thumb_path else None + duration = round(res.num_frames / res.fps, 3) if res.fps > 0 else 0.0 + self.sanitize_b64(req) + params = {k: v for k, v in vars(req).items() if k != 'extra'} + return ResVideo( + video=video_b64, + video_path=res.video_path, + thumbnail=thumb_b64, + thumbnail_path=res.thumb_path, + frames=b64_frames, + frames_count=res.num_frames, + fps=res.fps, + duration=duration, + has_audio=res.has_audio, + still=res.still, + params=params, + info=res.processed.info, + ) + + def get_video_models(self, engine: str | None = None): + """List video engines and models; optionally filter by engine.""" + items = [] + for family, rows in models_def.models.items(): + if engine is not None and family.lower() != engine.lower(): + continue + for m in rows: + if m.name == 'None': + continue + items.append(ItemVideoModel( + engine=family, + name=m.name, + repo=m.repo or '', + url=m.url or '', + mode=model_mode(m), + workflow=m.workflow, + base=m.base, + loaded=(m.name == video_load.loaded_model), + )) + return items + + def get_video_file(self, file: str): + """Serve a video artifact produced by this endpoint; the path must resolve inside the video output directory.""" + import mimetypes + from pathlib import Path + from starlette.responses import FileResponse + if not file or not file.strip(): + raise HTTPException(status_code=400, detail="file path is required") + root = Path(resolve_output_path(shared.opts.outdir_samples, shared.opts.outdir_video)).resolve() + target = Path(file).resolve() + if root not in target.parents: + raise HTTPException(status_code=403, detail=f"file {file}: must be inside the video output directory") + if not target.is_file(): + raise HTTPException(status_code=404, detail=f"file not found: {file}") + media_type = mimetypes.guess_type(target.name)[0] or 'application/octet-stream' + return FileResponse(str(target), media_type=media_type, filename=target.name) diff --git a/modules/processing_diffusers.py b/modules/processing_diffusers.py index 1b84ba775..ba16e0e02 100644 --- a/modules/processing_diffusers.py +++ b/modules/processing_diffusers.py @@ -564,18 +564,8 @@ def update_pipeline(sd_model, p: processing.StableDiffusionProcessing): def validate_pipeline(p: processing.StableDiffusionProcessing): - from modules.video_models.models_def import models as video_models - models_cls = [] - for family in video_models: - for m in video_models[family]: - if m.repo_cls is not None: - if isinstance(m.repo_cls, str): - models_cls.append(m.repo_cls) - else: - models_cls.append(m.repo_cls.__name__) - if m.custom is not None: - models_cls.append(m.custom) - is_video_model = shared.sd_model.__class__.__name__ in models_cls + from modules.video_models import models_def + is_video_model = shared.sd_model.__class__.__name__ in models_def.pipeline_classes() override_video_pipelines = ['WanPipeline', 'WanImageToVideoPipeline', 'WanVACEPipeline', 'MiniMaxH3ModularPipeline'] is_video_pipeline = ('video' in p.__class__.__name__.lower()) or (shared.sd_model.__class__.__name__ in override_video_pipelines) if is_video_model and not is_video_pipeline: diff --git a/modules/video_models/models_def.py b/modules/video_models/models_def.py index 8edb5a255..db7b1d42e 100644 --- a/modules/video_models/models_def.py +++ b/modules/video_models/models_def.py @@ -688,3 +688,47 @@ try: except Exception as e: models = {} log.error(f'Networks: type="video" {e}') + + +def engines() -> list[str]: + """Engine families with at least one real model, sentinel rows excluded.""" + return [engine for engine, rows in models.items() if any(row.name != 'None' for row in rows)] + + +def model_names(engine: str) -> list[str]: + """Real model names for an engine, sentinel rows excluded.""" + return [row.name for row in models.get(engine, []) if row.name != 'None'] + + +def find(engine: str, name: str) -> Model | None: + """Case-insensitive exact-name lookup; the 'None' sentinel rows are not models.""" + for family, rows in models.items(): + if family.lower() != (engine or '').lower(): + continue + for row in rows: + if row.name != 'None' and row.name.lower() == (name or '').lower(): + return row + return None + + +def workflow_for_class(cls_name: str) -> str | None: + """Workflow of the first registry row whose pipeline class matches, if any.""" + for rows in models.values(): + for row in rows: + if row.workflow is not None and row.repo_cls is not None: + row_cls = row.repo_cls if isinstance(row.repo_cls, str) else row.repo_cls.__name__ + if row_cls == cls_name: + return row.workflow + return None + + +def pipeline_classes() -> set[str]: + """Class names of every registry pipeline; repo_cls is a string before load and a class after video_load resolves it in place.""" + classes = set() + for rows in models.values(): + for row in rows: + if row.repo_cls is not None: + classes.add(row.repo_cls if isinstance(row.repo_cls, str) else row.repo_cls.__name__) + if row.custom is not None: + classes.add(row.custom) + return classes diff --git a/modules/video_models/video_modular.py b/modules/video_models/video_modular.py index efd8e04cc..0732defe8 100644 --- a/modules/video_models/video_modular.py +++ b/modules/video_models/video_modular.py @@ -61,6 +61,7 @@ def load_modular_pipe(repo_cls, repo: str, workflow: str | None = None, revision **offline_args, ) loaded = [name for name, component in pipe.components.items() if component is not None] + pipe.sdnext_video_workflow = workflow # lets a pipe loaded outside the video registry report its own workflow if hasattr(pipe, 'min_duration') and hasattr(pipe, 'fps'): pipe.sdnext_supported_min_frames = int(pipe.min_duration * pipe.fps) # fresh pipes report the true floor; still mode gates per instance log.debug(f'Load modular: cls={pipe.__class__.__name__} workflow={workflow} components={loaded} time={time.time()-t0:.2f}') diff --git a/modules/video_models/video_run.py b/modules/video_models/video_run.py index edd0686ba..3c1e10d35 100644 --- a/modules/video_models/video_run.py +++ b/modules/video_models/video_run.py @@ -1,53 +1,129 @@ import os import copy import time +from dataclasses import dataclass from modules import shared, errors, sd_models, processing, devices, images, ui_common, scripts_manager from modules.logger import log -from modules.video_models import models_def, video_utils, video_load, video_vae, video_overrides, video_save +from modules.video_models import models_def, video_utils, video_load, video_vae, video_overrides, video_save, video_modular from modules.paths import resolve_output_path debug = log.trace if os.environ.get('SD_VIDEO_DEBUG', None) is not None else lambda *args, **kwargs: None -def generate(task_id, ui_state, - engine, model, - prompt, negative, styles, - width, height, frames, steps, - sampler_index, sampler_shift, dynamic_shift, - seed, guidance_scale, guidance_true, - init_image, init_strength, last_image, - vae_type, vae_tile_frames, audio, - mp4_fps, mp4_interpolate, mp4_codec, mp4_ext, mp4_opt, mp4_video, mp4_frames, mp4_sf, mp4_thumb, - override_settings, - *args, **kwargs - ): +class VideoError(Exception): + """Video generation failure; code follows HTTP semantics so API callers can map it directly.""" + def __init__(self, msg: str, code: int = 500): + super().__init__(msg) + self.code = code - if engine is None or model is None or engine == 'None' or model == 'None': - return video_utils.queue_err('model not selected') - # videojob = shared.state.begin('Video') - found = [model.name for model in models_def.models.get(engine, [])] - selected: models_def.Model = [m for m in models_def.models[engine] if m.name == model][0] if len(found) > 0 else None - if not shared.sd_loaded: - debug('Video: model not yet loaded') - video_load.load_model(selected) - if selected.name != video_load.loaded_model: - debug('Video: force reload') - video_load.load_model(selected) - if not shared.sd_loaded: - debug('Video: model still not loaded') - return video_utils.queue_err('model not loaded') - debug(f'Video generate: task={task_id} args={args} kwargs={kwargs}') + +@dataclass +class VideoResult: + images: list # PIL frames as produced; callers decide what to surface + video_path: str | None + thumb_path: str | None + num_frames: int + fps: float # effective save fps after interpolation + has_audio: bool + still: bool + processed: processing.Processed + + +def resolve_model(engine: str | None, model: str | None) -> tuple[models_def.Model, bool]: + """Return (selected, needs_load): a registry row when both names are given, or a synthesized + row describing the already-loaded pipeline when both are omitted.""" + engine_given = engine not in (None, '', 'None') + model_given = model not in (None, '', 'None') + if engine_given != model_given: + raise VideoError('video model selection requires both engine and model', 400) + if engine_given: + selected = models_def.find(engine, model) + if selected is None: + available = models_def.model_names(engine) or models_def.engines() + raise VideoError(f'video model not found: engine="{engine}" model="{model}" available={available}', 404) + return selected, True + cls = shared.sd_model.__class__.__name__ if shared.sd_loaded else None + if not shared.sd_loaded or cls not in models_def.pipeline_classes(): + raise VideoError(f'no video model loaded: cls={cls} select engine and model or load a video-capable checkpoint first', 400) + pipe = shared.sd_model + workflow = getattr(pipe, 'sdnext_video_workflow', None) + if workflow is None and video_modular.is_modular(pipe): + workflow = models_def.workflow_for_class(cls) or 'auto' # modular pipes dispatch on inputs, so any workflow marker selects the modular branch + ckpt = getattr(pipe, 'sd_checkpoint_info', None) + selected = models_def.Model( + name=getattr(ckpt, 'title', None) or cls, + repo=getattr(ckpt, 'name', None), + repo_cls=type(pipe), + workflow=workflow, + base=True, + ) + return selected, False + + +def run(selected: models_def.Model, *, + prompt: str, + negative: str = '', + styles: list | None = None, + width: int = 832, + height: int = 480, + frames: int = 17, + steps: int = 50, + sampler_name: str = 'Default', + sampler_shift: float = -1.0, + dynamic_shift: bool = False, + seed: int = -1, + guidance_scale: float = -1.0, + guidance_true: float = -1.0, + init_image=None, + init_strength: float = 0.8, + last_image=None, + vae_type: str = 'Default', + vae_tile_frames: int = 16, + audio: bool = True, + mp4_fps: int = 24, + mp4_interpolate: int = 0, + mp4_codec: str = 'libx264', + mp4_ext: str = 'mp4', + mp4_opt: str = 'crf:16', + mp4_video: bool = True, + mp4_frames: bool = False, + mp4_sf: bool = False, + mp4_thumb: bool = True, + override_settings=None, + engine: str | None = None, + ui_state=None, + scripts=None, + script_args=(), + per_script_args: dict | None = None, + extra_p: dict | None = None, + needs_load: bool = True, + ) -> VideoResult: + + if needs_load: + if not shared.sd_loaded: + debug('Video: model not yet loaded') + video_load.load_model(selected) + if selected.name != video_load.loaded_model: + debug('Video: force reload') + video_load.load_model(selected) + if not shared.sd_loaded: + debug('Video: model still not loaded') + raise VideoError('model not loaded', 500) + + if isinstance(override_settings, (list, tuple)): # the ui override control emits "setting: value" pairs; always empty on the video tab since the control stays hidden + from modules.generation_parameters_copypaste import create_override_settings_dict + override_settings = create_override_settings_dict(override_settings) p = processing.StableDiffusionProcessingVideo( sd_model=shared.sd_model, - video_engine=engine, - video_model=model, + video_engine=engine or 'Loaded', + video_model=selected.name, prompt=prompt, negative_prompt=negative, - styles=styles, + styles=styles or [], seed=int(seed), - sampler_name = processing.get_sampler_name(sampler_index), + sampler_name=sampler_name, sampler_shift=float(sampler_shift), steps=int(steps), width=16 * int(width // 16), @@ -67,9 +143,13 @@ def generate(task_id, ui_state, p.vae_type = 'Default' p.state = ui_state - p.scripts = scripts_manager.scripts_video - p.script_args = args - processed: processing.Processed = scripts_manager.scripts_video.run(p, *args) + if per_script_args: + p.per_script_args.update(per_script_args) + for k, v in (extra_p or {}).items(): + setattr(p, k, v) + p.scripts = scripts if scripts is not None else scripts_manager.scripts_video + p.script_args = tuple(script_args) + p.scripts.run(p, *script_args) p.do_not_save_grid = True p.do_not_save_samples = not mp4_frames @@ -85,44 +165,44 @@ def generate(task_id, ui_state, if p.video_still: p.do_not_save_samples = False # the still is the product; save it like an image result elif int(mp4_fps) != 24: - log.warning(f'Video: model="{model}" fps={mp4_fps} model output is fixed at 24') + log.warning(f'Video: model="{selected.name}" fps={mp4_fps} model output is fixed at 24') log.debug(f'Video: op=modular workflow={selected.workflow} still={p.video_still} init={init_image} last={last_image}') - elif 'T2V' in model: + elif 'T2V' in selected.name: if init_image is not None: log.warning('Video: op=T2V init image not supported') - elif 'I2V' in model: + elif 'I2V' in selected.name: if init_image is None: - return video_utils.queue_err('No input image provided. Please upload or select an image.') + raise VideoError('No input image provided. Please upload or select an image.', 400) p.task_args['image'] = images.resize_image(resize_mode=2, im=init_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil') if last_image is not None and video_utils.supports_last_frame(shared.sd_model): p.task_args['last_image'] = images.resize_image(resize_mode=2, im=last_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil') log.debug(f'Video: op=FLF2V init={init_image} last={last_image} resized={p.task_args["image"]}') elif last_image is not None: - log.warning(f'Video: op=I2V model="{model}" last frame not supported, ignoring') + log.warning(f'Video: op=I2V model="{selected.name}" last frame not supported, ignoring') else: log.debug(f'Video: op=I2V init={init_image} resized={p.task_args["image"]}') - elif 'FLF2V' in model: + elif 'FLF2V' in selected.name: if init_image is None: - return video_utils.queue_err('No input image provided. Please upload or select an image.') + raise VideoError('No input image provided. Please upload or select an image.', 400) if last_image is None: - return video_utils.queue_err('No last frame image provided. Please upload or select an image.') + raise VideoError('No last frame image provided. Please upload or select an image.', 400) p.task_args['image'] = images.resize_image(resize_mode=2, im=init_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil') p.task_args['last_image'] = images.resize_image(resize_mode=2, im=last_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil') log.debug(f'Video: op=FLF2V init={init_image} last={last_image} resized={p.task_args["image"]}') - elif 'VACE' in model: + elif 'VACE' in selected.name: if init_image is not None: p.task_args['reference_images'] = [images.resize_image(resize_mode=2, im=init_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil')] log.debug(f'Video: op=VACE reference={init_image} resized={p.task_args["reference_images"]}') - elif 'Animate' in model: + elif 'Animate' in selected.name: if init_image is None: - return video_utils.queue_err('No input image provided. Please upload or select an image.') + raise VideoError('No input image provided. Please upload or select an image.', 400) p.task_args['image'] = images.resize_image(resize_mode=2, im=init_image, width=p.width, height=p.height, upscaler_name=None, output_type='pil') p.task_args['mode'] = 'animate' p.task_args['pose_video'] = [] # input pose video to condition the generation on. must be a list of PIL images. p.task_args['face_video'] = [] # input face video to condition the generation on. must be a list of PIL images. log.debug(f'Video: op=Animate init={p.task_args["image"]} pose={p.task_args["pose_video"]} face={p.task_args["face_video"]}') else: - log.warning(f'Video: unknown model type "{model}"') + log.warning(f'Video: unknown model type "{selected.name}"') # cleanup memory shared.sd_model = sd_models.apply_balanced_offload(shared.sd_model) @@ -176,24 +256,23 @@ def generate(task_id, ui_state, # done if err: - return video_utils.queue_err(err) + raise VideoError(err, 500) if processed is None or (len(processed.images) == 0 and processed.bytes is None): - return video_utils.queue_err('processing failed') + raise VideoError('processing failed', 500) log.info(f'Video: name="{selected.name}" cls={shared.sd_model.__class__.__name__} frames={len(processed.images)} time={t1-t0:.2f}') if getattr(p, 'video_still', False): - processed.images = processed.images[:1] # already trimmed in process_decode; defensive - generation_info_js = processed.js() if processed is not None else '' - return processed.images, None, generation_info_js, processed.info, ui_common.plaintext_to_html(processed.comments) + stills = processed.images[:1] # already trimmed in process_decode; defensive + return VideoResult(images=stills, video_path=None, thumb_path=None, num_frames=len(stills), fps=0.0, has_audio=False, still=True, processed=processed) if hasattr(processed, 'images') and processed.images is not None: pixels = video_save.images_to_tensor(processed.images) else: pixels = None if hasattr(processed, 'audio') and processed.audio is not None: - audio = processed.audio[0].float().cpu() + waveform = processed.audio[0].float().cpu() else: - audio = None + waveform = None if mp4_interpolate > 0 and pixels is not None: p.video_interpolate = mp4_interpolate @@ -206,10 +285,10 @@ def generate(task_id, ui_state, pixels = x.permute(1, 0, 2, 3).unsqueeze(0) from modules.processing_video import interpolation_factor save_fps = mp4_fps * interpolation_factor(p) - _num_frames, video_file, _thumb = video_save.save_video( + num_frames, video_file, thumb_file = video_save.save_video( p=p, pixels=pixels, - audio=audio, + audio=waveform, aac_sample_rate=getattr(p, 'audio_sampling_rate', None) or 24000, binary=processed.bytes, mp4_fps=save_fps, @@ -223,9 +302,68 @@ def generate(task_id, ui_state, mp4_interpolate=mp4_interpolate, metadata={}, ) - if not mp4_frames: - processed.images = [] + return VideoResult(images=processed.images, video_path=video_file, thumb_path=thumb_file, num_frames=num_frames, fps=float(save_fps), has_audio=waveform is not None, still=False, processed=processed) - generation_info_js = processed.js() if processed is not None else '' - # shared.state.end(videojob) - return processed.images, video_file, generation_info_js, processed.info, ui_common.plaintext_to_html(processed.comments) + +def generate(task_id, ui_state, + engine, model, + prompt, negative, styles, + width, height, frames, steps, + sampler_index, sampler_shift, dynamic_shift, + seed, guidance_scale, guidance_true, + init_image, init_strength, last_image, + vae_type, vae_tile_frames, audio, + mp4_fps, mp4_interpolate, mp4_codec, mp4_ext, mp4_opt, mp4_video, mp4_frames, mp4_sf, mp4_thumb, + override_settings, + *args, **kwargs + ): + # gradio adapter around run(): the positional signature is frozen since external callers bind to it + if engine is None or model is None or engine == 'None' or model == 'None': + return video_utils.queue_err('model not selected') + selected = models_def.find(engine, model) + if selected is None: + return video_utils.queue_err(f'model not found: engine="{engine}" model="{model}"') + debug(f'Video generate: task={task_id} args={args} kwargs={kwargs}') + try: + res = run(selected, + prompt=prompt, + negative=negative, + styles=styles, + width=width, + height=height, + frames=frames, + steps=steps, + sampler_name=processing.get_sampler_name(sampler_index), + sampler_shift=sampler_shift, + dynamic_shift=dynamic_shift, + seed=seed, + guidance_scale=guidance_scale, + guidance_true=guidance_true, + init_image=init_image, + init_strength=init_strength, + last_image=last_image, + vae_type=vae_type, + vae_tile_frames=vae_tile_frames, + audio=audio, + mp4_fps=mp4_fps, + mp4_interpolate=mp4_interpolate, + mp4_codec=mp4_codec, + mp4_ext=mp4_ext, + mp4_opt=mp4_opt, + mp4_video=mp4_video, + mp4_frames=mp4_frames, + mp4_sf=mp4_sf, + mp4_thumb=mp4_thumb, + override_settings=override_settings, + engine=engine, + ui_state=ui_state, + script_args=args, + ) + except VideoError as e: + return video_utils.queue_err(str(e)) + generation_info_js = res.processed.js() + html_log = ui_common.plaintext_to_html(res.processed.comments) + if res.still: + return res.images, None, generation_info_js, res.processed.info, html_log + result_images = res.images if mp4_frames else [] + return result_images, res.video_path, generation_info_js, res.processed.info, html_log diff --git a/modules/video_models/video_ui.py b/modules/video_models/video_ui.py index be1a1c6fe..e3bdf2a78 100644 --- a/modules/video_models/video_ui.py +++ b/modules/video_models/video_ui.py @@ -58,34 +58,6 @@ def model_load(engine, model): return msg -def run_video(*args): - engine, model = args[2], args[3] - debug(f'Video run: engine="{engine}" model="{model}"') - selected = get_selected(engine, model) - if not selected or engine is None or model is None or engine == 'None' or model == 'None': - return video_utils.queue_err('model not selected') - debug(f'Video run: {str(selected)}') - if selected and 'Hunyuan' in selected.name: - return video_run.generate(*args) - elif selected and 'LTX' in selected.name: - return video_run.generate(*args) - elif selected and 'Mochi' in selected.name: - return video_run.generate(*args) - elif selected and 'Cog' in selected.name: - return video_run.generate(*args) - elif selected and 'Allegro' in selected.name: - return video_run.generate(*args) - elif selected and 'WAN' in selected.name: - return video_run.generate(*args) - elif selected and 'Latte' in selected.name: - return video_run.generate(*args) - elif selected and 'anisora' in selected.name.lower(): - return video_run.generate(*args) - elif selected and 'Kandinsky' in selected.name: - return video_run.generate(*args) - return video_utils.queue_err(f'model not found: engine="{engine}" model="{model}"') - - def create_ui_outputs(): with gr.Row(): with gr.Column(variant='compact', elem_id="video_outputs", elem_classes=['settings-column'], scale=1): diff --git a/test/full-test.sh b/test/full-test.sh index bf129f336..5813b16f3 100644 --- a/test/full-test.sh +++ b/test/full-test.sh @@ -30,3 +30,7 @@ echo control-preprocess python cli/api-preprocess.py --input ui/assets/logo-bg-0.jpg --model "Zoe Depth" echo control-controlnet python cli/api-control.py --prompt "cute robot" --input ui/assets/logo-bg-0.jpg --type controlnet --control "Zoe Depth:Xinsir Union XL:0.5" +echo video-models +python cli/api-video.py --list +echo video +python cli/api-video.py --prompt "a paper boat drifting down a rain gutter" --frames 17 --steps 8 --output /tmp/sdnext-test-video.mp4 diff --git a/test/test-video-api.py b/test/test-video-api.py new file mode 100644 index 000000000..f57cd6067 --- /dev/null +++ b/test/test-video-api.py @@ -0,0 +1,272 @@ +#!/usr/bin/env python +""" +API tests for video generation. + +Tests: +- GET /sdapi/v1/video/models — engine/model enumeration and mode derivation +- POST /sdapi/v1/video — request validation errors (partial pair, unknown model/sampler, checkpoint override, unknown script) +- POST /sdapi/v1/video — still mode (frames=1) against the currently loaded model +- POST /sdapi/v1/video — video generation against the currently loaded model +- POST /sdapi/v1/video — wire switches and GET /sdapi/v1/video/file serving + +Requires a running SD.Next instance. Generation categories require a video-capable +model loaded (for example MiniMax-H3 via the base checkpoint dropdown) and are +skipped otherwise; enumeration and validation run against any instance. + +Usage: + python test/test-video-api.py [--url URL] [--steps STEPS] [--frames FRAMES] +""" + +import os +import sys +import base64 +import time +import argparse +import requests +import urllib3 + +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + +VALID_MODES = {'workflow', 't2v', 'i2v', 'flf2v', 'vace', 'animate'} + + +class VideoAPITest: + """Test harness for the video generation API.""" + + def __init__(self, base_url, steps=8, frames=17, timeout=3600): + self.base_url = base_url.rstrip('/') + self.steps = steps + self.frames = frames + self.timeout = timeout + self.video_capable = None # set by the still-mode probe + self.results = { + 'enumerate': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []}, + 'validation': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []}, + 'still': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []}, + 'generation': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []}, + 'wire': {'passed': 0, 'failed': 0, 'skipped': 0, 'tests': []}, + } + self._category = 'enumerate' + + def _get(self, endpoint, params=None): + try: + r = requests.get(f'{self.base_url}{endpoint}', params=params, timeout=self.timeout, verify=False) + if r.status_code != 200: + res = {'error': r.status_code, 'reason': r.reason} + try: + res['detail'] = r.json().get('detail', None) + except Exception: + pass + return res + return r.json() + except requests.exceptions.ConnectionError: + return {'error': 'connection_refused', 'reason': 'Server not running'} + except Exception as e: + return {'error': 'exception', 'reason': str(e)} + + def _post(self, endpoint, data): + try: + r = requests.post(f'{self.base_url}{endpoint}', json=data, timeout=self.timeout, verify=False) + if r.status_code != 200: + res = {'error': r.status_code, 'reason': r.reason} + try: + res['detail'] = r.json().get('detail', None) + except Exception: + pass + return res + return r.json() + except requests.exceptions.ConnectionError: + return {'error': 'connection_refused', 'reason': 'Server not running'} + except Exception as e: + return {'error': 'exception', 'reason': str(e)} + + def record(self, passed, name, detail=''): + status = 'PASS' if passed else 'FAIL' + self.results[self._category]['passed' if passed else 'failed'] += 1 + self.results[self._category]['tests'].append((status, name)) + msg = f' {status}: {name}' + if detail: + msg += f' ({detail})' + print(msg) + + def skip(self, name, reason): + self.results[self._category]['skipped'] += 1 + self.results[self._category]['tests'].append(('SKIP', name)) + print(f' SKIP: {name} ({reason})') + + def _video(self, extra_params=None, prompt='a red fox in the snow'): + payload = { + 'prompt': prompt, + 'steps': self.steps, + 'frames': self.frames, + 'width': 640, + 'height': 384, + 'seed': 42, + } + if extra_params: + payload.update(extra_params) + t0 = time.time() + data = self._post('/sdapi/v1/video', payload) + return data, time.time() - t0 + + # ========================================================================= + # Tests: Enumeration + # ========================================================================= + + def test_enumerate(self): + self._category = 'enumerate' + print("\n--- Enumeration Tests ---") + data = self._get('/sdapi/v1/video/models') + if isinstance(data, dict) and 'error' in data: + self.record(False, 'models_list', f'error: {data}') + return [] + self.record(len(data) > 0, 'models_list', f'{len(data)} models') + bad_modes = [item['name'] for item in data if item.get('mode') not in VALID_MODES] + self.record(len(bad_modes) == 0, 'models_modes', 'all valid' if not bad_modes else f'invalid: {bad_modes}') + minimax = [item for item in data if item['engine'] == 'MiniMax'] + if minimax: + self.record(all(item['base'] for item in minimax), 'models_minimax_base', f'{len(minimax)} rows') + self.record(all(item['mode'] == 'workflow' for item in minimax), 'models_minimax_workflow') + else: + self.skip('models_minimax', 'no MiniMax rows in registry') + filtered = self._get('/sdapi/v1/video/models', params={'engine': 'MiniMax'}) + if isinstance(filtered, list): + self.record(all(item['engine'] == 'MiniMax' for item in filtered), 'models_engine_filter', f'{len(filtered)} rows') + else: + self.record(False, 'models_engine_filter', f'error: {filtered}') + return data + + # ========================================================================= + # Tests: Validation + # ========================================================================= + + def test_validation(self, models): + self._category = 'validation' + print("\n--- Validation Tests ---") + data, _elapsed = self._video({'engine': 'MiniMax'}) + self.record(data.get('error') == 400, 'partial_pair_rejected', f'code={data.get("error")}') + data, _elapsed = self._video({'engine': 'NoSuchEngine', 'model': 'NoSuchModel'}) + self.record(data.get('error') == 404, 'unknown_model_rejected', f'code={data.get("error")} detail={data.get("detail")}') + # a valid registry pair fails on the sampler before any model load happens + if models: + pair = {'engine': models[0]['engine'], 'model': models[0]['name']} + data, _elapsed = self._video({**pair, 'sampler_name': 'NoSuchSampler'}) + self.record(data.get('error') == 404, 'unknown_sampler_rejected', f'code={data.get("error")}') + data, _elapsed = self._video({**pair, 'override_settings': {'sd_model_checkpoint': 'other-model'}}) + self.record(data.get('error') == 400, 'checkpoint_override_rejected', f'code={data.get("error")}') + data, _elapsed = self._video({**pair, 'alwayson_scripts': {'no-such-script': {'args': []}}}) + self.record(data.get('error') == 422, 'unknown_script_rejected', f'code={data.get("error")}') + else: + self.skip('unknown_sampler_rejected', 'no registry models to pair with') + + # ========================================================================= + # Tests: Still mode (doubles as the video-capability probe) + # ========================================================================= + + def test_still(self): + self._category = 'still' + print("\n--- Still Mode Tests ---") + data, elapsed = self._video({'frames': 1}) + if data.get('error') == 400: + self.video_capable = False + self.skip('still_generation', f'no video-capable model loaded: {data.get("detail")}') + return + if 'error' in data: + self.video_capable = False + self.record(False, 'still_generation', f'error: {data}') + return + self.video_capable = True + self.record(data.get('still') is True, 'still_flag', f'time={elapsed:.1f}s') + self.record(data.get('video') is None, 'still_no_video') + self.record(len(data.get('frames') or []) == 1, 'still_single_frame', f'frames={len(data.get("frames") or [])}') + + # ========================================================================= + # Tests: Generation with the loaded model + # ========================================================================= + + def test_generation(self): + self._category = 'generation' + print("\n--- Generation Tests ---") + if not self.video_capable: + self.skip('video_generation', 'no video-capable model loaded') + return None + data, elapsed = self._video() + if 'error' in data: + self.record(False, 'video_generation', f'error: {data}') + return None + self.record(data.get('frames_count', 0) > 0, 'video_frames_count', f'frames={data.get("frames_count")} time={elapsed:.1f}s') + video_b64 = data.get('video') + decoded = len(base64.b64decode(video_b64)) if video_b64 else 0 + self.record(decoded > 1000, 'video_payload', f'bytes={decoded}') + self.record(data.get('fps', 0) > 0 and data.get('duration', 0) > 0, 'video_timing', f'fps={data.get("fps")} duration={data.get("duration")}') + self.record(isinstance(data.get('has_audio'), bool), 'video_audio_flag', f'has_audio={data.get("has_audio")}') + self.record(bool(data.get('info')), 'video_info') + return data + + # ========================================================================= + # Tests: Wire switches and file serving + # ========================================================================= + + def test_wire(self): + self._category = 'wire' + print("\n--- Wire Tests ---") + if not self.video_capable: + self.skip('wire_all', 'no video-capable model loaded') + return + data, _elapsed = self._video({'send_video': False, 'send_thumbnail': False}) + if 'error' in data: + self.record(False, 'wire_send_video_off', f'error: {data}') + return + self.record(data.get('video') is None, 'wire_send_video_off') + path = data.get('video_path') + self.record(bool(path), 'wire_video_path', f'path={path}') + if path: + r = requests.get(f'{self.base_url}/sdapi/v1/video/file', params={'file': path}, timeout=300, verify=False) + ctype = r.headers.get('content-type', '') + self.record(r.status_code == 200 and ctype.startswith('video/'), 'wire_file_endpoint', f'code={r.status_code} type={ctype} bytes={len(r.content)}') + r = requests.get(f'{self.base_url}/sdapi/v1/video/file', params={'file': '/etc/passwd'}, timeout=60, verify=False) + self.record(r.status_code == 403, 'wire_file_jail', f'code={r.status_code}') + + # ========================================================================= + # Runner + # ========================================================================= + + def run_all(self): + print("=" * 60) + print("Video API Test Suite") + print(f"Server: {self.base_url}") + print(f"Steps: {self.steps} Frames: {self.frames}") + print("=" * 60) + + models = self.test_enumerate() + self.test_validation(models) + self.test_still() + self.test_generation() + self.test_wire() + + print("\n" + "=" * 60) + print("Results") + print("=" * 60) + total_passed = 0 + total_failed = 0 + total_skipped = 0 + for cat, data in self.results.items(): + total_passed += data['passed'] + total_failed += data['failed'] + total_skipped += data['skipped'] + status = 'PASS' if data['failed'] == 0 else 'FAIL' + print(f" {cat}: {data['passed']} passed, {data['failed']} failed, {data['skipped']} skipped [{status}]") + print(f" Total: {total_passed} passed, {total_failed} failed, {total_skipped} skipped") + print("=" * 60) + return total_failed == 0 + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description='Video API Tests (enumeration, validation, generation, file serving)') + parser.add_argument('--url', default=os.environ.get('SDAPI_URL', 'http://127.0.0.1:7860'), help='server URL') + parser.add_argument('--steps', type=int, default=8, help='generation steps (lower = faster tests)') + parser.add_argument('--frames', type=int, default=17, help='frame count for video tests') + args = parser.parse_args() + test = VideoAPITest(args.url, args.steps, args.frames) + success = test.run_all() + sys.exit(0 if success else 1)