646 lines
23 KiB
Python
646 lines
23 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
"""
|
|
Central registry for FastVideo pipelines and model configuration discovery.
|
|
|
|
This module mirrors the organization of sglang's registry while keeping
|
|
FastVideo's legacy behavior and mappings intact.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import os
|
|
from collections.abc import Callable
|
|
from functools import lru_cache
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from fastvideo.configs.pipelines.base import PipelineConfig
|
|
from fastvideo.configs.pipelines.cosmos import CosmosConfig
|
|
from fastvideo.configs.pipelines.cosmos2_5 import Cosmos25Config
|
|
from fastvideo.configs.pipelines.hunyuan import FastHunyuanConfig, HunyuanConfig
|
|
from fastvideo.configs.pipelines.hunyuangamecraft import HunyuanGameCraftPipelineConfig
|
|
from fastvideo.configs.pipelines.hunyuan15 import (Hunyuan15T2V480PConfig, Hunyuan15I2V480PStepDistilledConfig,
|
|
Hunyuan15T2V720PConfig, Hunyuan15I2V720PConfig,
|
|
Hunyuan15SR1080PConfig)
|
|
from fastvideo.configs.pipelines.hyworld import HYWorldConfig
|
|
from fastvideo.configs.pipelines.lingbotworld import LingBotWorldI2V480PConfig
|
|
from fastvideo.configs.pipelines.longcat import LongCatT2V480PConfig
|
|
from fastvideo.configs.pipelines.ltx2 import LTX2T2VConfig
|
|
from fastvideo.configs.pipelines.turbodiffusion import (
|
|
TurboDiffusionI2V_A14B_Config,
|
|
TurboDiffusionT2V_14B_Config,
|
|
TurboDiffusionT2V_1_3B_Config,
|
|
)
|
|
from fastvideo.configs.pipelines.wan import (
|
|
FastWan2_1_T2V_480P_Config,
|
|
FastWan2_2_TI2V_5B_Config,
|
|
MatrixGameI2V480PConfig,
|
|
SelfForcingWan2_2_T2V480PConfig,
|
|
SelfForcingWanT2V480PConfig,
|
|
WANV2VConfig,
|
|
Wan2_2_I2V_A14B_Config,
|
|
Wan2_2_T2V_A14B_Config,
|
|
Wan2_2_TI2V_5B_Config,
|
|
WanI2V480PConfig,
|
|
WanI2V720PConfig,
|
|
WanT2V480PConfig,
|
|
WanT2V720PConfig,
|
|
)
|
|
from fastvideo.configs.pipelines.sd35 import SD35Config
|
|
from fastvideo.configs.sample.base import SamplingParam
|
|
from fastvideo.configs.sample.cosmos import (
|
|
Cosmos_Predict2_2B_Video2World_SamplingParam, )
|
|
from fastvideo.configs.sample.cosmos2_5 import Cosmos25SamplingParamBase
|
|
from fastvideo.configs.sample.hunyuan import (FastHunyuanSamplingParam, HunyuanSamplingParam)
|
|
from fastvideo.configs.sample.hunyuan15 import (Hunyuan15_480P_SamplingParam,
|
|
Hunyuan15_480P_StepDistilled_I2V_SamplingParam,
|
|
Hunyuan15_720P_SamplingParam,
|
|
Hunyuan15_720P_Distilled_I2V_SamplingParam,
|
|
Hunyuan15_SR_1080P_SamplingParam)
|
|
from fastvideo.configs.sample.hyworld import HYWorld_SamplingParam
|
|
from fastvideo.configs.sample.hunyuangamecraft import HunyuanGameCraftSamplingParam
|
|
from fastvideo.configs.sample.lingbotworld import LingBotWorld_SamplingParam
|
|
from fastvideo.configs.sample.ltx2 import (LTX2BaseSamplingParam, LTX2DistilledSamplingParam)
|
|
from fastvideo.configs.sample.turbodiffusion import (
|
|
TurboDiffusionI2V_A14B_SamplingParam,
|
|
TurboDiffusionT2V_14B_SamplingParam,
|
|
TurboDiffusionT2V_1_3B_SamplingParam,
|
|
)
|
|
from fastvideo.configs.sample.wan import (
|
|
FastWanT2V480P_SamplingParam,
|
|
MatrixGame2_SamplingParam,
|
|
SelfForcingWan2_1_T2V_1_3B_480P_SamplingParam,
|
|
SelfForcingWan2_2_T2V_A14B_480P_SamplingParam,
|
|
Wan2_1_Fun_1_3B_Control_SamplingParam,
|
|
Wan2_1_Fun_1_3B_InP_SamplingParam,
|
|
Wan2_2_I2V_A14B_SamplingParam,
|
|
Wan2_2_T2V_A14B_SamplingParam,
|
|
Wan2_2_TI2V_5B_SamplingParam,
|
|
WanI2V_14B_480P_SamplingParam,
|
|
WanI2V_14B_720P_SamplingParam,
|
|
WanT2V_14B_SamplingParam,
|
|
WanT2V_1_3B_SamplingParam,
|
|
)
|
|
from fastvideo.configs.sample.sd35 import SD35SamplingParam
|
|
|
|
from fastvideo.fastvideo_args import WorkloadType
|
|
from fastvideo.logger import init_logger
|
|
from fastvideo.utils import (maybe_download_model_index, verify_model_config_and_directory)
|
|
|
|
logger = init_logger(__name__)
|
|
|
|
if TYPE_CHECKING:
|
|
from fastvideo.pipelines.composed_pipeline_base import ComposedPipelineBase
|
|
from fastvideo.pipelines.pipeline_registry import PipelineType
|
|
|
|
# --- Part 1: Pipeline Discovery ---
|
|
|
|
_PIPELINE_REGISTRY: dict[str, dict[str, type[ComposedPipelineBase]]] = {}
|
|
|
|
# Registry for pipeline configuration classes (for single-file weights without
|
|
# model_index.json). Maps pipeline_class_name -> (PipelineConfig, SamplingParam)
|
|
_PIPELINE_CONFIG_REGISTRY: dict[str, tuple[type[PipelineConfig], type[SamplingParam]]] = {}
|
|
|
|
|
|
def _discover_and_register_pipelines() -> None:
|
|
if _PIPELINE_REGISTRY:
|
|
return
|
|
|
|
from fastvideo.pipelines.pipeline_registry import import_pipeline_classes
|
|
|
|
pipeline_classes = import_pipeline_classes()
|
|
for pipeline_type, pipeline_dict in pipeline_classes.items():
|
|
_PIPELINE_REGISTRY[pipeline_type] = pipeline_dict
|
|
for pipeline_cls in pipeline_dict.values():
|
|
if pipeline_cls is None:
|
|
continue
|
|
if hasattr(pipeline_cls, "pipeline_config_cls") and hasattr(pipeline_cls, "sampling_params_cls"):
|
|
_PIPELINE_CONFIG_REGISTRY[pipeline_cls.__name__] = (
|
|
pipeline_cls.pipeline_config_cls,
|
|
pipeline_cls.sampling_params_cls,
|
|
)
|
|
|
|
|
|
def get_pipeline_config_classes(pipeline_class_name: str) -> tuple[type[PipelineConfig], type[SamplingParam]] | None:
|
|
_discover_and_register_pipelines()
|
|
return _PIPELINE_CONFIG_REGISTRY.get(pipeline_class_name)
|
|
|
|
|
|
# --- Part 2: Config Registration ---
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class ConfigInfo:
|
|
"""Encapsulates sampling + pipeline config classes for a model family."""
|
|
|
|
sampling_param_cls: type[SamplingParam] | None
|
|
pipeline_config_cls: type[PipelineConfig]
|
|
|
|
|
|
# The central registry mapping a model name to its configuration information
|
|
_CONFIG_REGISTRY: dict[str, ConfigInfo] = {}
|
|
|
|
# Mappings from Hugging Face model paths to our internal model names
|
|
_MODEL_HF_PATH_TO_NAME: dict[str, str] = {}
|
|
|
|
# Detectors to identify model families from paths or class names
|
|
_MODEL_NAME_DETECTORS: list[tuple[str, Callable[[str], bool]]] = []
|
|
|
|
|
|
def register_configs(
|
|
sampling_param_cls: type[SamplingParam] | None,
|
|
pipeline_config_cls: type[PipelineConfig],
|
|
hf_model_paths: list[str] | None = None,
|
|
model_detectors: list[Callable[[str], bool]] | None = None,
|
|
) -> None:
|
|
"""Register config classes for a model family."""
|
|
model_id = str(len(_CONFIG_REGISTRY))
|
|
|
|
_CONFIG_REGISTRY[model_id] = ConfigInfo(
|
|
sampling_param_cls=sampling_param_cls,
|
|
pipeline_config_cls=pipeline_config_cls,
|
|
)
|
|
|
|
if hf_model_paths:
|
|
for path in hf_model_paths:
|
|
if path in _MODEL_HF_PATH_TO_NAME:
|
|
logger.warning("Model path '%s' is already mapped to '%s' and will be overwritten by '%s'.", path,
|
|
_MODEL_HF_PATH_TO_NAME[path], model_id)
|
|
_MODEL_HF_PATH_TO_NAME[path] = model_id
|
|
|
|
if model_detectors:
|
|
for detector in model_detectors:
|
|
_MODEL_NAME_DETECTORS.append((model_id, detector))
|
|
|
|
|
|
def get_model_short_name(model_id: str) -> str:
|
|
if "/" in model_id:
|
|
return model_id.split("/")[-1]
|
|
return model_id
|
|
|
|
|
|
def _get_config_info(
|
|
model_path: str,
|
|
*,
|
|
raise_on_missing: bool = True,
|
|
) -> ConfigInfo | None:
|
|
# 1. Exact match
|
|
if model_path in _MODEL_HF_PATH_TO_NAME:
|
|
model_id = _MODEL_HF_PATH_TO_NAME[model_path]
|
|
logger.debug("Resolved model path '%s' from exact path match.", model_path)
|
|
return _CONFIG_REGISTRY.get(model_id)
|
|
|
|
# 2. Partial match: use short model name.
|
|
model_name = get_model_short_name(model_path.lower())
|
|
all_model_hf_paths = sorted(_MODEL_HF_PATH_TO_NAME.keys(), key=len, reverse=True)
|
|
for registered_model_hf_id in all_model_hf_paths:
|
|
registered_model_name = get_model_short_name(registered_model_hf_id.lower())
|
|
if registered_model_name == model_name:
|
|
logger.debug("Resolved model name '%s' from partial path match.", registered_model_hf_id)
|
|
model_id = _MODEL_HF_PATH_TO_NAME[registered_model_hf_id]
|
|
return _CONFIG_REGISTRY.get(model_id)
|
|
|
|
# 3. Use detectors (path or model_index pipeline name).
|
|
if os.path.exists(model_path):
|
|
config = verify_model_config_and_directory(model_path)
|
|
else:
|
|
config = maybe_download_model_index(model_path)
|
|
|
|
pipeline_name = config.get("_class_name", "").lower()
|
|
|
|
matched_model_names: list[str] = []
|
|
for model_id, detector in _MODEL_NAME_DETECTORS:
|
|
if detector(model_path.lower()) or detector(pipeline_name):
|
|
logger.debug("Matched model name '%s' using a registered detector.", model_id)
|
|
matched_model_names.append(model_id)
|
|
|
|
if matched_model_names:
|
|
if len(matched_model_names) > 1:
|
|
logger.warning(
|
|
"Multiple models matched for path '%s': %s. Using the first matched: '%s'.",
|
|
model_path,
|
|
matched_model_names,
|
|
matched_model_names[0],
|
|
)
|
|
model_id = matched_model_names[0]
|
|
return _CONFIG_REGISTRY.get(model_id)
|
|
|
|
if raise_on_missing:
|
|
raise RuntimeError(f"No model info found for model path: {model_path}")
|
|
return None
|
|
|
|
|
|
def _register_configs() -> None:
|
|
# LTX-2 (base)
|
|
register_configs(
|
|
sampling_param_cls=LTX2BaseSamplingParam,
|
|
pipeline_config_cls=LTX2T2VConfig,
|
|
hf_model_paths=[
|
|
"Lightricks/LTX-2",
|
|
"FastVideo/LTX2-base",
|
|
"FastVideo/LTX2-Diffusers",
|
|
],
|
|
model_detectors=[
|
|
lambda path: ("ltx2" in path.lower() or "ltx-2" in path.lower()) and "distilled" not in path.lower(),
|
|
],
|
|
)
|
|
# LTX-2 (distilled)
|
|
register_configs(
|
|
sampling_param_cls=LTX2DistilledSamplingParam,
|
|
pipeline_config_cls=LTX2T2VConfig,
|
|
hf_model_paths=[
|
|
"FastVideo/LTX2-Distilled-Diffusers",
|
|
],
|
|
model_detectors=[
|
|
lambda path: ("ltx2" in path.lower() or "ltx-2" in path.lower()) and "distilled" in path.lower(),
|
|
],
|
|
)
|
|
|
|
# Hunyuan 1.5 (specific)
|
|
register_configs(
|
|
sampling_param_cls=Hunyuan15_480P_SamplingParam,
|
|
pipeline_config_cls=Hunyuan15T2V480PConfig,
|
|
hf_model_paths=[
|
|
"hunyuanvideo-community/HunyuanVideo-1.5-Diffusers-480p_t2v",
|
|
],
|
|
model_detectors=[
|
|
lambda path: any(token in path.lower() for token in (
|
|
"hunyuan15",
|
|
"hunyuanvideo15",
|
|
"hunyuanvideo-1.5",
|
|
"hunyuanvideo_1.5",
|
|
)),
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Hunyuan15_480P_StepDistilled_I2V_SamplingParam,
|
|
pipeline_config_cls=Hunyuan15I2V480PStepDistilledConfig,
|
|
hf_model_paths=[
|
|
"hunyuanvideo-community/HunyuanVideo-1.5-Diffusers-480p_i2v_step_distilled",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Hunyuan15_720P_SamplingParam,
|
|
pipeline_config_cls=Hunyuan15T2V720PConfig,
|
|
hf_model_paths=[
|
|
"hunyuanvideo-community/HunyuanVideo-1.5-Diffusers-720p_t2v",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Hunyuan15_720P_Distilled_I2V_SamplingParam,
|
|
pipeline_config_cls=Hunyuan15I2V720PConfig,
|
|
hf_model_paths=[
|
|
"hunyuanvideo-community/HunyuanVideo-1.5-Diffusers-720p_i2v_distilled",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Hunyuan15_SR_1080P_SamplingParam,
|
|
pipeline_config_cls=Hunyuan15SR1080PConfig,
|
|
hf_model_paths=["weizhou03/HunyuanVideo-1.5-Diffusers-1080p", "weizhou03/HunyuanVideo-1.5-Diffusers-1080p-2SR"],
|
|
)
|
|
|
|
# Hunyuan (excludes gamecraft, hyworld, and versioned models)
|
|
register_configs(
|
|
sampling_param_cls=HunyuanSamplingParam,
|
|
pipeline_config_cls=HunyuanConfig,
|
|
hf_model_paths=[
|
|
"hunyuanvideo-community/HunyuanVideo",
|
|
],
|
|
model_detectors=[
|
|
lambda path: "hunyuan" in path.lower() and "gamecraft" not in path.lower() and "hyworld" not in path.lower(
|
|
) and "1.5" not in path.lower() and "1-5" not in path.lower()
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=FastHunyuanSamplingParam,
|
|
pipeline_config_cls=FastHunyuanConfig,
|
|
hf_model_paths=[
|
|
"FastVideo/FastHunyuan-diffusers",
|
|
],
|
|
)
|
|
|
|
# HYWorld
|
|
register_configs(
|
|
sampling_param_cls=HYWorld_SamplingParam,
|
|
pipeline_config_cls=HYWorldConfig,
|
|
hf_model_paths=[
|
|
"FastVideo/HY-WorldPlay-Bidirectional-Diffusers",
|
|
],
|
|
model_detectors=[lambda path: "hyworld" in path.lower()],
|
|
)
|
|
|
|
# HunyuanGameCraft
|
|
register_configs(
|
|
sampling_param_cls=HunyuanGameCraftSamplingParam,
|
|
pipeline_config_cls=HunyuanGameCraftPipelineConfig,
|
|
hf_model_paths=[
|
|
"FastVideo/HunyuanGameCraft-Diffusers",
|
|
],
|
|
model_detectors=[lambda path: "gamecraft" in path.lower()],
|
|
)
|
|
# LingBotWorld
|
|
register_configs(
|
|
sampling_param_cls=LingBotWorld_SamplingParam,
|
|
pipeline_config_cls=LingBotWorldI2V480PConfig,
|
|
hf_model_paths=[
|
|
"FastVideo/LingBot-World-Base-Cam-Diffusers",
|
|
],
|
|
model_detectors=[lambda path: ("lingbotworld" in path.lower() or "lingbot-world" in path.lower())],
|
|
)
|
|
|
|
# Kandinsky5 Lite T2V
|
|
register_configs(
|
|
sampling_param_cls=None,
|
|
pipeline_config_cls=PipelineConfig,
|
|
hf_model_paths=[
|
|
"kandinskylab/Kandinsky-5.0-T2V-Lite-sft-5s-Diffusers",
|
|
],
|
|
model_detectors=[
|
|
lambda path: any(token in path.lower() for token in ("kandinsky5", "kandinsky-5")),
|
|
],
|
|
)
|
|
|
|
# LongCat
|
|
register_configs(
|
|
sampling_param_cls=None,
|
|
pipeline_config_cls=LongCatT2V480PConfig,
|
|
hf_model_paths=[
|
|
"FastVideo/LongCat-Video-T2V-Diffusers",
|
|
"FastVideo/LongCat-Video-I2V-Diffusers",
|
|
"FastVideo/LongCat-Video-VC-Diffusers",
|
|
],
|
|
model_detectors=[
|
|
lambda path: "longcatimagetovideo" in path.lower(),
|
|
lambda path: "longcatvideocontinuation" in path.lower(),
|
|
lambda path: "longcat" in path.lower(),
|
|
],
|
|
)
|
|
|
|
# MatrixGame
|
|
register_configs(
|
|
sampling_param_cls=MatrixGame2_SamplingParam,
|
|
pipeline_config_cls=MatrixGameI2V480PConfig,
|
|
hf_model_paths=[
|
|
"FastVideo/Matrix-Game-2.0-Base-Diffusers",
|
|
"FastVideo/Matrix-Game-2.0-GTA-Diffusers",
|
|
"FastVideo/Matrix-Game-2.0-TempleRun-Diffusers",
|
|
],
|
|
model_detectors=[
|
|
lambda path: "matrix-game" in path.lower() or "matrixgame" in path.lower(),
|
|
],
|
|
)
|
|
|
|
# Cosmos 2.5
|
|
register_configs(
|
|
sampling_param_cls=Cosmos25SamplingParamBase,
|
|
pipeline_config_cls=Cosmos25Config,
|
|
hf_model_paths=[
|
|
"KyleShao/Cosmos-Predict2.5-2B-Diffusers",
|
|
],
|
|
model_detectors=[
|
|
lambda path: any(token in path.lower() for token in (
|
|
"cosmos25",
|
|
"cosmos2_5",
|
|
"cosmos2.5",
|
|
)),
|
|
],
|
|
)
|
|
|
|
# Cosmos 2
|
|
register_configs(
|
|
sampling_param_cls=Cosmos_Predict2_2B_Video2World_SamplingParam,
|
|
pipeline_config_cls=CosmosConfig,
|
|
hf_model_paths=[
|
|
"nvidia/Cosmos-Predict2-2B-Video2World",
|
|
],
|
|
model_detectors=[
|
|
lambda path: "cosmos" in path.lower() and
|
|
("2.5" not in path.lower() and "2_5" not in path.lower() and "25" not in path.lower()),
|
|
],
|
|
)
|
|
|
|
# TurboDiffusion
|
|
register_configs(
|
|
sampling_param_cls=TurboDiffusionT2V_1_3B_SamplingParam,
|
|
pipeline_config_cls=TurboDiffusionT2V_1_3B_Config,
|
|
hf_model_paths=[
|
|
"loayrashid/TurboWan2.1-T2V-1.3B-Diffusers",
|
|
],
|
|
model_detectors=[lambda path: "turbodiffusion" in path.lower() or "turbowan" in path.lower()],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=TurboDiffusionT2V_14B_SamplingParam,
|
|
pipeline_config_cls=TurboDiffusionT2V_14B_Config,
|
|
hf_model_paths=[
|
|
"loayrashid/TurboWan2.1-T2V-14B-Diffusers",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=TurboDiffusionI2V_A14B_SamplingParam,
|
|
pipeline_config_cls=TurboDiffusionI2V_A14B_Config,
|
|
hf_model_paths=[
|
|
"loayrashid/TurboWan2.2-I2V-A14B-Diffusers",
|
|
],
|
|
)
|
|
|
|
# Wan
|
|
register_configs(
|
|
sampling_param_cls=WanT2V_1_3B_SamplingParam,
|
|
pipeline_config_cls=WanT2V480PConfig,
|
|
hf_model_paths=[
|
|
"Wan-AI/Wan2.1-T2V-1.3B-Diffusers",
|
|
],
|
|
model_detectors=[lambda path: "wanpipeline" in path.lower()],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=WanT2V_14B_SamplingParam,
|
|
pipeline_config_cls=WanT2V720PConfig,
|
|
hf_model_paths=[
|
|
"Wan-AI/Wan2.1-T2V-14B-Diffusers",
|
|
"FastVideo/Wan2.1-VSA-T2V-14B-720P-Diffusers",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=WanI2V_14B_480P_SamplingParam,
|
|
pipeline_config_cls=WanI2V480PConfig,
|
|
hf_model_paths=[
|
|
"Wan-AI/Wan2.1-I2V-14B-480P-Diffusers",
|
|
],
|
|
model_detectors=[lambda path: "wanimagetovideo" in path.lower()],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=WanI2V_14B_720P_SamplingParam,
|
|
pipeline_config_cls=WanI2V720PConfig,
|
|
hf_model_paths=[
|
|
"Wan-AI/Wan2.1-I2V-14B-720P-Diffusers",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Wan2_1_Fun_1_3B_InP_SamplingParam,
|
|
pipeline_config_cls=WanI2V480PConfig,
|
|
hf_model_paths=[
|
|
"weizhou03/Wan2.1-Fun-1.3B-InP-Diffusers",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Wan2_1_Fun_1_3B_Control_SamplingParam,
|
|
pipeline_config_cls=WANV2VConfig,
|
|
hf_model_paths=[
|
|
"IRMChen/Wan2.1-Fun-1.3B-Control-Diffusers",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=FastWanT2V480P_SamplingParam,
|
|
pipeline_config_cls=FastWan2_1_T2V_480P_Config,
|
|
hf_model_paths=[
|
|
"FastVideo/FastWan2.1-T2V-1.3B-Diffusers",
|
|
"FastVideo/FastWan2.1-T2V-14B-480P-Diffusers",
|
|
],
|
|
model_detectors=[lambda path: "wandmdpipeline" in path.lower()],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Wan2_2_TI2V_5B_SamplingParam,
|
|
pipeline_config_cls=Wan2_2_TI2V_5B_Config,
|
|
hf_model_paths=[
|
|
"Wan-AI/Wan2.2-TI2V-5B-Diffusers",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Wan2_2_TI2V_5B_SamplingParam,
|
|
pipeline_config_cls=FastWan2_2_TI2V_5B_Config,
|
|
hf_model_paths=[
|
|
"FastVideo/FastWan2.2-TI2V-5B-FullAttn-Diffusers",
|
|
"FastVideo/FastWan2.2-TI2V-5B-Diffusers",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Wan2_2_T2V_A14B_SamplingParam,
|
|
pipeline_config_cls=Wan2_2_T2V_A14B_Config,
|
|
hf_model_paths=[
|
|
"Wan-AI/Wan2.2-T2V-A14B-Diffusers",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=Wan2_2_I2V_A14B_SamplingParam,
|
|
pipeline_config_cls=Wan2_2_I2V_A14B_Config,
|
|
hf_model_paths=[
|
|
"Wan-AI/Wan2.2-I2V-A14B-Diffusers",
|
|
],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=SelfForcingWan2_1_T2V_1_3B_480P_SamplingParam,
|
|
pipeline_config_cls=SelfForcingWanT2V480PConfig,
|
|
hf_model_paths=[
|
|
"wlsaidhi/SFWan2.1-T2V-1.3B-Diffusers",
|
|
],
|
|
model_detectors=[lambda path: "wancausaldmdpipeline" in path.lower()],
|
|
)
|
|
register_configs(
|
|
sampling_param_cls=SelfForcingWan2_2_T2V_A14B_480P_SamplingParam,
|
|
pipeline_config_cls=SelfForcingWan2_2_T2V480PConfig,
|
|
hf_model_paths=[
|
|
"rand0nmr/SFWan2.2-T2V-A14B-Diffusers",
|
|
"FastVideo/SFWan2.2-I2V-A14B-Preview-Diffusers",
|
|
],
|
|
)
|
|
|
|
# SD3.5
|
|
register_configs(
|
|
sampling_param_cls=SD35SamplingParam,
|
|
pipeline_config_cls=SD35Config,
|
|
hf_model_paths=[
|
|
"stabilityai/stable-diffusion-3.5-medium",
|
|
],
|
|
model_detectors=[
|
|
lambda path: any(token in path.lower() for token in (
|
|
"sd35",
|
|
"stablediffusion3",
|
|
"stabilityai__stable-diffusion-3.5-medium",
|
|
)),
|
|
],
|
|
)
|
|
|
|
|
|
# --- Part 3: Main Resolver ---
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class ModelInfo:
|
|
pipeline_cls: type[ComposedPipelineBase]
|
|
sampling_param_cls: type[SamplingParam]
|
|
pipeline_config_cls: type[PipelineConfig]
|
|
|
|
|
|
@lru_cache(maxsize=32)
|
|
def get_model_info(
|
|
model_path: str,
|
|
pipeline_type: PipelineType | str | None = None,
|
|
workload_type: WorkloadType | None = None,
|
|
override_pipeline_cls_name: str | None = None,
|
|
) -> ModelInfo:
|
|
from fastvideo.pipelines.pipeline_registry import (PipelineType, get_pipeline_registry)
|
|
|
|
if pipeline_type is None:
|
|
pipeline_type = PipelineType.BASIC
|
|
elif isinstance(pipeline_type, str):
|
|
pipeline_type = PipelineType.from_string(pipeline_type)
|
|
|
|
if workload_type is None:
|
|
workload_type = WorkloadType.T2V
|
|
|
|
if os.path.exists(model_path):
|
|
config = verify_model_config_and_directory(model_path)
|
|
else:
|
|
config = maybe_download_model_index(model_path)
|
|
|
|
pipeline_name = config.get("_class_name")
|
|
if override_pipeline_cls_name:
|
|
logger.info("Overriding pipeline class name from %s to %s", pipeline_name, override_pipeline_cls_name)
|
|
pipeline_name = override_pipeline_cls_name
|
|
|
|
if pipeline_name is None:
|
|
raise ValueError("Model config does not contain a _class_name attribute. "
|
|
"Only diffusers format is supported.")
|
|
|
|
pipeline_registry = get_pipeline_registry(pipeline_type)
|
|
pipeline_cls = pipeline_registry.resolve_pipeline_cls(pipeline_name, pipeline_type, workload_type)
|
|
|
|
config_info = _get_config_info(model_path, raise_on_missing=True)
|
|
assert config_info is not None, "config_info must be resolved"
|
|
|
|
sampling_param_cls = config_info.sampling_param_cls or SamplingParam
|
|
|
|
return ModelInfo(
|
|
pipeline_cls=pipeline_cls,
|
|
sampling_param_cls=sampling_param_cls,
|
|
pipeline_config_cls=config_info.pipeline_config_cls,
|
|
)
|
|
|
|
|
|
def get_pipeline_config_cls_from_name(pipeline_name_or_path: str) -> type[PipelineConfig]:
|
|
config_info = _get_config_info(pipeline_name_or_path, raise_on_missing=False)
|
|
if config_info is None:
|
|
raise ValueError(
|
|
f"No match found for pipeline {pipeline_name_or_path}, please check the pipeline name or path.")
|
|
return config_info.pipeline_config_cls
|
|
|
|
|
|
def get_sampling_param_cls_for_name(pipeline_name_or_path: str) -> Any | None:
|
|
config_info = _get_config_info(pipeline_name_or_path, raise_on_missing=False)
|
|
if config_info is None:
|
|
logger.warning("No match found for pipeline %s, using default sampling param.", pipeline_name_or_path)
|
|
return None
|
|
return config_info.sampling_param_cls
|
|
|
|
|
|
_register_configs()
|
|
|
|
__all__ = [
|
|
"ConfigInfo",
|
|
"ModelInfo",
|
|
"get_model_info",
|
|
"get_pipeline_config_cls_from_name",
|
|
"get_sampling_param_cls_for_name",
|
|
"get_pipeline_config_classes",
|
|
]
|