Pipeline

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline

Bases: ConfigMixin, PushToHubMixin

Base class for all Modular pipelines.

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

PARAMETER	DESCRIPTION
blocks	ModularPipelineBlocks, the blocks to be used in the pipeline TYPE: Optional[ModularPipelineBlocks] DEFAULT: None

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

class ModularPipeline(ConfigMixin, PushToHubMixin):
    """
    Base class for all Modular pipelines.

    <Tip warning={true}>

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

    </Tip>

    Args:
        blocks: ModularPipelineBlocks, the blocks to be used in the pipeline
    """

    config_name = "modular_model_index.json"
    hf_device_map = None

    # YiYi TODO: add warning for passing multiple ComponentSpec/ConfigSpec with the same name
    def __init__(
        self,
        blocks: Optional[ModularPipelineBlocks] = None,
        pretrained_model_name_or_path: Optional[Union[str, os.PathLike]] = None,
        components_manager: Optional[ComponentsManager] = None,
        collection: Optional[str] = None,
        **kwargs,
    ):
        """
        Initialize a ModularPipeline instance.

        This method sets up the pipeline by:
        - creating default pipeline blocks if not provided
        - gather component and config specifications based on the pipeline blocks's requirement (e.g.
           expected_components, expected_configs)
        - update the loading specs of from_pretrained components based on the modular_model_index.json file from
           huggingface hub if `pretrained_model_name_or_path` is provided
        - create defaultfrom_config components and register everything

        Args:
            blocks: `ModularPipelineBlocks` instance. If None, will attempt to load
                   default blocks based on the pipeline class name.
            pretrained_model_name_or_path: Path to a pretrained pipeline configuration. If provided,
                    will load component specs (only for from_pretrained components) and config values from the saved
                    modular_model_index.json file.
            components_manager:
                Optional ComponentsManager for managing multiple component cross different pipelines and apply
                offloading strategies.
            collection: Optional collection name for organizing components in the ComponentsManager.
            **kwargs: Additional arguments passed to `load_config()` when loading pretrained configuration.

        Examples:
            ```python
            # Initialize with custom blocks
            pipeline = ModularPipeline(blocks=my_custom_blocks)

            # Initialize from pretrained configuration
            pipeline = ModularPipeline(blocks=my_blocks, pretrained_model_name_or_path="my-repo/modular-pipeline")

            # Initialize with components manager
            pipeline = ModularPipeline(
                blocks=my_blocks, components_manager=ComponentsManager(), collection="my_collection"
            )
            ```

        Notes:
            - If blocks is None, the method will try to find default blocks based on the pipeline class name
            - Components with default_creation_method="from_config" are created immediately, its specs are not included
              in config dict and will not be saved in `modular_model_index.json`
            - Components with default_creation_method="from_pretrained" are set to None and can be loaded later with
              `load_default_components()`/`load_components()`
            - The pipeline's config dict is populated with component specs (only for from_pretrained components) and
              config values, which will be saved as `modular_model_index.json` during `save_pretrained`
            - The pipeline's config dict is also used to store the pipeline blocks's class name, which will be saved as
              `_blocks_class_name` in the config dict
        """
        if blocks is None:
            blocks_class_name = MODULAR_PIPELINE_BLOCKS_MAPPING.get(self.__class__.__name__)
            if blocks_class_name is not None:
                diffusers_module = importlib.import_module("mindone.diffusers")
                blocks_class = getattr(diffusers_module, blocks_class_name)
                blocks = blocks_class()
            else:
                logger.warning(f"`blocks` is `None`, no default blocks class found for {self.__class__.__name__}")

        self.blocks = blocks
        self._components_manager = components_manager
        self._collection = collection
        self._component_specs = {spec.name: deepcopy(spec) for spec in self.blocks.expected_components}
        self._config_specs = {spec.name: deepcopy(spec) for spec in self.blocks.expected_configs}

        # update component_specs and config_specs from modular_repo
        if pretrained_model_name_or_path is not None:
            config_dict = self.load_config(pretrained_model_name_or_path, **kwargs)

            for name, value in config_dict.items():
                # all the components in modular_model_index.json are from_pretrained components
                if name in self._component_specs and isinstance(value, (tuple, list)) and len(value) == 3:
                    library, class_name, component_spec_dict = value
                    component_spec = self._dict_to_component_spec(name, component_spec_dict)
                    component_spec.default_creation_method = "from_pretrained"
                    self._component_specs[name] = component_spec

                elif name in self._config_specs:
                    self._config_specs[name].default = value

        register_components_dict = {}
        for name, component_spec in self._component_specs.items():
            if component_spec.default_creation_method == "from_config":
                component = component_spec.create()
            else:
                component = None
            register_components_dict[name] = component
        self.register_components(**register_components_dict)

        default_configs = {}
        for name, config_spec in self._config_specs.items():
            default_configs[name] = config_spec.default
        self.register_to_config(**default_configs)

        self.register_to_config(_blocks_class_name=self.blocks.__class__.__name__ if self.blocks is not None else None)

    @property
    def default_call_parameters(self) -> Dict[str, Any]:
        """
        Returns:
            - Dictionary mapping input names to their default values
        """
        params = {}
        for input_param in self.blocks.inputs:
            params[input_param.name] = input_param.default
        return params

    def load_default_components(self, **kwargs):
        """
        Load from_pretrained components using the loading specs in the config dict.

        Args:
            **kwargs: Additional arguments passed to `from_pretrained` method, e.g. mindspore_dtype, cache_dir, etc.
        """
        names = [
            name
            for name in self._component_specs.keys()
            if self._component_specs[name].default_creation_method == "from_pretrained"
        ]
        self.load_components(names=names, **kwargs)

    @classmethod
    @validate_hf_hub_args
    def from_pretrained(
        cls,
        pretrained_model_name_or_path: Optional[Union[str, os.PathLike]],
        trust_remote_code: Optional[bool] = None,
        components_manager: Optional[ComponentsManager] = None,
        collection: Optional[str] = None,
        **kwargs,
    ):
        """
        Load a ModularPipeline from a huggingface hub repo.

        Args:
            pretrained_model_name_or_path (`str` or `os.PathLike`, optional):
                Path to a pretrained pipeline configuration. If provided, will load component specs (only for
                from_pretrained components) and config values from the modular_model_index.json file.
            trust_remote_code (`bool`, optional):
                Whether to trust remote code when loading the pipeline, need to be set to True if you want to create
                pipeline blocks based on the custom code in `pretrained_model_name_or_path`
            components_manager (`ComponentsManager`, optional):
                ComponentsManager instance for managing multiple component cross different pipelines and apply
                offloading strategies.
            collection (`str`, optional):`
                Collection name for organizing components in the ComponentsManager.
        """
        from ..pipelines.pipeline_loading_utils import _get_pipeline_class

        try:
            blocks = ModularPipelineBlocks.from_pretrained(
                pretrained_model_name_or_path, trust_remote_code=trust_remote_code, **kwargs
            )
        except EnvironmentError:
            blocks = None

        cache_dir = kwargs.pop("cache_dir", None)
        force_download = kwargs.pop("force_download", False)
        proxies = kwargs.pop("proxies", None)
        token = kwargs.pop("token", None)
        local_files_only = kwargs.pop("local_files_only", False)
        revision = kwargs.pop("revision", None)

        load_config_kwargs = {
            "cache_dir": cache_dir,
            "force_download": force_download,
            "proxies": proxies,
            "token": token,
            "local_files_only": local_files_only,
            "revision": revision,
        }

        try:
            config_dict = cls.load_config(pretrained_model_name_or_path, **load_config_kwargs)
            pipeline_class = _get_pipeline_class(cls, config=config_dict)
        except EnvironmentError:
            pipeline_class = cls
            pretrained_model_name_or_path = None

        pipeline = pipeline_class(
            blocks=blocks,
            pretrained_model_name_or_path=pretrained_model_name_or_path,
            components_manager=components_manager,
            collection=collection,
            **kwargs,
        )
        return pipeline

    def save_pretrained(self, save_directory: Union[str, os.PathLike], push_to_hub: bool = False, **kwargs):
        """
        Save the pipeline to a directory. It does not save components, you need to save them separately.

        Args:
            save_directory (`str` or `os.PathLike`):
                Path to the directory where the pipeline will be saved.
            push_to_hub (`bool`, optional):
                Whether to push the pipeline to the huggingface hub.
            **kwargs: Additional arguments passed to `save_config()` method
        """
        if push_to_hub:
            commit_message = kwargs.pop("commit_message", None)
            private = kwargs.pop("private", None)
            create_pr = kwargs.pop("create_pr", False)
            token = kwargs.pop("token", None)
            repo_id = kwargs.pop("repo_id", save_directory.split(os.path.sep)[-1])
            repo_id = create_repo(repo_id, exist_ok=True, private=private, token=token).repo_id

            # Create a new empty model card and eventually tag it
            model_card = load_or_create_model_card(repo_id, token=token, is_pipeline=True)
            model_card = populate_model_card(model_card)
            model_card.save(os.path.join(save_directory, "README.md"))

        # YiYi TODO: maybe order the json file to make it more readable: configs first, then components
        self.save_config(save_directory=save_directory)

        if push_to_hub:
            self._upload_folder(
                save_directory,
                repo_id,
                token=token,
                commit_message=commit_message,
                create_pr=create_pr,
            )

    @property
    def doc(self):
        """
        Returns:
            - The docstring of the pipeline blocks
        """
        return self.blocks.doc

    def register_components(self, **kwargs):
        """
        Register components with their corresponding specifications.

        This method is responsible for:
        1. Sets component objects as attributes on the loader (e.g., self.unet = unet)
        2. Updates the config dict, which will be saved as `modular_model_index.json` during `save_pretrained` (only
           for from_pretrained components)
        3. Adds components to the component manager if one is attached (only for from_pretrained components)

        This method is called when:
        - Components are first initialized in __init__:
           - from_pretrained components not loaded during __init__ so they are registered as None;
           - non from_pretrained components are created during __init__ and registered as the object itself
        - Components are updated with the `update_components()` method: e.g. loader.update_components(unet=unet) or
          loader.update_components(guider=guider_spec)
        - (from_pretrained) Components are loaded with the `load_default_components()` method: e.g.
          loader.load_default_components(names=["unet"])

        Args:
            **kwargs: Keyword arguments where keys are component names and values are component objects.
                      E.g., register_components(unet=unet_model, text_encoder=encoder_model)

        Notes:
            - When registering None for a component, it sets attribute to None but still syncs specs with the config
              dict, which will be saved as `modular_model_index.json` during `save_pretrained`
            - component_specs are updated to match the new component outside of this method, e.g. in
              `update_components()` method
        """
        for name, module in kwargs.items():
            # current component spec
            component_spec = self._component_specs.get(name)
            if component_spec is None:
                logger.warning(f"ModularPipeline.register_components: skipping unknown component '{name}'")
                continue

            # check if it is the first time registration, i.e. calling from __init__
            is_registered = hasattr(self, name)
            is_from_pretrained = component_spec.default_creation_method == "from_pretrained"

            if module is not None:
                # actual library and class name of the module
                library, class_name = _fetch_class_library_tuple(module)  # e.g. ("diffusers", "UNet2DConditionModel")
            else:
                # if module is None, e.g. self.register_components(unet=None) during __init__
                # we do not update the spec,
                # but we still need to update the modular_model_index.json config based on component spec
                library, class_name = None, None

            # extract the loading spec from the updated component spec that'll be used as part of modular_model_index.json config
            # e.g. {"repo": "stabilityai/stable-diffusion-2-1",
            #       "type_hint": ("diffusers", "UNet2DConditionModel"),
            #       "subfolder": "unet",
            #       "variant": None,
            #       "revision": None}
            component_spec_dict = self._component_spec_to_dict(component_spec)

            register_dict = {name: (library, class_name, component_spec_dict)}

            # set the component as attribute
            # if it is not set yet, just set it and skip the process to check and warn below
            if not is_registered:
                if is_from_pretrained:
                    self.register_to_config(**register_dict)
                setattr(self, name, module)
                if module is not None and is_from_pretrained and self._components_manager is not None:
                    self._components_manager.add(name, module, self._collection)
                continue

            current_module = getattr(self, name, None)
            # skip if the component is already registered with the same object
            if current_module is module:
                logger.info(
                    f"ModularPipeline.register_components: {name} is already registered with same object, skipping"
                )
                continue

            # warn if unregister
            if current_module is not None and module is None:
                logger.info(
                    f"ModularPipeline.register_components: setting '{name}' to None "
                    f"(was {current_module.__class__.__name__})"
                )
            # same type, new instance → replace but send debug log
            elif (
                current_module is not None
                and module is not None
                and isinstance(module, current_module.__class__)
                and current_module != module
            ):
                logger.debug(
                    f"ModularPipeline.register_components: replacing existing '{name}' "
                    f"(same type {type(current_module).__name__}, new instance)"
                )

            # update modular_model_index.json config
            if is_from_pretrained:
                self.register_to_config(**register_dict)
            # finally set models
            setattr(self, name, module)
            # add to component manager if one is attached
            if module is not None and is_from_pretrained and self._components_manager is not None:
                self._components_manager.add(name, module, self._collection)

    @property
    def dtype(self) -> ms.Type:
        r"""
        Returns:
            `ms.Type`: The mindspore dtype on which the pipeline is located.
        """
        modules = self.components.values()
        modules = [m for m in modules if isinstance(m, ms.nn.Cell)]

        for module in modules:
            return module.dtype

        return ms.float32

    @property
    def null_component_names(self) -> List[str]:
        """
        Returns:
            - List of names for components that needs to be loaded
        """
        return [name for name in self._component_specs.keys() if hasattr(self, name) and getattr(self, name) is None]

    @property
    def component_names(self) -> List[str]:
        """
        Returns:
            - List of names for all components
        """
        return list(self.components.keys())

    @property
    def pretrained_component_names(self) -> List[str]:
        """
        Returns:
            - List of names for from_pretrained components
        """
        return [
            name
            for name in self._component_specs.keys()
            if self._component_specs[name].default_creation_method == "from_pretrained"
        ]

    @property
    def config_component_names(self) -> List[str]:
        """
        Returns:
            - List of names for from_config components
        """
        return [
            name
            for name in self._component_specs.keys()
            if self._component_specs[name].default_creation_method == "from_config"
        ]

    @property
    def components(self) -> Dict[str, Any]:
        """
        Returns:
            - Dictionary mapping component names to their objects (include both from_pretrained and from_config
              components)
        """
        # return only components we've actually set as attributes on self
        return {name: getattr(self, name) for name in self._component_specs.keys() if hasattr(self, name)}

    def get_component_spec(self, name: str) -> ComponentSpec:
        """
        Returns:
            - a copy of the ComponentSpec object for the given component name
        """
        return deepcopy(self._component_specs[name])

    def update_components(self, **kwargs):
        """
        Update components and configuration values and specs after the pipeline has been instantiated.

        This method allows you to:
        1. Replace existing components with new ones (e.g., updating `self.unet` or `self.text_encoder`)
        2. Update configuration values (e.g., changing `self.requires_safety_checker` flag)

        In addition to updating the components and configuration values as pipeline attributes, the method also
        updates:
        - the corresponding specs in `_component_specs` and `_config_specs`
        - the `config` dict, which will be saved as `modular_model_index.json` during `save_pretrained`

        Args:
            **kwargs: Component objects, ComponentSpec objects, or configuration values to update:
                - Component objects: Only supports components we can extract specs using
                  `ComponentSpec.from_component()` method i.e. components created with ComponentSpec.load() or
                  ConfigMixin subclasses that aren't nn.Modules (e.g., `unet=new_unet, text_encoder=new_encoder`)
                - ComponentSpec objects: Only supports default_creation_method == "from_config", will call create()
                  method to create a new component (e.g., `guider=ComponentSpec(name="guider",
                  type_hint=ClassifierFreeGuidance, config={...}, default_creation_method="from_config")`)
                - Configuration values: Simple values to update configuration settings (e.g.,
                  `requires_safety_checker=False`)

        Raises:
            ValueError: If a component object is not supported in ComponentSpec.from_component() method:
                - nn.Module components without a valid `_diffusers_load_id` attribute
                - Non-ConfigMixin components without a valid `_diffusers_load_id` attribute

        Examples:
            ```python
            # Update multiple components at once
            pipeline.update_components(unet=new_unet_model, text_encoder=new_text_encoder)

            # Update configuration values
            pipeline.update_components(requires_safety_checker=False)

            # Update both components and configs together
            pipeline.update_components(unet=new_unet_model, requires_safety_checker=False)

            # Update with ComponentSpec objects (from_config only)
            pipeline.update_components(
                guider=ComponentSpec(
                    name="guider",
                    type_hint=ClassifierFreeGuidance,
                    config={"guidance_scale": 5.0},
                    default_creation_method="from_config",
                )
            )
            ```

        Notes:
            - Components with trained weights must be created using ComponentSpec.load(). If the component has not been
              shared in huggingface hub and you don't have loading specs, you can upload it using `push_to_hub()`
            - ConfigMixin objects without weights (e.g., schedulers, guiders) can be passed directly
            - ComponentSpec objects with default_creation_method="from_pretrained" are not supported in
              update_components()
        """

        # extract component_specs_updates & config_specs_updates from `specs`
        passed_component_specs = {
            k: kwargs.pop(k) for k in self._component_specs if k in kwargs and isinstance(kwargs[k], ComponentSpec)
        }
        passed_components = {
            k: kwargs.pop(k) for k in self._component_specs if k in kwargs and not isinstance(kwargs[k], ComponentSpec)
        }
        passed_config_values = {k: kwargs.pop(k) for k in self._config_specs if k in kwargs}

        for name, component in passed_components.items():
            current_component_spec = self._component_specs[name]

            # warn if type changed
            if current_component_spec.type_hint is not None and not isinstance(
                component, current_component_spec.type_hint
            ):
                logger.warning(
                    f"ModularPipeline.update_components: adding {name} with new type: {component.__class__.__name__}, previous type: {current_component_spec.type_hint.__name__}"  # noqa
                )
            # update _component_specs based on the new component
            new_component_spec = ComponentSpec.from_component(name, component)
            if new_component_spec.default_creation_method != current_component_spec.default_creation_method:
                logger.warning(
                    f"ModularPipeline.update_components: changing the default_creation_method of {name} from {current_component_spec.default_creation_method} to {new_component_spec.default_creation_method}."  # noqa
                )

            self._component_specs[name] = new_component_spec

        if len(kwargs) > 0:
            logger.warning(f"Unexpected keyword arguments, will be ignored: {kwargs.keys()}")

        created_components = {}
        for name, component_spec in passed_component_specs.items():
            if component_spec.default_creation_method == "from_pretrained":
                raise ValueError(
                    "ComponentSpec object with default_creation_method == 'from_pretrained' is not supported in update_components() method"
                )
            created_components[name] = component_spec.create()
            current_component_spec = self._component_specs[name]
            # warn if type changed
            if current_component_spec.type_hint is not None and not isinstance(
                created_components[name], current_component_spec.type_hint
            ):
                logger.warning(
                    f"ModularPipeline.update_components: adding {name} with new type: {created_components[name].__class__.__name__}, previous type: {current_component_spec.type_hint.__name__}"  # noqa
                )
            # update _component_specs based on the user passed component_spec
            self._component_specs[name] = component_spec
        self.register_components(**passed_components, **created_components)

        config_to_register = {}
        for name, new_value in passed_config_values.items():
            # e.g. requires_aesthetics_score = False
            self._config_specs[name].default = new_value
            config_to_register[name] = new_value
        self.register_to_config(**config_to_register)

    # YiYi TODO: support map for additional from_pretrained kwargs
    # YiYi/Dhruv TODO: consolidate load_components and load_default_components?
    def load_components(self, names: Union[List[str], str], **kwargs):
        """
        Load selected components from specs.

        Args:
            names: List of component names to load; by default will not load any components
            **kwargs: additional kwargs to be passed to `from_pretrained()`.Can be:
             - a single value to be applied to all components to be loaded, e.g. mindspore_dtype=ms.bfloat16
             - a dict, e.g. mindspore_dtype={"unet": ms.bfloat16, "default": ms.float32}
             - if potentially override ComponentSpec if passed a different loading field in kwargs, e.g. `repo`,
               `variant`, `revision`, etc.
        """

        if isinstance(names, str):
            names = [names]
        elif not isinstance(names, list):
            raise ValueError(f"Invalid type for names: {type(names)}")

        components_to_load = {name for name in names if name in self._component_specs}
        unknown_names = {name for name in names if name not in self._component_specs}
        if len(unknown_names) > 0:
            logger.warning(f"Unknown components will be ignored: {unknown_names}")

        components_to_register = {}
        for name in components_to_load:
            spec = self._component_specs[name]
            component_load_kwargs = {}
            for key, value in kwargs.items():
                if not isinstance(value, dict):
                    # if the value is a single value, apply it to all components
                    component_load_kwargs[key] = value
                else:
                    if name in value:
                        # if it is a dict, check if the component name is in the dict
                        component_load_kwargs[key] = value[name]
                    elif "default" in value:
                        # check if the default is specified
                        component_load_kwargs[key] = value["default"]
            try:
                components_to_register[name] = spec.load(**component_load_kwargs)
            except Exception as e:
                logger.warning(f"Failed to create component '{name}': {e}")

        # Register all components at once
        self.register_components(**components_to_register)

    # Copied from diffusers.pipelines.pipeline_utils.DiffusionPipeline._maybe_raise_error_if_group_offload_active
    def _maybe_raise_error_if_group_offload_active(
        self, raise_error: bool = False, module: Optional[ms.nn.Cell] = None
    ) -> bool:
        return False

    # Modified from diffusers.pipelines.pipeline_utils.DiffusionPipeline.to
    def to(self, *args, **kwargs) -> Self:
        r"""
        Performs Pipeline dtype conversion. A ms.Type is inferred from the
        arguments of `self.to(*args, **kwargs).`

        <Tip>

            If the pipeline already has the correct ms.Type, then it is returned as is. Otherwise,
            the returned pipeline is a copy of self with the desired ms.Type.

        </Tip>


        Here are the ways to call `to`:

        - `to(dtype, silence_dtype_warnings=False) → DiffusionPipeline` to return a pipeline with the specified
          [`dtype`](https://www.mindspore.cn/docs/zh-CN/r2.7.0/api_python/mindspore/mindspore.dtype.html#mindspore.dtype)

        Arguments:
            dtype (`ms.Type`, *optional*):
                Returns a pipeline with the specified
                [`dtype`](https://www.mindspore.cn/docs/zh-CN/r2.7.0/api_python/mindspore/mindspore.dtype.html#mindspore.dtype)

        Returns:
            [`DiffusionPipeline`]: The pipeline converted to specified `dtype` and/or `dtype`.
        """
        dtype = kwargs.pop("dtype", None)

        dtype_arg = None
        if len(args) == 1:
            if isinstance(args[0], ms.Type):
                dtype_arg = args[0]
        elif len(args) == 2:
            if isinstance(args[0], ms.Type):
                raise ValueError("When passing two arguments, make sure the second to `dtype`.")
            dtype_arg = args[1]
        elif len(args) > 2:
            raise ValueError("Please make sure to pass at most two arguments (`dtype`) `.to(...)`")

        if dtype is not None and dtype_arg is not None:
            raise ValueError(
                "You have passed `dtype` both as an argument and as a keyword argument. Please only pass one of the two."
            )

        dtype = dtype or dtype_arg

        modules = self.components.values()
        modules = [m for m in modules if isinstance(m, ms.nn.Cell)]

        for module in modules:
            module.to(dtype)

        return self

    @staticmethod
    def _component_spec_to_dict(component_spec: ComponentSpec) -> Any:
        """
        Convert a ComponentSpec into a JSON‐serializable dict for saving as an entry in `modular_model_index.json`. If
        the `default_creation_method` is not `from_pretrained`, return None.

        This dict contains:
          - "type_hint": Tuple[str, str]
              Library name and class name of the component. (e.g. ("diffusers", "UNet2DConditionModel"))
          - All loading fields defined by `component_spec.loading_fields()`, typically:
              - "repo": Optional[str]
                  The model repository (e.g., "stabilityai/stable-diffusion-xl").
              - "subfolder": Optional[str]
                  A subfolder within the repo where this component lives.
              - "variant": Optional[str]
                  An optional variant identifier for the model.
              - "revision": Optional[str]
                  A specific git revision (commit hash, tag, or branch).
              - ... any other loading fields defined on the spec.

        Args:
            component_spec (ComponentSpec):
                The spec object describing one pipeline component.

        Returns:
            Dict[str, Any]: A mapping suitable for JSON serialization.

        Example:
            >>> from mindone.diffusers.pipelines.modular_pipeline_utils import ComponentSpec >>> from mindone.diffusers import
            UNet2DConditionModel >>> spec = ComponentSpec(
                ... name="unet", ... type_hint=UNet2DConditionModel, ... config=None, ... repo="path/to/repo", ...
                subfolder="subfolder", ... variant=None, ... revision=None, ...
                default_creation_method="from_pretrained",
            ... ) >>> ModularPipeline._component_spec_to_dict(spec) {
                "type_hint": ("diffusers", "UNet2DConditionModel"), "repo": "path/to/repo", "subfolder": "subfolder",
                "variant": None, "revision": None,
            }
        """
        if component_spec.default_creation_method != "from_pretrained":
            return None

        if component_spec.type_hint is not None:
            lib_name, cls_name = _fetch_class_library_tuple(component_spec.type_hint)
        else:
            lib_name = None
            cls_name = None
        load_spec_dict = {k: getattr(component_spec, k) for k in component_spec.loading_fields()}
        return {
            "type_hint": (lib_name, cls_name),
            **load_spec_dict,
        }

    @staticmethod
    def _dict_to_component_spec(
        name: str,
        spec_dict: Dict[str, Any],
    ) -> ComponentSpec:
        """
        Reconstruct a ComponentSpec from a loading specdict.

        This method converts a dictionary representation back into a ComponentSpec object. The dict should contain:
          - "type_hint": Tuple[str, str]
              Library name and class name of the component. (e.g. ("diffusers", "UNet2DConditionModel"))
          - All loading fields defined by `component_spec.loading_fields()`, typically:
              - "repo": Optional[str]
                  The model repository (e.g., "stabilityai/stable-diffusion-xl").
              - "subfolder": Optional[str]
                  A subfolder within the repo where this component lives.
              - "variant": Optional[str]
                  An optional variant identifier for the model.
              - "revision": Optional[str]
                  A specific git revision (commit hash, tag, or branch).
              - ... any other loading fields defined on the spec.

        Args:
            name (str):
                The name of the component.
            specdict (Dict[str, Any]):
                A dictionary containing the component specification data.

        Returns:
            ComponentSpec: A reconstructed ComponentSpec object.

        Example:
            >>> spec_dict = { ... "type_hint": ("diffusers", "UNet2DConditionModel"), ... "repo":
            "stabilityai/stable-diffusion-xl", ... "subfolder": "unet", ... "variant": None, ... "revision": None, ...
            } >>> ModularPipeline._dict_to_component_spec("unet", spec_dict) ComponentSpec(
                name="unet", type_hint=UNet2DConditionModel, config=None, repo="stabilityai/stable-diffusion-xl",
                subfolder="unet", variant=None, revision=None, default_creation_method="from_pretrained"
            )
        """
        # make a shallow copy so we can pop() safely
        spec_dict = spec_dict.copy()
        # pull out and resolve the stored type_hint
        lib_name, cls_name = spec_dict.pop("type_hint")
        if lib_name is not None and cls_name is not None:
            if "Tokenizer" not in cls_name:
                lib_name = f"mindone.{lib_name}"
            type_hint = simple_get_class_obj(lib_name, cls_name)
        else:
            type_hint = None

        # re‐assemble the ComponentSpec
        return ComponentSpec(
            name=name,
            type_hint=type_hint,
            **spec_dict,
        )

    def set_progress_bar_config(self, **kwargs):
        for sub_block_name, sub_block in self.blocks.sub_blocks.items():
            if hasattr(sub_block, "set_progress_bar_config"):
                sub_block.set_progress_bar_config(**kwargs)

    def __call__(self, state: PipelineState = None, output: Union[str, List[str]] = None, **kwargs):
        """
        Execute the pipeline by running the pipeline blocks with the given inputs.

        Args:
            state (`PipelineState`, optional):
                PipelineState instance contains inputs and intermediate values. If None, a new `PipelineState` will be
                created based on the user inputs and the pipeline blocks's requirement.
            output (`str` or `List[str]`, optional):
                Optional specification of what to return:
                   - None: Returns the complete `PipelineState` with all inputs and intermediates (default)
                   - str: Returns a specific intermediate value from the state (e.g. `output="image"`)
                   - List[str]: Returns a dictionary of specific intermediate values (e.g. `output=["image",
                     "latents"]`)


        Examples:
            ```python
            # Get complete pipeline state
            state = pipeline(prompt="A beautiful sunset", num_inference_steps=20)
            print(state.intermediates)  # All intermediate outputs

            # Get specific output
            image = pipeline(prompt="A beautiful sunset", output="image")

            # Get multiple specific outputs
            results = pipeline(prompt="A beautiful sunset", output=["image", "latents"])
            image, latents = results["image"], results["latents"]

            # Continue from previous state
            state = pipeline(prompt="A beautiful sunset")
            new_state = pipeline(state=state, output="image")  # Continue processing
            ```

        Returns:
            - If `output` is None: Complete `PipelineState` containing all inputs and intermediates
            - If `output` is str: The specific intermediate value from the state (e.g. `output="image"`)
            - If `output` is List[str]: Dictionary mapping output names to their values from the state (e.g.
              `output=["image", "latents"]`)
        """
        if state is None:
            state = PipelineState()

        # Make a copy of the input kwargs
        passed_kwargs = kwargs.copy()

        # Add inputs to state, using defaults if not provided in the kwargs or the state
        # if same input already in the state, will override it if provided in the kwargs
        for expected_input_param in self.blocks.inputs:
            name = expected_input_param.name
            default = expected_input_param.default
            kwargs_type = expected_input_param.kwargs_type
            if name in passed_kwargs:
                state.set(name, passed_kwargs.pop(name), kwargs_type)
            elif name not in state.values:
                state.set(name, default, kwargs_type)

        # Warn about unexpected inputs
        if len(passed_kwargs) > 0:
            warnings.warn(f"Unexpected input '{passed_kwargs.keys()}' provided. This input will be ignored.")
        # Run the pipeline
        try:
            _, state = self.blocks(self, state)
        except Exception:
            error_msg = f"Error in block: ({self.blocks.__class__.__name__}):\n"
            logger.error(error_msg)
            raise

        if output is None:
            return state

        if isinstance(output, str):
            return state.get(output)

        elif isinstance(output, (list, tuple)):
            return state.get(output)
        else:
            raise ValueError(f"Output '{output}' is not a valid output type")

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.component_names property

