Components and configs

mindone.diffusers.modular_pipelines.modular_pipeline.ComponentSpec dataclass

Specification for a pipeline component.

A component can be created in two ways: 1. From scratch using init with a config dict 2. using from_pretrained

ATTRIBUTE	DESCRIPTION
name	Name of the component TYPE: Optional[str]
type_hint	Type of the component (e.g. UNet2DConditionModel) TYPE: Optional[Type]
description	Optional description of the component TYPE: Optional[str]
config	Optional config dict for init creation TYPE: Optional[FrozenDict]
repo	Optional repo path for from_pretrained creation TYPE: Optional[Union[str, List[str]]]
subfolder	Optional subfolder in repo TYPE: Optional[str]
variant	Optional variant in repo TYPE: Optional[str]
revision	Optional revision in repo TYPE: Optional[str]
default_creation_method	Preferred creation method - "from_config" or "from_pretrained" TYPE: Literal['from_config', 'from_pretrained']

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

@dataclass
class ComponentSpec:
    """Specification for a pipeline component.

    A component can be created in two ways:
    1. From scratch using __init__ with a config dict
    2. using `from_pretrained`

    Attributes:
        name: Name of the component
        type_hint: Type of the component (e.g. UNet2DConditionModel)
        description: Optional description of the component
        config: Optional config dict for __init__ creation
        repo: Optional repo path for from_pretrained creation
        subfolder: Optional subfolder in repo
        variant: Optional variant in repo
        revision: Optional revision in repo
        default_creation_method: Preferred creation method - "from_config" or "from_pretrained"
    """

    name: Optional[str] = None
    type_hint: Optional[Type] = None
    description: Optional[str] = None
    config: Optional[FrozenDict] = None
    # YiYi Notes: should we change it to pretrained_model_name_or_path for consistency? a bit long for a field name
    repo: Optional[Union[str, List[str]]] = field(default=None, metadata={"loading": True})
    subfolder: Optional[str] = field(default="", metadata={"loading": True})
    variant: Optional[str] = field(default=None, metadata={"loading": True})
    revision: Optional[str] = field(default=None, metadata={"loading": True})
    default_creation_method: Literal["from_config", "from_pretrained"] = "from_pretrained"

    def __hash__(self):
        """Make ComponentSpec hashable, using load_id as the hash value."""
        return hash((self.name, self.load_id, self.default_creation_method))

    def __eq__(self, other):
        """Compare ComponentSpec objects based on name and load_id."""
        if not isinstance(other, ComponentSpec):
            return False
        return (
            self.name == other.name
            and self.load_id == other.load_id
            and self.default_creation_method == other.default_creation_method
        )

    @classmethod
    def from_component(cls, name: str, component: Any) -> Any:
        """Create a ComponentSpec from a Component.

        Currently supports:
        - Components created with `ComponentSpec.load()` method
        - Components that are ConfigMixin subclasses but not nn.Modules (e.g. schedulers, guiders)

        Args:
            name: Name of the component
            component: Component object to create spec from

        Returns:
            ComponentSpec object

        Raises:
            ValueError: If component is not supported (e.g. nn.Module without load_id, non-ConfigMixin)
        """

        # Check if component was created with ComponentSpec.load()
        if hasattr(component, "_diffusers_load_id") and component._diffusers_load_id != "null":
            # component has a usable load_id -> from_pretrained, no warning needed
            default_creation_method = "from_pretrained"
        else:
            # Component doesn't have a usable load_id, check if it's a nn.Module
            if isinstance(component, ms.nn.Cell):
                raise ValueError(
                    "Cannot create ComponentSpec from a nn.Module that was not created with `ComponentSpec.load()` method."
                )
            # ConfigMixin objects without weights (e.g. scheduler & guider) can be recreated with from_config
            elif isinstance(component, ConfigMixin):
                # warn if component was not created with `ComponentSpec`
                if not hasattr(component, "_diffusers_load_id"):
                    logger.warning(
                        "Component was not created using `ComponentSpec`, defaulting to `from_config` creation method"
                    )
                default_creation_method = "from_config"
            else:
                # Not a ConfigMixin and not created with `ComponentSpec.load()` method -> throw error
                raise ValueError(
                    f"Cannot create ComponentSpec from {name}({component.__class__.__name__}). Currently ComponentSpec.from_component() only supports: "
                    f" - components created with `ComponentSpec.load()` method"
                    f" - components that are a subclass of ConfigMixin but not a nn.Module (e.g. guider, scheduler)."
                )

        type_hint = component.__class__

        if isinstance(component, ConfigMixin) and default_creation_method == "from_config":
            config = component.config
        else:
            config = None
        if hasattr(component, "_diffusers_load_id") and component._diffusers_load_id != "null":
            load_spec = cls.decode_load_id(component._diffusers_load_id)
        else:
            load_spec = {}

        return cls(
            name=name, type_hint=type_hint, config=config, default_creation_method=default_creation_method, **load_spec
        )

    @classmethod
    def loading_fields(cls) -> List[str]:
        """
        Return the names of all loading‐related fields (i.e. those whose field.metadata["loading"] is True).
        """
        return [f.name for f in fields(cls) if f.metadata.get("loading", False)]

    @property
    def load_id(self) -> str:
        """
        Unique identifier for this spec's pretrained load, composed of repo|subfolder|variant|revision (no empty
        segments).
        """
        if self.default_creation_method == "from_config":
            return "null"
        parts = [getattr(self, k) for k in self.loading_fields()]
        parts = ["null" if p is None else p for p in parts]
        return "|".join(p for p in parts if p)

    @classmethod
    def decode_load_id(cls, load_id: str) -> Dict[str, Optional[str]]:
        """
        Decode a load_id string back into a dictionary of loading fields and values.

        Args:
            load_id: The load_id string to decode, format: "repo|subfolder|variant|revision"
                     where None values are represented as "null"

        Returns:
            Dict mapping loading field names to their values. e.g. {
                "repo": "path/to/repo", "subfolder": "subfolder", "variant": "variant", "revision": "revision"
            } If a segment value is "null", it's replaced with None. Returns None if load_id is "null" (indicating
            component not created with `load` method).
        """

        # Get all loading fields in order
        loading_fields = cls.loading_fields()
        result = {f: None for f in loading_fields}

        if load_id == "null":
            return result

        # Split the load_id
        parts = load_id.split("|")

        # Map parts to loading fields by position
        for i, part in enumerate(parts):
            if i < len(loading_fields):
                # Convert "null" string back to None
                result[loading_fields[i]] = None if part == "null" else part

        return result

    # YiYi TODO: I think we should only support ConfigMixin for this method (after we make guider and image_processors config mixin)
    # otherwise we cannot do spec -> spec.create() -> component -> ComponentSpec.from_component(component)
    # the config info is lost in the process
    # remove error check in from_component spec and ModularPipeline.update_components() if we remove support for non configmixin in `create()` method
    def create(self, config: Optional[Union[FrozenDict, Dict[str, Any]]] = None, **kwargs) -> Any:
        """Create component using from_config with config."""

        if self.type_hint is None or not isinstance(self.type_hint, type):
            raise ValueError("`type_hint` is required when using from_config creation method.")

        config = config or self.config or {}

        if issubclass(self.type_hint, ConfigMixin):
            component = self.type_hint.from_config(config, **kwargs)
        else:
            signature_params = inspect.signature(self.type_hint.__init__).parameters
            init_kwargs = {}
            for k, v in config.items():
                if k in signature_params:
                    init_kwargs[k] = v
            for k, v in kwargs.items():
                if k in signature_params:
                    init_kwargs[k] = v
            component = self.type_hint(**init_kwargs)

        component._diffusers_load_id = "null"
        if hasattr(component, "config"):
            self.config = component.config

        return component

    # YiYi TODO: add guard for type of model, if it is supported by from_pretrained
    def load(self, **kwargs) -> Any:
        """Load component using from_pretrained."""

        # select loading fields from kwargs passed from user: e.g. repo, subfolder, variant, revision, note the list could change
        passed_loading_kwargs = {key: kwargs.pop(key) for key in self.loading_fields() if key in kwargs}
        # merge loading field value in the spec with user passed values to create load_kwargs
        load_kwargs = {key: passed_loading_kwargs.get(key, getattr(self, key)) for key in self.loading_fields()}
        # repo is a required argument for from_pretrained, a.k.a. pretrained_model_name_or_path
        repo = load_kwargs.pop("repo", None)
        if repo is None:
            raise ValueError(
                "`repo` info is required when using `load` method (you can directly set it in `repo` field of the ComponentSpec or pass it as an argument)"
            )

        if self.type_hint is None:
            try:
                from mindone.diffusers import AutoModel

                component = AutoModel.from_pretrained(repo, **load_kwargs, **kwargs)
            except Exception as e:
                raise ValueError(f"Unable to load {self.name} without `type_hint`: {e}")
            # update type_hint if AutoModel load successfully
            self.type_hint = component.__class__
        else:
            try:
                component = self.type_hint.from_pretrained(repo, **load_kwargs, **kwargs)
            except Exception as e:
                raise ValueError(f"Unable to load {self.name} using load method: {e}")

        self.repo = repo
        for k, v in load_kwargs.items():
            setattr(self, k, v)
        component._diffusers_load_id = self.load_id

        return component

mindone.diffusers.modular_pipelines.modular_pipeline.ComponentSpec.load_id property

Unique identifier for this spec's pretrained load, composed of repo|subfolder|variant|revision (no empty segments).

mindone.diffusers.modular_pipelines.modular_pipeline.ComponentSpec.__eq__(other)

Compare ComponentSpec objects based on name and load_id.

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

def __eq__(self, other):
    """Compare ComponentSpec objects based on name and load_id."""
    if not isinstance(other, ComponentSpec):
        return False
    return (
        self.name == other.name
        and self.load_id == other.load_id
        and self.default_creation_method == other.default_creation_method
    )

mindone.diffusers.modular_pipelines.modular_pipeline.ComponentSpec.__hash__()

Make ComponentSpec hashable, using load_id as the hash value.

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

def __hash__(self):
    """Make ComponentSpec hashable, using load_id as the hash value."""
    return hash((self.name, self.load_id, self.default_creation_method))

mindone.diffusers.modular_pipelines.modular_pipeline.ComponentSpec.create(config=None, **kwargs)

Create component using from_config with config.

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

def create(self, config: Optional[Union[FrozenDict, Dict[str, Any]]] = None, **kwargs) -> Any:
    """Create component using from_config with config."""

    if self.type_hint is None or not isinstance(self.type_hint, type):
        raise ValueError("`type_hint` is required when using from_config creation method.")

    config = config or self.config or {}

    if issubclass(self.type_hint, ConfigMixin):
        component = self.type_hint.from_config(config, **kwargs)
    else:
        signature_params = inspect.signature(self.type_hint.__init__).parameters
        init_kwargs = {}
        for k, v in config.items():
            if k in signature_params:
                init_kwargs[k] = v
        for k, v in kwargs.items():
            if k in signature_params:
                init_kwargs[k] = v
        component = self.type_hint(**init_kwargs)

    component._diffusers_load_id = "null"
    if hasattr(component, "config"):
        self.config = component.config

    return component

mindone.diffusers.modular_pipelines.modular_pipeline.ComponentSpec.decode_load_id(load_id) classmethod

Decode a load_id string back into a dictionary of loading fields and values.

PARAMETER	DESCRIPTION
load_id	The load_id string to decode, format: "repo|subfolder|variant|revision" where None values are represented as "null" TYPE: str

RETURNS	DESCRIPTION
Dict[str, Optional[str]]	Dict mapping loading field names to their values. e.g. { "repo": "path/to/repo", "subfolder": "subfolder", "variant": "variant", "revision": "revision"
Dict[str, Optional[str]]	} If a segment value is "null", it's replaced with None. Returns None if load_id is "null" (indicating
Dict[str, Optional[str]]	component not created with load method).

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

@classmethod
def decode_load_id(cls, load_id: str) -> Dict[str, Optional[str]]:
    """
    Decode a load_id string back into a dictionary of loading fields and values.

    Args:
        load_id: The load_id string to decode, format: "repo|subfolder|variant|revision"
                 where None values are represented as "null"

    Returns:
        Dict mapping loading field names to their values. e.g. {
            "repo": "path/to/repo", "subfolder": "subfolder", "variant": "variant", "revision": "revision"
        } If a segment value is "null", it's replaced with None. Returns None if load_id is "null" (indicating
        component not created with `load` method).
    """

    # Get all loading fields in order
    loading_fields = cls.loading_fields()
    result = {f: None for f in loading_fields}

    if load_id == "null":
        return result

    # Split the load_id
    parts = load_id.split("|")

    # Map parts to loading fields by position
    for i, part in enumerate(parts):
        if i < len(loading_fields):
            # Convert "null" string back to None
            result[loading_fields[i]] = None if part == "null" else part

    return result

mindone.diffusers.modular_pipelines.modular_pipeline.ComponentSpec.from_component(name, component) classmethod

Create a ComponentSpec from a Component.

Currently supports: - Components created with ComponentSpec.load() method - Components that are ConfigMixin subclasses but not nn.Modules (e.g. schedulers, guiders)

PARAMETER	DESCRIPTION
name	Name of the component TYPE: str
component	Component object to create spec from TYPE: Any

RETURNS	DESCRIPTION
Any	ComponentSpec object

RAISES	DESCRIPTION
ValueError	If component is not supported (e.g. nn.Module without load_id, non-ConfigMixin)

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

@classmethod
def from_component(cls, name: str, component: Any) -> Any:
    """Create a ComponentSpec from a Component.

    Currently supports:
    - Components created with `ComponentSpec.load()` method
    - Components that are ConfigMixin subclasses but not nn.Modules (e.g. schedulers, guiders)

    Args:
        name: Name of the component
        component: Component object to create spec from

    Returns:
        ComponentSpec object

    Raises:
        ValueError: If component is not supported (e.g. nn.Module without load_id, non-ConfigMixin)
    """

    # Check if component was created with ComponentSpec.load()
    if hasattr(component, "_diffusers_load_id") and component._diffusers_load_id != "null":
        # component has a usable load_id -> from_pretrained, no warning needed
        default_creation_method = "from_pretrained"
    else:
        # Component doesn't have a usable load_id, check if it's a nn.Module
        if isinstance(component, ms.nn.Cell):
            raise ValueError(
                "Cannot create ComponentSpec from a nn.Module that was not created with `ComponentSpec.load()` method."
            )
        # ConfigMixin objects without weights (e.g. scheduler & guider) can be recreated with from_config
        elif isinstance(component, ConfigMixin):
            # warn if component was not created with `ComponentSpec`
            if not hasattr(component, "_diffusers_load_id"):
                logger.warning(
                    "Component was not created using `ComponentSpec`, defaulting to `from_config` creation method"
                )
            default_creation_method = "from_config"
        else:
            # Not a ConfigMixin and not created with `ComponentSpec.load()` method -> throw error
            raise ValueError(
                f"Cannot create ComponentSpec from {name}({component.__class__.__name__}). Currently ComponentSpec.from_component() only supports: "
                f" - components created with `ComponentSpec.load()` method"
                f" - components that are a subclass of ConfigMixin but not a nn.Module (e.g. guider, scheduler)."
            )

    type_hint = component.__class__

    if isinstance(component, ConfigMixin) and default_creation_method == "from_config":
        config = component.config
    else:
        config = None
    if hasattr(component, "_diffusers_load_id") and component._diffusers_load_id != "null":
        load_spec = cls.decode_load_id(component._diffusers_load_id)
    else:
        load_spec = {}

    return cls(
        name=name, type_hint=type_hint, config=config, default_creation_method=default_creation_method, **load_spec
    )

mindone.diffusers.modular_pipelines.modular_pipeline.ComponentSpec.load(**kwargs)

Load component using from_pretrained.

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

