fastvideo ¶

@classmethod
def from_kwargs(cls, kwargs: dict[str, Any], config_cli_prefix: str = "") -> "PipelineConfig":
    """
    Load PipelineConfig from kwargs Dictionary.
    kwargs: dictionary of kwargs
    config_cli_prefix: prefix of CLI arguments for this PipelineConfig instance
    """
    from fastvideo.registry import get_pipeline_config_cls_from_name

    prefix_with_dot = f"{config_cli_prefix}." if (config_cli_prefix.strip() != "") else ""
    model_path: str | None = kwargs.get(prefix_with_dot + 'model_path', None) or kwargs.get('model_path')
    pipeline_config_or_path: str | PipelineConfig | dict[str, Any] | None = kwargs.get(
        prefix_with_dot + 'pipeline_config', None) or kwargs.get('pipeline_config')
    if model_path is None:
        raise ValueError("model_path is required in kwargs")

    # 1. Get the pipeline config class from the registry
    pipeline_config_cls = get_pipeline_config_cls_from_name(model_path)

    # 2. Instantiate PipelineConfig
    if pipeline_config_cls is None:
        logger.warning("Couldn't find pipeline config for %s. Using the default pipeline config.", model_path)
        pipeline_config = cls()
    else:
        pipeline_config = pipeline_config_cls()

    # 3. Load PipelineConfig from a json file or a PipelineConfig object if provided
    if isinstance(pipeline_config_or_path, str):
        pipeline_config.load_from_json(pipeline_config_or_path)
        kwargs[prefix_with_dot + 'pipeline_config_path'] = pipeline_config_or_path
    elif isinstance(pipeline_config_or_path, PipelineConfig):
        pipeline_config = pipeline_config_or_path
    elif isinstance(pipeline_config_or_path, dict):
        pipeline_config.update_pipeline_config(pipeline_config_or_path)

    # 4. Update PipelineConfig from CLI arguments if provided
    kwargs[prefix_with_dot + 'model_path'] = model_path
    pipeline_config.update_config_from_dict(kwargs, config_cli_prefix)

    return pipeline_config

@classmethod
def from_pretrained(cls, model_path: str) -> "PipelineConfig":
    """
    use the pipeline class setting from model_path to match the pipeline config
    """
    from fastvideo.registry import get_pipeline_config_cls_from_name
    pipeline_config_cls = get_pipeline_config_cls_from_name(model_path)

    return cast(PipelineConfig, pipeline_config_cls(model_path=model_path))

