





"""simple docstring"""
import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import (
    PIL_INTERPOLATION,
    randn_tensor,
)




def       __snake_case    (   __A			,__A			,__A     )		->					int:


   if isinstance(_UpperCamelCase			,torch.Tensor     ):
      return image
   elif isinstance(_UpperCamelCase			,PIL.Image.Image     ):
      lowercase     :      Union[str, Any]	=							[image]

   if isinstance(image[0]			,PIL.Image.Image     ):
      lowercase     :      Tuple	=							[np.array(i.resize((w, h)			,resample=PIL_INTERPOLATION["""lanczos"""]     )     )[None, :] for i in image]
      lowercase     :      List[str]	=							np.concatenate(_UpperCamelCase			,axis=0     )
      lowercase     :      Any	=							np.array(_UpperCamelCase     ).astype(np.floataa     ) / 255.0
      lowercase     :      Optional[int]	=							image.transpose(0			,3			,1			,2     )
      lowercase     :      str	=							2.0 * image - 1.0
      lowercase     :      Any	=							torch.from_numpy(_UpperCamelCase     )
   elif isinstance(image[0]			,torch.Tensor     ):
      lowercase     :      Any	=							torch.cat(_UpperCamelCase			,dim=0     )
   return image




def       __snake_case    (   __A			,__A			,__A			,__A=0.99_95     )		->					str:


   if not isinstance(_UpperCamelCase			,np.ndarray     ):
      lowercase     :      Dict	=							True
      lowercase     :      Dict	=							va.device
      lowercase     :      Tuple	=							va.cpu().numpy()
      lowercase     :      int	=							va.cpu().numpy()

   lowercase     :      Dict	=							np.sum(va * va / (np.linalg.norm(_UpperCamelCase     ) * np.linalg.norm(_UpperCamelCase     ))     )
   if np.abs(_UpperCamelCase     ) > DOT_THRESHOLD:
      lowercase     :      Any	=							(1 - t) * va + t * va
   else:
      lowercase     :      Optional[int]	=							np.arccos(_UpperCamelCase     )
      lowercase     :      Any	=							np.sin(_UpperCamelCase     )
      lowercase     :      Optional[Any]	=							theta_a * t
      lowercase     :      str	=							np.sin(_UpperCamelCase     )
      lowercase     :      Dict	=							np.sin(theta_a - theta_t     ) / sin_theta_a
      lowercase     :      Dict	=							sin_theta_t / sin_theta_a
      lowercase     :      Optional[int]	=							sa * va + sa * va

   if inputs_are_torch:
      lowercase     :      List[str]	=							torch.from_numpy(_UpperCamelCase     ).to(_UpperCamelCase     )

   return va




def       __snake_case    (   __A			,__A     )		->					List[str]:
   lowercase     :      Union[str, Any]	=							F.normalize(_UpperCamelCase			,dim=-1     )
   lowercase     :      Optional[Any]	=							F.normalize(_UpperCamelCase			,dim=-1     )
   return (x - y).norm(dim=-1     ).div(2     ).arcsin().pow(2     ).mul(2     )




def       __snake_case    (   __A			,__A     )		->					Optional[int]:


   for param in model.parameters():
      lowercase     :      List[str]	=							value







