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)

Methods:

fastvideo.entrypoints.video_generator.VideoGenerator.default_health_check_request staticmethod

default_health_check_request() -> GenerationRequest

Return the minimal typed request Dynamo uses for probes.

256x256, 8 frames, 1 inference step -- fast enough to be a viable liveness check, non-trivial enough to exercise the DiT -> VAE -> decode path. Consumers adapt this shape to their transport's health-check payload (see docs/design/server_contracts/dynamo.md).

Source code in fastvideo/entrypoints/video_generator.py

@staticmethod
def default_health_check_request() -> GenerationRequest:
    """Return the minimal typed request Dynamo uses for probes.

    256x256, 8 frames, 1 inference step -- fast enough to be a
    viable liveness check, non-trivial enough to exercise the
    DiT -> VAE -> decode path. Consumers adapt this shape to their
    transport's health-check payload (see
    ``docs/design/server_contracts/dynamo.md``).
    """
    return GenerationRequest(
        prompt="health check",
        inputs=InputConfig(),
        sampling=SamplingConfig(
            num_frames=8,
            height=256,
            width=256,
            fps=24,
            num_inference_steps=1,
            guidance_scale=1.0,
        ),
        output=OutputConfig(save_video=False, return_frames=False),
    )

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)
    if kwargs.pop("nvfp4_fa4", False):
        import os
        os.environ["FASTVIDEO_NVFP4_FA4"] = "1"
        os.environ.setdefault("CUTE_DSL_ENABLE_TVM_FFI", "1")
    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_async async

generate_async(request: GenerationRequest | Mapping[str, Any], *, log_queue=None)

Async generation that yields typed :class:VideoEvents.

Three consumers share this substrate:

• Streaming server (:mod:fastvideo.entrypoints.streaming) — pipes :class:VideoPartialEvent frames into fMP4.
• Stateless OpenAI server — ignores progress events, forwards :class:VideoFinalEvent as the HTTP response body.
• Dynamo native backend (components/src/dynamo/fastvideo/) — wraps each event as an NvVideosResponse chunk.

The aggregated code path shipped here yields a single :class:VideoProgressEvent at start and one :class:VideoFinalEvent at end. Future work will thread per-step progress events through the pipeline's denoise loop so streaming consumers don't have to wait on a materialized final.

Source code in fastvideo/entrypoints/video_generator.py

async def generate_async(
    self,
    request: GenerationRequest | Mapping[str, Any],
    *,
    log_queue=None,
):
    """Async generation that yields typed :class:`VideoEvent`s.

    Three consumers share this substrate:

    * Streaming server (:mod:`fastvideo.entrypoints.streaming`) —
      pipes :class:`VideoPartialEvent` frames into fMP4.
    * Stateless OpenAI server — ignores progress events, forwards
      :class:`VideoFinalEvent` as the HTTP response body.
    * Dynamo native backend
      (``components/src/dynamo/fastvideo/``) — wraps each event as
      an ``NvVideosResponse`` chunk.

    The aggregated code path shipped here yields a single
    :class:`VideoProgressEvent` at start and one
    :class:`VideoFinalEvent` at end. Future work will thread
    per-step progress events through the pipeline's denoise loop
    so streaming consumers don't have to wait on a materialized
    final.
    """
    import asyncio

    normalized = normalize_generation_request(request)
    total_steps = max(1, normalized.sampling.num_inference_steps)
    yield VideoProgressEvent(step=0, total_steps=total_steps, stage="denoise")

    if log_queue:
        self.executor.set_log_queue(log_queue)
    try:
        result = await asyncio.to_thread(self._generate_request_impl, normalized)
    finally:
        if log_queue:
            self.executor.clear_log_queue()

    if isinstance(result, list):
        # Prompt-batch expansion — emit one Final per sub-result.
        for sub in result:
            yield await asyncio.to_thread(_final_event_from_result, sub)
        return
    yield await asyncio.to_thread(_final_event_from_result, result)

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:
        extra_overrides: dict[str, Any] = {}
        for _ek in _BATCH_EXTRA_PASSTHROUGH_KEYS:
            if _ek in kwargs:
                extra_overrides[_ek] = kwargs.pop(_ek)

        request = legacy_generate_call_to_request(
            prompt,
            sampling_param,
            mouse_cond=mouse_cond,
            keyboard_cond=keyboard_cond,
            grid_sizes=grid_sizes,
            legacy_kwargs=kwargs,
        )

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

        resolved_sampling_param = request_to_sampling_param(
            request,
            model_path=self.fastvideo_args.model_path,
        )
        return self._generate_video_impl(
            prompt=request.prompt,
            sampling_param=resolved_sampling_param,
            fastvideo_args=fastvideo_args,
            **extra_overrides,
        )
    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: