video_generator

VideoGenerator module for FastVideo.

This module provides a consolidated interface for generating videos using diffusion models.

Classes

fastvideo.entrypoints.video_generator.VideoGenerator

VideoGenerator(fastvideo_args: FastVideoArgs, executor_class: type[Executor], log_stats: bool, *, log_queue=None)

A unified class for generating videos using diffusion models.

This class provides a simple interface for video generation with rich customization options, similar to popular frameworks like HF Diffusers.

Initialize the video generator.

Parameters:

Name	Type	Description	Default
fastvideo_args	FastVideoArgs	The inference arguments	required
executor_class	type[Executor]	The executor class to use for inference	required
log_stats	bool	Whether to log statistics	required
log_queue		Optional multiprocessing.Queue to forward worker logs to	None

Source code in fastvideo/entrypoints/video_generator.py

def __init__(
    self,
    fastvideo_args: FastVideoArgs,
    executor_class: type[Executor],
    log_stats: bool,
    *,
    log_queue=None,
):
    """
    Initialize the video generator.

    Args:
        fastvideo_args: The inference arguments
        executor_class: The executor class to use for inference
        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)

Functions

fastvideo.entrypoints.video_generator.VideoGenerator.from_fastvideo_args classmethod

from_fastvideo_args(fastvideo_args: FastVideoArgs, *, log_queue=None) -> VideoGenerator

Create a video generator with the specified arguments.

Parameters:

Name	Type	Description	Default
fastvideo_args	FastVideoArgs	The inference arguments	required
log_queue		Optional multiprocessing.Queue to forward worker logs to	None

Returns:

Type	Description
VideoGenerator	The created video generator

Source code in fastvideo/entrypoints/video_generator.py

@classmethod
def from_fastvideo_args(
    cls,
    fastvideo_args: FastVideoArgs,
    *,
    log_queue=None,
) -> "VideoGenerator":
    """
    Create a video generator with the specified arguments.

    Args:
        fastvideo_args: The inference arguments
        log_queue: Optional multiprocessing.Queue to forward worker logs to

    Returns:
        The created video generator
    """
    # Initialize distributed environment if needed
    # initialize_distributed_and_parallelism(fastvideo_args)

    executor_class = Executor.get_class(fastvideo_args)
    return cls(
        fastvideo_args=fastvideo_args,
        executor_class=executor_class,
        log_stats=False,  # TODO: implement
        log_queue=log_queue,
    )

fastvideo.entrypoints.video_generator.VideoGenerator.from_pretrained classmethod

from_pretrained(model_path: str | GeneratorConfig | Mapping[str, Any] | None = None, **kwargs) -> VideoGenerator

Create a video generator from a pretrained model.

Parameters:

Name	Type	Description	Default
model_path	str | GeneratorConfig | Mapping[str, Any] | None	Path or identifier for the pretrained model	None
pipeline_config		Pipeline config to use for inference	required
**kwargs		Additional arguments to customize model loading, set any FastVideoArgs or PipelineConfig attributes here.	{}

Returns:

Type	Description
VideoGenerator	The created video generator

Priority level: Default pipeline config < User's pipeline config < User's kwargs

Stable convenience kwargs remain supported here for common engine and offload settings. Advanced model- or pipeline-specific options should move to VideoGenerator.from_config(...).

Source code in fastvideo/entrypoints/video_generator.py

@classmethod
def from_pretrained(
    cls,
    model_path: str | GeneratorConfig | Mapping[str, Any] | None = None,
    **kwargs,
) -> "VideoGenerator":
    """
    Create a video generator from a pretrained model.

    Args:
        model_path: Path or identifier for the pretrained model
        pipeline_config: Pipeline config to use for inference
        **kwargs: Additional arguments to customize model loading, set any FastVideoArgs or PipelineConfig attributes here.

    Returns:
        The created video generator

    Priority level: Default pipeline config < User's pipeline config < User's kwargs

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

fastvideo.entrypoints.video_generator.VideoGenerator.generate

generate(request: GenerationRequest | Mapping[str, Any], *, log_queue=None) -> GenerationResult | list[GenerationResult]

Generate video or image outputs from a typed inference request.

Parameters:

Name	Type	Description	Default
request	GenerationRequest | Mapping[str, Any]	A GenerationRequest instance or a mapping that can be parsed into one. This is the primary public inference entrypoint for the typed API.	required
log_queue		Optional multiprocessing.Queue to forward worker logs to during this request.	None

Returns:

Type	Description
GenerationResult | list[GenerationResult]	A GenerationResult for single-request generation, or a list of
GenerationResult | list[GenerationResult]	GenerationResult objects when the request expands into multiple
GenerationResult | list[GenerationResult]	prompts.

Source code in fastvideo/entrypoints/video_generator.py

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

fastvideo.entrypoints.video_generator.VideoGenerator.generate_video

generate_video(prompt: str | None = None, sampling_param: SamplingParam | None = None, mouse_cond: Tensor | None = None, keyboard_cond: Tensor | None = None, grid_sizes: tuple[int, int, int] | list[int] | Tensor | None = None, **kwargs) -> dict[str, Any] | list[dict[str, Any]]

Generate a video based on the given prompt.

Parameters:

Name	Type	Description	Default
prompt	str | None	The prompt to use for generation (optional if prompt_txt is provided)	None
negative_prompt		The negative prompt to use (overrides the one in fastvideo_args)	required
output_path		Path to save the video (overrides the one in fastvideo_args)	required
prompt_path		Path to prompt file	required
save_video		Whether to save the video to disk	required
return_frames		Whether to include raw frames in the result dict	required
num_inference_steps		Number of denoising steps (overrides fastvideo_args)	required
guidance_scale		Classifier-free guidance scale (overrides fastvideo_args)	required
num_frames		Number of frames to generate (overrides fastvideo_args)	required
height		Height of generated video (overrides fastvideo_args)	required
width		Width of generated video (overrides fastvideo_args)	required
fps		Frames per second for saved video (overrides fastvideo_args)	required
seed		Random seed for generation (overrides fastvideo_args)	required
callback		Callback function called after each step	required
callback_steps		Number of steps between each callback	required

Returns:

Type	Description
dict[str, Any] | list[dict[str, Any]]	A metadata dictionary for single-prompt generation, or a list of
dict[str, Any] | list[dict[str, Any]]	metadata dictionaries for prompt-file batch generation.

Source code in fastvideo/entrypoints/video_generator.py

def generate_video(
    self,
    prompt: str | None = None,
    sampling_param: SamplingParam | None = None,
    # Action control inputs (Matrix-Game)
    mouse_cond: torch.Tensor | None = None,
    keyboard_cond: torch.Tensor | None = None,
    grid_sizes: tuple[int, int, int] | list[int] | torch.Tensor
    | None = None,
    **kwargs,
) -> dict[str, Any] | list[dict[str, Any]]:
    """
    Generate a video based on the given prompt.

    Args:
        prompt: The prompt to use for generation (optional if prompt_txt is provided)
        negative_prompt: The negative prompt to use (overrides the one in fastvideo_args)
        output_path: Path to save the video (overrides the one in fastvideo_args)
        prompt_path: Path to prompt file
        save_video: Whether to save the video to disk
        return_frames: Whether to include raw frames in the result dict
        num_inference_steps: Number of denoising steps (overrides fastvideo_args)
        guidance_scale: Classifier-free guidance scale (overrides fastvideo_args)
        num_frames: Number of frames to generate (overrides fastvideo_args)
        height: Height of generated video (overrides fastvideo_args)
        width: Width of generated video (overrides fastvideo_args)
        fps: Frames per second for saved video (overrides fastvideo_args)
        seed: Random seed for generation (overrides fastvideo_args)
        callback: Callback function called after each step
        callback_steps: Number of steps between each callback

    Returns:
        A metadata dictionary for single-prompt generation, or a list of
        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)

    try:
        return self._generate_video_impl(
            prompt=prompt,
            sampling_param=sampling_param,
            mouse_cond=mouse_cond,
            keyboard_cond=keyboard_cond,
            grid_sizes=grid_sizes,
            **kwargs,
        )
    finally:
        if log_queue:
            self.executor.clear_log_queue()

fastvideo.entrypoints.video_generator.VideoGenerator.shutdown

shutdown()

Shutdown the video generator.

Source code in fastvideo/entrypoints/video_generator.py

def shutdown(self):
    """
    Shutdown the video generator.
    """
    self.executor.shutdown()
    del self.executor

fastvideo.entrypoints.video_generator.VideoGenerator.unmerge_lora_weights

unmerge_lora_weights() -> None

Use unmerged weights for inference to produce videos that align with validation videos generated during training.

Source code in fastvideo/entrypoints/video_generator.py

def unmerge_lora_weights(self) -> None:
    """
    Use unmerged weights for inference to produce videos that align with 
    validation videos generated during training.
    """
    self.executor.unmerge_lora_weights()

Functions