


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


SCREAMING_SNAKE_CASE       :		Optional[int]         =				logging.get_logger(__name__)  # pylint: disable=invalid-name






class      UpperCamelCase					(		lowercase__     ):


   '''simple docstring'''







   def __init__(     self   ,							UpperCamelCase_   ,							UpperCamelCase_   ,							UpperCamelCase_   ,							UpperCamelCase_   ,							UpperCamelCase_   ,							UpperCamelCase_   ,							UpperCamelCase_   ,							):
        super().__init__()
        self.register_modules(
            vae=UpperCamelCase_   ,							text_encoder=UpperCamelCase_   ,							tokenizer=UpperCamelCase_   ,							unet=UpperCamelCase_   ,							scheduler=UpperCamelCase_   ,							safety_checker=UpperCamelCase_   ,							feature_extractor=UpperCamelCase_   ,							)
   def 		UpperCamelCase  (     self   ,							UpperCamelCase_ = "auto"		):
        if slice_size == "auto":
             # half the attention head size is usually a good trade-off between
             # speed and memory
             lowercase_       :Optional[int]     =       self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(UpperCamelCase_		)
   def 		UpperCamelCase  (     self		):
        self.enable_attention_slicing(UpperCamelCase_		)




   @torch.no_grad()
   def __call__(     self   ,							UpperCamelCase_   ,							UpperCamelCase_ = 512   ,							UpperCamelCase_ = 512   ,							UpperCamelCase_ = 50   ,							UpperCamelCase_ = 7.5   ,							UpperCamelCase_ = None   ,							UpperCamelCase_ = 1   ,							UpperCamelCase_ = 0.0   ,							UpperCamelCase_ = None   ,							UpperCamelCase_ = None   ,							UpperCamelCase_ = "pil"   ,							UpperCamelCase_ = True   ,							UpperCamelCase_ = None   ,							UpperCamelCase_ = 1   ,							UpperCamelCase_ = None   ,							**UpperCamelCase_   ,							):

        if isinstance(UpperCamelCase_   ,							UpperCamelCase_		):
             lowercase_       :Optional[Any]     =       1
        elif isinstance(UpperCamelCase_   ,							UpperCamelCase_		):
             lowercase_       :Union[str, Any]     =       len(UpperCamelCase_		)
        else:
             raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(UpperCamelCase_		)}"		)

        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(UpperCamelCase_   ,							UpperCamelCase_		) or callback_steps <= 0)
        ):
             raise ValueError(
                 f"`callback_steps` has to be a positive integer but is {callback_steps} of type"
                 f" {type(UpperCamelCase_		)}."		)

        # get prompt text embeddings
        lowercase_       :List[str]     =       self.tokenizer(
            UpperCamelCase_   ,							padding='''max_length'''   ,							max_length=self.tokenizer.model_max_length   ,							return_tensors='''pt'''   ,							)
        lowercase_       :Dict     =       text_inputs.input_ids

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
             lowercase_       :List[str]     =       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_       :Optional[Any]     =       text_input_ids[:, : self.tokenizer.model_max_length]

        if text_embeddings is None:
             lowercase_       :Optional[Any]     =       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_       :List[Any]     =       text_embeddings.shape
        lowercase_       :Optional[Any]     =       text_embeddings.repeat(1   ,							UpperCamelCase_   ,							1		)
        lowercase_       :Union[str, Any]     =       text_embeddings.view(bs_embed * num_images_per_prompt   ,							UpperCamelCase_   ,							-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_       :Optional[Any]     =       guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
             lowercase_       :List[str]
             if negative_prompt is None:
                  lowercase_       :Any     =       ['''''']
             elif type(UpperCamelCase_		) is not type(UpperCamelCase_		):
                  raise TypeError(
                      f"`negative_prompt` should be the same type to `prompt`, but got {type(UpperCamelCase_		)} !="
                      f" {type(UpperCamelCase_		)}."		)
             elif isinstance(UpperCamelCase_   ,							UpperCamelCase_		):
                  lowercase_       :Tuple     =       [negative_prompt]
             elif batch_size != len(UpperCamelCase_		):
                  raise ValueError(
                      f"`negative_prompt`: {negative_prompt} has batch size {len(UpperCamelCase_		)}, but `prompt`:"
                      f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"
                      ''' the batch size of `prompt`.'''		)
             else:
                  lowercase_       :Optional[int]     =       negative_prompt

             lowercase_       :Optional[int]     =       text_input_ids.shape[-1]
             lowercase_       :Tuple     =       self.tokenizer(
                 UpperCamelCase_   ,							padding='''max_length'''   ,							max_length=UpperCamelCase_   ,							truncation=UpperCamelCase_   ,							return_tensors='''pt'''   ,							)
             lowercase_       :Dict     =       self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]

             # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
             lowercase_       :Tuple     =       uncond_embeddings.shape[1]
             lowercase_       :Optional[int]     =       uncond_embeddings.repeat(UpperCamelCase_   ,							UpperCamelCase_   ,							1		)
             lowercase_       :Optional[Any]     =       uncond_embeddings.view(batch_size * num_images_per_prompt   ,							UpperCamelCase_   ,							-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_       :Optional[int]     =       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_       :int     =       (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
        lowercase_       :Union[str, Any]     =       (batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
        lowercase_       :List[str]     =       text_embeddings.dtype
        if latents is None:
             if self.device.type == "mps":
                  # randn does not exist on mps
                  lowercase_       :Tuple     =       torch.randn(
                      UpperCamelCase_   ,							generator=UpperCamelCase_   ,							device='''cpu'''   ,							dtype=UpperCamelCase_		).to(self.device		)
                  lowercase_       :Union[str, Any]     =       torch.randn(UpperCamelCase_   ,							generator=UpperCamelCase_   ,							device='''cpu'''   ,							dtype=UpperCamelCase_		).to(
                      self.device		)
             else:
                  lowercase_       :int     =       torch.randn(
                      UpperCamelCase_   ,							generator=UpperCamelCase_   ,							device=self.device   ,							dtype=UpperCamelCase_		)
                  lowercase_       :int     =       torch.randn(UpperCamelCase_   ,							generator=UpperCamelCase_   ,							device=self.device   ,							dtype=UpperCamelCase_		)
        else:
             if latents_reference.shape != latents_shape:
                  raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}"		)
             lowercase_       :int     =       latents_reference.to(self.device		)
             lowercase_       :Dict     =       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_       :Optional[int]     =       (latents_shape[3] - latents_shape_reference[3]) // 2
        lowercase_       :Optional[Any]     =       (latents_shape[2] - latents_shape_reference[2]) // 2
        lowercase_       :Tuple     =       latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
        lowercase_       :Dict     =       latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
        lowercase_       :Union[str, Any]     =       0 if dx < 0 else dx
        lowercase_       :Union[str, Any]     =       0 if dy < 0 else dy
        lowercase_       :Any     =       max(-dx   ,							0		)
        lowercase_       :List[Any]     =       max(-dy   ,							0		)
        # import pdb
        # pdb.set_trace()
        lowercase_       :Optional[Any]     =       latents_reference[:, :, dy : dy + h, dx : dx + w]

        # set timesteps
        self.scheduler.set_timesteps(UpperCamelCase_		)

        # Some schedulers like PNDM have timesteps as arrays
        # It's more optimized to move all timesteps to correct device beforehand
        lowercase_       :List[str]     =       self.scheduler.timesteps.to(self.device		)

        # scale the initial noise by the standard deviation required by the scheduler
        lowercase_       :Dict     =       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_       :Tuple     =       '''eta''' in set(inspect.signature(self.scheduler.step		).parameters.keys()		)
        lowercase_       :Dict     =       {}
        if accepts_eta:
             lowercase_       :int     =       eta

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

             # predict the noise residual
             lowercase_       :Any     =       self.unet(UpperCamelCase_   ,							UpperCamelCase_   ,							encoder_hidden_states=UpperCamelCase_		).sample

             # perform guidance
             if do_classifier_free_guidance:
                  lowercase_    ,      lowercase_       :List[Any]     =       noise_pred.chunk(2		)
                  lowercase_       :Union[str, Any]     =       noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

             # compute the previous noisy sample x_t -> x_t-1
             lowercase_       :Tuple     =       self.scheduler.step(UpperCamelCase_   ,							UpperCamelCase_   ,							UpperCamelCase_   ,							**UpperCamelCase_		).prev_sample

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

        lowercase_       :List[Any]     =       1 / 0.1_8215 * latents
        lowercase_       :Tuple     =       self.vae.decode(UpperCamelCase_		).sample

        lowercase_       :Tuple     =       (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_       :Optional[int]     =       image.cpu().permute(0   ,							2   ,							3   ,							1		).float().numpy()

        if self.safety_checker is not None:
             lowercase_       :int     =       self.feature_extractor(self.numpy_to_pil(UpperCamelCase_		)   ,							return_tensors='''pt'''		).to(
                 self.device		)
             lowercase_    ,      lowercase_       :Union[str, Any]     =       self.safety_checker(
                 images=UpperCamelCase_   ,							clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype		)		)
        else:
             lowercase_       :Tuple     =       None

        if output_type == "pil":
             lowercase_       :List[Any]     =       self.numpy_to_pil(UpperCamelCase_		)

        if not return_dict:
             return (image, has_nsfw_concept)

        return StableDiffusionPipelineOutput(images=UpperCamelCase_   ,							nsfw_content_detected=UpperCamelCase_		)





def     UpperCamelCase    (      _a       ,		_a		)					->							int:
     '''simple docstring'''





     while a != 0:
          lowercase_    ,      lowercase_       :Union[str, Any]     =       b % a, a
     return b




def     UpperCamelCase    (      _a       ,		_a		)					->							int:
     '''simple docstring'''





     if gcd(_a       ,		_a		) != 1:
          lowercase_       :Union[str, Any]     =       f"mod inverse of {a!r} and {m!r} does not exist"
          raise ValueError(_a		)
     lowercase_    ,      lowercase_    ,      lowercase_       :Any     =       1, 0, a
     lowercase_    ,      lowercase_    ,      lowercase_       :List[str]     =       0, 1, m
     while va != 0:
          lowercase_       :Tuple     =       ua // va
          lowercase_    ,      lowercase_    ,      lowercase_    ,      lowercase_    ,      lowercase_    ,      lowercase_       :Optional[int]     =       (ua - q * va), (ua - q * va), (ua - q * va), va, va, va
     return ua % m