class lowerCamelCase__	(     __UpperCamelCase					):



   def __init__(							self   , snake_case   , snake_case   , snake_case   , snake_case   , snake_case   , snake_case   , snake_case   , snake_case=None   , snake_case=None   , snake_case=None   , )			->     str:







      """simple docstring"""
      super().__init__()
      self.register_modules(
          vae=snake_case__   , text_encoder=snake_case__   , clip_model=snake_case__   , tokenizer=snake_case__   , unet=snake_case__   , scheduler=snake_case__   , feature_extractor=snake_case__   , coca_model=snake_case__   , coca_tokenizer=snake_case__   , coca_transform=snake_case__   , )
      lowercase     :      Tuple	=							(
          feature_extractor.size
          if isinstance(feature_extractor.size   , snake_case__						)
          else feature_extractor.size["""shortest_edge"""]
      )
      lowercase     :      Optional[int]	=							transforms.Normalize(mean=feature_extractor.image_mean   , std=feature_extractor.image_std						)
      set_requires_grad(self.text_encoder   , snake_case__						)
      set_requires_grad(self.clip_model   , snake_case__						)



   def  _UpperCAmelCase						(							self   , snake_case = "auto"						)			->     List[str]:







      """simple docstring"""






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



   def  _UpperCAmelCase						(							self						)			->     int:







      """simple docstring"""
      self.enable_attention_slicing(snake_case__						)



   def  _UpperCAmelCase						(							self						)			->     Optional[int]:







      """simple docstring"""
      set_requires_grad(self.vae   , snake_case__						)



   def  _UpperCAmelCase						(							self						)			->     Optional[int]:







      """simple docstring"""
      set_requires_grad(self.vae   , snake_case__						)



   def  _UpperCAmelCase						(							self						)			->     int:







      """simple docstring"""
      set_requires_grad(self.unet   , snake_case__						)



   def  _UpperCAmelCase						(							self						)			->     Optional[Any]:







      """simple docstring"""
      set_requires_grad(self.unet   , snake_case__						)



   def  _UpperCAmelCase						(							self   , snake_case   , snake_case   , snake_case						)			->     Any:







      """simple docstring"""
      lowercase     :      Dict	=							min(int(num_inference_steps * strength						)   , snake_case__						)

      lowercase     :      int	=							max(num_inference_steps - init_timestep   , 0						)
      lowercase     :      Tuple	=							self.scheduler.timesteps[t_start:]

      return timesteps, num_inference_steps - t_start



   def  _UpperCAmelCase						(							self   , snake_case   , snake_case   , snake_case   , snake_case   , snake_case   , snake_case=None						)			->     Tuple:







      """simple docstring"""






      if not isinstance(snake_case__   , torch.Tensor						):
         raise ValueError(f'''`image` has to be of type `torch.Tensor` but is {type(snake_case__						)}'''						)

      lowercase     :      List[Any]	=							image.to(device=snake_case__   , dtype=snake_case__						)

      if isinstance(snake_case__   , snake_case__						):
         lowercase     :      str	=							[
             self.vae.encode(image[i : i + 1]						).latent_dist.sample(generator[i]						) for i in range(snake_case__						)
         ]
         lowercase     :      Union[str, Any]	=							torch.cat(snake_case__   , dim=0						)
      else:
         lowercase     :      int	=							self.vae.encode(snake_case__						).latent_dist.sample(snake_case__						)

      # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      lowercase     :      Any	=							0.1_82_15 * init_latents
      lowercase     :      str	=							init_latents.repeat_interleave(snake_case__   , dim=0						)

      lowercase     :      Any	=							randn_tensor(init_latents.shape   , generator=snake_case__   , device=snake_case__   , dtype=snake_case__						)

      # get latents
      lowercase     :      Union[str, Any]	=							self.scheduler.add_noise(snake_case__   , snake_case__   , snake_case__						)
      lowercase     :      Optional[int]	=							init_latents

      return latents



   def  _UpperCAmelCase						(							self   , snake_case						)			->     str:







      """simple docstring"""
      lowercase     :      List[str]	=							self.coca_transform(snake_case__						).unsqueeze(0						)
      with torch.no_grad(), torch.cuda.amp.autocast():
         lowercase     :      Dict	=							self.coca_model.generate(transformed_image.to(device=self.device   , dtype=self.coca_model.dtype						)						)
      lowercase     :      Optional[Any]	=							self.coca_tokenizer.decode(generated[0].cpu().numpy()						)
      return generated.split("""<end_of_text>"""						)[0].replace("""<start_of_text>"""   , """"""						).rstrip(""" .,"""						)



   def  _UpperCAmelCase						(							self   , snake_case   , snake_case						)			->     Any:







      """simple docstring"""
      lowercase     :      int	=							self.feature_extractor.preprocess(snake_case__						)
      lowercase     :      Any	=							torch.from_numpy(clip_image_input["""pixel_values"""][0]						).unsqueeze(0						).to(self.device						).half()
      lowercase     :      Tuple	=							self.clip_model.get_image_features(snake_case__						)
      lowercase     :      Dict	=							image_embeddings_clip / image_embeddings_clip.norm(p=2   , dim=-1   , keepdim=snake_case__						)
      lowercase     :      Any	=							image_embeddings_clip.repeat_interleave(snake_case__   , dim=0						)
      return image_embeddings_clip



   @torch.enable_grad()
   def  _UpperCAmelCase						(							self   , snake_case   , snake_case   , snake_case   , snake_case   , snake_case   , snake_case   , snake_case   , )			->     Dict:







      """simple docstring"""
      lowercase     :      List[Any]	=							latents.detach().requires_grad_()

      lowercase     :      Any	=							self.scheduler.scale_model_input(snake_case__   , snake_case__						)

      # predict the noise residual
      lowercase     :      List[str]	=							self.unet(snake_case__   , snake_case__   , encoder_hidden_states=snake_case__						).sample

      if isinstance(self.scheduler   , (PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)						):
         lowercase     :      int	=							self.scheduler.alphas_cumprod[timestep]
         lowercase     :      Optional[int]	=							1 - alpha_prod_t
         # compute predicted original sample from predicted noise also called
         # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
         lowercase     :      str	=							(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

         lowercase     :      Union[str, Any]	=							torch.sqrt(snake_case__						)
         lowercase     :      int	=							pred_original_sample * (fac) + latents * (1 - fac)
      elif isinstance(self.scheduler   , snake_case__						):
         lowercase     :      str	=							self.scheduler.sigmas[index]
         lowercase     :      str	=							latents - sigma * noise_pred
      else:
         raise ValueError(f'''scheduler type {type(self.scheduler						)} not supported'''						)

      # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      lowercase     :      Dict	=							1 / 0.1_82_15 * sample
      lowercase     :      Dict	=							self.vae.decode(snake_case__						).sample
      lowercase     :      Tuple	=							(image / 2 + 0.5).clamp(0   , 1						)

      lowercase     :      Union[str, Any]	=							transforms.Resize(self.feature_extractor_size						)(snake_case__						)
      lowercase     :      List[Any]	=							self.normalize(snake_case__						).to(latents.dtype						)

      lowercase     :      Tuple	=							self.clip_model.get_image_features(snake_case__						)
      lowercase     :      Optional[Any]	=							image_embeddings_clip / image_embeddings_clip.norm(p=2   , dim=-1   , keepdim=snake_case__						)

      lowercase     :      str	=							spherical_dist_loss(snake_case__   , snake_case__						).mean() * clip_guidance_scale

      lowercase     :      Dict	=							-torch.autograd.grad(snake_case__   , snake_case__						)[0]

      if isinstance(self.scheduler   , snake_case__						):
         lowercase     :      List[Any]	=							latents.detach() + grads * (sigma**2)
         lowercase     :      Tuple	=							noise_pred_original
      else:
         lowercase     :      List[str]	=							noise_pred_original - torch.sqrt(snake_case__						) * grads
      return noise_pred, latents





   @torch.no_grad()
   def __call__(							self   , snake_case   , snake_case   , snake_case = None   , snake_case = None   , snake_case = 5_1_2   , snake_case = 5_1_2   , snake_case = 0.6   , snake_case = 5_0   , snake_case = 7.5   , snake_case = 1   , snake_case = 0.0   , snake_case = 1_0_0   , snake_case = None   , snake_case = "pil"   , snake_case = True   , snake_case = 0.8   , snake_case = 0.1   , snake_case = 0.1   , )			->     Union[str, Any]:







      """simple docstring"""






      if isinstance(snake_case__   , snake_case__						) and len(snake_case__						) != batch_size:
         raise ValueError(f'''You have passed {batch_size} batch_size, but only {len(snake_case__						)} generators.'''						)

      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 isinstance(snake_case__   , torch.Generator						) and batch_size > 1:
         lowercase     :      int	=							[generator] + [None] * (batch_size - 1)

      lowercase     :      List[Any]	=							[
          ("""model""", self.coca_model is None),
          ("""tokenizer""", self.coca_tokenizer is None),
          ("""transform""", self.coca_transform is None),
      ]
      lowercase     :      Tuple	=							[x[0] for x in coca_is_none if x[1]]
      lowercase     :      Any	=							""", """.join(snake_case__						)
      # generate prompts with coca model if prompt is None
      if content_prompt is None:
         if len(snake_case__						):
            raise ValueError(
                f'''Content prompt is None and CoCa [{coca_is_none_str}] is None.'''
                f'''Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.'''						)
         lowercase     :      Any	=							self.get_image_description(snake_case__						)
      if style_prompt is None:
         if len(snake_case__						):
            raise ValueError(
                f'''Style prompt is None and CoCa [{coca_is_none_str}] is None.'''
                f''' Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.'''						)
         lowercase     :      List[str]	=							self.get_image_description(snake_case__						)

      # get prompt text embeddings for content and style
      lowercase     :      Tuple	=							self.tokenizer(
          snake_case__   , padding="""max_length"""   , max_length=self.tokenizer.model_max_length   , truncation=snake_case__   , return_tensors="""pt"""   , )
      lowercase     :      Tuple	=							self.text_encoder(content_text_input.input_ids.to(self.device						)						)[0]

      lowercase     :      Union[str, Any]	=							self.tokenizer(
          snake_case__   , padding="""max_length"""   , max_length=self.tokenizer.model_max_length   , truncation=snake_case__   , return_tensors="""pt"""   , )
      lowercase     :      Tuple	=							self.text_encoder(style_text_input.input_ids.to(self.device						)						)[0]

      lowercase     :      str	=							slerp(snake_case__   , snake_case__   , snake_case__						)

      # duplicate text embeddings for each generation per prompt
      lowercase     :      Dict	=							text_embeddings.repeat_interleave(snake_case__   , dim=0						)

      # set timesteps
      lowercase     :      int	=							"""offset""" in set(inspect.signature(self.scheduler.set_timesteps						).parameters.keys()						)
      lowercase     :      List[Any]	=							{}
      if accepts_offset:
         lowercase     :      Tuple	=							1

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

      lowercase	,       lowercase     :      Any	=							self.get_timesteps(snake_case__   , snake_case__   , self.device						)
      lowercase     :      Optional[Any]	=							timesteps[:1].repeat(snake_case__						)

      # Preprocess image
      lowercase     :      Tuple	=							preprocess(snake_case__   , snake_case__   , snake_case__						)
      lowercase     :      Optional[Any]	=							self.prepare_latents(
          snake_case__   , snake_case__   , snake_case__   , text_embeddings.dtype   , self.device   , snake_case__						)

      lowercase     :      int	=							preprocess(snake_case__   , snake_case__   , snake_case__						)
      lowercase     :      List[str]	=							self.prepare_latents(
          snake_case__   , snake_case__   , snake_case__   , text_embeddings.dtype   , self.device   , snake_case__						)

      lowercase     :      Any	=							slerp(snake_case__   , snake_case__   , snake_case__						)

      if clip_guidance_scale > 0:
         lowercase     :      Union[str, Any]	=							self.get_clip_image_embeddings(snake_case__   , snake_case__						)
         lowercase     :      str	=							self.get_clip_image_embeddings(snake_case__   , snake_case__						)
         lowercase     :      int	=							slerp(
             snake_case__   , snake_case__   , snake_case__						)

      # 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     :      Dict	=							guidance_scale > 1.0
      # get unconditional embeddings for classifier free guidance
      if do_classifier_free_guidance:
         lowercase     :      List[str]	=							content_text_input.input_ids.shape[-1]
         lowercase     :      int	=							self.tokenizer([""""""]   , padding="""max_length"""   , max_length=snake_case__   , return_tensors="""pt"""						)
         lowercase     :      Union[str, Any]	=							self.text_encoder(uncond_input.input_ids.to(self.device						)						)[0]
         # duplicate unconditional embeddings for each generation per prompt
         lowercase     :      Optional[int]	=							uncond_embeddings.repeat_interleave(snake_case__   , dim=0						)

         # 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     :      Dict	=							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     :      List[str]	=							(batch_size, self.unet.config.in_channels, height // 8, width // 8)
      lowercase     :      str	=							text_embeddings.dtype
      if latents is None:
         if self.device.type == "mps":
            # randn does not work reproducibly on mps
            lowercase     :      List[Any]	=							torch.randn(snake_case__   , generator=snake_case__   , device="""cpu"""   , dtype=snake_case__						).to(
                self.device						)
         else:
            lowercase     :      int	=							torch.randn(snake_case__   , generator=snake_case__   , device=self.device   , dtype=snake_case__						)
      else:
         if latents.shape != latents_shape:
            raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''						)
         lowercase     :      Any	=							latents.to(self.device						)

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

      # check if the scheduler accepts generator
      lowercase     :      int	=							"""generator""" in set(inspect.signature(self.scheduler.step						).parameters.keys()						)
      if accepts_generator:
         lowercase     :      Optional[Any]	=							generator

      with self.progress_bar(total=snake_case__						):
         for i, t in enumerate(snake_case__						):
            # expand the latents if we are doing classifier free guidance
            lowercase     :      Union[str, Any]	=							torch.cat([latents] * 2						) if do_classifier_free_guidance else latents
            lowercase     :      Dict	=							self.scheduler.scale_model_input(snake_case__   , snake_case__						)

            # predict the noise residual
            lowercase     :      Tuple	=							self.unet(snake_case__   , snake_case__   , encoder_hidden_states=snake_case__						).sample

            # perform classifier free guidance
            if do_classifier_free_guidance:
               lowercase	,       lowercase     :      Optional[Any]	=							noise_pred.chunk(2						)
               lowercase     :      Dict	=							noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

            # perform clip guidance
            if clip_guidance_scale > 0:
               lowercase     :      str	=							(
                   text_embeddings.chunk(2						)[1] if do_classifier_free_guidance else text_embeddings
               )
               lowercase	,       lowercase     :      Union[str, Any]	=							self.cond_fn(
                   snake_case__   , snake_case__   , snake_case__   , snake_case__   , snake_case__   , snake_case__   , snake_case__   , )

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

             # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      lowercase     :      int	=							1 / 0.1_82_15 * latents
      lowercase     :      List[Any]	=							self.vae.decode(snake_case__						).sample

      lowercase     :      str	=							(image / 2 + 0.5).clamp(0   , 1						)
      lowercase     :      Any	=							image.cpu().permute(0   , 2   , 3   , 1						).numpy()

      if output_type == "pil":
         lowercase     :      Optional[Any]	=							self.numpy_to_pil(snake_case__						)

      if not return_dict:
         return (image, None)

      return StableDiffusionPipelineOutput(images=snake_case__   , nsfw_content_detected=snake_case__						)