RETURNS	DESCRIPTION
List[str]	List of names for all components

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.components property

RETURNS	DESCRIPTION
Dict[str, Any]	Dictionary mapping component names to their objects (include both from_pretrained and from_config components)

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.config_component_names property

RETURNS	DESCRIPTION
List[str]	List of names for from_config components

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.default_call_parameters property

RETURNS	DESCRIPTION
Dict[str, Any]	Dictionary mapping input names to their default values

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.doc property

RETURNS	DESCRIPTION
	The docstring of the pipeline blocks

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.dtype property

RETURNS	DESCRIPTION
Type	ms.Type: The mindspore dtype on which the pipeline is located.

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.null_component_names property

RETURNS	DESCRIPTION
List[str]	List of names for components that needs to be loaded

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.pretrained_component_names property

RETURNS	DESCRIPTION
List[str]	List of names for from_pretrained components

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.__call__(state=None, output=None, **kwargs)

Execute the pipeline by running the pipeline blocks with the given inputs.

PARAMETER	DESCRIPTION
state	PipelineState instance contains inputs and intermediate values. If None, a new PipelineState will be created based on the user inputs and the pipeline blocks's requirement. TYPE: `PipelineState` DEFAULT: None
output	Optional specification of what to return: - None: Returns the complete PipelineState with all inputs and intermediates (default) - str: Returns a specific intermediate value from the state (e.g. output="image") - List[str]: Returns a dictionary of specific intermediate values (e.g. output=["image", "latents"]) TYPE: `str` or `List[str]` DEFAULT: None

Examples:

# Get complete pipeline state
state = pipeline(prompt="A beautiful sunset", num_inference_steps=20)
print(state.intermediates)  # All intermediate outputs

# Get specific output
image = pipeline(prompt="A beautiful sunset", output="image")

# Get multiple specific outputs
results = pipeline(prompt="A beautiful sunset", output=["image", "latents"])
image, latents = results["image"], results["latents"]

# Continue from previous state
state = pipeline(prompt="A beautiful sunset")
new_state = pipeline(state=state, output="image")  # Continue processing

RETURNS	DESCRIPTION
	If output is None: Complete PipelineState containing all inputs and intermediates
	If output is str: The specific intermediate value from the state (e.g. output="image")
	If output is List[str]: Dictionary mapping output names to their values from the state (e.g. output=["image", "latents"])

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

def __call__(self, state: PipelineState = None, output: Union[str, List[str]] = None, **kwargs):
    """
    Execute the pipeline by running the pipeline blocks with the given inputs.

    Args:
        state (`PipelineState`, optional):
            PipelineState instance contains inputs and intermediate values. If None, a new `PipelineState` will be
            created based on the user inputs and the pipeline blocks's requirement.
        output (`str` or `List[str]`, optional):
            Optional specification of what to return:
               - None: Returns the complete `PipelineState` with all inputs and intermediates (default)
               - str: Returns a specific intermediate value from the state (e.g. `output="image"`)
               - List[str]: Returns a dictionary of specific intermediate values (e.g. `output=["image",
                 "latents"]`)


    Examples:
        ```python
        # Get complete pipeline state
        state = pipeline(prompt="A beautiful sunset", num_inference_steps=20)
        print(state.intermediates)  # All intermediate outputs

        # Get specific output
        image = pipeline(prompt="A beautiful sunset", output="image")

        # Get multiple specific outputs
        results = pipeline(prompt="A beautiful sunset", output=["image", "latents"])
        image, latents = results["image"], results["latents"]

        # Continue from previous state
        state = pipeline(prompt="A beautiful sunset")
        new_state = pipeline(state=state, output="image")  # Continue processing
        ```

    Returns:
        - If `output` is None: Complete `PipelineState` containing all inputs and intermediates
        - If `output` is str: The specific intermediate value from the state (e.g. `output="image"`)
        - If `output` is List[str]: Dictionary mapping output names to their values from the state (e.g.
          `output=["image", "latents"]`)
    """
    if state is None:
        state = PipelineState()

    # Make a copy of the input kwargs
    passed_kwargs = kwargs.copy()

    # Add inputs to state, using defaults if not provided in the kwargs or the state
    # if same input already in the state, will override it if provided in the kwargs
    for expected_input_param in self.blocks.inputs:
        name = expected_input_param.name
        default = expected_input_param.default
        kwargs_type = expected_input_param.kwargs_type
        if name in passed_kwargs:
            state.set(name, passed_kwargs.pop(name), kwargs_type)
        elif name not in state.values:
            state.set(name, default, kwargs_type)

    # Warn about unexpected inputs
    if len(passed_kwargs) > 0:
        warnings.warn(f"Unexpected input '{passed_kwargs.keys()}' provided. This input will be ignored.")
    # Run the pipeline
    try:
        _, state = self.blocks(self, state)
    except Exception:
        error_msg = f"Error in block: ({self.blocks.__class__.__name__}):\n"
        logger.error(error_msg)
        raise

    if output is None:
        return state

    if isinstance(output, str):
        return state.get(output)

    elif isinstance(output, (list, tuple)):
        return state.get(output)
    else:
        raise ValueError(f"Output '{output}' is not a valid output type")

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.__init__(blocks=None, pretrained_model_name_or_path=None, components_manager=None, collection=None, **kwargs)

