





import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


snake_case 				=       logging.get_logger(__name__)  # pylint: disable=invalid-name





class 				A_				(      UpperCAmelCase	):







       """simple docstring"""





       def __init__(    self	:	Any						,__A	:	AutoencoderKL						,__A	:	CLIPTextModel						,__A	:	CLIPTokenizer						,__A	:	UNetaDConditionModel						,__A	:	Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]						,__A	:	StableDiffusionSafetyChecker						,__A	:	CLIPImageProcessor						,)						->					Optional[Any]:
         super().__init__()
         self.register_modules(
             vae=__A						,text_encoder=__A						,tokenizer=__A						,unet=__A						,scheduler=__A						,safety_checker=__A						,feature_extractor=__A						,)







       def    __UpperCAmelCase (    self	:	Any						,__A	:	Optional[Union[str, int]] = "auto"  )						->					str:
         if slice_size == "auto":
           # half the attention head size is usually a good trade-off between
           # speed and memory
           _lowercase							= self.unet.config.attention_head_dim // 2
         self.unet.set_attention_slice(__A  )







       def    __UpperCAmelCase (    self	:	List[str]  )						->					Tuple:
         self.enable_attention_slicing(__A  )







       @torch.no_grad()
       def __call__(    self	:	Tuple						,__A	:	Union[str, List[str]]						,__A	:	int = 512						,__A	:	int = 512						,__A	:	int = 50						,__A	:	float = 7.5						,__A	:	Optional[Union[str, List[str]]] = None						,__A	:	Optional[int] = 1						,__A	:	float = 0.0						,__A	:	Optional[torch.Generator] = None						,__A	:	Optional[torch.FloatTensor] = None						,__A	:	Optional[str] = "pil"						,__A	:	bool = True						,__A	:	Optional[Callable[[int, int, torch.FloatTensor], None]] = None						,__A	:	int = 1						,__A	:	Optional[torch.FloatTensor] = None						,**__A	:	Union[str, Any]						,)						->					str:

         if isinstance(__A						,__A  ):
           _lowercase							= 1
         elif isinstance(__A						,__A  ):
           _lowercase							= len(__A  )
         else:
           raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(__A  )}"""  )

         if height % 8 != 0 or width % 8 != 0:
           raise ValueError(F"""`height` and `width` have to be divisible by 8 but are {height} and {width}."""  )

         if (callback_steps is None) or (
             callback_steps is not None and (not isinstance(__A						,__A  ) or callback_steps <= 0)
         ):
           raise ValueError(
               F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
               F""" {type(__A  )}."""  )

         # get prompt text embeddings
         _lowercase							= self.tokenizer(
             __A						,padding='max_length'						,max_length=self.tokenizer.model_max_length						,return_tensors='pt'						,)
         _lowercase							= text_inputs.input_ids

         if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
           _lowercase							= self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]  )
           logger.warning(
               'The following part of your input was truncated because CLIP can only handle sequences up to'
               F""" {self.tokenizer.model_max_length} tokens: {removed_text}"""  )
           _lowercase							= text_input_ids[:, : self.tokenizer.model_max_length]

         if text_embeddings is None:
           _lowercase							= self.text_encoder(text_input_ids.to(self.device  )  )[0]

         # duplicate text embeddings for each generation per prompt, using mps friendly method
         _lowercase						,		_lowercase						,		_lowercase							= text_embeddings.shape
         _lowercase							= text_embeddings.repeat(1						,__A						,1  )
         _lowercase							= text_embeddings.view(bs_embed * num_images_per_prompt						,__A						,-1  )

         # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
         # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
         # corresponds to doing no classifier free guidance.
         _lowercase							= guidance_scale > 1.0
         # get unconditional embeddings for classifier free guidance
         if do_classifier_free_guidance:
           _lowercase							= 42
           if negative_prompt is None:
             _lowercase							= ['']
           elif type(__A  ) is not type(__A  ):
             raise TypeError(
                 F"""`negative_prompt` should be the same type to `prompt`, but got {type(__A  )} !="""
                 F""" {type(__A  )}."""  )
           elif isinstance(__A						,__A  ):
             _lowercase							= [negative_prompt]
           elif batch_size != len(__A  ):
             raise ValueError(
                 F"""`negative_prompt`: {negative_prompt} has batch size {len(__A  )}, but `prompt`:"""
                 F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
                 ' the batch size of `prompt`.'  )
           else:
             _lowercase							= negative_prompt

           _lowercase							= text_input_ids.shape[-1]
           _lowercase							= self.tokenizer(
               __A						,padding='max_length'						,max_length=__A						,truncation=__A						,return_tensors='pt'						,)
           _lowercase							= self.text_encoder(uncond_input.input_ids.to(self.device  )  )[0]

           # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
           _lowercase							= uncond_embeddings.shape[1]
           _lowercase							= uncond_embeddings.repeat(__A						,__A						,1  )
           _lowercase							= uncond_embeddings.view(batch_size * num_images_per_prompt						,__A						,-1  )

           # For classifier free guidance, we need to do two forward passes.
           # Here we concatenate the unconditional and text embeddings into a single batch
           # to avoid doing two forward passes
           _lowercase							= torch.cat([uncond_embeddings, text_embeddings]  )

         # get the initial random noise unless the user supplied it

         # Unlike in other pipelines, latents need to be generated in the target device
         # for 1-to-1 results reproducibility with the CompVis implementation.
         # However this currently doesn't work in `mps`.
         _lowercase							= (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
         _lowercase							= (batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
         _lowercase							= text_embeddings.dtype
         if latents is None:
           if self.device.type == "mps":
             # randn does not exist on mps
             _lowercase							= torch.randn(
                 __A						,generator=__A						,device='cpu'						,dtype=__A  ).to(self.device  )
             _lowercase							= torch.randn(__A						,generator=__A						,device='cpu'						,dtype=__A  ).to(
                 self.device  )
           else:
             _lowercase							= torch.randn(
                 __A						,generator=__A						,device=self.device						,dtype=__A  )
             _lowercase							= torch.randn(__A						,generator=__A						,device=self.device						,dtype=__A  )
         else:
           if latents_reference.shape != latents_shape:
             raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""  )
           _lowercase							= latents_reference.to(self.device  )
           _lowercase							= latents.to(self.device  )

         # This is the key part of the pipeline where we
         # try to ensure that the generated images w/ the same seed
         # but different sizes actually result in similar images
         _lowercase							= (latents_shape[3] - latents_shape_reference[3]) // 2
         _lowercase							= (latents_shape[2] - latents_shape_reference[2]) // 2
         _lowercase							= latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
         _lowercase							= latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
         _lowercase							= 0 if dx < 0 else dx
         _lowercase							= 0 if dy < 0 else dy
         _lowercase							= max(-dx						,0  )
         _lowercase							= max(-dy						,0  )
         # import pdb
         # pdb.set_trace()
         _lowercase							= latents_reference[:, :, dy : dy + h, dx : dx + w]

         # set timesteps
         self.scheduler.set_timesteps(__A  )

         # Some schedulers like PNDM have timesteps as arrays
         # It's more optimized to move all timesteps to correct device beforehand
         _lowercase							= self.scheduler.timesteps.to(self.device  )

         # scale the initial noise by the standard deviation required by the scheduler
         _lowercase							= latents * self.scheduler.init_noise_sigma

         # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
         # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
         # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
         # and should be between [0, 1]
         _lowercase							= 'eta' in set(inspect.signature(self.scheduler.step  ).parameters.keys()  )
         _lowercase							= {}
         if accepts_eta:
           _lowercase							= eta

         for i, t in enumerate(self.progress_bar(__A  )  ):
           # expand the latents if we are doing classifier free guidance
           _lowercase							= torch.cat([latents] * 2  ) if do_classifier_free_guidance else latents
           _lowercase							= self.scheduler.scale_model_input(__A						,__A  )

           # predict the noise residual
           _lowercase							= self.unet(__A						,__A						,encoder_hidden_states=__A  ).sample

           # perform guidance
           if do_classifier_free_guidance:
             _lowercase						,		_lowercase							= noise_pred.chunk(2  )
             _lowercase							= noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

           # compute the previous noisy sample x_t -> x_t-1
           _lowercase							= self.scheduler.step(__A						,__A						,__A						,**__A  ).prev_sample

           # call the callback, if provided
           if callback is not None and i % callback_steps == 0:
             callback(__A						,__A						,__A  )

         _lowercase							= 1 / 0.18215 * latents
         _lowercase							= self.vae.decode(__A  ).sample

         _lowercase							= (image / 2 + 0.5).clamp(0						,1  )

         # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
         _lowercase							= image.cpu().permute(0						,2						,3						,1  ).float().numpy()

         if self.safety_checker is not None:
           _lowercase							= self.feature_extractor(self.numpy_to_pil(__A  )						,return_tensors='pt'  ).to(
               self.device  )
           _lowercase						,		_lowercase							= self.safety_checker(
               images=__A						,clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype  )  )
         else:
           _lowercase							= None

         if output_type == "pil":
           _lowercase							= self.numpy_to_pil(__A  )

         if not return_dict:
           return (image, has_nsfw_concept)

         return StableDiffusionPipelineOutput(images=__A						,nsfw_content_detected=__A  )






def 							SCREAMING_SNAKE_CASE__       (  snake_case__  :str			,							snake_case__  :int     )      ->						list[str]:
  return [sentence[i : i + ngram_size] for i in range(len(snake_case__     ) - ngram_size + 1     )]


if __name__ == "__main__":
      from doctest import testmod

      testmod()