def load(self, **kwargs) -> Any:
    """Load component using from_pretrained."""

    # select loading fields from kwargs passed from user: e.g. repo, subfolder, variant, revision, note the list could change
    passed_loading_kwargs = {key: kwargs.pop(key) for key in self.loading_fields() if key in kwargs}
    # merge loading field value in the spec with user passed values to create load_kwargs
    load_kwargs = {key: passed_loading_kwargs.get(key, getattr(self, key)) for key in self.loading_fields()}
    # repo is a required argument for from_pretrained, a.k.a. pretrained_model_name_or_path
    repo = load_kwargs.pop("repo", None)
    if repo is None:
        raise ValueError(
            "`repo` info is required when using `load` method (you can directly set it in `repo` field of the ComponentSpec or pass it as an argument)"
        )

    if self.type_hint is None:
        try:
            from mindone.diffusers import AutoModel

            component = AutoModel.from_pretrained(repo, **load_kwargs, **kwargs)
        except Exception as e:
            raise ValueError(f"Unable to load {self.name} without `type_hint`: {e}")
        # update type_hint if AutoModel load successfully
        self.type_hint = component.__class__
    else:
        try:
            component = self.type_hint.from_pretrained(repo, **load_kwargs, **kwargs)
        except Exception as e:
            raise ValueError(f"Unable to load {self.name} using load method: {e}")

    self.repo = repo
    for k, v in load_kwargs.items():
        setattr(self, k, v)
    component._diffusers_load_id = self.load_id

    return component

mindone.diffusers.modular_pipelines.modular_pipeline.ComponentSpec.loading_fields() classmethod

Return the names of all loading‐related fields (i.e. those whose field.metadata["loading"] is True).

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

@classmethod
def loading_fields(cls) -> List[str]:
    """
    Return the names of all loading‐related fields (i.e. those whose field.metadata["loading"] is True).
    """
    return [f.name for f in fields(cls) if f.metadata.get("loading", False)]

mindone.diffusers.modular_pipelines.modular_pipeline.ConfigSpec dataclass

Specification for a pipeline configuration parameter.

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

@dataclass
class ConfigSpec:
    """Specification for a pipeline configuration parameter."""

    name: str
    default: Any
    description: Optional[str] = None

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager

A central registry and management system for model components across multiple pipelines.

[ComponentsManager] provides a unified way to register, track, and reuse model components (like UNet, VAE, text encoders, etc.) across different modular pipelines. It includes features for duplicate detection, memory management, and component organization.

This is an experimental feature and is likely to change in the future.

Example

from mindone.diffusers import ComponentsManager

# Create a components manager
cm = ComponentsManager()

# Add components
cm.add("unet", unet_model, collection="sdxl")
cm.add("vae", vae_model, collection="sdxl")

# Retrieve components
unet = cm.get_one(name="unet", collection="sdxl")

Source code in mindone/diffusers/modular_pipelines/components_manager.py