Initialize a ModularPipeline instance.

This method sets up the pipeline by: - creating default pipeline blocks if not provided - gather component and config specifications based on the pipeline blocks's requirement (e.g. expected_components, expected_configs) - update the loading specs of from_pretrained components based on the modular_model_index.json file from huggingface hub if pretrained_model_name_or_path is provided - create defaultfrom_config components and register everything

PARAMETER	DESCRIPTION
blocks	ModularPipelineBlocks instance. If None, will attempt to load default blocks based on the pipeline class name. TYPE: Optional[ModularPipelineBlocks] DEFAULT: None
pretrained_model_name_or_path	Path to a pretrained pipeline configuration. If provided, will load component specs (only for from_pretrained components) and config values from the saved modular_model_index.json file. TYPE: Optional[Union[str, PathLike]] DEFAULT: None
components_manager	Optional ComponentsManager for managing multiple component cross different pipelines and apply offloading strategies. TYPE: Optional[ComponentsManager] DEFAULT: None
collection	Optional collection name for organizing components in the ComponentsManager. TYPE: Optional[str] DEFAULT: None
**kwargs	Additional arguments passed to load_config() when loading pretrained configuration. DEFAULT: {}

Examples:

# Initialize with custom blocks
pipeline = ModularPipeline(blocks=my_custom_blocks)

