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)

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

Source code in fastvideo/entrypoints/video_generator.py

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

    Args:
        fastvideo_args: The inference arguments
        executor_class: The executor class to use for inference
    """
    self.fastvideo_args = fastvideo_args
    self.executor = executor_class(fastvideo_args)

Functions

fastvideo.entrypoints.video_generator.VideoGenerator.from_fastvideo_args classmethod

from_fastvideo_args(fastvideo_args: FastVideoArgs) -> VideoGenerator

Create a video generator with the specified arguments.

Parameters:

Name	Type	Description	Default
fastvideo_args	FastVideoArgs	The inference arguments	required

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) -> "VideoGenerator":
    """
    Create a video generator with the specified arguments.

    Args:
        fastvideo_args: The inference arguments

    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
    )

fastvideo.entrypoints.video_generator.VideoGenerator.from_pretrained classmethod

from_pretrained(model_path: str, device: str | None = None, torch_dtype: dtype | None = None, **kwargs) -> VideoGenerator

Create a video generator from a pretrained model.

Parameters:

Name	Type	Description	Default
model_path	str	Path or identifier for the pretrained model	required
device	str | None	Device to load the model on (e.g., "cuda", "cuda:0", "cpu")	None
torch_dtype	dtype | None	Data type for model weights (e.g., torch.float16)	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

Source code in fastvideo/entrypoints/video_generator.py

@classmethod
def from_pretrained(cls,
                    model_path: str,
                    device: str | None = None,
                    torch_dtype: torch.dtype | None = None,
                    **kwargs) -> "VideoGenerator":
    """
    Create a video generator from a pretrained model.

    Args:
        model_path: Path or identifier for the pretrained model
        device: Device to load the model on (e.g., "cuda", "cuda:0", "cpu")
        torch_dtype: Data type for model weights (e.g., torch.float16)
        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
    """
    # If users also provide some kwargs, it will override the FastVideoArgs and PipelineConfig.
    kwargs['model_path'] = model_path
    fastvideo_args = FastVideoArgs.from_kwargs(**kwargs)

    return cls.from_fastvideo_args(fastvideo_args)

fastvideo.entrypoints.video_generator.VideoGenerator.generate_video

generate_video(prompt: str | None = None, sampling_param: SamplingParam | None = None, **kwargs) -> dict[str, Any] | list[ndarray] | 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 return the raw frames	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[ndarray] | list[dict[str, Any]]	Either the output dictionary, list of frames, or list of results for batch processing

Source code in fastvideo/entrypoints/video_generator.py

def generate_video(
    self,
    prompt: str | None = None,
    sampling_param: SamplingParam | None = None,
    **kwargs,
) -> dict[str, Any] | list[np.ndarray] | 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 return the raw frames
        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:
        Either the output dictionary, list of frames, or list of results for batch processing
    """
    # Handle batch processing from text file
    if sampling_param is None:
        sampling_param = SamplingParam.from_pretrained(
            self.fastvideo_args.model_path)
    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 not os.path.exists(prompt_txt_path):
            raise FileNotFoundError(
                f"Prompt text file not found: {prompt_txt_path}")

        # Read prompts from file
        with open(prompt_txt_path, encoding='utf-8') as f:
            prompts = [line.strip() for line in f if line.strip()]

        if not prompts:
            raise ValueError(f"No prompts found in file: {prompt_txt_path}")

        logger.info("Found %d prompts in %s", len(prompts), prompt_txt_path)

        results = []
        for i, batch_prompt in enumerate(prompts):
            logger.info("Processing prompt %d/%d: %s...", i + 1,
                        len(prompts), batch_prompt[:100])
            try:
                # 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)

                # Add prompt info to result
                if isinstance(result, dict):
                    result["prompt_index"] = i
                    result["prompt"] = batch_prompt

                results.append(result)
                logger.info("Successfully generated video for prompt %d",
                            i + 1)

            except Exception as e:
                logger.error("Failed to generate video for prompt %d: %s",
                             i + 1, e)
                continue

        logger.info(
            "Completed batch processing. Generated %d videos successfully.",
            len(results))
        return results

    # Single prompt generation (original behavior)
    if prompt is None:
        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)

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