diffusers/community-pipelines-mirror

1"""2modified based on diffusion library from Huggingface: https://github.com/huggingface/diffusers/blob/main/src/diffusers/pipelines/stable_diffusion/pipeline_stable_diffusion.py3"""4 5import inspect6from typing import Callable, List, Optional, Union7 8import torch9from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer10 11from diffusers import DiffusionPipeline12from diffusers.models import AutoencoderKL, UNet2DConditionModel13from diffusers.pipelines.pipeline_utils import StableDiffusionMixin14from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput15from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker16from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler17from diffusers.utils import logging18 19 20logger = logging.get_logger(__name__)  # pylint: disable=invalid-name21 22 23class SeedResizeStableDiffusionPipeline(DiffusionPipeline, StableDiffusionMixin):24    r"""25    Pipeline for text-to-image generation using Stable Diffusion.26 27    This model inherits from [`DiffusionPipeline`]. Check the superclass documentation for the generic methods the28    library implements for all the pipelines (such as downloading or saving, running on a particular device, etc.)29 30    Args:31        vae ([`AutoencoderKL`]):32            Variational Auto-Encoder (VAE) Model to encode and decode images to and from latent representations.33        text_encoder ([`CLIPTextModel`]):34            Frozen text-encoder. Stable Diffusion uses the text portion of35            [CLIP](https://huggingface.co/docs/transformers/model_doc/clip#transformers.CLIPTextModel), specifically36            the [clip-vit-large-patch14](https://huggingface.co/openai/clip-vit-large-patch14) variant.37        tokenizer (`CLIPTokenizer`):38            Tokenizer of class39            [CLIPTokenizer](https://huggingface.co/docs/transformers/v4.21.0/en/model_doc/clip#transformers.CLIPTokenizer).40        unet ([`UNet2DConditionModel`]): Conditional U-Net architecture to denoise the encoded image latents.41        scheduler ([`SchedulerMixin`]):42            A scheduler to be used in combination with `unet` to denoise the encoded image latents. Can be one of43            [`DDIMScheduler`], [`LMSDiscreteScheduler`], or [`PNDMScheduler`].44        safety_checker ([`StableDiffusionSafetyChecker`]):45            Classification module that estimates whether generated images could be considered offensive or harmful.46            Please, refer to the [model card](https://huggingface.co/CompVis/stable-diffusion-v1-4) for details.47        feature_extractor ([`CLIPImageProcessor`]):48            Model that extracts features from generated images to be used as inputs for the `safety_checker`.49    """50 51    def __init__(52        self,53        vae: AutoencoderKL,54        text_encoder: CLIPTextModel,55        tokenizer: CLIPTokenizer,56        unet: UNet2DConditionModel,57        scheduler: Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler],58        safety_checker: StableDiffusionSafetyChecker,59        feature_extractor: CLIPImageProcessor,60    ):61        super().__init__()62        self.register_modules(63            vae=vae,64            text_encoder=text_encoder,65            tokenizer=tokenizer,66            unet=unet,67            scheduler=scheduler,68            safety_checker=safety_checker,69            feature_extractor=feature_extractor,70        )71 72    @torch.no_grad()73    def __call__(74        self,75        prompt: Union[str, List[str]],76        height: int = 512,77        width: int = 512,78        num_inference_steps: int = 50,79        guidance_scale: float = 7.5,80        negative_prompt: Optional[Union[str, List[str]]] = None,81        num_images_per_prompt: Optional[int] = 1,82        eta: float = 0.0,83        generator: Optional[torch.Generator] = None,84        latents: Optional[torch.Tensor] = None,85        output_type: Optional[str] = "pil",86        return_dict: bool = True,87        callback: Optional[Callable[[int, int, torch.Tensor], None]] = None,88        callback_steps: int = 1,89        text_embeddings: Optional[torch.Tensor] = None,90        **kwargs,91    ):92        r"""93        Function invoked when calling the pipeline for generation.94 95        Args:96            prompt (`str` or `List[str]`):97                The prompt or prompts to guide the image generation.98            height (`int`, *optional*, defaults to 512):99                The height in pixels of the generated image.100            width (`int`, *optional*, defaults to 512):101                The width in pixels of the generated image.102            num_inference_steps (`int`, *optional*, defaults to 50):103                The number of denoising steps. More denoising steps usually lead to a higher quality image at the104                expense of slower inference.105            guidance_scale (`float`, *optional*, defaults to 7.5):106                Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).107                `guidance_scale` is defined as `w` of equation 2. of [Imagen108                Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale >109                1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`,110                usually at the expense of lower image quality.111            negative_prompt (`str` or `List[str]`, *optional*):112                The prompt or prompts not to guide the image generation. Ignored when not using guidance (i.e., ignored113                if `guidance_scale` is less than `1`).114            num_images_per_prompt (`int`, *optional*, defaults to 1):115                The number of images to generate per prompt.116            eta (`float`, *optional*, defaults to 0.0):117                Corresponds to parameter eta (η) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to118                [`schedulers.DDIMScheduler`], will be ignored for others.119            generator (`torch.Generator`, *optional*):120                A [torch generator](https://pytorch.org/docs/stable/generated/torch.Generator.html) to make generation121                deterministic.122            latents (`torch.Tensor`, *optional*):123                Pre-generated noisy latents, sampled from a Gaussian distribution, to be used as inputs for image124                generation. Can be used to tweak the same generation with different prompts. If not provided, a latents125                tensor will ge generated by sampling using the supplied random `generator`.126            output_type (`str`, *optional*, defaults to `"pil"`):127                The output format of the generate image. Choose between128                [PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`.129            return_dict (`bool`, *optional*, defaults to `True`):130                Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a131                plain tuple.132            callback (`Callable`, *optional*):133                A function that will be called every `callback_steps` steps during inference. The function will be134                called with the following arguments: `callback(step: int, timestep: int, latents: torch.Tensor)`.135            callback_steps (`int`, *optional*, defaults to 1):136                The frequency at which the `callback` function will be called. If not specified, the callback will be137                called at every step.138 139        Returns:140            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`:141            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] if `return_dict` is True, otherwise a `tuple.142            When returning a tuple, the first element is a list with the generated images, and the second element is a143            list of `bool`s denoting whether the corresponding generated image likely represents "not-safe-for-work"144            (nsfw) content, according to the `safety_checker`.145        """146 147        if isinstance(prompt, str):148            batch_size = 1149        elif isinstance(prompt, list):150            batch_size = len(prompt)151        else:152            raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(prompt)}")153 154        if height % 8 != 0 or width % 8 != 0:155            raise ValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}.")156 157        if (callback_steps is None) or (158            callback_steps is not None and (not isinstance(callback_steps, int) or callback_steps <= 0)159        ):160            raise ValueError(161                f"`callback_steps` has to be a positive integer but is {callback_steps} of type"162                f" {type(callback_steps)}."163            )164 165        # get prompt text embeddings166        text_inputs = self.tokenizer(167            prompt,168            padding="max_length",169            max_length=self.tokenizer.model_max_length,170            return_tensors="pt",171        )172        text_input_ids = text_inputs.input_ids173 174        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:175            removed_text = self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :])176            logger.warning(177                "The following part of your input was truncated because CLIP can only handle sequences up to"178                f" {self.tokenizer.model_max_length} tokens: {removed_text}"179            )180            text_input_ids = text_input_ids[:, : self.tokenizer.model_max_length]181 182        if text_embeddings is None:183            text_embeddings = self.text_encoder(text_input_ids.to(self.device))[0]184 185        # duplicate text embeddings for each generation per prompt, using mps friendly method186        bs_embed, seq_len, _ = text_embeddings.shape187        text_embeddings = text_embeddings.repeat(1, num_images_per_prompt, 1)188        text_embeddings = text_embeddings.view(bs_embed * num_images_per_prompt, seq_len, -1)189 190        # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)191        # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`192        # corresponds to doing no classifier free guidance.193        do_classifier_free_guidance = guidance_scale > 1.0194        # get unconditional embeddings for classifier free guidance195        if do_classifier_free_guidance:196            uncond_tokens: List[str]197            if negative_prompt is None:198                uncond_tokens = [""]199            elif type(prompt) is not type(negative_prompt):200                raise TypeError(201                    f"`negative_prompt` should be the same type to `prompt`, but got {type(negative_prompt)} !="202                    f" {type(prompt)}."203                )204            elif isinstance(negative_prompt, str):205                uncond_tokens = [negative_prompt]206            elif batch_size != len(negative_prompt):207                raise ValueError(208                    f"`negative_prompt`: {negative_prompt} has batch size {len(negative_prompt)}, but `prompt`:"209                    f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"210                    " the batch size of `prompt`."211                )212            else:213                uncond_tokens = negative_prompt214 215            max_length = text_input_ids.shape[-1]216            uncond_input = self.tokenizer(217                uncond_tokens,218                padding="max_length",219                max_length=max_length,220                truncation=True,221                return_tensors="pt",222            )223            uncond_embeddings = self.text_encoder(uncond_input.input_ids.to(self.device))[0]224 225            # duplicate unconditional embeddings for each generation per prompt, using mps friendly method226            seq_len = uncond_embeddings.shape[1]227            uncond_embeddings = uncond_embeddings.repeat(batch_size, num_images_per_prompt, 1)228            uncond_embeddings = uncond_embeddings.view(batch_size * num_images_per_prompt, seq_len, -1)229 230            # For classifier free guidance, we need to do two forward passes.231            # Here we concatenate the unconditional and text embeddings into a single batch232            # to avoid doing two forward passes233            text_embeddings = torch.cat([uncond_embeddings, text_embeddings])234 235        # get the initial random noise unless the user supplied it236 237        # Unlike in other pipelines, latents need to be generated in the target device238        # for 1-to-1 results reproducibility with the CompVis implementation.239        # However this currently doesn't work in `mps`.240        latents_shape = (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)241        latents_shape_reference = (batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)242        latents_dtype = text_embeddings.dtype243        if latents is None:244            if self.device.type == "mps":245                # randn does not exist on mps246                latents_reference = torch.randn(247                    latents_shape_reference, generator=generator, device="cpu", dtype=latents_dtype248                ).to(self.device)249                latents = torch.randn(latents_shape, generator=generator, device="cpu", dtype=latents_dtype).to(250                    self.device251                )252            else:253                latents_reference = torch.randn(254                    latents_shape_reference, generator=generator, device=self.device, dtype=latents_dtype255                )256                latents = torch.randn(latents_shape, generator=generator, device=self.device, dtype=latents_dtype)257        else:258            if latents_reference.shape != latents_shape:259                raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}")260            latents_reference = latents_reference.to(self.device)261            latents = latents.to(self.device)262 263        # This is the key part of the pipeline where we264        # try to ensure that the generated images w/ the same seed265        # but different sizes actually result in similar images266        dx = (latents_shape[3] - latents_shape_reference[3]) // 2267        dy = (latents_shape[2] - latents_shape_reference[2]) // 2268        w = latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx269        h = latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy270        tx = 0 if dx < 0 else dx271        ty = 0 if dy < 0 else dy272        dx = max(-dx, 0)273        dy = max(-dy, 0)274        # import pdb275        # pdb.set_trace()276        latents[:, :, ty : ty + h, tx : tx + w] = latents_reference[:, :, dy : dy + h, dx : dx + w]277 278        # set timesteps279        self.scheduler.set_timesteps(num_inference_steps)280 281        # Some schedulers like PNDM have timesteps as arrays282        # It's more optimized to move all timesteps to correct device beforehand283        timesteps_tensor = self.scheduler.timesteps.to(self.device)284 285        # scale the initial noise by the standard deviation required by the scheduler286        latents = latents * self.scheduler.init_noise_sigma287 288        # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature289        # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.290        # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502291        # and should be between [0, 1]292        accepts_eta = "eta" in set(inspect.signature(self.scheduler.step).parameters.keys())293        extra_step_kwargs = {}294        if accepts_eta:295            extra_step_kwargs["eta"] = eta296 297        for i, t in enumerate(self.progress_bar(timesteps_tensor)):298            # expand the latents if we are doing classifier free guidance299            latent_model_input = torch.cat([latents] * 2) if do_classifier_free_guidance else latents300            latent_model_input = self.scheduler.scale_model_input(latent_model_input, t)301 302            # predict the noise residual303            noise_pred = self.unet(latent_model_input, t, encoder_hidden_states=text_embeddings).sample304 305            # perform guidance306            if do_classifier_free_guidance:307                noise_pred_uncond, noise_pred_text = noise_pred.chunk(2)308                noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)309 310            # compute the previous noisy sample x_t -> x_t-1311            latents = self.scheduler.step(noise_pred, t, latents, **extra_step_kwargs).prev_sample312 313            # call the callback, if provided314            if callback is not None and i % callback_steps == 0:315                step_idx = i // getattr(self.scheduler, "order", 1)316                callback(step_idx, t, latents)317 318        latents = 1 / 0.18215 * latents319        image = self.vae.decode(latents).sample320 321        image = (image / 2 + 0.5).clamp(0, 1)322 323        # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16324        image = image.cpu().permute(0, 2, 3, 1).float().numpy()325 326        if self.safety_checker is not None:327            safety_checker_input = self.feature_extractor(self.numpy_to_pil(image), return_tensors="pt").to(328                self.device329            )330            image, has_nsfw_concept = self.safety_checker(331                images=image, clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype)332            )333        else:334            has_nsfw_concept = None335 336        if output_type == "pil":337            image = self.numpy_to_pil(image)338 339        if not return_dict:340            return (image, has_nsfw_concept)341 342        return StableDiffusionPipelineOutput(images=image, nsfw_content_detected=has_nsfw_concept)343