class ComponentsManager:
    """
    A central registry and management system for model components across multiple pipelines.

    [`ComponentsManager`] provides a unified way to register, track, and reuse model components (like UNet, VAE, text
    encoders, etc.) across different modular pipelines. It includes features for duplicate detection, memory
    management, and component organization.

    <Tip warning={true}>

        This is an experimental feature and is likely to change in the future.

    </Tip>

    Example:
        ```python
        from mindone.diffusers import ComponentsManager

        # Create a components manager
        cm = ComponentsManager()

        # Add components
        cm.add("unet", unet_model, collection="sdxl")
        cm.add("vae", vae_model, collection="sdxl")

        # Retrieve components
        unet = cm.get_one(name="unet", collection="sdxl")
        ```
    """

    _available_info_fields = [
        "model_id",
        "added_time",
        "collection",
        "class_name",
        "size_gb",
        "adapters",
        "has_hook",
        "ip_adapter",
    ]

    def __init__(self):
        self.components = OrderedDict()
        # YiYi TODO: can remove once confirm we don't need this in mellon
        self.added_time = OrderedDict()  # Store when components were added
        self.collections = OrderedDict()  # collection_name -> set of component_names
        self.model_hooks = None
        self._auto_offload_enabled = False

    def _lookup_ids(
        self,
        name: Optional[str] = None,
        collection: Optional[str] = None,
        load_id: Optional[str] = None,
        components: Optional[OrderedDict] = None,
    ):
        """
        Lookup component_ids by name, collection, or load_id. Does not support pattern matching. Returns a set of
        component_ids
        """
        if components is None:
            components = self.components

        if name:
            ids_by_name = set()
            for component_id, component in components.items():
                comp_name = self._id_to_name(component_id)
                if comp_name == name:
                    ids_by_name.add(component_id)
        else:
            ids_by_name = set(components.keys())
        if collection:
            ids_by_collection = set()
            for component_id, component in components.items():
                if component_id in self.collections[collection]:
                    ids_by_collection.add(component_id)
        else:
            ids_by_collection = set(components.keys())
        if load_id:
            ids_by_load_id = set()
            for name, component in components.items():
                if hasattr(component, "_diffusers_load_id") and component._diffusers_load_id == load_id:
                    ids_by_load_id.add(name)
        else:
            ids_by_load_id = set(components.keys())

        ids = ids_by_name.intersection(ids_by_collection).intersection(ids_by_load_id)
        return ids

    @staticmethod
    def _id_to_name(component_id: str):
        return "_".join(component_id.split("_")[:-1])

    def add(self, name: str, component: Any, collection: Optional[str] = None):
        """
        Add a component to the ComponentsManager.

        Args:
            name (str): The name of the component
            component (Any): The component to add
            collection (Optional[str]): The collection to add the component to

        Returns:
            str: The unique component ID, which is generated as "{name}_{id(component)}" where
                 id(component) is Python's built-in unique identifier for the object
        """
        component_id = f"{name}_{id(component)}"
        is_new_component = True  # noqa

        # check for duplicated components
        for comp_id, comp in self.components.items():
            if comp == component:
                comp_name = self._id_to_name(comp_id)
                if comp_name == name:
                    logger.warning(f"ComponentsManager: component '{name}' already exists as '{comp_id}'")
                    component_id = comp_id
                    is_new_component = False  # noqa
                    break
                else:
                    logger.warning(
                        f"ComponentsManager: adding component '{name}' as '{component_id}', but it is duplicate of '{comp_id}'"
                        f"To remove a duplicate, call `components_manager.remove('<component_id>')`."
                    )

        # check for duplicated load_id and warn (we do not delete for you)
        if hasattr(component, "_diffusers_load_id") and component._diffusers_load_id != "null":
            components_with_same_load_id = self._lookup_ids(load_id=component._diffusers_load_id)
            components_with_same_load_id = [id for id in components_with_same_load_id if id != component_id]

            if components_with_same_load_id:
                existing = ", ".join(components_with_same_load_id)
                logger.warning(
                    f"ComponentsManager: adding component '{component_id}', "
                    f"but it has duplicate load_id '{component._diffusers_load_id}' with existing components: {existing}. "
                    f"To remove a duplicate, call `components_manager.remove('<component_id>')`."
                )

        # add component to components manager
        self.components[component_id] = component
        self.added_time[component_id] = time.time()

        if collection:
            if collection not in self.collections:
                self.collections[collection] = set()
            if component_id not in self.collections[collection]:
                comp_ids_in_collection = self._lookup_ids(name=name, collection=collection)
                for comp_id in comp_ids_in_collection:
                    logger.warning(
                        f"ComponentsManager: removing existing {name} from collection '{collection}': {comp_id}"
                    )
                    # remove existing component from this collection (if it is not in any other collection,
                    # will be removed from ComponentsManager)
                    self.remove_from_collection(comp_id, collection)

                self.collections[collection].add(component_id)
                logger.info(f"ComponentsManager: added component '{name}' in collection '{collection}': {component_id}")
        else:
            logger.info(f"ComponentsManager: added component '{name}' as '{component_id}'")

        return component_id

    def remove_from_collection(self, component_id: str, collection: str):
        """
        Remove a component from a collection.
        """
        if collection not in self.collections:
            logger.warning(f"Collection '{collection}' not found in ComponentsManager")
            return
        if component_id not in self.collections[collection]:
            logger.warning(f"Component '{component_id}' not found in collection '{collection}'")
            return
        # remove from the collection
        self.collections[collection].remove(component_id)
        # check if this component is in any other collection
        comp_colls = [coll for coll, comps in self.collections.items() if component_id in comps]
        if not comp_colls:  # only if no other collection contains this component, remove it
            logger.warning(f"ComponentsManager: removing component '{component_id}' from ComponentsManager")
            self.remove(component_id)

    def remove(self, component_id: str = None):
        """
        Remove a component from the ComponentsManager.

        Args:
            component_id (str): The ID of the component to remove
        """
        if component_id not in self.components:
            logger.warning(f"Component '{component_id}' not found in ComponentsManager")
            return

        component = self.components.pop(component_id)  # noqa
        self.added_time.pop(component_id)

        for collection in self.collections:
            if component_id in self.collections[collection]:
                self.collections[collection].remove(component_id)

    # YiYi TODO: rename to search_components for now, may remove this method
    def search_components(
        self,
        names: Optional[str] = None,
        collection: Optional[str] = None,
        load_id: Optional[str] = None,
        return_dict_with_names: bool = True,
    ):
        """
        Search components by name with simple pattern matching. Optionally filter by collection or load_id.

        Args:
            names: Component name(s) or pattern(s)
                Patterns:
                - "unet" : match any component with base name "unet" (e.g., unet_123abc)
                - "!unet" : everything except components with base name "unet"
                - "unet*" : anything with base name starting with "unet"
                - "!unet*" : anything with base name NOT starting with "unet"
                - "*unet*" : anything with base name containing "unet"
                - "!*unet*" : anything with base name NOT containing "unet"
                - "refiner|vae|unet" : anything with base name exactly matching "refiner", "vae", or "unet"
                - "!refiner|vae|unet" : anything with base name NOT exactly matching "refiner", "vae", or "unet"
                - "unet*|vae*" : anything with base name starting with "unet" OR starting with "vae"
            collection: Optional collection to filter by
            load_id: Optional load_id to filter by
            return_dict_with_names:
                                    If True, returns a dictionary with component names as keys, throw an error if
                                    multiple components with the same name are found If False, returns a dictionary
                                    with component IDs as keys

        Returns:
            Dictionary mapping component names to components if return_dict_with_names=True, or a dictionary mapping
            component IDs to components if return_dict_with_names=False
        """

        # select components based on collection and load_id filters
        selected_ids = self._lookup_ids(collection=collection, load_id=load_id)
        components = {k: self.components[k] for k in selected_ids}

        def get_return_dict(components, return_dict_with_names):
            """
            Create a dictionary mapping component names to components if return_dict_with_names=True, or a dictionary
            mapping component IDs to components if return_dict_with_names=False, throw an error if duplicate component
            names are found when return_dict_with_names=True
            """
            if return_dict_with_names:
                dict_to_return = {}
                for comp_id, comp in components.items():
                    comp_name = self._id_to_name(comp_id)
                    if comp_name in dict_to_return:
                        raise ValueError(
                            f"Duplicate component names found in the search results: {comp_name}, please set `return_dict_with_names=False` to return a dictionary with component IDs as keys"  # noqa
                        )
                    dict_to_return[comp_name] = comp
                return dict_to_return
            else:
                return components

        # if no names are provided, return the filtered components as it is
        if names is None:
            return get_return_dict(components, return_dict_with_names)

        # if names is not a string, raise an error
        elif not isinstance(names, str):
            raise ValueError(f"Invalid type for `names: {type(names)}, only support string")

        # Create mapping from component_id to base_name for components to be used for pattern matching
        base_names = {comp_id: self._id_to_name(comp_id) for comp_id in components.keys()}

        # Helper function to check if a component matches a pattern based on its base name
        def matches_pattern(component_id, pattern, exact_match=False):
            """
            Helper function to check if a component matches a pattern based on its base name.

            Args:
                component_id: The component ID to check
                pattern: The pattern to match against
                exact_match: If True, only exact matches to base_name are considered
            """
            base_name = base_names[component_id]

            # Exact match with base name
            if exact_match:
                return pattern == base_name

            # Prefix match (ends with *)
            elif pattern.endswith("*"):
                prefix = pattern[:-1]
                return base_name.startswith(prefix)

            # Contains match (starts with *)
            elif pattern.startswith("*"):
                search = pattern[1:-1] if pattern.endswith("*") else pattern[1:]
                return search in base_name

            # Exact match (no wildcards)
            else:
                return pattern == base_name

        # Check if this is a "not" pattern
        is_not_pattern = names.startswith("!")
        if is_not_pattern:
            names = names[1:]  # Remove the ! prefix

        # Handle OR patterns (containing |)
        if "|" in names:
            terms = names.split("|")
            matches = {}

            for comp_id, comp in components.items():
                # For OR patterns with exact names (no wildcards), we do exact matching on base names
                exact_match = all(not (term.startswith("*") or term.endswith("*")) for term in terms)

                # Check if any of the terms match this component
                should_include = any(matches_pattern(comp_id, term, exact_match) for term in terms)

                # Flip the decision if this is a NOT pattern
                if is_not_pattern:
                    should_include = not should_include

                if should_include:
                    matches[comp_id] = comp

            log_msg = "NOT " if is_not_pattern else ""
            match_type = "exactly matching" if exact_match else "matching any of patterns"
            logger.info(f"Getting components {log_msg}{match_type} {terms}: {list(matches.keys())}")

        # Try exact match with a base name
        elif any(names == base_name for base_name in base_names.values()):
            # Find all components with this base name
            matches = {
                comp_id: comp
                for comp_id, comp in components.items()
                if (base_names[comp_id] == names) != is_not_pattern
            }

            if is_not_pattern:
                logger.info(f"Getting all components except those with base name '{names}': {list(matches.keys())}")
            else:
                logger.info(f"Getting components with base name '{names}': {list(matches.keys())}")

        # Prefix match (ends with *)
        elif names.endswith("*"):
            prefix = names[:-1]
            matches = {
                comp_id: comp
                for comp_id, comp in components.items()
                if base_names[comp_id].startswith(prefix) != is_not_pattern
            }
            if is_not_pattern:
                logger.info(f"Getting components NOT starting with '{prefix}': {list(matches.keys())}")
            else:
                logger.info(f"Getting components starting with '{prefix}': {list(matches.keys())}")

        # Contains match (starts with *)
        elif names.startswith("*"):
            search = names[1:-1] if names.endswith("*") else names[1:]
            matches = {
                comp_id: comp
                for comp_id, comp in components.items()
                if (search in base_names[comp_id]) != is_not_pattern
            }
            if is_not_pattern:
                logger.info(f"Getting components NOT containing '{search}': {list(matches.keys())}")
            else:
                logger.info(f"Getting components containing '{search}': {list(matches.keys())}")

        # Substring match (no wildcards, but not an exact component name)
        elif any(names in base_name for base_name in base_names.values()):
            matches = {
                comp_id: comp
                for comp_id, comp in components.items()
                if (names in base_names[comp_id]) != is_not_pattern
            }
            if is_not_pattern:
                logger.info(f"Getting components NOT containing '{names}': {list(matches.keys())}")
            else:
                logger.info(f"Getting components containing '{names}': {list(matches.keys())}")

        else:
            raise ValueError(f"Component or pattern '{names}' not found in ComponentsManager")

        if not matches:
            raise ValueError(f"No components found matching pattern '{names}'")

        return get_return_dict(matches, return_dict_with_names)

    # YiYi TODO: (1) add quantization info
    def get_model_info(
        self,
        component_id: str,
        fields: Optional[Union[str, List[str]]] = None,
    ) -> Optional[Dict[str, Any]]:
        """Get comprehensive information about a component.

        Args:
            component_id (str): Name of the component to get info for
            fields (Optional[Union[str, List[str]]]):
                   Field(s) to return. Can be a string for single field or list of fields. If None, uses the
                   available_info_fields setting.

        Returns:
            Dictionary containing requested component metadata. If fields is specified, returns only those fields.
            Otherwise, returns all fields.
        """
        if component_id not in self.components:
            raise ValueError(f"Component '{component_id}' not found in ComponentsManager")

        component = self.components[component_id]

        # Validate fields if specified
        if fields is not None:
            if isinstance(fields, str):
                fields = [fields]
            for field in fields:
                if field not in self._available_info_fields:
                    raise ValueError(f"Field '{field}' not found in available_info_fields")

        # Build complete info dict first
        info = {
            "model_id": component_id,
            "added_time": self.added_time[component_id],
            "collection": ", ".join([coll for coll, comps in self.collections.items() if component_id in comps])
            or None,
        }

        # Additional info for ms.nn.Cell components
        if isinstance(component, ms.nn.Cell):
            # Check for hook information
            has_hook = hasattr(component, "_hf_hook")

            info.update(
                {
                    "class_name": component.__class__.__name__,
                    "size_gb": component.get_memory_footprint() / (1024**3),
                    "adapters": None,  # Default to None
                    "has_hook": has_hook,
                }
            )

            # Get adapters if applicable
            if hasattr(component, "peft_config"):
                info["adapters"] = list(component.peft_config.keys())

            # Check for IP-Adapter scales
            if hasattr(component, "_load_ip_adapter_weights") and hasattr(component, "attn_processors"):
                processors = copy.deepcopy(component.attn_processors)
                # First check if any processor is an IP-Adapter
                processor_types = [v.__class__.__name__ for v in processors.values()]
                if any("IPAdapter" in ptype for ptype in processor_types):
                    # Then get scales only from IP-Adapter processors
                    scales = {
                        k: v.scale
                        for k, v in processors.items()
                        if hasattr(v, "scale") and "IPAdapter" in v.__class__.__name__
                    }
                    if scales:
                        info["ip_adapter"] = summarize_dict_by_value_and_parts(scales)

        # If fields specified, filter info
        if fields is not None:
            return {k: v for k, v in info.items() if k in fields}
        else:
            return info

    # YiYi TODO: (1) add display fields, allow user to set which fields to display in the comnponents table
    def __repr__(self):
        # Handle empty components case
        if not self.components:
            return "Components:\n" + "=" * 50 + "\nNo components registered.\n" + "=" * 50

        # Extract load_id if available
        def get_load_id(component):
            if hasattr(component, "_diffusers_load_id"):
                return component._diffusers_load_id
            return "N/A"

        # Get max length of load_ids for models
        load_ids = [
            get_load_id(component)
            for component in self.components.values()
            if isinstance(component, ms.nn.Cell) and hasattr(component, "_diffusers_load_id")
        ]
        max_load_id_len = max([15] + [len(str(lid)) for lid in load_ids]) if load_ids else 15

        # Get all collections for each component
        component_collections = {}
        for name in self.components.keys():
            component_collections[name] = []
            for coll, comps in self.collections.items():
                if name in comps:
                    component_collections[name].append(coll)
            if not component_collections[name]:
                component_collections[name] = ["N/A"]

        # Find the maximum collection name length
        all_collections = [coll for colls in component_collections.values() for coll in colls]
        max_collection_len = max(10, max(len(str(c)) for c in all_collections)) if all_collections else 10

        col_widths = {
            "id": max(15, max(len(name) for name in self.components.keys())),
            "class": max(25, max(len(component.__class__.__name__) for component in self.components.values())),
            "dtype": 15,
            "size": 10,
            "load_id": max_load_id_len,
            "collection": max_collection_len,
        }

        # Create the header lines
        sep_line = "=" * (sum(col_widths.values()) + len(col_widths) * 3 - 1) + "\n"
        dash_line = "-" * (sum(col_widths.values()) + len(col_widths) * 3 - 1) + "\n"

        output = "Components:\n" + sep_line

        # Separate components into models and others
        models = {k: v for k, v in self.components.items() if isinstance(v, ms.nn.Cell)}
        others = {k: v for k, v in self.components.items() if not isinstance(v, ms.nn.Cell)}

        # Models section
        if models:
            output += "Models:\n" + dash_line
            # Column headers
            output += f"{'Name_ID':<{col_widths['id']}} | {'Class':<{col_widths['class']}} | "
            # output += f"{'Device: act(exec)':<{col_widths['device']}} | {'Dtype':<{col_widths['dtype']}} | "
            output += f"{'Size (GB)':<{col_widths['size']}} | {'Load ID':<{col_widths['load_id']}} | Collection\n"
            output += dash_line

            # Model entries
            for name, component in models.items():
                info = self.get_model_info(name)
                # device_str = format_device(component, info)
                dtype = str(component.dtype) if hasattr(component, "dtype") else "N/A"
                load_id = get_load_id(component)

                # Print first collection on the main line
                first_collection = component_collections[name][0] if component_collections[name] else "N/A"

                output += f"{name:<{col_widths['id']}} | {info['class_name']:<{col_widths['class']}} | "
                # output += f"{device_str:<{col_widths['device']}} | {dtype:<{col_widths['dtype']}} | "
                output += f"{dtype:<{col_widths['dtype']}} | "
                output += f"{info['size_gb']:<{col_widths['size']}.2f} | {load_id:<{col_widths['load_id']}} | {first_collection}\n"

                # Print additional collections on separate lines if they exist
                for i in range(1, len(component_collections[name])):
                    collection = component_collections[name][i]
                    output += f"{'':<{col_widths['id']}} | {'':<{col_widths['class']}} | "
                    # output += f"{'':<{col_widths['device']}} | {'':<{col_widths['dtype']}} | "
                    output += f"{'':<{col_widths['dtype']}} | "
                    output += f"{'':<{col_widths['size']}} | {'':<{col_widths['load_id']}} | {collection}\n"

            output += dash_line

        # Other components section
        if others:
            if models:  # Add extra newline if we had models section
                output += "\n"
            output += "Other Components:\n" + dash_line
            # Column headers for other components
            output += f"{'ID':<{col_widths['id']}} | {'Class':<{col_widths['class']}} | Collection\n"
            output += dash_line

            # Other component entries
            for name, component in others.items():
                info = self.get_model_info(name)

                # Print first collection on the main line
                first_collection = component_collections[name][0] if component_collections[name] else "N/A"

                output += f"{name:<{col_widths['id']}} | {component.__class__.__name__:<{col_widths['class']}} | {first_collection}\n"

                # Print additional collections on separate lines if they exist
                for i in range(1, len(component_collections[name])):
                    collection = component_collections[name][i]
                    output += f"{'':<{col_widths['id']}} | {'':<{col_widths['class']}} | {collection}\n"

            output += dash_line

        # Add additional component info
        output += "\nAdditional Component Info:\n" + "=" * 50 + "\n"
        for name in self.components:
            info = self.get_model_info(name)
            if info is not None and (info.get("adapters") is not None or info.get("ip_adapter")):
                output += f"\n{name}:\n"
                if info.get("adapters") is not None:
                    output += f"  Adapters: {info['adapters']}\n"
                if info.get("ip_adapter"):
                    output += "  IP-Adapter: Enabled\n"

        return output

    def get_one(
        self,
        component_id: Optional[str] = None,
        name: Optional[str] = None,
        collection: Optional[str] = None,
        load_id: Optional[str] = None,
    ) -> Any:
        """
        Get a single component by either:
        - searching name (pattern matching), collection, or load_id.
        - passing in a component_id
        Raises an error if multiple components match or none are found.

        Args:
            component_id (Optional[str]): Optional component ID to get
            name (Optional[str]): Component name or pattern
            collection (Optional[str]): Optional collection to filter by
            load_id (Optional[str]): Optional load_id to filter by

        Returns:
            A single component

        Raises:
            ValueError: If no components match or multiple components match
        """

        if component_id is not None and (name is not None or collection is not None or load_id is not None):
            raise ValueError("If searching by component_id, do not pass name, collection, or load_id")

        # search by component_id
        if component_id is not None:
            if component_id not in self.components:
                raise ValueError(f"Component '{component_id}' not found in ComponentsManager")
            return self.components[component_id]
        # search with name/collection/load_id
        results = self.search_components(name, collection, load_id)

        if not results:
            raise ValueError(f"No components found matching '{name}'")

        if len(results) > 1:
            raise ValueError(f"Multiple components found matching '{name}': {list(results.keys())}")

        return next(iter(results.values()))

    def get_ids(self, names: Union[str, List[str]] = None, collection: Optional[str] = None):
        """
        Get component IDs by a list of names, optionally filtered by collection.

        Args:
            names (Union[str, List[str]]): List of component names
            collection (Optional[str]): Optional collection to filter by

        Returns:
            List[str]: List of component IDs
        """
        ids = set()
        if not isinstance(names, list):
            names = [names]
        for name in names:
            ids.update(self._lookup_ids(name=name, collection=collection))
        return list(ids)

    def get_components_by_ids(self, ids: List[str], return_dict_with_names: Optional[bool] = True):
        """
        Get components by a list of IDs.

        Args:
            ids (List[str]):
                List of component IDs
            return_dict_with_names (Optional[bool]):
                Whether to return a dictionary with component names as keys:

        Returns:
            Dict[str, Any]: Dictionary of components.
                - If return_dict_with_names=True, keys are component names.
                - If return_dict_with_names=False, keys are component IDs.

        Raises:
            ValueError: If duplicate component names are found in the search results when return_dict_with_names=True
        """
        components = {id: self.components[id] for id in ids}

        if return_dict_with_names:
            dict_to_return = {}
            for comp_id, comp in components.items():
                comp_name = self._id_to_name(comp_id)
                if comp_name in dict_to_return:
                    raise ValueError(
                        f"Duplicate component names found in the search results: {comp_name}, please set `return_dict_with_names=False` to return a dictionary with component IDs as keys"  # noqa
                    )
                dict_to_return[comp_name] = comp
            return dict_to_return
        else:
            return components

    def get_components_by_names(self, names: List[str], collection: Optional[str] = None):
        """
        Get components by a list of names, optionally filtered by collection.

        Args:
            names (List[str]): List of component names
            collection (Optional[str]): Optional collection to filter by

        Returns:
            Dict[str, Any]: Dictionary of components with component names as keys

        Raises:
            ValueError: If duplicate component names are found in the search results
        """
        ids = self.get_ids(names, collection)
        return self.get_components_by_ids(ids)

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager.add(name, component, collection=None)