# Initialize from pretrained configuration
pipeline = ModularPipeline(blocks=my_blocks, pretrained_model_name_or_path="my-repo/modular-pipeline")

# Initialize with components manager
pipeline = ModularPipeline(
    blocks=my_blocks, components_manager=ComponentsManager(), collection="my_collection"
)

Notes

• If blocks is None, the method will try to find default blocks based on the pipeline class name
• Components with default_creation_method="from_config" are created immediately, its specs are not included in config dict and will not be saved in modular_model_index.json
• Components with default_creation_method="from_pretrained" are set to None and can be loaded later with load_default_components()/load_components()
• The pipeline's config dict is populated with component specs (only for from_pretrained components) and config values, which will be saved as modular_model_index.json during save_pretrained
• The pipeline's config dict is also used to store the pipeline blocks's class name, which will be saved as _blocks_class_name in the config dict

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

def __init__(
    self,
    blocks: Optional[ModularPipelineBlocks] = None,
    pretrained_model_name_or_path: Optional[Union[str, os.PathLike]] = None,
    components_manager: Optional[ComponentsManager] = None,
    collection: Optional[str] = None,
    **kwargs,
):
    """
    Initialize a ModularPipeline instance.

    This method sets up the pipeline by:
    - creating default pipeline blocks if not provided
    - gather component and config specifications based on the pipeline blocks's requirement (e.g.
       expected_components, expected_configs)
    - update the loading specs of from_pretrained components based on the modular_model_index.json file from
       huggingface hub if `pretrained_model_name_or_path` is provided
    - create defaultfrom_config components and register everything

    Args:
        blocks: `ModularPipelineBlocks` instance. If None, will attempt to load
               default blocks based on the pipeline class name.
        pretrained_model_name_or_path: Path to a pretrained pipeline configuration. If provided,
                will load component specs (only for from_pretrained components) and config values from the saved
                modular_model_index.json file.
        components_manager:
            Optional ComponentsManager for managing multiple component cross different pipelines and apply
            offloading strategies.
        collection: Optional collection name for organizing components in the ComponentsManager.
        **kwargs: Additional arguments passed to `load_config()` when loading pretrained configuration.

    Examples:
        ```python
        # Initialize with custom blocks
        pipeline = ModularPipeline(blocks=my_custom_blocks)

        # Initialize from pretrained configuration
        pipeline = ModularPipeline(blocks=my_blocks, pretrained_model_name_or_path="my-repo/modular-pipeline")

        # Initialize with components manager
        pipeline = ModularPipeline(
            blocks=my_blocks, components_manager=ComponentsManager(), collection="my_collection"
        )
        ```

    Notes:
        - If blocks is None, the method will try to find default blocks based on the pipeline class name
        - Components with default_creation_method="from_config" are created immediately, its specs are not included
          in config dict and will not be saved in `modular_model_index.json`
        - Components with default_creation_method="from_pretrained" are set to None and can be loaded later with
          `load_default_components()`/`load_components()`
        - The pipeline's config dict is populated with component specs (only for from_pretrained components) and
          config values, which will be saved as `modular_model_index.json` during `save_pretrained`
        - The pipeline's config dict is also used to store the pipeline blocks's class name, which will be saved as
          `_blocks_class_name` in the config dict
    """
    if blocks is None:
        blocks_class_name = MODULAR_PIPELINE_BLOCKS_MAPPING.get(self.__class__.__name__)
        if blocks_class_name is not None:
            diffusers_module = importlib.import_module("mindone.diffusers")
            blocks_class = getattr(diffusers_module, blocks_class_name)
            blocks = blocks_class()
        else:
            logger.warning(f"`blocks` is `None`, no default blocks class found for {self.__class__.__name__}")

    self.blocks = blocks
    self._components_manager = components_manager
    self._collection = collection
    self._component_specs = {spec.name: deepcopy(spec) for spec in self.blocks.expected_components}
    self._config_specs = {spec.name: deepcopy(spec) for spec in self.blocks.expected_configs}

    # update component_specs and config_specs from modular_repo
    if pretrained_model_name_or_path is not None:
        config_dict = self.load_config(pretrained_model_name_or_path, **kwargs)

        for name, value in config_dict.items():
            # all the components in modular_model_index.json are from_pretrained components
            if name in self._component_specs and isinstance(value, (tuple, list)) and len(value) == 3:
                library, class_name, component_spec_dict = value
                component_spec = self._dict_to_component_spec(name, component_spec_dict)
                component_spec.default_creation_method = "from_pretrained"
                self._component_specs[name] = component_spec

            elif name in self._config_specs:
                self._config_specs[name].default = value

    register_components_dict = {}
    for name, component_spec in self._component_specs.items():
        if component_spec.default_creation_method == "from_config":
            component = component_spec.create()
        else:
            component = None
        register_components_dict[name] = component
    self.register_components(**register_components_dict)

    default_configs = {}
    for name, config_spec in self._config_specs.items():
        default_configs[name] = config_spec.default
    self.register_to_config(**default_configs)

    self.register_to_config(_blocks_class_name=self.blocks.__class__.__name__ if self.blocks is not None else None)

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.from_pretrained(pretrained_model_name_or_path, trust_remote_code=None, components_manager=None, collection=None, **kwargs) classmethod

