Repository: huggingface/diffusers
File: src/diffusers/pipelines/pipeline_utils.py (Python)

Code before:
            new_pipeline.to(dtype=torch_dtype)

        return new_pipeline

    def _maybe_raise_error_if_group_offload_active(
        self, raise_error: bool = False, module: Optional[torch.nn.Module] = None
    ) -> bool:
        from ..hooks.group_offloading import _is_group_offload_enabled

        components = self.components.values() if module is None else [module]
        components = [component for component in components if isinstance(component, torch.nn.Module)]
        for component in components:
            if _is_group_offload_enabled(component):
                if raise_error:
                    raise ValueError(
                        "You are trying to apply model/sequential CPU offloading to a pipeline that contains components "
                        "with group offloading enabled. This is not supported. Please disable group offloading for "
                        "components of the pipeline to use other offloading methods."
                    )
                return True
        return False

    def compile(
        self,
        compile_regions_for_transformer: bool = True,
        transformer_module_name: str = "transformer",
        other_modules_names: List[str] = [],
        **compile_kwargs,
    ):
        transformer = getattr(self, transformer_module_name, None)
        if transformer is None:
            raise ValueError(
                f"{transformer_module_name} not found in the pipeline. Set `transformer_module_name` to the correct module name."
            )

        if compile_regions_for_transformer:
            compile_region_classes = getattr(transformer, "compile_region_classes", None)
            if compile_region_classes is None:
                raise ValueError(
                    f"{transformer_module_name} does not have `compile_region_classes` attribute. Set `compile_regions_for_transformer` to False."
                )

            for submod in transformer.modules():
                if isinstance(submod, compile_region_classes):
                    submod.compile(**compile_kwargs)
        else:
            transformer.compile(**compile_kwargs)

        for module_name in other_modules_names:
            module = getattr(self, module_name, None)
            if module is None:

Review comment (suggestion):
These two will have to have changed if we proceed to have it inside `ModelMixin`.

Code after:

        return new_pipeline

    def _maybe_raise_error_if_group_offload_active(
        self, raise_error: bool = False, module: Optional[torch.nn.Module] = None
    ) -> bool:
        from ..hooks.group_offloading import _is_group_offload_enabled

        components = self.components.values() if module is None else [module]
        components = [component for component in components if isinstance(component, torch.nn.Module)]
        for component in components:
            if _is_group_offload_enabled(component):
                if raise_error:
                    raise ValueError(
                        "You are trying to apply model/sequential CPU offloading to a pipeline that contains components "
                        "with group offloading enabled. This is not supported. Please disable group offloading for "
                        "components of the pipeline to use other offloading methods."
                    )
                return True
        return False


class StableDiffusionMixin:
    r"""
    Helper for DiffusionPipeline with vae and unet.(mainly for LDM such as stable diffusion)
    """

    def enable_vae_slicing(self):
        r"""
        Enable sliced VAE decoding. When this option is enabled, the VAE will split the input tensor in slices to
        compute decoding in several steps. This is useful to save some memory and allow larger batch sizes.
        """
        self.vae.enable_slicing()

    def disable_vae_slicing(self):
        r"""
        Disable sliced VAE decoding. If `enable_vae_slicing` was previously enabled, this method will go back to
        computing decoding in one step.
        """
        self.vae.disable_slicing()

    def enable_vae_tiling(self):
        r"""
        Enable tiled VAE decoding. When this option is enabled, the VAE will split the input tensor into tiles to
        compute decoding and encoding in several steps. This is useful for saving a large amount of memory and to allow
        processing larger images.
        """
        self.vae.enable_tiling()

    def disable_vae_tiling(self):
        r"""