





'''simple docstring'''

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


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







class __a (   _snake_case			):





   def __init__(					self				:			str				,lowerCamelCase				:			AutoencoderKL				,lowerCamelCase				:			CLIPTextModel				,lowerCamelCase				:			CLIPTokenizer				,lowerCamelCase				:			UNetaDConditionModel				,lowerCamelCase				:			Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]				,lowerCamelCase				:			StableDiffusionSafetyChecker				,lowerCamelCase				:			CLIPImageProcessor				,):

      '''simple docstring'''

      super().__init__()
      self.register_modules(
          vae=lowerCamelCase				,text_encoder=lowerCamelCase				,tokenizer=lowerCamelCase				,unet=lowerCamelCase				,scheduler=lowerCamelCase				,safety_checker=lowerCamelCase				,feature_extractor=lowerCamelCase				,)





   def      UpperCAmelCase__     (					self				:			Any				,lowerCamelCase				:			Optional[Union[str, int]] = "auto"    ):

      '''simple docstring'''

      if slice_size == "auto":
         # half the attention head size is usually a good trade-off between
         # speed and memory
         __SCREAMING_SNAKE_CASE      							=       self.unet.config.attention_head_dim // 2
      self.unet.set_attention_slice(lowerCamelCase    )





   def      UpperCAmelCase__     (					self				:			Dict    ):

      '''simple docstring'''

      self.enable_attention_slicing(lowerCamelCase    )





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

      '''simple docstring'''

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

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

      # get prompt text embeddings
      __SCREAMING_SNAKE_CASE      							=       self.tokenizer(
          lowerCamelCase				,padding="""max_length"""				,max_length=self.tokenizer.model_max_length				,return_tensors="""pt"""				,)
      __SCREAMING_SNAKE_CASE      							=       text_inputs.input_ids

      if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
         __SCREAMING_SNAKE_CASE      							=       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}"""    )
         __SCREAMING_SNAKE_CASE      							=       text_input_ids[:, : self.tokenizer.model_max_length]

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

      # duplicate text embeddings for each generation per prompt, using mps friendly method
      __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE      							=       text_embeddings.shape
      __SCREAMING_SNAKE_CASE      							=       text_embeddings.repeat(1				,lowerCamelCase				,1    )
      __SCREAMING_SNAKE_CASE      							=       text_embeddings.view(bs_embed * num_images_per_prompt				,lowerCamelCase				,-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.
      __SCREAMING_SNAKE_CASE      							=       guidance_scale > 1.0
      # get unconditional embeddings for classifier free guidance
      if do_classifier_free_guidance:
         __SCREAMING_SNAKE_CASE      							=       42
         if negative_prompt is None:
            __SCREAMING_SNAKE_CASE      							=       [""""""]
         elif type(lowerCamelCase    ) is not type(lowerCamelCase    ):
            raise TypeError(
                f"""`negative_prompt` should be the same type to `prompt`, but got {type(lowerCamelCase    )} !="""
                f""" {type(lowerCamelCase    )}."""    )
         elif isinstance(lowerCamelCase				,lowerCamelCase    ):
            __SCREAMING_SNAKE_CASE      							=       [negative_prompt]
         elif batch_size != len(lowerCamelCase    ):
            raise ValueError(
                f"""`negative_prompt`: {negative_prompt} has batch size {len(lowerCamelCase    )}, but `prompt`:"""
                f""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
                """ the batch size of `prompt`."""    )
         else:
            __SCREAMING_SNAKE_CASE      							=       negative_prompt

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

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

      # set timesteps
      self.scheduler.set_timesteps(lowerCamelCase    )

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

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

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

         # predict the noise residual
         __SCREAMING_SNAKE_CASE      							=       self.unet(lowerCamelCase				,lowerCamelCase				,encoder_hidden_states=lowerCamelCase    ).sample

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

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

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

      __SCREAMING_SNAKE_CASE      							=       1 / 0.18_215 * latents
      __SCREAMING_SNAKE_CASE      							=       self.vae.decode(lowerCamelCase    ).sample

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

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

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

      if output_type == "pil":
         __SCREAMING_SNAKE_CASE      							=       self.numpy_to_pil(lowerCamelCase    )

      if not return_dict:
         return (image, has_nsfw_concept)

      return StableDiffusionPipelineOutput(images=lowerCamelCase				,nsfw_content_detected=lowerCamelCase    )












'''simple docstring'''

import math
from enum import Enum
from typing import Optional, Union

from torch.optim import Optimizer
from torch.optim.lr_scheduler import LambdaLR

from .utils import logging


a        =						logging.get_logger(__name__)







class __a (   _snake_case			):
   __UpperCamelCase   : int	      =	'linear'
   __UpperCamelCase   : Tuple	      =	'cosine'
   __UpperCamelCase   : Tuple	      =	'cosine_with_restarts'
   __UpperCamelCase   : List[Any]	      =	'polynomial'
   __UpperCamelCase   : Optional[Any]	      =	'constant'
   __UpperCamelCase   : Optional[int]	      =	'constant_with_warmup'
   __UpperCamelCase   : List[Any]	      =	'piecewise_constant'





def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase = -1      )						->					int:

   '''simple docstring'''







   return LambdaLR(__UpperCAmelCase							,	lambda __UpperCAmelCase      : 1							,	last_epoch=__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = -1      )						->					List[Any]:

   '''simple docstring'''







   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1.0							,	__UpperCAmelCase      )      )
      return 1.0

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = -1      )						->					int:

   '''simple docstring'''







   __SCREAMING_SNAKE_CASE      							=       {}
   __SCREAMING_SNAKE_CASE      							=       step_rules.split(""","""      )
   for rule_str in rule_list[:-1]:
      __SCREAMING_SNAKE_CASE							,     __SCREAMING_SNAKE_CASE      							=       rule_str.split(""":"""      )
      __SCREAMING_SNAKE_CASE      							=       int(__UpperCAmelCase      )
      __SCREAMING_SNAKE_CASE      							=       float(__UpperCAmelCase      )
      __SCREAMING_SNAKE_CASE      							=       value
   __SCREAMING_SNAKE_CASE      							=       float(rule_list[-1]      )

   def create_rules_function(__UpperCAmelCase							,	__UpperCAmelCase      ):
      def rule_func(__UpperCAmelCase      ) -> float:
         __SCREAMING_SNAKE_CASE      							=       sorted(rules_dict.keys()      )
         for i, sorted_step in enumerate(__UpperCAmelCase      ):
            if steps < sorted_step:
               return rules_dict[sorted_steps[i]]
         return last_lr_multiple

      return rule_func

   __SCREAMING_SNAKE_CASE      							=       create_rules_function(__UpperCAmelCase							,	__UpperCAmelCase      )

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase=-1      )						->					int:

   '''simple docstring'''







   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1							,	__UpperCAmelCase      )      )
      return max(
          0.0							,	float(num_training_steps - current_step      ) / float(max(1							,	num_training_steps - num_warmup_steps      )      )      )

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = 0.5							,	__UpperCAmelCase = -1      )						->					Dict:

   '''simple docstring'''







   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1							,	__UpperCAmelCase      )      )
      __SCREAMING_SNAKE_CASE      							=       float(current_step - num_warmup_steps      ) / float(max(1							,	num_training_steps - num_warmup_steps      )      )
      return max(0.0							,	0.5 * (1.0 + math.cos(math.pi * float(__UpperCAmelCase      ) * 2.0 * progress      ))      )

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = 1							,	__UpperCAmelCase = -1      )						->					Tuple:

   '''simple docstring'''







   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1							,	__UpperCAmelCase      )      )
      __SCREAMING_SNAKE_CASE      							=       float(current_step - num_warmup_steps      ) / float(max(1							,	num_training_steps - num_warmup_steps      )      )
      if progress >= 1.0:
         return 0.0
      return max(0.0							,	0.5 * (1.0 + math.cos(math.pi * ((float(__UpperCAmelCase      ) * progress) % 1.0)      ))      )

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase      )

def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase=1e-7							,	__UpperCAmelCase=1.0							,	__UpperCAmelCase=-1      )						->					Tuple:

   '''simple docstring'''







   __SCREAMING_SNAKE_CASE      							=       optimizer.defaults["""lr"""]
   if not (lr_init > lr_end):
      raise ValueError(f"""lr_end ({lr_end}) must be be smaller than initial lr ({lr_init})"""      )

   def lr_lambda(__UpperCAmelCase      ):
      if current_step < num_warmup_steps:
         return float(__UpperCAmelCase      ) / float(max(1							,	__UpperCAmelCase      )      )
      elif current_step > num_training_steps:
         return lr_end / lr_init  # as LambdaLR multiplies by lr_init
      else:
         __SCREAMING_SNAKE_CASE      							=       lr_init - lr_end
         __SCREAMING_SNAKE_CASE      							=       num_training_steps - num_warmup_steps
         __SCREAMING_SNAKE_CASE      							=       1 - (current_step - num_warmup_steps) / decay_steps
         __SCREAMING_SNAKE_CASE      							=       lr_range * pct_remaining**power + lr_end
         return decay / lr_init  # as LambdaLR multiplies by lr_init

   return LambdaLR(__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase      )


a        =						{
    SchedulerType.LINEAR: get_linear_schedule_with_warmup,
    SchedulerType.COSINE: get_cosine_schedule_with_warmup,
    SchedulerType.COSINE_WITH_RESTARTS: get_cosine_with_hard_restarts_schedule_with_warmup,
    SchedulerType.POLYNOMIAL: get_polynomial_decay_schedule_with_warmup,
    SchedulerType.CONSTANT: get_constant_schedule,
    SchedulerType.CONSTANT_WITH_WARMUP: get_constant_schedule_with_warmup,
    SchedulerType.PIECEWISE_CONSTANT: get_piecewise_constant_schedule,
}







def 						__magic_name__       (							__UpperCAmelCase							,	__UpperCAmelCase							,	__UpperCAmelCase = None							,	__UpperCAmelCase = None							,	__UpperCAmelCase = None							,	__UpperCAmelCase = 1							,	__UpperCAmelCase = 1.0							,	__UpperCAmelCase = -1							,	)						->					str:

   '''simple docstring'''







   __SCREAMING_SNAKE_CASE      							=       SchedulerType(__UpperCAmelCase      )
   __SCREAMING_SNAKE_CASE      							=       TYPE_TO_SCHEDULER_FUNCTION[name]
   if name == SchedulerType.CONSTANT:
      return schedule_func(__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

   if name == SchedulerType.PIECEWISE_CONSTANT:
      return schedule_func(__UpperCAmelCase							,	step_rules=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

   # All other schedulers require `num_warmup_steps`
   if num_warmup_steps is None:
      raise ValueError(f"""{name} requires `num_warmup_steps`, please provide that argument."""      )

   if name == SchedulerType.CONSTANT_WITH_WARMUP:
      return schedule_func(__UpperCAmelCase							,	num_warmup_steps=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )

   # All other schedulers require `num_training_steps`
   if num_training_steps is None:
      raise ValueError(f"""{name} requires `num_training_steps`, please provide that argument."""      )

   if name == SchedulerType.COSINE_WITH_RESTARTS:
      return schedule_func(
          __UpperCAmelCase							,	num_warmup_steps=__UpperCAmelCase							,	num_training_steps=__UpperCAmelCase							,	num_cycles=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase							,	)

   if name == SchedulerType.POLYNOMIAL:
      return schedule_func(
          __UpperCAmelCase							,	num_warmup_steps=__UpperCAmelCase							,	num_training_steps=__UpperCAmelCase							,	power=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase							,	)

   return schedule_func(
       __UpperCAmelCase							,	num_warmup_steps=__UpperCAmelCase							,	num_training_steps=__UpperCAmelCase							,	last_epoch=__UpperCAmelCase      )





