feat(api): add video generation endpoint

Add POST /sdapi/v1/video plus GET /sdapi/v1/video/models and
GET /sdapi/v1/video/file. The generation body is extracted from the
gradio handler into a keyword-only core, video_run.run, which returns a
structured result and raises typed errors; the positional generate
signature is unchanged and now adapts to the core. Omitting engine and
model drives the currently loaded checkpoint when it is video-capable,
which covers models loaded from local folders without a registry entry.

- registry helpers in models_def (find, engines, pipeline_classes,
  workflow_for_class); validate_pipeline reuses the shared class set
- modular pipes stamp their workflow so out-of-registry loads dispatch
  onto the modular branch
- disk switches (mp4_*) and wire switches (send_*) are independent;
  artifacts above the base64 cap fall back to path plus the file route,
  which is jailed to the video output directory and serves video/mp4
  with range support
- always-on video scripts get bootstrapped default args, matching the
  txt2img handler; missing bootstrap raised a TypeError per frame
- checkpoint overrides are rejected with a pointer to the checkpoint
  endpoint; unknown engine, model and sampler names return 404 with the
  valid choices
- cli/api-video.py client, test/test-video-api.py suite and a
  full-test.sh entry; video mimetypes registered; rate-limit cost set
- remove the unreferenced video_ui.run_video dispatcher

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)

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"])

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())

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')

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,

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"<str {len(args[idx])}>"
+        for name in ('init_image', 'last_image'):
+            val = getattr(req, name, None)
+            if isinstance(val, str) and len(val) >= 1000:
+                setattr(req, name, f"<str {len(val)}>")
+        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)

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:

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

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}')

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

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):

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

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)