Add a component to the ComponentsManager.

PARAMETER	DESCRIPTION
name	The name of the component TYPE: str
component	The component to add TYPE: Any
collection	The collection to add the component to TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
str	The unique component ID, which is generated as "{name}_{id(component)}" where id(component) is Python's built-in unique identifier for the object

Source code in mindone/diffusers/modular_pipelines/components_manager.py

def add(self, name: str, component: Any, collection: Optional[str] = None):
    """
    Add a component to the ComponentsManager.

    Args:
        name (str): The name of the component
        component (Any): The component to add
        collection (Optional[str]): The collection to add the component to

    Returns:
        str: The unique component ID, which is generated as "{name}_{id(component)}" where
             id(component) is Python's built-in unique identifier for the object
    """
    component_id = f"{name}_{id(component)}"
    is_new_component = True  # noqa

    # check for duplicated components
    for comp_id, comp in self.components.items():
        if comp == component:
            comp_name = self._id_to_name(comp_id)
            if comp_name == name:
                logger.warning(f"ComponentsManager: component '{name}' already exists as '{comp_id}'")
                component_id = comp_id
                is_new_component = False  # noqa
                break
            else:
                logger.warning(
                    f"ComponentsManager: adding component '{name}' as '{component_id}', but it is duplicate of '{comp_id}'"
                    f"To remove a duplicate, call `components_manager.remove('<component_id>')`."
                )

    # check for duplicated load_id and warn (we do not delete for you)
    if hasattr(component, "_diffusers_load_id") and component._diffusers_load_id != "null":
        components_with_same_load_id = self._lookup_ids(load_id=component._diffusers_load_id)
        components_with_same_load_id = [id for id in components_with_same_load_id if id != component_id]

        if components_with_same_load_id:
            existing = ", ".join(components_with_same_load_id)
            logger.warning(
                f"ComponentsManager: adding component '{component_id}', "
                f"but it has duplicate load_id '{component._diffusers_load_id}' with existing components: {existing}. "
                f"To remove a duplicate, call `components_manager.remove('<component_id>')`."
            )

    # add component to components manager
    self.components[component_id] = component
    self.added_time[component_id] = time.time()

    if collection:
        if collection not in self.collections:
            self.collections[collection] = set()
        if component_id not in self.collections[collection]:
            comp_ids_in_collection = self._lookup_ids(name=name, collection=collection)
            for comp_id in comp_ids_in_collection:
                logger.warning(
                    f"ComponentsManager: removing existing {name} from collection '{collection}': {comp_id}"
                )
                # remove existing component from this collection (if it is not in any other collection,
                # will be removed from ComponentsManager)
                self.remove_from_collection(comp_id, collection)

            self.collections[collection].add(component_id)
            logger.info(f"ComponentsManager: added component '{name}' in collection '{collection}': {component_id}")
    else:
        logger.info(f"ComponentsManager: added component '{name}' as '{component_id}'")

    return component_id

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager.get_components_by_ids(ids, return_dict_with_names=True)