Load a ModularPipeline from a huggingface hub repo.

PARAMETER	DESCRIPTION
pretrained_model_name_or_path	Path to a pretrained pipeline configuration. If provided, will load component specs (only for from_pretrained components) and config values from the modular_model_index.json file. TYPE: `str` or `os.PathLike`
trust_remote_code	Whether to trust remote code when loading the pipeline, need to be set to True if you want to create pipeline blocks based on the custom code in pretrained_model_name_or_path TYPE: `bool` DEFAULT: None
components_manager	ComponentsManager instance for managing multiple component cross different pipelines and apply offloading strategies. TYPE: `ComponentsManager` DEFAULT: None
collection	` Collection name for organizing components in the ComponentsManager. TYPE: `str` DEFAULT: None

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

@classmethod
@validate_hf_hub_args
def from_pretrained(
    cls,
    pretrained_model_name_or_path: Optional[Union[str, os.PathLike]],
    trust_remote_code: Optional[bool] = None,
    components_manager: Optional[ComponentsManager] = None,
    collection: Optional[str] = None,
    **kwargs,
):
    """
    Load a ModularPipeline from a huggingface hub repo.

    Args:
        pretrained_model_name_or_path (`str` or `os.PathLike`, optional):
            Path to a pretrained pipeline configuration. If provided, will load component specs (only for
            from_pretrained components) and config values from the modular_model_index.json file.
        trust_remote_code (`bool`, optional):
            Whether to trust remote code when loading the pipeline, need to be set to True if you want to create
            pipeline blocks based on the custom code in `pretrained_model_name_or_path`
        components_manager (`ComponentsManager`, optional):
            ComponentsManager instance for managing multiple component cross different pipelines and apply
            offloading strategies.
        collection (`str`, optional):`
            Collection name for organizing components in the ComponentsManager.
    """
    from ..pipelines.pipeline_loading_utils import _get_pipeline_class

    try:
        blocks = ModularPipelineBlocks.from_pretrained(
            pretrained_model_name_or_path, trust_remote_code=trust_remote_code, **kwargs
        )
    except EnvironmentError:
        blocks = None

    cache_dir = kwargs.pop("cache_dir", None)
    force_download = kwargs.pop("force_download", False)
    proxies = kwargs.pop("proxies", None)
    token = kwargs.pop("token", None)
    local_files_only = kwargs.pop("local_files_only", False)
    revision = kwargs.pop("revision", None)

    load_config_kwargs = {
        "cache_dir": cache_dir,
        "force_download": force_download,
        "proxies": proxies,
        "token": token,
        "local_files_only": local_files_only,
        "revision": revision,
    }

    try:
        config_dict = cls.load_config(pretrained_model_name_or_path, **load_config_kwargs)
        pipeline_class = _get_pipeline_class(cls, config=config_dict)
    except EnvironmentError:
        pipeline_class = cls
        pretrained_model_name_or_path = None

    pipeline = pipeline_class(
        blocks=blocks,
        pretrained_model_name_or_path=pretrained_model_name_or_path,
        components_manager=components_manager,
        collection=collection,
        **kwargs,
    )
    return pipeline

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.get_component_spec(name)

RETURNS	DESCRIPTION
ComponentSpec	a copy of the ComponentSpec object for the given component name

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

def get_component_spec(self, name: str) -> ComponentSpec:
    """
    Returns:
        - a copy of the ComponentSpec object for the given component name
    """
    return deepcopy(self._component_specs[name])

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.load_components(names, **kwargs)

Load selected components from specs.

PARAMETER	DESCRIPTION
names	List of component names to load; by default will not load any components TYPE: Union[List[str], str]
**kwargs	additional kwargs to be passed to from_pretrained().Can be: - a single value to be applied to all components to be loaded, e.g. mindspore_dtype=ms.bfloat16 - a dict, e.g. mindspore_dtype={"unet": ms.bfloat16, "default": ms.float32} - if potentially override ComponentSpec if passed a different loading field in kwargs, e.g. repo, variant, revision, etc. DEFAULT: {}

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

def load_components(self, names: Union[List[str], str], **kwargs):
    """
    Load selected components from specs.

    Args:
        names: List of component names to load; by default will not load any components
        **kwargs: additional kwargs to be passed to `from_pretrained()`.Can be:
         - a single value to be applied to all components to be loaded, e.g. mindspore_dtype=ms.bfloat16
         - a dict, e.g. mindspore_dtype={"unet": ms.bfloat16, "default": ms.float32}
         - if potentially override ComponentSpec if passed a different loading field in kwargs, e.g. `repo`,
           `variant`, `revision`, etc.
    """

    if isinstance(names, str):
        names = [names]
    elif not isinstance(names, list):
        raise ValueError(f"Invalid type for names: {type(names)}")

    components_to_load = {name for name in names if name in self._component_specs}
    unknown_names = {name for name in names if name not in self._component_specs}
    if len(unknown_names) > 0:
        logger.warning(f"Unknown components will be ignored: {unknown_names}")

    components_to_register = {}
    for name in components_to_load:
        spec = self._component_specs[name]
        component_load_kwargs = {}
        for key, value in kwargs.items():
            if not isinstance(value, dict):
                # if the value is a single value, apply it to all components
                component_load_kwargs[key] = value
            else:
                if name in value:
                    # if it is a dict, check if the component name is in the dict
                    component_load_kwargs[key] = value[name]
                elif "default" in value:
                    # check if the default is specified
                    component_load_kwargs[key] = value["default"]
        try:
            components_to_register[name] = spec.load(**component_load_kwargs)
        except Exception as e:
            logger.warning(f"Failed to create component '{name}': {e}")

    # Register all components at once
    self.register_components(**components_to_register)

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.load_default_components(**kwargs)

Load from_pretrained components using the loading specs in the config dict.

PARAMETER	DESCRIPTION
**kwargs	Additional arguments passed to from_pretrained method, e.g. mindspore_dtype, cache_dir, etc. DEFAULT: {}

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

def load_default_components(self, **kwargs):
    """
    Load from_pretrained components using the loading specs in the config dict.

    Args:
        **kwargs: Additional arguments passed to `from_pretrained` method, e.g. mindspore_dtype, cache_dir, etc.
    """
    names = [
        name
        for name in self._component_specs.keys()
        if self._component_specs[name].default_creation_method == "from_pretrained"
    ]
    self.load_components(names=names, **kwargs)

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.register_components(**kwargs)

Register components with their corresponding specifications.

This method is responsible for: 1. Sets component objects as attributes on the loader (e.g., self.unet = unet) 2. Updates the config dict, which will be saved as modular_model_index.json during save_pretrained (only for from_pretrained components) 3. Adds components to the component manager if one is attached (only for from_pretrained components)

This method is called when: - Components are first initialized in init: - from_pretrained components not loaded during init so they are registered as None; - non from_pretrained components are created during init and registered as the object itself - Components are updated with the update_components() method: e.g. loader.update_components(unet=unet) or loader.update_components(guider=guider_spec) - (from_pretrained) Components are loaded with the load_default_components() method: e.g. loader.load_default_components(names=["unet"])

PARAMETER	DESCRIPTION
**kwargs	Keyword arguments where keys are component names and values are component objects. E.g., register_components(unet=unet_model, text_encoder=encoder_model) DEFAULT: {}

Notes

• When registering None for a component, it sets attribute to None but still syncs specs with the config dict, which will be saved as modular_model_index.json during save_pretrained
• component_specs are updated to match the new component outside of this method, e.g. in update_components() method

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

def register_components(self, **kwargs):
    """
    Register components with their corresponding specifications.

    This method is responsible for:
    1. Sets component objects as attributes on the loader (e.g., self.unet = unet)
    2. Updates the config dict, which will be saved as `modular_model_index.json` during `save_pretrained` (only
       for from_pretrained components)
    3. Adds components to the component manager if one is attached (only for from_pretrained components)

    This method is called when:
    - Components are first initialized in __init__:
       - from_pretrained components not loaded during __init__ so they are registered as None;
       - non from_pretrained components are created during __init__ and registered as the object itself
    - Components are updated with the `update_components()` method: e.g. loader.update_components(unet=unet) or
      loader.update_components(guider=guider_spec)
    - (from_pretrained) Components are loaded with the `load_default_components()` method: e.g.
      loader.load_default_components(names=["unet"])

    Args:
        **kwargs: Keyword arguments where keys are component names and values are component objects.
                  E.g., register_components(unet=unet_model, text_encoder=encoder_model)

    Notes:
        - When registering None for a component, it sets attribute to None but still syncs specs with the config
          dict, which will be saved as `modular_model_index.json` during `save_pretrained`
        - component_specs are updated to match the new component outside of this method, e.g. in
          `update_components()` method
    """
    for name, module in kwargs.items():
        # current component spec
        component_spec = self._component_specs.get(name)
        if component_spec is None:
            logger.warning(f"ModularPipeline.register_components: skipping unknown component '{name}'")
            continue

        # check if it is the first time registration, i.e. calling from __init__
        is_registered = hasattr(self, name)
        is_from_pretrained = component_spec.default_creation_method == "from_pretrained"

        if module is not None:
            # actual library and class name of the module
            library, class_name = _fetch_class_library_tuple(module)  # e.g. ("diffusers", "UNet2DConditionModel")
        else:
            # if module is None, e.g. self.register_components(unet=None) during __init__
            # we do not update the spec,
            # but we still need to update the modular_model_index.json config based on component spec
            library, class_name = None, None

        # extract the loading spec from the updated component spec that'll be used as part of modular_model_index.json config
        # e.g. {"repo": "stabilityai/stable-diffusion-2-1",
        #       "type_hint": ("diffusers", "UNet2DConditionModel"),
        #       "subfolder": "unet",
        #       "variant": None,
        #       "revision": None}
        component_spec_dict = self._component_spec_to_dict(component_spec)

        register_dict = {name: (library, class_name, component_spec_dict)}

        # set the component as attribute
        # if it is not set yet, just set it and skip the process to check and warn below
        if not is_registered:
            if is_from_pretrained:
                self.register_to_config(**register_dict)
            setattr(self, name, module)
            if module is not None and is_from_pretrained and self._components_manager is not None:
                self._components_manager.add(name, module, self._collection)
            continue

        current_module = getattr(self, name, None)
        # skip if the component is already registered with the same object
        if current_module is module:
            logger.info(
                f"ModularPipeline.register_components: {name} is already registered with same object, skipping"
            )
            continue

        # warn if unregister
        if current_module is not None and module is None:
            logger.info(
                f"ModularPipeline.register_components: setting '{name}' to None "
                f"(was {current_module.__class__.__name__})"
            )
        # same type, new instance → replace but send debug log
        elif (
            current_module is not None
            and module is not None
            and isinstance(module, current_module.__class__)
            and current_module != module
        ):
            logger.debug(
                f"ModularPipeline.register_components: replacing existing '{name}' "
                f"(same type {type(current_module).__name__}, new instance)"
            )

        # update modular_model_index.json config
        if is_from_pretrained:
            self.register_to_config(**register_dict)
        # finally set models
        setattr(self, name, module)
        # add to component manager if one is attached
        if module is not None and is_from_pretrained and self._components_manager is not None:
            self._components_manager.add(name, module, self._collection)

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.save_pretrained(save_directory, push_to_hub=False, **kwargs)

Save the pipeline to a directory. It does not save components, you need to save them separately.

PARAMETER	DESCRIPTION
save_directory	Path to the directory where the pipeline will be saved. TYPE: `str` or `os.PathLike`
push_to_hub	Whether to push the pipeline to the huggingface hub. TYPE: `bool` DEFAULT: False
**kwargs	Additional arguments passed to save_config() method DEFAULT: {}

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

def save_pretrained(self, save_directory: Union[str, os.PathLike], push_to_hub: bool = False, **kwargs):
    """
    Save the pipeline to a directory. It does not save components, you need to save them separately.

    Args:
        save_directory (`str` or `os.PathLike`):
            Path to the directory where the pipeline will be saved.
        push_to_hub (`bool`, optional):
            Whether to push the pipeline to the huggingface hub.
        **kwargs: Additional arguments passed to `save_config()` method
    """
    if push_to_hub:
        commit_message = kwargs.pop("commit_message", None)
        private = kwargs.pop("private", None)
        create_pr = kwargs.pop("create_pr", False)
        token = kwargs.pop("token", None)
        repo_id = kwargs.pop("repo_id", save_directory.split(os.path.sep)[-1])
        repo_id = create_repo(repo_id, exist_ok=True, private=private, token=token).repo_id

        # Create a new empty model card and eventually tag it
        model_card = load_or_create_model_card(repo_id, token=token, is_pipeline=True)
        model_card = populate_model_card(model_card)
        model_card.save(os.path.join(save_directory, "README.md"))

    # YiYi TODO: maybe order the json file to make it more readable: configs first, then components
    self.save_config(save_directory=save_directory)

    if push_to_hub:
        self._upload_folder(
            save_directory,
            repo_id,
            token=token,
            commit_message=commit_message,
            create_pr=create_pr,
        )

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.to(*args, **kwargs)

Performs Pipeline dtype conversion. A ms.Type is inferred from the arguments of self.to(*args, **kwargs).

If the pipeline already has the correct ms.Type, then it is returned as is. Otherwise,
the returned pipeline is a copy of self with the desired ms.Type.

Here are the ways to call to:

• to(dtype, silence_dtype_warnings=False) → DiffusionPipeline to return a pipeline with the specified dtype

PARAMETER	DESCRIPTION
dtype	Returns a pipeline with the specified dtype TYPE: `ms.Type`, *optional*

RETURNS	DESCRIPTION
Self	[DiffusionPipeline]: The pipeline converted to specified dtype and/or dtype.

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

def to(self, *args, **kwargs) -> Self:
    r"""
    Performs Pipeline dtype conversion. A ms.Type is inferred from the
    arguments of `self.to(*args, **kwargs).`

    <Tip>

        If the pipeline already has the correct ms.Type, then it is returned as is. Otherwise,
        the returned pipeline is a copy of self with the desired ms.Type.

    </Tip>


    Here are the ways to call `to`:

    - `to(dtype, silence_dtype_warnings=False) → DiffusionPipeline` to return a pipeline with the specified
      [`dtype`](https://www.mindspore.cn/docs/zh-CN/r2.7.0/api_python/mindspore/mindspore.dtype.html#mindspore.dtype)

    Arguments:
        dtype (`ms.Type`, *optional*):
            Returns a pipeline with the specified
            [`dtype`](https://www.mindspore.cn/docs/zh-CN/r2.7.0/api_python/mindspore/mindspore.dtype.html#mindspore.dtype)

    Returns:
        [`DiffusionPipeline`]: The pipeline converted to specified `dtype` and/or `dtype`.
    """
    dtype = kwargs.pop("dtype", None)

    dtype_arg = None
    if len(args) == 1:
        if isinstance(args[0], ms.Type):
            dtype_arg = args[0]
    elif len(args) == 2:
        if isinstance(args[0], ms.Type):
            raise ValueError("When passing two arguments, make sure the second to `dtype`.")
        dtype_arg = args[1]
    elif len(args) > 2:
        raise ValueError("Please make sure to pass at most two arguments (`dtype`) `.to(...)`")

    if dtype is not None and dtype_arg is not None:
        raise ValueError(
            "You have passed `dtype` both as an argument and as a keyword argument. Please only pass one of the two."
        )

    dtype = dtype or dtype_arg

    modules = self.components.values()
    modules = [m for m in modules if isinstance(m, ms.nn.Cell)]

    for module in modules:
        module.to(dtype)

    return self

mindone.diffusers.modular_pipelines.modular_pipeline.ModularPipeline.update_components(**kwargs)

Update components and configuration values and specs after the pipeline has been instantiated.

This method allows you to: 1. Replace existing components with new ones (e.g., updating self.unet or self.text_encoder) 2. Update configuration values (e.g., changing self.requires_safety_checker flag)

In addition to updating the components and configuration values as pipeline attributes, the method also updates: - the corresponding specs in _component_specs and _config_specs - the config dict, which will be saved as modular_model_index.json during save_pretrained

PARAMETER

DESCRIPTION

**kwargs

Component objects, ComponentSpec objects, or configuration values to update: - Component objects: Only supports components we can extract specs using ComponentSpec.from_component() method i.e. components created with ComponentSpec.load() or ConfigMixin subclasses that aren't nn.Modules (e.g., unet=new_unet, text_encoder=new_encoder) - ComponentSpec objects: Only supports default_creation_method == "from_config", will call create() method to create a new component (e.g., guider=ComponentSpec(name="guider", type_hint=ClassifierFreeGuidance, config={...}, default_creation_method="from_config")) - Configuration values: Simple values to update configuration settings (e.g., requires_safety_checker=False) DEFAULT: {}

RAISES	DESCRIPTION
ValueError	If a component object is not supported in ComponentSpec.from_component() method: - nn.Module components without a valid _diffusers_load_id attribute - Non-ConfigMixin components without a valid _diffusers_load_id attribute

Examples:

# Update multiple components at once
pipeline.update_components(unet=new_unet_model, text_encoder=new_text_encoder)

# Update configuration values
pipeline.update_components(requires_safety_checker=False)

# Update both components and configs together
pipeline.update_components(unet=new_unet_model, requires_safety_checker=False)

# Update with ComponentSpec objects (from_config only)
pipeline.update_components(
    guider=ComponentSpec(
        name="guider",
        type_hint=ClassifierFreeGuidance,
        config={"guidance_scale": 5.0},
        default_creation_method="from_config",
    )
)

Notes

• Components with trained weights must be created using ComponentSpec.load(). If the component has not been shared in huggingface hub and you don't have loading specs, you can upload it using push_to_hub()
• ConfigMixin objects without weights (e.g., schedulers, guiders) can be passed directly
• ComponentSpec objects with default_creation_method="from_pretrained" are not supported in update_components()

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

def update_components(self, **kwargs):
    """
    Update components and configuration values and specs after the pipeline has been instantiated.

    This method allows you to:
    1. Replace existing components with new ones (e.g., updating `self.unet` or `self.text_encoder`)
    2. Update configuration values (e.g., changing `self.requires_safety_checker` flag)

    In addition to updating the components and configuration values as pipeline attributes, the method also
    updates:
    - the corresponding specs in `_component_specs` and `_config_specs`
    - the `config` dict, which will be saved as `modular_model_index.json` during `save_pretrained`

    Args:
        **kwargs: Component objects, ComponentSpec objects, or configuration values to update:
            - Component objects: Only supports components we can extract specs using
              `ComponentSpec.from_component()` method i.e. components created with ComponentSpec.load() or
              ConfigMixin subclasses that aren't nn.Modules (e.g., `unet=new_unet, text_encoder=new_encoder`)
            - ComponentSpec objects: Only supports default_creation_method == "from_config", will call create()
              method to create a new component (e.g., `guider=ComponentSpec(name="guider",
              type_hint=ClassifierFreeGuidance, config={...}, default_creation_method="from_config")`)
            - Configuration values: Simple values to update configuration settings (e.g.,
              `requires_safety_checker=False`)

    Raises:
        ValueError: If a component object is not supported in ComponentSpec.from_component() method:
            - nn.Module components without a valid `_diffusers_load_id` attribute
            - Non-ConfigMixin components without a valid `_diffusers_load_id` attribute

    Examples:
        ```python
        # Update multiple components at once
        pipeline.update_components(unet=new_unet_model, text_encoder=new_text_encoder)

        # Update configuration values
        pipeline.update_components(requires_safety_checker=False)

        # Update both components and configs together
        pipeline.update_components(unet=new_unet_model, requires_safety_checker=False)

        # Update with ComponentSpec objects (from_config only)
        pipeline.update_components(
            guider=ComponentSpec(
                name="guider",
                type_hint=ClassifierFreeGuidance,
                config={"guidance_scale": 5.0},
                default_creation_method="from_config",
            )
        )
        ```

    Notes:
        - Components with trained weights must be created using ComponentSpec.load(). If the component has not been
          shared in huggingface hub and you don't have loading specs, you can upload it using `push_to_hub()`
        - ConfigMixin objects without weights (e.g., schedulers, guiders) can be passed directly
        - ComponentSpec objects with default_creation_method="from_pretrained" are not supported in
          update_components()
    """

    # extract component_specs_updates & config_specs_updates from `specs`
    passed_component_specs = {
        k: kwargs.pop(k) for k in self._component_specs if k in kwargs and isinstance(kwargs[k], ComponentSpec)
    }
    passed_components = {
        k: kwargs.pop(k) for k in self._component_specs if k in kwargs and not isinstance(kwargs[k], ComponentSpec)
    }
    passed_config_values = {k: kwargs.pop(k) for k in self._config_specs if k in kwargs}

    for name, component in passed_components.items():
        current_component_spec = self._component_specs[name]

        # warn if type changed
        if current_component_spec.type_hint is not None and not isinstance(
            component, current_component_spec.type_hint
        ):
            logger.warning(
                f"ModularPipeline.update_components: adding {name} with new type: {component.__class__.__name__}, previous type: {current_component_spec.type_hint.__name__}"  # noqa
            )
        # update _component_specs based on the new component
        new_component_spec = ComponentSpec.from_component(name, component)
        if new_component_spec.default_creation_method != current_component_spec.default_creation_method:
            logger.warning(
                f"ModularPipeline.update_components: changing the default_creation_method of {name} from {current_component_spec.default_creation_method} to {new_component_spec.default_creation_method}."  # noqa
            )

        self._component_specs[name] = new_component_spec

    if len(kwargs) > 0:
        logger.warning(f"Unexpected keyword arguments, will be ignored: {kwargs.keys()}")

    created_components = {}
    for name, component_spec in passed_component_specs.items():
        if component_spec.default_creation_method == "from_pretrained":
            raise ValueError(
                "ComponentSpec object with default_creation_method == 'from_pretrained' is not supported in update_components() method"
            )
        created_components[name] = component_spec.create()
        current_component_spec = self._component_specs[name]
        # warn if type changed
        if current_component_spec.type_hint is not None and not isinstance(
            created_components[name], current_component_spec.type_hint
        ):
            logger.warning(
                f"ModularPipeline.update_components: adding {name} with new type: {created_components[name].__class__.__name__}, previous type: {current_component_spec.type_hint.__name__}"  # noqa
            )
        # update _component_specs based on the user passed component_spec
        self._component_specs[name] = component_spec
    self.register_components(**passed_components, **created_components)

    config_to_register = {}
    for name, new_value in passed_config_values.items():
        # e.g. requires_aesthetics_score = False
        self._config_specs[name].default = new_value
        config_to_register[name] = new_value
    self.register_to_config(**config_to_register)