nodes/hao-ai-lab-FastVideo
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[feat] [2/n] Improve API: add initial support in video_generator (#1220)

Browse Source
This commit is contained in:
William Lin
2026-04-06 10:33:54 -07:00
committed by GitHub
parent 65dff806a8
commit c591d6d2a6
7 changed files with 1290 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/inference/basic/README.md
+5
View File
@@ -28,6 +28,11 @@ For an example running DMD+VSA inference:
python examples/inference/basic/basic_dmd.py
```
For the typed config/request path added during the inference API refactor:
```
python examples/inference/basic/basic_dmd_new_api.py
```
## Basic Walkthrough
All you need to generate videos using multi-gpus from state-of-the-art diffusion pipelines is the following few lines!
examples/inference/basic/basic_dmd_new_api.py
+98
View File
@@ -0,0 +1,98 @@
import os
import time
from fastvideo import VideoGenerator
from fastvideo.api import (
EngineConfig,
GenerationRequest,
GeneratorConfig,
OffloadConfig,
OutputConfig,
PipelineSelection,
)
OUTPUT_PATH = "video_samples_dmd2_typed"
def main():
os.environ["FASTVIDEO_ATTENTION_BACKEND"] = "VIDEO_SPARSE_ATTN"
model_name = "FastVideo/FastWan2.1-T2V-1.3B-Diffusers"
generator_config = GeneratorConfig(
model_path=model_name,
engine=EngineConfig(
num_gpus=1,
use_fsdp_inference=False,
offload=OffloadConfig(
text_encoder=True,
pin_cpu_memory=True,
dit=False,
vae=False,
),
),
# PR 2 still routes a few advanced inference knobs through the
# compatibility bridge until they get first-class typed fields.
pipeline=PipelineSelection(
experimental={
"VSA_sparsity": 0.8,
},
),
)
load_start_time = time.perf_counter()
generator = VideoGenerator.from_config(generator_config)
load_end_time = time.perf_counter()
load_time = load_end_time - load_start_time
prompt = (
"A neon-lit alley in futuristic Tokyo during a heavy rainstorm at night. "
"The puddles reflect glowing signs in kanji, advertising ramen, karaoke, "
"and VR arcades. A woman in a translucent raincoat walks briskly with an "
"LED umbrella. Steam rises from a street food cart, and a cat darts "
"across the screen. Raindrops are visible on the camera lens, creating "
"a cinematic bokeh effect."
)
request = GenerationRequest(
prompt=prompt,
output=OutputConfig(
output_path=OUTPUT_PATH,
save_video=True,
return_frames=False,
),
)
start_time = time.perf_counter()
result = generator.generate(request)
end_time = time.perf_counter()
gen_time = end_time - start_time
prompt2 = (
"A majestic lion strides across the golden savanna, its powerful frame "
"glistening under the warm afternoon sun. The tall grass ripples gently "
"in the breeze, enhancing the lion's commanding presence. The tone is "
"vibrant, embodying the raw energy of the wild. Low angle, steady "
"tracking shot, cinematic."
)
request2 = GenerationRequest(
prompt=prompt2,
output=OutputConfig(
output_path=OUTPUT_PATH,
save_video=True,
return_frames=False,
),
)
start_time = time.perf_counter()
result2 = generator.generate(request2)
end_time = time.perf_counter()
gen_time2 = end_time - start_time
print(f"Time taken to load model: {load_time} seconds")
print(f"Time taken to generate video: {gen_time} seconds")
print(f"First output written to: {result.video_path}")
print(f"Time taken to generate video2: {gen_time2} seconds")
print(f"Second output written to: {result2.video_path}")
if __name__ == "__main__":
main()
fastvideo/api/__init__.py
+2
View File
@@ -30,6 +30,7 @@ from fastvideo.api.parser import (
load_serve_config,
parse_config,
)
from fastvideo.api.results import GenerationResult
__all__ = [
"CompileConfig",
@@ -37,6 +38,7 @@ __all__ = [
"ContinuationState",
"ConfigValidationError",
"EngineConfig",
"GenerationResult",
"GenerationPlan",
"GenerationRequest",
"GeneratorConfig",
fastvideo/api/compat.py
+501
View File
@@ -0,0 +1,501 @@
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
from collections.abc import Mapping
from copy import deepcopy
from dataclasses import fields, is_dataclass
from pathlib import Path
from typing import Any
from fastvideo.api.overrides import apply_overrides, parse_cli_overrides
from fastvideo.api.parser import config_to_dict, load_raw_config, parse_config
from fastvideo.api.schema import (
GenerationRequest,
GeneratorConfig,
InputConfig,
OutputConfig,
RequestRuntimeConfig,
SamplingConfig,
)
from fastvideo.configs.sample import SamplingParam
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.utils import shallow_asdict
_EXPLICIT_REQUEST_ATTR = "_fastvideo_explicit_request"
_INPUT_FIELD_NAMES = {field.name for field in fields(InputConfig)}
_SAMPLING_FIELD_NAMES = {field.name for field in fields(SamplingConfig)}
_RUNTIME_FIELD_NAMES = {field.name for field in fields(RequestRuntimeConfig)}
_OUTPUT_FIELD_NAMES = {field.name for field in fields(OutputConfig)}
_MISSING = object()
_LEGACY_REQUEST_ALIASES = {
"neg_prompt": "negative_prompt",
}
_REQUEST_PIPELINE_OVERRIDE_FIELDS = frozenset({
"embedded_cfg_scale",
})
def normalize_generator_config(config: GeneratorConfig | Mapping[str, Any], ) -> GeneratorConfig:
if isinstance(config, GeneratorConfig):
return config
return parse_config(GeneratorConfig, config)
def load_generator_config_from_file(
path: str | Path,
overrides: list[str] | Mapping[str, Any] | None = None,
) -> GeneratorConfig:
raw = load_raw_config(path)
normalized_overrides = _normalize_overrides(overrides)
if _looks_like_run_or_serve_config(raw):
if normalized_overrides:
raw = apply_overrides(raw, normalized_overrides)
return parse_config(GeneratorConfig, raw["generator"])
if normalized_overrides:
adjusted = normalized_overrides
if all(key.startswith("generator.") for key in adjusted):
adjusted = {key[len("generator."):]: value for key, value in adjusted.items()}
raw = apply_overrides(raw, adjusted)
return parse_config(GeneratorConfig, raw)
def legacy_from_pretrained_to_config(
model_path: str,
kwargs: Mapping[str, Any],
) -> GeneratorConfig:
raw: dict[str, Any] = {"model_path": model_path}
engine: dict[str, Any] = {}
parallelism: dict[str, Any] = {}
offload: dict[str, Any] = {}
compile_config: dict[str, Any] = {}
pipeline: dict[str, Any] = {}
components: dict[str, Any] = {}
quantization: dict[str, Any] = {}
experimental: dict[str, Any] = {}
for key, value in kwargs.items():
if key == "revision":
raw["revision"] = value
elif key == "trust_remote_code":
raw["trust_remote_code"] = value
elif key == "num_gpus":
engine["num_gpus"] = value
elif key == "distributed_executor_backend":
engine["execution_backend"] = value
elif key in {"tp_size", "sp_size", "hsdp_replicate_dim", "hsdp_shard_dim", "dist_timeout"}:
parallelism[key] = value
elif key == "dit_cpu_offload":
offload["dit"] = value
elif key == "dit_layerwise_offload":
offload["dit_layerwise"] = value
elif key == "text_encoder_cpu_offload":
offload["text_encoder"] = value
elif key == "image_encoder_cpu_offload":
offload["image_encoder"] = value
elif key == "vae_cpu_offload":
offload["vae"] = value
elif key == "pin_cpu_memory":
offload["pin_cpu_memory"] = value
elif key == "enable_torch_compile":
compile_config["enabled"] = value
elif key == "torch_compile_kwargs":
compile_config["kwargs"] = deepcopy(value)
elif key in {"enable_stage_verification", "use_fsdp_inference", "disable_autocast"}:
engine[key] = value
elif key == "override_text_encoder_quant":
quantization["text_encoder_quant"] = value
elif key == "workload_type":
pipeline["workload_type"] = value
elif key == "lora_path":
components["lora_path"] = value
elif key == "override_pipeline_cls_name":
components["override_pipeline_cls_name"] = value
elif key == "override_transformer_cls_name":
components["override_transformer_cls_name"] = value
elif key == "pipeline_config":
if isinstance(value, str):
components["pipeline_config_path"] = value
else:
experimental[key] = deepcopy(value)
elif key == "override_text_encoder_safetensors":
components["text_encoder_weights"] = value
elif key == "init_weights_from_safetensors":
components["transformer_weights"] = value
elif key == "init_weights_from_safetensors_2":
components["transformer_2_weights"] = value
else:
experimental[key] = deepcopy(value)
if parallelism:
engine["parallelism"] = parallelism
if offload:
engine["offload"] = offload
if compile_config:
engine["compile"] = compile_config
if quantization:
engine["quantization"] = quantization
if engine:
raw["engine"] = engine
if components:
pipeline["components"] = components
if experimental:
pipeline["experimental"] = experimental
if pipeline:
raw["pipeline"] = pipeline
return parse_config(GeneratorConfig, raw)
def generator_config_to_fastvideo_args(config: GeneratorConfig | Mapping[str, Any], ) -> FastVideoArgs:
normalized = normalize_generator_config(config)
unsupported = []
if normalized.pipeline.profile is not None:
unsupported.append("pipeline.profile")
if normalized.pipeline.profile_version is not None:
unsupported.append("pipeline.profile_version")
if normalized.pipeline.components.config_root is not None:
unsupported.append("pipeline.components.config_root")
if normalized.pipeline.components.vae_weights is not None:
unsupported.append("pipeline.components.vae_weights")
if normalized.pipeline.components.upsampler_weights is not None:
unsupported.append("pipeline.components.upsampler_weights")
if unsupported:
joined = ", ".join(unsupported)
raise NotImplementedError(f"VideoGenerator compatibility adapter does not support {joined} yet")
engine = normalized.engine
kwargs: dict[str, Any] = {
"model_path": normalized.model_path,
"revision": normalized.revision,
"trust_remote_code": normalized.trust_remote_code,
"num_gpus": engine.num_gpus,
"distributed_executor_backend": engine.execution_backend,
"tp_size": engine.parallelism.tp_size,
"sp_size": engine.parallelism.sp_size,
"hsdp_replicate_dim": engine.parallelism.hsdp_replicate_dim,
"hsdp_shard_dim": engine.parallelism.hsdp_shard_dim,
"dist_timeout": engine.parallelism.dist_timeout,
"dit_cpu_offload": engine.offload.dit,
"dit_layerwise_offload": engine.offload.dit_layerwise,
"text_encoder_cpu_offload": engine.offload.text_encoder,
"image_encoder_cpu_offload": engine.offload.image_encoder,
"vae_cpu_offload": engine.offload.vae,
"pin_cpu_memory": engine.offload.pin_cpu_memory,
"enable_torch_compile": engine.compile.enabled,
"torch_compile_kwargs": deepcopy(engine.compile.kwargs),
"enable_stage_verification": engine.enable_stage_verification,
"use_fsdp_inference": engine.use_fsdp_inference,
"disable_autocast": engine.disable_autocast,
}
if normalized.pipeline.workload_type is not None:
kwargs["workload_type"] = normalized.pipeline.workload_type
quantization = engine.quantization
if quantization is not None and quantization.text_encoder_quant is not None:
kwargs["override_text_encoder_quant"] = quantization.text_encoder_quant
if quantization is not None and quantization.transformer_quant is not None:
kwargs["transformer_quant"] = quantization.transformer_quant
components = normalized.pipeline.components
if components.pipeline_config_path is not None:
kwargs["pipeline_config"] = components.pipeline_config_path
if components.lora_path is not None:
kwargs["lora_path"] = components.lora_path
if components.override_pipeline_cls_name is not None:
kwargs["override_pipeline_cls_name"] = components.override_pipeline_cls_name
if components.override_transformer_cls_name is not None:
kwargs["override_transformer_cls_name"] = components.override_transformer_cls_name
if components.text_encoder_weights is not None:
kwargs["override_text_encoder_safetensors"] = components.text_encoder_weights
if components.transformer_weights is not None:
kwargs["init_weights_from_safetensors"] = components.transformer_weights
if components.transformer_2_weights is not None:
kwargs["init_weights_from_safetensors_2"] = components.transformer_2_weights
kwargs.update(deepcopy(normalized.pipeline.profile_overrides))
kwargs.update(deepcopy(normalized.pipeline.experimental))
return FastVideoArgs.from_kwargs(**kwargs)
def normalize_generation_request(request: GenerationRequest | Mapping[str, Any], ) -> GenerationRequest:
normalized = (request if isinstance(request, GenerationRequest) else parse_config(GenerationRequest, request))
if not hasattr(normalized, _EXPLICIT_REQUEST_ATTR):
setattr(normalized, _EXPLICIT_REQUEST_ATTR, _serialize_generation_request(normalized))
return normalized
def legacy_generate_call_to_request(
prompt: str | None,
sampling_param: SamplingParam | None,
*,
mouse_cond: Any | None = None,
keyboard_cond: Any | None = None,
grid_sizes: Any | None = None,
legacy_kwargs: Mapping[str, Any] | None = None,
) -> GenerationRequest:
raw = _sampling_param_to_request_raw(sampling_param)
if prompt is not None:
raw["prompt"] = prompt
for key, value in (legacy_kwargs or {}).items():
_apply_request_field(raw, key, value)
if mouse_cond is not None:
raw.setdefault("inputs", {})["mouse_cond"] = mouse_cond
if keyboard_cond is not None:
raw.setdefault("inputs", {})["keyboard_cond"] = keyboard_cond
if grid_sizes is not None:
raw.setdefault("inputs", {})["grid_sizes"] = grid_sizes
normalized = parse_config(GenerationRequest, raw)
setattr(normalized, _EXPLICIT_REQUEST_ATTR, deepcopy(raw))
return normalized
def request_to_sampling_param(
request: GenerationRequest,
*,
model_path: str,
) -> SamplingParam:
if request.plan is not None:
raise NotImplementedError("GenerationRequest.plan is not wired into VideoGenerator yet")
if request.state is not None:
raise NotImplementedError("GenerationRequest.state is not wired into VideoGenerator yet")
sampling_param = SamplingParam.from_pretrained(model_path)
updates = _explicit_request_updates(request)
for key, value in updates.items():
if hasattr(sampling_param, key):
setattr(sampling_param, key, deepcopy(value))
elif key in _REQUEST_PIPELINE_OVERRIDE_FIELDS or _is_supported_as_default_only(key, value):
continue
else:
raise ValueError(f"Request field {key!r} is not supported by sampling params for {model_path}")
sampling_param.__post_init__()
sampling_param.check_sampling_param()
return sampling_param
def expand_request_prompt_batch(request: GenerationRequest, ) -> list[GenerationRequest]:
if not isinstance(request.prompt, list):
return [request]
requests: list[GenerationRequest] = []
for index, prompt in enumerate(request.prompt):
single_request = deepcopy(request)
single_request.prompt = prompt
_fan_out_batched_input_value(request, single_request, "image_path", index)
_fan_out_batched_input_value(request, single_request, "video_path", index)
_fan_out_explicit_request_metadata(request, single_request, index, prompt)
requests.append(single_request)
return requests
def _looks_like_run_or_serve_config(raw: Mapping[str, Any]) -> bool:
return isinstance(raw.get("generator"), Mapping)
def _normalize_overrides(overrides: list[str] | Mapping[str, Any] | None, ) -> dict[str, Any] | None:
if not overrides:
return None
if isinstance(overrides, list):
return parse_cli_overrides(overrides)
return dict(overrides)
def _sampling_param_to_request_raw(sampling_param: SamplingParam | None, ) -> dict[str, Any]:
if sampling_param is None:
return {}
raw: dict[str, Any] = {}
for key, value in shallow_asdict(sampling_param).items():
if key == "prompt":
continue
_apply_request_field(raw, key, deepcopy(value))
return raw
def _apply_request_field(
raw: dict[str, Any],
key: str,
value: Any,
) -> None:
key = _LEGACY_REQUEST_ALIASES.get(key, key)
if key == "negative_prompt":
raw["negative_prompt"] = value
return
if key in _INPUT_FIELD_NAMES:
raw.setdefault("inputs", {})[key] = value
return
if key in _SAMPLING_FIELD_NAMES:
raw.setdefault("sampling", {})[key] = value
return
if key in _RUNTIME_FIELD_NAMES:
raw.setdefault("runtime", {})[key] = value
return
if key in _OUTPUT_FIELD_NAMES:
raw.setdefault("output", {})[key] = value
return
raw.setdefault("extensions", {})[key] = value
def request_to_pipeline_overrides(request: GenerationRequest) -> dict[str, Any]:
overrides: dict[str, Any] = {}
for key, value in _explicit_request_updates(request).items():
if key in _REQUEST_PIPELINE_OVERRIDE_FIELDS:
overrides[key] = deepcopy(value)
return overrides
def _explicit_request_updates(request: GenerationRequest) -> dict[str, Any]:
raw = getattr(request, _EXPLICIT_REQUEST_ATTR, None)
if raw is None:
raw = _serialize_generation_request(request)
return _extract_request_updates(raw)
def _extract_request_updates(raw: Mapping[str, Any]) -> dict[str, Any]:
updates: dict[str, Any] = {}
if "negative_prompt" in raw:
updates["negative_prompt"] = deepcopy(raw["negative_prompt"])
for section_name in ("inputs", "sampling", "runtime", "output"):
section = raw.get(section_name)
if not isinstance(section, Mapping):
continue
for key, value in section.items():
updates[key] = deepcopy(value)
stage_overrides = raw.get("stage_overrides")
if stage_overrides:
updates.update(_flatten_stage_overrides(stage_overrides))
extensions = raw.get("extensions")
if isinstance(extensions, Mapping):
for key, value in extensions.items():
updates[key] = deepcopy(value)
return updates
def _flatten_stage_overrides(stage_overrides: Any) -> dict[str, Any]:
if not isinstance(stage_overrides, Mapping):
raise ValueError("GenerationRequest.stage_overrides must be a mapping")
flattened: dict[str, Any] = {}
for stage_name, overrides in stage_overrides.items():
if not isinstance(overrides, Mapping):
raise ValueError(f"GenerationRequest.stage_overrides.{stage_name} must be a mapping")
for key, value in overrides.items():
if key in flattened and flattened[key] != value:
raise ValueError(f"Conflicting stage override for {key!r} across stages")
flattened[key] = deepcopy(value)
return flattened
def _serialize_generation_request(request: GenerationRequest) -> dict[str, Any]:
return deepcopy(config_to_dict(request))
def _fan_out_batched_input_value(
source_request: GenerationRequest,
target_request: GenerationRequest,
field_name: str,
index: int,
) -> None:
value = getattr(source_request.inputs, field_name)
if not isinstance(value, list):
return
_validate_batched_input_length(source_request.prompt, value, field_name)
setattr(target_request.inputs, field_name, deepcopy(value[index]))
def _fan_out_explicit_request_metadata(
source_request: GenerationRequest,
target_request: GenerationRequest,
index: int,
prompt: str,
) -> None:
raw = getattr(source_request, _EXPLICIT_REQUEST_ATTR, None)
if raw is None:
return
raw = deepcopy(raw)
raw["prompt"] = prompt
inputs = raw.get("inputs")
if isinstance(inputs, dict):
for field_name in ("image_path", "video_path"):
value = inputs.get(field_name)
if isinstance(value, list):
_validate_batched_input_length(source_request.prompt, value, field_name)
inputs[field_name] = deepcopy(value[index])
setattr(target_request, _EXPLICIT_REQUEST_ATTR, raw)
def _validate_batched_input_length(
prompts: str | list[str] | None,
values: list[Any],
field_name: str,
) -> None:
if not isinstance(prompts, list):
return
if len(values) != len(prompts):
raise ValueError(f"GenerationRequest.inputs.{field_name} must have the same length as request.prompt")
def _is_supported_as_default_only(key: str, value: Any) -> bool:
default_value = _DEFAULT_REQUEST_UPDATES.get(key, _MISSING)
return default_value is not _MISSING and _values_equal(value, default_value)
def _collect_non_default_fields(
value: Any,
default: Any,
) -> dict[str, Any]:
if not (is_dataclass(value) and is_dataclass(default)):
return {}
result: dict[str, Any] = {}
for field in fields(value):
current = getattr(value, field.name)
default_value = getattr(default, field.name)
if is_dataclass(current) and is_dataclass(default_value):
nested = _collect_non_default_fields(current, default_value)
if nested:
result[field.name] = nested
continue
if not _values_equal(current, default_value):
result[field.name] = deepcopy(current)
return result
def _values_equal(left: Any, right: Any) -> bool:
if left is right:
return True
try:
return bool(left == right)
except Exception:
return False
_DEFAULT_REQUEST_UPDATES = _extract_request_updates(config_to_dict(GenerationRequest()))
__all__ = [
"generator_config_to_fastvideo_args",
"legacy_from_pretrained_to_config",
"legacy_generate_call_to_request",
"load_generator_config_from_file",
"normalize_generation_request",
"normalize_generator_config",
"request_to_pipeline_overrides",
"request_to_sampling_param",
]
fastvideo/api/results.py
+101
View File
@@ -0,0 +1,101 @@
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from collections.abc import Mapping
from fastvideo.api.schema import ContinuationState
@dataclass
class GenerationResult:
prompt: str | None = None
prompt_index: int | None = None
samples: Any | None = None
frames: Any | None = None
audio: Any | None = None
size: tuple[int, int, int] | None = None
generation_time: float | None = None
logging_info: Any | None = None
trajectory: Any | None = None
trajectory_timesteps: Any | None = None
trajectory_decoded: Any | None = None
video_path: str | None = None
peak_memory_mb: float | None = None
state: ContinuationState | None = None
extra: dict[str, Any] = field(default_factory=dict)
@classmethod
def from_legacy_result(
cls,
result: Mapping[str, Any],
) -> GenerationResult:
prompt = result.get("prompt")
if prompt is None:
prompt = result.get("prompts")
extra = {
key: value
for key, value in result.items() if key not in {
"prompt",
"prompt_index",
"prompts",
"samples",
"frames",
"audio",
"size",
"generation_time",
"logging_info",
"trajectory",
"trajectory_timesteps",
"trajectory_decoded",
"video_path",
"peak_memory_mb",
"state",
}
}
return cls(
prompt=prompt,
prompt_index=result.get("prompt_index"),
samples=result.get("samples"),
frames=result.get("frames"),
audio=result.get("audio"),
size=result.get("size"),
generation_time=result.get("generation_time"),
logging_info=result.get("logging_info"),
trajectory=result.get("trajectory"),
trajectory_timesteps=result.get("trajectory_timesteps"),
trajectory_decoded=result.get("trajectory_decoded"),
video_path=result.get("video_path"),
peak_memory_mb=result.get("peak_memory_mb"),
state=result.get("state"),
extra=extra,
)
def to_legacy_dict(self) -> dict[str, Any]:
result = {
"prompts": self.prompt,
"samples": self.samples,
"frames": self.frames,
"audio": self.audio,
"size": self.size,
"generation_time": self.generation_time,
"logging_info": self.logging_info,
"trajectory": self.trajectory,
"trajectory_timesteps": self.trajectory_timesteps,
"trajectory_decoded": self.trajectory_decoded,
"video_path": self.video_path,
"peak_memory_mb": self.peak_memory_mb,
}
if self.prompt_index is not None:
result["prompt_index"] = self.prompt_index
result["prompt"] = self.prompt
if self.state is not None:
result["state"] = self.state
result.update(self.extra)
return result
__all__ = ["GenerationResult"]
fastvideo/entrypoints/video_generator.py
+226 -16
View File
@@ -8,8 +8,12 @@ diffusion models.
import os
import re
import shutil
import threading
import time
import tempfile
import warnings
from collections.abc import Mapping
from copy import deepcopy
from typing import Any
@@ -18,9 +22,19 @@ import numpy as np
import torch
import torchvision
from einops import rearrange
import shutil
import tempfile
from fastvideo.api.compat import (
expand_request_prompt_batch,
generator_config_to_fastvideo_args,
legacy_from_pretrained_to_config,
load_generator_config_from_file,
normalize_generation_request,
normalize_generator_config,
request_to_pipeline_overrides,
request_to_sampling_param,
)
from fastvideo.api.results import GenerationResult
from fastvideo.api.schema import GenerationRequest, GeneratorConfig
from fastvideo.configs.sample import SamplingParam
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.logger import init_logger
@@ -30,6 +44,29 @@ from fastvideo.worker.executor import Executor
logger = init_logger(__name__)
_FROM_PRETRAINED_CONVENIENCE_KWARGS = frozenset({
"num_gpus",
"revision",
"trust_remote_code",
"distributed_executor_backend",
"tp_size",
"sp_size",
"hsdp_replicate_dim",
"hsdp_shard_dim",
"dist_timeout",
"use_fsdp_inference",
"disable_autocast",
"enable_stage_verification",
"dit_cpu_offload",
"dit_layerwise_offload",
"text_encoder_cpu_offload",
"image_encoder_cpu_offload",
"vae_cpu_offload",
"pin_cpu_memory",
"enable_torch_compile",
"torch_compile_kwargs",
})
def _infer_latent_batch_size(batch: ForwardBatch) -> int:
if isinstance(batch.prompt, list):
@@ -69,11 +106,16 @@ class VideoGenerator:
log_stats: Whether to log statistics
log_queue: Optional multiprocessing.Queue to forward worker logs to
"""
self.config: GeneratorConfig | None = None
self.fastvideo_args = fastvideo_args
self.executor = executor_class(fastvideo_args, log_queue=log_queue)
@classmethod
def from_pretrained(cls, model_path: str, **kwargs) -> "VideoGenerator":
def from_pretrained(
cls,
model_path: str | GeneratorConfig | Mapping[str, Any] | None = None,
**kwargs,
) -> "VideoGenerator":
"""
Create a video generator from a pretrained model.
@@ -86,13 +128,69 @@ class VideoGenerator:
The created video generator
Priority level: Default pipeline config < User's pipeline config < User's kwargs
"""
# If users also provide some kwargs, it will override the FastVideoArgs and PipelineConfig.
kwargs['model_path'] = model_path
log_queue = kwargs.pop("log_queue", None)
fastvideo_args = FastVideoArgs.from_kwargs(**kwargs)
return cls.from_fastvideo_args(fastvideo_args, log_queue=log_queue)
Stable convenience kwargs remain supported here for common engine and
offload settings. Advanced model- or pipeline-specific options should
move to VideoGenerator.from_config(...).
"""
log_queue = kwargs.pop("log_queue", None)
typed_config = kwargs.pop("config", None)
if typed_config is not None:
if model_path is not None:
raise TypeError("Pass either model_path or config to from_pretrained, not both")
if kwargs:
unexpected = ", ".join(sorted(kwargs))
raise TypeError(f"Unexpected keyword arguments with config: {unexpected}")
return cls.from_config(typed_config, log_queue=log_queue)
if isinstance(model_path, GeneratorConfig | Mapping):
if kwargs:
unexpected = ", ".join(sorted(kwargs))
raise TypeError(f"Unexpected keyword arguments with typed config: {unexpected}")
return cls.from_config(model_path, log_queue=log_queue)
if model_path is None:
raise TypeError("model_path or config is required")
legacy_only_kwargs = sorted(set(kwargs) - _FROM_PRETRAINED_CONVENIENCE_KWARGS)
if legacy_only_kwargs:
warnings.warn(
"VideoGenerator.from_pretrained(...) received legacy-only kwargs "
f"({', '.join(legacy_only_kwargs)}); prefer VideoGenerator.from_config(...) "
"for advanced configuration.",
DeprecationWarning,
stacklevel=2,
)
return cls.from_config(
legacy_from_pretrained_to_config(model_path, kwargs),
log_queue=log_queue,
)
@classmethod
def from_config(
cls,
config: GeneratorConfig | Mapping[str, Any],
*,
log_queue=None,
) -> "VideoGenerator":
normalized = normalize_generator_config(config)
fastvideo_args = generator_config_to_fastvideo_args(normalized)
generator = cls.from_fastvideo_args(fastvideo_args, log_queue=log_queue)
generator.config = normalized
return generator
@classmethod
def from_file(
cls,
path: str,
overrides: list[str] | Mapping[str, Any] | None = None,
*,
log_queue=None,
) -> "VideoGenerator":
return cls.from_config(
load_generator_config_from_file(path, overrides=overrides),
log_queue=log_queue,
)
@classmethod
def from_fastvideo_args(
@@ -122,6 +220,37 @@ class VideoGenerator:
log_queue=log_queue,
)
def generate(
self,
request: GenerationRequest | Mapping[str, Any],
*,
log_queue=None,
) -> GenerationResult | list[GenerationResult]:
"""
Generate video or image outputs from a typed inference request.
Args:
request: A `GenerationRequest` instance or a mapping that can be
parsed into one. This is the primary public inference
entrypoint for the typed API.
log_queue: Optional multiprocessing.Queue to forward worker logs to
during this request.
Returns:
A `GenerationResult` for single-request generation, or a list of
`GenerationResult` objects when the request expands into multiple
prompts.
"""
normalized_request = normalize_generation_request(request)
if log_queue:
self.executor.set_log_queue(log_queue)
try:
return self._generate_request_impl(normalized_request)
finally:
if log_queue:
self.executor.clear_log_queue()
def generate_video(
self,
prompt: str | None = None,
@@ -158,6 +287,12 @@ class VideoGenerator:
metadata dictionaries for prompt-file batch generation.
"""
log_queue = kwargs.pop("log_queue", None)
warnings.warn(
"VideoGenerator.generate_video(...) is deprecated; use "
"VideoGenerator.generate(request=...) instead.",
DeprecationWarning,
stacklevel=2,
)
if log_queue:
self.executor.set_log_queue(log_queue)
@@ -174,6 +309,52 @@ class VideoGenerator:
if log_queue:
self.executor.clear_log_queue()
def _generate_request_impl(
self,
request: GenerationRequest,
) -> GenerationResult | list[GenerationResult]:
if isinstance(request.prompt, list):
if request.inputs.prompt_path is not None:
raise ValueError("request.prompt list cannot be combined with request.inputs.prompt_path")
results: list[GenerationResult] = []
for index, single_request in enumerate(expand_request_prompt_batch(request)):
prompt = single_request.prompt
wrapped = self._generate_single_request(single_request)
if isinstance(wrapped, list):
results.extend(wrapped)
continue
wrapped.prompt_index = index
if wrapped.prompt is None:
wrapped.prompt = prompt
results.append(wrapped)
return results
return self._generate_single_request(request)
def _generate_single_request(
self,
request: GenerationRequest,
) -> GenerationResult | list[GenerationResult]:
fastvideo_args = self.fastvideo_args
pipeline_overrides = request_to_pipeline_overrides(request)
if pipeline_overrides:
fastvideo_args = deepcopy(self.fastvideo_args)
for key, value in pipeline_overrides.items():
if not hasattr(fastvideo_args.pipeline_config, key):
raise ValueError(f"Request field {key!r} is not supported by pipeline config overrides")
setattr(fastvideo_args.pipeline_config, key, deepcopy(value))
sampling_param = request_to_sampling_param(
request,
model_path=self.fastvideo_args.model_path,
)
result = self._generate_video_impl(
prompt=request.prompt,
sampling_param=sampling_param,
fastvideo_args=fastvideo_args,
)
return self._wrap_legacy_result(result)
def _generate_video_impl(
self,
prompt: str | None = None,
@@ -182,12 +363,16 @@ class VideoGenerator:
keyboard_cond: torch.Tensor | None = None,
grid_sizes: tuple[int, int, int] | list[int] | torch.Tensor
| None = None,
fastvideo_args: FastVideoArgs | None = None,
**kwargs,
) -> dict[str, Any] | list[np.ndarray] | list[dict[str, Any]]:
"""Internal implementation of generate_video."""
if fastvideo_args is None:
fastvideo_args = self.fastvideo_args
# Handle batch processing from text file
if sampling_param is None:
sampling_param = SamplingParam.from_pretrained(self.fastvideo_args.model_path)
sampling_param = SamplingParam.from_pretrained(fastvideo_args.model_path)
# Add action control inputs to kwargs if provided
if mouse_cond is not None:
@@ -199,8 +384,8 @@ class VideoGenerator:
sampling_param.update(kwargs)
if self.fastvideo_args.prompt_txt is not None or sampling_param.prompt_path is not None:
prompt_txt_path = sampling_param.prompt_path or self.fastvideo_args.prompt_txt
if fastvideo_args.prompt_txt is not None or sampling_param.prompt_path is not None:
prompt_txt_path = sampling_param.prompt_path or fastvideo_args.prompt_txt
if not prompt_txt_path or not os.path.exists(prompt_txt_path):
raise FileNotFoundError(f"Prompt text file not found: {prompt_txt_path}")
@@ -220,7 +405,12 @@ class VideoGenerator:
# Generate video for this prompt using the same logic below
output_path = self._prepare_output_path(sampling_param.output_path, batch_prompt)
kwargs["output_path"] = output_path
result = self._generate_single_video(prompt=batch_prompt, sampling_param=sampling_param, **kwargs)
result = self._generate_single_video(
prompt=batch_prompt,
sampling_param=sampling_param,
fastvideo_args=fastvideo_args,
**kwargs,
)
# Add prompt info to result
result["prompt_index"] = i
@@ -241,7 +431,12 @@ class VideoGenerator:
raise ValueError("Either prompt or prompt_txt must be provided")
output_path = self._prepare_output_path(sampling_param.output_path, prompt)
kwargs["output_path"] = output_path
return self._generate_single_video(prompt=prompt, sampling_param=sampling_param, **kwargs)
return self._generate_single_video(
prompt=prompt,
sampling_param=sampling_param,
fastvideo_args=fastvideo_args,
**kwargs,
)
def _is_image_workload(self) -> bool:
"""Return True when the workload produces a single image (t2i, i2i …)."""
@@ -325,11 +520,12 @@ class VideoGenerator:
self,
prompt: str,
sampling_param: SamplingParam | None = None,
fastvideo_args: FastVideoArgs | None = None,
**kwargs,
) -> dict[str, Any]:
"""Internal method for single video generation"""
# Create a copy of inference args to avoid modifying the original
fastvideo_args = self.fastvideo_args
if fastvideo_args is None:
fastvideo_args = self.fastvideo_args
# Validate inputs
if not isinstance(prompt, str):
@@ -472,6 +668,20 @@ class VideoGenerator:
return result
@staticmethod
def _wrap_legacy_result(
result: dict[str, Any] | list[dict[str, Any]], ) -> GenerationResult | list[GenerationResult]:
if isinstance(result, list):
return [GenerationResult.from_legacy_result(item) for item in result]
return GenerationResult.from_legacy_result(result)
@staticmethod
def _unwrap_typed_result(
result: GenerationResult | list[GenerationResult], ) -> dict[str, Any] | list[dict[str, Any]]:
if isinstance(result, list):
return [item.to_legacy_dict() for item in result]
return result.to_legacy_dict()
@staticmethod
def _mux_audio(
video_path: str,
fastvideo/tests/entrypoints/test_video_generator.py
+357
View File
@@ -1,6 +1,19 @@
import os
from types import SimpleNamespace
import warnings
import pytest
from fastvideo.api import (
GenerationRequest,
GenerationResult,
GeneratorConfig,
InputConfig,
SamplingConfig,
)
from fastvideo.configs.sample import SamplingParam
from fastvideo.entrypoints.video_generator import VideoGenerator
from fastvideo.fastvideo_args import WorkloadType
def _new_video_generator() -> VideoGenerator:
@@ -8,6 +21,66 @@ def _new_video_generator() -> VideoGenerator:
return VideoGenerator.__new__(VideoGenerator)
def _new_runtime_video_generator() -> VideoGenerator:
generator = _new_video_generator()
generator.fastvideo_args = SimpleNamespace(
model_path="test-model",
prompt_txt=None,
workload_type=SimpleNamespace(value="t2v"),
)
generator.executor = SimpleNamespace(
set_log_queue=lambda queue: None,
clear_log_queue=lambda: None,
)
generator.config = None
return generator
def _patch_from_fastvideo_args(monkeypatch):
captured = {}
def fake_from_fastvideo_args(cls, fastvideo_args, *, log_queue=None):
generator = cls.__new__(cls)
generator.fastvideo_args = fastvideo_args
generator.executor = None
generator.config = None
captured["fastvideo_args"] = fastvideo_args
captured["log_queue"] = log_queue
return generator
monkeypatch.setattr(
VideoGenerator,
"from_fastvideo_args",
classmethod(fake_from_fastvideo_args),
)
return captured
def _patch_fastvideo_args_from_kwargs(monkeypatch):
captured = {}
def fake_from_kwargs(cls, **kwargs):
captured["kwargs"] = kwargs
return SimpleNamespace(
model_path=kwargs["model_path"],
num_gpus=kwargs["num_gpus"],
workload_type=WorkloadType.from_string(kwargs.get("workload_type", "t2v")),
)
monkeypatch.setattr(
"fastvideo.api.compat.FastVideoArgs.from_kwargs",
classmethod(fake_from_kwargs),
)
return captured
def _patch_sampling_param_from_pretrained(monkeypatch):
def fake_from_pretrained(cls, model_path):
return cls()
monkeypatch.setattr(SamplingParam, "from_pretrained", classmethod(fake_from_pretrained))
def test_prepare_output_path_file_sanitization(tmp_path):
vg = _new_video_generator()
target_dir = tmp_path / "dir"
@@ -76,3 +149,287 @@ def test_prepare_output_path_empty_prompt_fallback(tmp_path):
assert os.path.dirname(result) == str(out_dir)
assert os.path.basename(result) == "output.mp4"
def test_from_config_normalizes_and_translates(monkeypatch):
captured = _patch_from_fastvideo_args(monkeypatch)
_patch_fastvideo_args_from_kwargs(monkeypatch)
config = GeneratorConfig(model_path="test-model")
config.engine.num_gpus = 2
config.pipeline.workload_type = "t2v"
generator = VideoGenerator.from_config(config)
assert captured["fastvideo_args"].model_path == "test-model"
assert captured["fastvideo_args"].num_gpus == 2
assert captured["fastvideo_args"].workload_type.value == "t2v"
assert generator.config == config
def test_from_file_loads_generator_from_run_config(tmp_path, monkeypatch):
captured = _patch_from_fastvideo_args(monkeypatch)
_patch_fastvideo_args_from_kwargs(monkeypatch)
config_path = tmp_path / "run.yaml"
config_path.write_text(
"generator:\n"
" model_path: test-model\n"
" engine:\n"
" num_gpus: 3\n"
"request:\n"
" prompt: hello\n",
encoding="utf-8",
)
VideoGenerator.from_file(str(config_path))
assert captured["fastvideo_args"].model_path == "test-model"
assert captured["fastvideo_args"].num_gpus == 3
def test_from_pretrained_convenience_kwargs_do_not_warn(monkeypatch):
captured = _patch_from_fastvideo_args(monkeypatch)
fastvideo_args_capture = _patch_fastvideo_args_from_kwargs(monkeypatch)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
generator = VideoGenerator.from_pretrained(
"test-model",
num_gpus=4,
use_fsdp_inference=False,
text_encoder_cpu_offload=True,
pin_cpu_memory=True,
dit_cpu_offload=False,
vae_cpu_offload=False,
)
assert not caught
assert captured["fastvideo_args"].model_path == "test-model"
assert captured["fastvideo_args"].num_gpus == 4
assert fastvideo_args_capture["kwargs"]["use_fsdp_inference"] is False
assert fastvideo_args_capture["kwargs"]["text_encoder_cpu_offload"] is True
assert fastvideo_args_capture["kwargs"]["pin_cpu_memory"] is True
assert fastvideo_args_capture["kwargs"]["dit_cpu_offload"] is False
assert fastvideo_args_capture["kwargs"]["vae_cpu_offload"] is False
assert generator.config is not None
assert generator.config.model_path == "test-model"
assert generator.config.engine.num_gpus == 4
def test_from_pretrained_legacy_only_kwargs_warn(monkeypatch):
captured = _patch_from_fastvideo_args(monkeypatch)
_patch_fastvideo_args_from_kwargs(monkeypatch)
with pytest.warns(DeprecationWarning, match="legacy-only kwargs"):
generator = VideoGenerator.from_pretrained(
"test-model",
num_gpus=4,
workload_type="t2v",
)
assert captured["fastvideo_args"].model_path == "test-model"
assert captured["fastvideo_args"].num_gpus == 4
assert captured["fastvideo_args"].workload_type.value == "t2v"
assert generator.config is not None
assert generator.config.pipeline.workload_type == "t2v"
def test_generate_uses_typed_request_path(monkeypatch):
generator = _new_runtime_video_generator()
_patch_sampling_param_from_pretrained(monkeypatch)
captured = {}
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
captured["prompt"] = prompt
captured["sampling_param"] = sampling_param
captured["kwargs"] = kwargs
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
result = generator.generate(
GenerationRequest(
prompt="hello world",
sampling=SamplingConfig(num_frames=81, height=480, width=832),
)
)
assert isinstance(result, GenerationResult)
assert captured["prompt"] == "hello world"
assert captured["sampling_param"].num_frames == 81
assert captured["sampling_param"].height == 480
assert captured["sampling_param"].width == 832
assert result.video_path == "outputs/test.mp4"
def test_generate_preserves_schema_defaults_for_dataclass_request(monkeypatch):
generator = _new_runtime_video_generator()
captured = {}
def fake_from_pretrained(cls, model_path):
return cls(
negative_prompt="model default",
num_frames=61,
height=448,
width=832,
)
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
captured["sampling_param"] = sampling_param
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
monkeypatch.setattr(SamplingParam, "from_pretrained", classmethod(fake_from_pretrained))
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
generator.generate(
GenerationRequest(
prompt="hello world",
negative_prompt=None,
sampling=SamplingConfig(num_frames=125, height=720, width=1280),
)
)
assert captured["sampling_param"].negative_prompt is None
assert captured["sampling_param"].num_frames == 125
assert captured["sampling_param"].height == 720
assert captured["sampling_param"].width == 1280
def test_generate_video_legacy_call_uses_legacy_impl(monkeypatch):
generator = _new_runtime_video_generator()
captured = {}
def fake_generate_video_impl(
prompt=None,
sampling_param=None,
mouse_cond=None,
keyboard_cond=None,
grid_sizes=None,
**kwargs,
):
captured["prompt"] = prompt
captured["sampling_param"] = sampling_param
captured["mouse_cond"] = mouse_cond
captured["keyboard_cond"] = keyboard_cond
captured["grid_sizes"] = grid_sizes
captured["kwargs"] = kwargs
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
with pytest.warns(DeprecationWarning):
result = generator.generate_video(
prompt="legacy prompt",
num_frames=49,
output_path="outputs/legacy",
save_video=False,
log_queue="queue-token",
)
assert captured["prompt"] == "legacy prompt"
assert captured["kwargs"]["num_frames"] == 49
assert captured["kwargs"]["output_path"] == "outputs/legacy"
assert captured["kwargs"]["save_video"] is False
assert result["video_path"] == "outputs/test.mp4"
def test_generate_video_legacy_call_preserves_unknown_kwargs(monkeypatch):
generator = _new_runtime_video_generator()
captured = {}
def fake_generate_video_impl(
prompt=None,
sampling_param=None,
mouse_cond=None,
keyboard_cond=None,
grid_sizes=None,
**kwargs,
):
captured["prompt"] = prompt
captured["sampling_param"] = sampling_param
captured["kwargs"] = kwargs
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
with pytest.warns(DeprecationWarning):
result = generator.generate_video(
prompt="legacy prompt",
neg_prompt="custom negative",
embedded_cfg_scale=7.5,
)
assert captured["prompt"] == "legacy prompt"
assert captured["kwargs"]["neg_prompt"] == "custom negative"
assert captured["kwargs"]["embedded_cfg_scale"] == 7.5
assert result["video_path"] == "outputs/test.mp4"
def test_generate_batch_prompt_file_returns_typed_results(tmp_path, monkeypatch):
generator = _new_runtime_video_generator()
_patch_sampling_param_from_pretrained(monkeypatch)
prompt_file = tmp_path / "prompts.txt"
prompt_file.write_text("first prompt\nsecond prompt\n", encoding="utf-8")
output_dir = tmp_path / "outputs"
captured_prompts = []
def fake_generate_single_video(prompt, sampling_param=None, **kwargs):
captured_prompts.append(prompt)
return {"prompts": prompt, "video_path": kwargs["output_path"]}
monkeypatch.setattr(generator, "_generate_single_video", fake_generate_single_video)
results = generator.generate(
{
"inputs": {"prompt_path": str(prompt_file)},
"output": {
"output_path": str(output_dir),
"save_video": False,
"return_frames": False,
},
}
)
assert isinstance(results, list)
assert [result.prompt for result in results] == ["first prompt", "second prompt"]
assert [result.prompt_index for result in results] == [0, 1]
assert captured_prompts == ["first prompt", "second prompt"]
def test_generate_batched_request_fans_out_media_inputs(monkeypatch):
generator = _new_runtime_video_generator()
_patch_sampling_param_from_pretrained(monkeypatch)
captured: list[tuple[str | None, str | None, str | None]] = []
def fake_generate_video_impl(prompt=None, sampling_param=None, **kwargs):
captured.append((prompt, sampling_param.image_path, sampling_param.video_path))
return {"prompts": prompt, "video_path": "outputs/test.mp4"}
monkeypatch.setattr(generator, "_generate_video_impl", fake_generate_video_impl)
results = generator.generate(
GenerationRequest(
prompt=["first prompt", "second prompt"],
inputs=InputConfig(
image_path=["first.png", "second.png"],
video_path=["first.mp4", "second.mp4"],
),
)
)
assert [result.prompt for result in results] == ["first prompt", "second prompt"]
assert captured == [
("first prompt", "first.png", "first.mp4"),
("second prompt", "second.png", "second.mp4"),
]
def test_generate_batched_request_rejects_mismatched_media_inputs(monkeypatch):
generator = _new_runtime_video_generator()
_patch_sampling_param_from_pretrained(monkeypatch)
with pytest.raises(ValueError, match="image_path"):
generator.generate(
GenerationRequest(
prompt=["first prompt", "second prompt"],
inputs=InputConfig(image_path=["first.png"]),
)
)