Get components by a list of IDs.

PARAMETER	DESCRIPTION
ids	List of component IDs TYPE: List[str]
return_dict_with_names	Whether to return a dictionary with component names as keys: TYPE: Optional[bool] DEFAULT: True

RETURNS	DESCRIPTION
	Dict[str, Any]: Dictionary of components. - If return_dict_with_names=True, keys are component names. - If return_dict_with_names=False, keys are component IDs.

RAISES	DESCRIPTION
ValueError	If duplicate component names are found in the search results when return_dict_with_names=True

Source code in mindone/diffusers/modular_pipelines/components_manager.py

def get_components_by_ids(self, ids: List[str], return_dict_with_names: Optional[bool] = True):
    """
    Get components by a list of IDs.

    Args:
        ids (List[str]):
            List of component IDs
        return_dict_with_names (Optional[bool]):
            Whether to return a dictionary with component names as keys:

    Returns:
        Dict[str, Any]: Dictionary of components.
            - If return_dict_with_names=True, keys are component names.
            - If return_dict_with_names=False, keys are component IDs.

    Raises:
        ValueError: If duplicate component names are found in the search results when return_dict_with_names=True
    """
    components = {id: self.components[id] for id in ids}

    if return_dict_with_names:
        dict_to_return = {}
        for comp_id, comp in components.items():
            comp_name = self._id_to_name(comp_id)
            if comp_name in dict_to_return:
                raise ValueError(
                    f"Duplicate component names found in the search results: {comp_name}, please set `return_dict_with_names=False` to return a dictionary with component IDs as keys"  # noqa
                )
            dict_to_return[comp_name] = comp
        return dict_to_return
    else:
        return components

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager.get_components_by_names(names, collection=None)

Get components by a list of names, optionally filtered by collection.

PARAMETER	DESCRIPTION
names	List of component names TYPE: List[str]
collection	Optional collection to filter by TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
	Dict[str, Any]: Dictionary of components with component names as keys

RAISES	DESCRIPTION
ValueError	If duplicate component names are found in the search results

Source code in mindone/diffusers/modular_pipelines/components_manager.py

def get_components_by_names(self, names: List[str], collection: Optional[str] = None):
    """
    Get components by a list of names, optionally filtered by collection.

    Args:
        names (List[str]): List of component names
        collection (Optional[str]): Optional collection to filter by

    Returns:
        Dict[str, Any]: Dictionary of components with component names as keys

    Raises:
        ValueError: If duplicate component names are found in the search results
    """
    ids = self.get_ids(names, collection)
    return self.get_components_by_ids(ids)

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager.get_ids(names=None, collection=None)

Get component IDs by a list of names, optionally filtered by collection.

PARAMETER	DESCRIPTION
names	List of component names TYPE: Union[str, List[str]] DEFAULT: None
collection	Optional collection to filter by TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
	List[str]: List of component IDs

Source code in mindone/diffusers/modular_pipelines/components_manager.py

