pipeline_registry

Classes

fastvideo.pipelines.pipeline_registry.PipelineType

Bases: str, Enum

Enumeration for different pipeline types.

Inherits from str to allow string comparison for backward compatibility.

Functions

fastvideo.pipelines.pipeline_registry.PipelineType.choices classmethod

choices() -> list[str]

Get all available choices as strings.

Source code in fastvideo/pipelines/pipeline_registry.py

@classmethod
def choices(cls) -> list[str]:
    """Get all available choices as strings."""
    return [pipeline_type.value for pipeline_type in cls]

fastvideo.pipelines.pipeline_registry.PipelineType.from_string classmethod

from_string(value: str) -> PipelineType

Convert string to PipelineType enum.

Source code in fastvideo/pipelines/pipeline_registry.py

@classmethod
def from_string(cls, value: str) -> "PipelineType":
    """Convert string to PipelineType enum."""
    try:
        return cls(value.lower())
    except ValueError:
        raise ValueError(
            f"Invalid pipeline type: {value}. Must be one of: {', '.join([t.value for t in cls])}"
        ) from None

Functions

fastvideo.pipelines.pipeline_registry.get_pipeline_registry

get_pipeline_registry(pipeline_type: PipelineType | str | None = None) -> _PipelineRegistry

Get a pipeline registry for the specified mode, pipeline type, and workload type.

Parameters:

Name	Type	Description	Default
pipeline_type	PipelineType | str | None	Pipeline type to load. If None and mode is provided, will be derived from mode.	None

Returns:

Type	Description
_PipelineRegistry	A pipeline registry instance.

Source code in fastvideo/pipelines/pipeline_registry.py

def get_pipeline_registry(
        pipeline_type: PipelineType | str | None = None) -> _PipelineRegistry:
    """
    Get a pipeline registry for the specified mode, pipeline type, and workload type.

    Args:
        pipeline_type: Pipeline type to load. If None and mode is provided, will be derived from mode.

    Returns:
        A pipeline registry instance.
    """
    if isinstance(pipeline_type, str):
        pipeline_type = PipelineType.from_string(pipeline_type)

    pipeline_classes = import_pipeline_classes(pipeline_type)
    return _PipelineRegistry(pipeline_classes)