@staticmethod
def add_cli_args(parser: Any) -> Any:
    """Add CLI arguments for SamplingParam fields"""
    parser.add_argument(
        "--prompt",
        type=str,
        default=SamplingParam.prompt,
        help="Text prompt for video generation",
    )
    parser.add_argument(
        "--negative-prompt",
        type=str,
        default=SamplingParam.negative_prompt,
        help="Negative text prompt for video generation",
    )
    parser.add_argument(
        "--prompt-path",
        type=str,
        default=SamplingParam.prompt_path,
        help="Path to a text file containing the prompt",
    )
    parser.add_argument(
        "--output-path",
        type=str,
        default=SamplingParam.output_path,
        help="Path to save the generated video",
    )
    parser.add_argument(
        "--output-video-name",
        type=str,
        default=SamplingParam.output_video_name,
        help="Name of the output video",
    )
    parser.add_argument(
        "--num-videos-per-prompt",
        type=int,
        default=SamplingParam.num_videos_per_prompt,
        help="Number of videos to generate per prompt",
    )
    parser.add_argument(
        "--seed",
        type=int,
        default=SamplingParam.seed,
        help="Random seed for generation",
    )
    parser.add_argument(
        "--num-frames",
        type=int,
        default=SamplingParam.num_frames,
        help="Number of frames to generate",
    )
    parser.add_argument(
        "--height",
        type=int,
        default=SamplingParam.height,
        help="Height of generated video",
    )
    parser.add_argument(
        "--width",
        type=int,
        default=SamplingParam.width,
        help="Width of generated video",
    )
    parser.add_argument(
        "--fps",
        type=int,
        default=SamplingParam.fps,
        help="Frames per second for saved video",
    )
    parser.add_argument(
        "--num-inference-steps",
        type=int,
        default=SamplingParam.num_inference_steps,
        help="Number of denoising steps",
    )
    parser.add_argument(
        "--guidance-scale",
        type=float,
        default=SamplingParam.guidance_scale,
        help="Classifier-free guidance scale",
    )
    parser.add_argument(
        "--cfg-normalization",
        action=StoreBoolean,
        default=SamplingParam.cfg_normalization,
        help="Cap Z-Image CFG prediction norm to the positive-prediction norm",
    )
    parser.add_argument(
        "--cfg-truncation",
        type=float,
        default=SamplingParam.cfg_truncation,
        help="Disable Z-Image CFG above this normalized-noise threshold",
    )
    parser.add_argument(
        "--batch-cfg",
        action=StoreBoolean,
        default=SamplingParam.batch_cfg,
        help="Evaluate conditional and unconditional CFG branches in one batch",
    )
    parser.add_argument(
        "--guidance-rescale",
        type=float,
        default=SamplingParam.guidance_rescale,
        help="Guidance rescale factor",
    )
    parser.add_argument(
        "--use-embedded-guidance",
        action="store_true",
        default=SamplingParam.use_embedded_guidance,
        help="Use embedded guidance scale (FLUX-style) instead of classic CFG",
    )
    parser.add_argument(
        "--true-cfg-scale",
        type=float,
        default=SamplingParam.true_cfg_scale,
        help="True CFG scale for FLUX when > 1 (requires negative prompt encoding)",
    )
    parser.add_argument(
        "--boundary-ratio",
        type=float,
        default=SamplingParam.boundary_ratio,
        help="Boundary timestep ratio",
    )
    parser.add_argument(
        "--save-video",
        action="store_true",
        default=SamplingParam.save_video,
        help="Whether to save the video to disk",
    )
    parser.add_argument(
        "--no-save-video",
        action="store_false",
        dest="save_video",
        help="Don't save the video to disk",
    )
    parser.add_argument(
        "--return-frames",
        action="store_true",
        default=False,
        help="Whether to return the raw frames",
    )
    parser.add_argument(
        "--image-path",
        type=str,
        default=SamplingParam.image_path,
        help="Path to input image for image-to-video generation",
    )
    parser.add_argument(
        "--video-path",
        type=str,
        default=SamplingParam.video_path,
        help="Path to input video for video-to-video generation",
    )
    parser.add_argument(
        "--refine-from",
        type=str,
        default=SamplingParam.refine_from,
        help="Path to stage1 video for refinement (LongCat 480p->720p)",
    )
    parser.add_argument(
        "--t-thresh",
        type=float,
        default=SamplingParam.t_thresh,
        help="Threshold for timestep scheduling in refinement (default: 0.5)",
    )
    parser.add_argument(
        "--spatial-refine-only",
        action=StoreBoolean,
        default=SamplingParam.spatial_refine_only,
        help="Only perform spatial super-resolution (no temporal doubling)",
    )
    parser.add_argument(
        "--num-cond-frames",
        type=int,
        default=SamplingParam.num_cond_frames,
        help="Number of conditioning frames for refinement",
    )
    parser.add_argument(
        "--moba-config-path",
        type=str,
        default=None,
        help="Path to a JSON file containing V-MoBA specific configurations.",
    )
    parser.add_argument(
        "--return-trajectory-latents",
        action="store_true",
        default=SamplingParam.return_trajectory_latents,
        help="Whether to return the trajectory",
    )
    parser.add_argument(
        "--return-trajectory-decoded",
        action="store_true",
        default=SamplingParam.return_trajectory_decoded,
        help="Whether to return the decoded trajectory",
    )
    return parser

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

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

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

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

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)

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

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

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