def get_ids(self, names: Union[str, List[str]] = None, collection: Optional[str] = None):
    """
    Get component IDs by a list of names, optionally filtered by collection.

    Args:
        names (Union[str, List[str]]): List of component names
        collection (Optional[str]): Optional collection to filter by

    Returns:
        List[str]: List of component IDs
    """
    ids = set()
    if not isinstance(names, list):
        names = [names]
    for name in names:
        ids.update(self._lookup_ids(name=name, collection=collection))
    return list(ids)

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager.get_model_info(component_id, fields=None)

Get comprehensive information about a component.

PARAMETER	DESCRIPTION
component_id	Name of the component to get info for TYPE: str
fields	Field(s) to return. Can be a string for single field or list of fields. If None, uses the available_info_fields setting. TYPE: Optional[Union[str, List[str]]] DEFAULT: None

RETURNS	DESCRIPTION
Optional[Dict[str, Any]]	Dictionary containing requested component metadata. If fields is specified, returns only those fields.
Optional[Dict[str, Any]]	Otherwise, returns all fields.

Source code in mindone/diffusers/modular_pipelines/components_manager.py

def get_model_info(
    self,
    component_id: str,
    fields: Optional[Union[str, List[str]]] = None,
) -> Optional[Dict[str, Any]]:
    """Get comprehensive information about a component.

    Args:
        component_id (str): Name of the component to get info for
        fields (Optional[Union[str, List[str]]]):
               Field(s) to return. Can be a string for single field or list of fields. If None, uses the
               available_info_fields setting.

    Returns:
        Dictionary containing requested component metadata. If fields is specified, returns only those fields.
        Otherwise, returns all fields.
    """
    if component_id not in self.components:
        raise ValueError(f"Component '{component_id}' not found in ComponentsManager")

    component = self.components[component_id]

    # Validate fields if specified
    if fields is not None:
        if isinstance(fields, str):
            fields = [fields]
        for field in fields:
            if field not in self._available_info_fields:
                raise ValueError(f"Field '{field}' not found in available_info_fields")

    # Build complete info dict first
    info = {
        "model_id": component_id,
        "added_time": self.added_time[component_id],
        "collection": ", ".join([coll for coll, comps in self.collections.items() if component_id in comps])
        or None,
    }

    # Additional info for ms.nn.Cell components
    if isinstance(component, ms.nn.Cell):
        # Check for hook information
        has_hook = hasattr(component, "_hf_hook")

        info.update(
            {
                "class_name": component.__class__.__name__,
                "size_gb": component.get_memory_footprint() / (1024**3),
                "adapters": None,  # Default to None
                "has_hook": has_hook,
            }
        )

        # Get adapters if applicable
        if hasattr(component, "peft_config"):
            info["adapters"] = list(component.peft_config.keys())

        # Check for IP-Adapter scales
        if hasattr(component, "_load_ip_adapter_weights") and hasattr(component, "attn_processors"):
            processors = copy.deepcopy(component.attn_processors)
            # First check if any processor is an IP-Adapter
            processor_types = [v.__class__.__name__ for v in processors.values()]
            if any("IPAdapter" in ptype for ptype in processor_types):
                # Then get scales only from IP-Adapter processors
                scales = {
                    k: v.scale
                    for k, v in processors.items()
                    if hasattr(v, "scale") and "IPAdapter" in v.__class__.__name__
                }
                if scales:
                    info["ip_adapter"] = summarize_dict_by_value_and_parts(scales)

    # If fields specified, filter info
    if fields is not None:
        return {k: v for k, v in info.items() if k in fields}
    else:
        return info

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager.get_one(component_id=None, name=None, collection=None, load_id=None)

Get a single component by either: - searching name (pattern matching), collection, or load_id. - passing in a component_id Raises an error if multiple components match or none are found.

PARAMETER	DESCRIPTION
component_id	Optional component ID to get TYPE: Optional[str] DEFAULT: None
name	Component name or pattern TYPE: Optional[str] DEFAULT: None
collection	Optional collection to filter by TYPE: Optional[str] DEFAULT: None
load_id	Optional load_id to filter by TYPE: Optional[str] DEFAULT: None

RETURNS	DESCRIPTION
Any	A single component

RAISES	DESCRIPTION
ValueError	If no components match or multiple components match

Source code in mindone/diffusers/modular_pipelines/components_manager.py

def get_one(
    self,
    component_id: Optional[str] = None,
    name: Optional[str] = None,
    collection: Optional[str] = None,
    load_id: Optional[str] = None,
) -> Any:
    """
    Get a single component by either:
    - searching name (pattern matching), collection, or load_id.
    - passing in a component_id
    Raises an error if multiple components match or none are found.

    Args:
        component_id (Optional[str]): Optional component ID to get
        name (Optional[str]): Component name or pattern
        collection (Optional[str]): Optional collection to filter by
        load_id (Optional[str]): Optional load_id to filter by

    Returns:
        A single component

    Raises:
        ValueError: If no components match or multiple components match
    """

    if component_id is not None and (name is not None or collection is not None or load_id is not None):
        raise ValueError("If searching by component_id, do not pass name, collection, or load_id")

    # search by component_id
    if component_id is not None:
        if component_id not in self.components:
            raise ValueError(f"Component '{component_id}' not found in ComponentsManager")
        return self.components[component_id]
    # search with name/collection/load_id
    results = self.search_components(name, collection, load_id)

    if not results:
        raise ValueError(f"No components found matching '{name}'")

    if len(results) > 1:
        raise ValueError(f"Multiple components found matching '{name}': {list(results.keys())}")

    return next(iter(results.values()))

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager.remove(component_id=None)

Remove a component from the ComponentsManager.

PARAMETER	DESCRIPTION
component_id	The ID of the component to remove TYPE: str DEFAULT: None

Source code in mindone/diffusers/modular_pipelines/components_manager.py

def remove(self, component_id: str = None):
    """
    Remove a component from the ComponentsManager.

    Args:
        component_id (str): The ID of the component to remove
    """
    if component_id not in self.components:
        logger.warning(f"Component '{component_id}' not found in ComponentsManager")
        return

    component = self.components.pop(component_id)  # noqa
    self.added_time.pop(component_id)

    for collection in self.collections:
        if component_id in self.collections[collection]:
            self.collections[collection].remove(component_id)

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager.remove_from_collection(component_id, collection)

Remove a component from a collection.

Source code in mindone/diffusers/modular_pipelines/components_manager.py

def remove_from_collection(self, component_id: str, collection: str):
    """
    Remove a component from a collection.
    """
    if collection not in self.collections:
        logger.warning(f"Collection '{collection}' not found in ComponentsManager")
        return
    if component_id not in self.collections[collection]:
        logger.warning(f"Component '{component_id}' not found in collection '{collection}'")
        return
    # remove from the collection
    self.collections[collection].remove(component_id)
    # check if this component is in any other collection
    comp_colls = [coll for coll, comps in self.collections.items() if component_id in comps]
    if not comp_colls:  # only if no other collection contains this component, remove it
        logger.warning(f"ComponentsManager: removing component '{component_id}' from ComponentsManager")
        self.remove(component_id)

mindone.diffusers.modular_pipelines.components_manager.ComponentsManager.search_components(names=None, collection=None, load_id=None, return_dict_with_names=True)

Search components by name with simple pattern matching. Optionally filter by collection or load_id.

PARAMETER	DESCRIPTION
names	Component name(s) or pattern(s) Patterns: - "unet" : match any component with base name "unet" (e.g., unet_123abc) - "!unet" : everything except components with base name "unet" - "unet*" : anything with base name starting with "unet" - "!unet*" : anything with base name NOT starting with "unet" - "unet" : anything with base name containing "unet" - "!unet" : anything with base name NOT containing "unet" - "refiner|vae|unet" : anything with base name exactly matching "refiner", "vae", or "unet" - "!refiner|vae|unet" : anything with base name NOT exactly matching "refiner", "vae", or "unet" - "unet*|vae*" : anything with base name starting with "unet" OR starting with "vae" TYPE: Optional[str] DEFAULT: None
collection	Optional collection to filter by TYPE: Optional[str] DEFAULT: None
load_id	Optional load_id to filter by TYPE: Optional[str] DEFAULT: None
return_dict_with_names	If True, returns a dictionary with component names as keys, throw an error if multiple components with the same name are found If False, returns a dictionary with component IDs as keys TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
	Dictionary mapping component names to components if return_dict_with_names=True, or a dictionary mapping
	component IDs to components if return_dict_with_names=False