import string


def 							__lowerCAmelCase     (							_UpperCamelCase		:	str						)		->				str:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE		    =      ''
 for i in sequence:
  SCREAMING_SNAKE_CASE		    =      ord(_UpperCamelCase						)
  if 65 <= extract <= 90:
   output += chr(1_55 - extract						)
  elif 97 <= extract <= 1_22:
   output += chr(2_19 - extract						)
  else:
   output += i
 return output


def 							__lowerCAmelCase     (							_UpperCamelCase		:	str						)		->				str:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE		    =      string.ascii_letters
 SCREAMING_SNAKE_CASE		    =      string.ascii_lowercase[::-1] + string.ascii_uppercase[::-1]
 return "".join(
     letters_reversed[letters.index(_UpperCamelCase						)] if c in letters else c for c in sequence						)


def 							__lowerCAmelCase     (							)		->				None:
 '''simple docstring'''


 from timeit import timeit

 print('Running performance benchmarks...'						)
 SCREAMING_SNAKE_CASE		    =      'from string import printable ; from __main__ import atbash, atbash_slow'
 print(f"""> atbash_slow(): {timeit('atbash_slow(printable)'		,				setup=_UpperCamelCase						)} seconds"""						)
 print(f""">      atbash(): {timeit('atbash(printable)'		,				setup=_UpperCamelCase						)} seconds"""						)


if __name__ == "__main__":
      for example in ("ABCDEFGH", "123GGjj", "testStringtest", "with space"):
            print(F"""{example} encrypted in atbash: {atbash(example)}""")
      benchmark()