Source code in mindone/diffusers/modular_pipelines/components_manager.py

def search_components(
    self,
    names: Optional[str] = None,
    collection: Optional[str] = None,
    load_id: Optional[str] = None,
    return_dict_with_names: bool = True,
):
    """
    Search components by name with simple pattern matching. Optionally filter by collection or load_id.

    Args:
        names: Component name(s) or pattern(s)
            Patterns:
            - "unet" : match any component with base name "unet" (e.g., unet_123abc)
            - "!unet" : everything except components with base name "unet"
            - "unet*" : anything with base name starting with "unet"
            - "!unet*" : anything with base name NOT starting with "unet"
            - "*unet*" : anything with base name containing "unet"
            - "!*unet*" : anything with base name NOT containing "unet"
            - "refiner|vae|unet" : anything with base name exactly matching "refiner", "vae", or "unet"
            - "!refiner|vae|unet" : anything with base name NOT exactly matching "refiner", "vae", or "unet"
            - "unet*|vae*" : anything with base name starting with "unet" OR starting with "vae"
        collection: Optional collection to filter by
        load_id: Optional load_id to filter by
        return_dict_with_names:
                                If True, returns a dictionary with component names as keys, throw an error if
                                multiple components with the same name are found If False, returns a dictionary
                                with component IDs as keys

    Returns:
        Dictionary mapping component names to components if return_dict_with_names=True, or a dictionary mapping
        component IDs to components if return_dict_with_names=False
    """

    # select components based on collection and load_id filters
    selected_ids = self._lookup_ids(collection=collection, load_id=load_id)
    components = {k: self.components[k] for k in selected_ids}

    def get_return_dict(components, return_dict_with_names):
        """
        Create a dictionary mapping component names to components if return_dict_with_names=True, or a dictionary
        mapping component IDs to components if return_dict_with_names=False, throw an error if duplicate component
        names are found when return_dict_with_names=True
        """
        if return_dict_with_names:
            dict_to_return = {}
            for comp_id, comp in components.items():
                comp_name = self._id_to_name(comp_id)
                if comp_name in dict_to_return:
                    raise ValueError(
                        f"Duplicate component names found in the search results: {comp_name}, please set `return_dict_with_names=False` to return a dictionary with component IDs as keys"  # noqa
                    )
                dict_to_return[comp_name] = comp
            return dict_to_return
        else:
            return components

    # if no names are provided, return the filtered components as it is
    if names is None:
        return get_return_dict(components, return_dict_with_names)

    # if names is not a string, raise an error
    elif not isinstance(names, str):
        raise ValueError(f"Invalid type for `names: {type(names)}, only support string")

    # Create mapping from component_id to base_name for components to be used for pattern matching
    base_names = {comp_id: self._id_to_name(comp_id) for comp_id in components.keys()}

    # Helper function to check if a component matches a pattern based on its base name
    def matches_pattern(component_id, pattern, exact_match=False):
        """
        Helper function to check if a component matches a pattern based on its base name.

        Args:
            component_id: The component ID to check
            pattern: The pattern to match against
            exact_match: If True, only exact matches to base_name are considered
        """
        base_name = base_names[component_id]

        # Exact match with base name
        if exact_match:
            return pattern == base_name

        # Prefix match (ends with *)
        elif pattern.endswith("*"):
            prefix = pattern[:-1]
            return base_name.startswith(prefix)

        # Contains match (starts with *)
        elif pattern.startswith("*"):
            search = pattern[1:-1] if pattern.endswith("*") else pattern[1:]
            return search in base_name

        # Exact match (no wildcards)
        else:
            return pattern == base_name

    # Check if this is a "not" pattern
    is_not_pattern = names.startswith("!")
    if is_not_pattern:
        names = names[1:]  # Remove the ! prefix

    # Handle OR patterns (containing |)
    if "|" in names:
        terms = names.split("|")
        matches = {}

        for comp_id, comp in components.items():
            # For OR patterns with exact names (no wildcards), we do exact matching on base names
            exact_match = all(not (term.startswith("*") or term.endswith("*")) for term in terms)

            # Check if any of the terms match this component
            should_include = any(matches_pattern(comp_id, term, exact_match) for term in terms)

            # Flip the decision if this is a NOT pattern
            if is_not_pattern:
                should_include = not should_include

            if should_include:
                matches[comp_id] = comp

        log_msg = "NOT " if is_not_pattern else ""
        match_type = "exactly matching" if exact_match else "matching any of patterns"
        logger.info(f"Getting components {log_msg}{match_type} {terms}: {list(matches.keys())}")

    # Try exact match with a base name
    elif any(names == base_name for base_name in base_names.values()):
        # Find all components with this base name
        matches = {
            comp_id: comp
            for comp_id, comp in components.items()
            if (base_names[comp_id] == names) != is_not_pattern
        }

        if is_not_pattern:
            logger.info(f"Getting all components except those with base name '{names}': {list(matches.keys())}")
        else:
            logger.info(f"Getting components with base name '{names}': {list(matches.keys())}")

    # Prefix match (ends with *)
    elif names.endswith("*"):
        prefix = names[:-1]
        matches = {
            comp_id: comp
            for comp_id, comp in components.items()
            if base_names[comp_id].startswith(prefix) != is_not_pattern
        }
        if is_not_pattern:
            logger.info(f"Getting components NOT starting with '{prefix}': {list(matches.keys())}")
        else:
            logger.info(f"Getting components starting with '{prefix}': {list(matches.keys())}")

    # Contains match (starts with *)
    elif names.startswith("*"):
        search = names[1:-1] if names.endswith("*") else names[1:]
        matches = {
            comp_id: comp
            for comp_id, comp in components.items()
            if (search in base_names[comp_id]) != is_not_pattern
        }
        if is_not_pattern:
            logger.info(f"Getting components NOT containing '{search}': {list(matches.keys())}")
        else:
            logger.info(f"Getting components containing '{search}': {list(matches.keys())}")

    # Substring match (no wildcards, but not an exact component name)
    elif any(names in base_name for base_name in base_names.values()):
        matches = {
            comp_id: comp
            for comp_id, comp in components.items()
            if (names in base_names[comp_id]) != is_not_pattern
        }
        if is_not_pattern:
            logger.info(f"Getting components NOT containing '{names}': {list(matches.keys())}")
        else:
            logger.info(f"Getting components containing '{names}': {list(matches.keys())}")

    else:
        raise ValueError(f"Component or pattern '{names}' not found in ComponentsManager")

    if not matches:
        raise ValueError(f"No components found matching pattern '{names}'")

    return get_return_dict(matches, return_dict_with_names)

mindone.diffusers.modular_pipelines.modular_pipeline_utils.InsertableDict

Bases: OrderedDict

Source code in mindone/diffusers/modular_pipelines/modular_pipeline_utils.py

class InsertableDict(OrderedDict):
    def insert(self, key, value, index):
        items = list(self.items())

        # Remove key if it already exists to avoid duplicates
        items = [(k, v) for k, v in items if k != key]

        # Insert at the specified index
        items.insert(index, (key, value))

        # Clear and update self
        self.clear()
        self.update(items)

        # Return self for method chaining
        return self

    def __repr__(self):
        if not self:
            return "InsertableDict()"

        items = []
        for i, (key, value) in enumerate(self.items()):
            if isinstance(value, type):
                # For classes, show class name and <class ...>
                obj_repr = f"<class '{value.__module__}.{value.__name__}'>"
            else:
                # For objects (instances) and other types, show class name and module
                obj_repr = f"<obj '{value.__class__.__module__}.{value.__class__.__name__}'>"
            items.append(f"{i}: ({repr(key)}, {obj_repr})")

        return "InsertableDict([\n  " + ",\n  ".join(items) + "\n])"