'''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   __UpperCAmelCase     (    a_:			Optional[Any],   a_:			Any,   a_:			Optional[int]				):
   if isinstance(snake_case__,   torch.Tensor				):
      return image
   elif isinstance(snake_case__,   PIL.Image.Image				):
      _UpperCAmelCase				:      Tuple				=  [image]

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






def   __UpperCAmelCase     (    a_:			int,   a_:			Dict,   a_:			Dict,   a_:			Optional[int]=0.99_95				):
   if not isinstance(snake_case__,   np.ndarray				):
      _UpperCAmelCase				:      Optional[int]				=  True
      _UpperCAmelCase				:      Dict				=  va.device
      _UpperCAmelCase				:      Optional[int]				=  va.cpu().numpy()
      _UpperCAmelCase				:      Tuple				=  va.cpu().numpy()

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

   if inputs_are_torch:
      _UpperCAmelCase				:      str				=  torch.from_numpy(snake_case__				).to(snake_case__				)

   return va






def   __UpperCAmelCase     (    a_:			str,   a_:			List[Any]				):
   _UpperCAmelCase				:      Dict				=  F.normalize(snake_case__,   dim=-1				)
   _UpperCAmelCase				:      Any				=  F.normalize(snake_case__,   dim=-1				)
   return (x - y).norm(dim=-1				).div(2				).arcsin().pow(2				).mul(2				)






def   __UpperCAmelCase     (    a_:			int,   a_:			str				):
   for param in model.parameters():
      _UpperCAmelCase				:      Optional[int]				=  value






class        A__				(			_SCREAMING_SNAKE_CASE  ):





   """simple docstring"""





   def __init__(					self	:						Dict		,	lowerCAmelCase__	:						Union[str, Any]		,	lowerCAmelCase__	:						int		,	lowerCAmelCase__	:						Dict		,	lowerCAmelCase__	:						str		,	lowerCAmelCase__	:						Any		,	lowerCAmelCase__	:						str		,	lowerCAmelCase__	:						List[Any]		,	lowerCAmelCase__	:						List[Any]=None		,	lowerCAmelCase__	:						Optional[Any]=None		,	lowerCAmelCase__	:						List[Any]=None		,	)			->       Any:







      """simple docstring"""




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




   def      _lowerCAmelCase						(					self	:						Any		,	lowerCAmelCase__	:						Optional[Any] = "auto"     )			->       List[Any]:







      """simple docstring"""




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




   def      _lowerCAmelCase						(					self	:						Union[str, Any]     )			->       int:







      """simple docstring"""




      self.enable_attention_slicing(lowerCAmelCase_     )




   def      _lowerCAmelCase						(					self	:						Any     )			->       str:







      """simple docstring"""




      set_requires_grad(self.vae		,	lowerCAmelCase_     )




   def      _lowerCAmelCase						(					self	:						int     )			->       List[str]:







      """simple docstring"""




      set_requires_grad(self.vae		,	lowerCAmelCase_     )




   def      _lowerCAmelCase						(					self	:						int     )			->       Optional[Any]:







      """simple docstring"""




      set_requires_grad(self.unet		,	lowerCAmelCase_     )




   def      _lowerCAmelCase						(					self	:						Any     )			->       List[Any]:







      """simple docstring"""




      set_requires_grad(self.unet		,	lowerCAmelCase_     )




   def      _lowerCAmelCase						(					self	:						List[str]		,	lowerCAmelCase__	:						Dict		,	lowerCAmelCase__	:						int		,	lowerCAmelCase__	:						Union[str, Any]     )			->       Optional[int]:







      """simple docstring"""




      _UpperCAmelCase				:      Any				=  min(int(num_inference_steps * strength     )		,	lowerCAmelCase_     )

      _UpperCAmelCase				:      List[str]				=  max(num_inference_steps - init_timestep		,	0     )
      _UpperCAmelCase				:      List[Any]				=  self.scheduler.timesteps[t_start:]

      return timesteps, num_inference_steps - t_start




   def      _lowerCAmelCase						(					self	:						Tuple		,	lowerCAmelCase__	:						Optional[int]		,	lowerCAmelCase__	:						int		,	lowerCAmelCase__	:						Optional[int]		,	lowerCAmelCase__	:						Optional[int]		,	lowerCAmelCase__	:						Optional[int]		,	lowerCAmelCase__	:						Union[str, Any]=None     )			->       List[str]:







      """simple docstring"""




      if not isinstance(lowerCAmelCase_		,	torch.Tensor     ):
         raise ValueError(F"""`image` has to be of type `torch.Tensor` but is {type(lowerCAmelCase_     )}"""     )

      _UpperCAmelCase				:      Tuple				=  image.to(device=lowerCAmelCase_		,	dtype=lowerCAmelCase_     )

      if isinstance(lowerCAmelCase_		,	lowerCAmelCase_     ):
         _UpperCAmelCase				:      Tuple				=  [
             self.vae.encode(image[i : i + 1]     ).latent_dist.sample(generator[i]     ) for i in range(lowerCAmelCase_     )
         ]
         _UpperCAmelCase				:      str				=  torch.cat(lowerCAmelCase_		,	dim=0     )
      else:
         _UpperCAmelCase				:      Dict				=  self.vae.encode(lowerCAmelCase_     ).latent_dist.sample(lowerCAmelCase_     )

      # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      _UpperCAmelCase				:      List[Any]				=  0.1_8215 * init_latents
      _UpperCAmelCase				:      Optional[int]				=  init_latents.repeat_interleave(lowerCAmelCase_		,	dim=0     )

      _UpperCAmelCase				:      Optional[int]				=  randn_tensor(init_latents.shape		,	generator=lowerCAmelCase_		,	device=lowerCAmelCase_		,	dtype=lowerCAmelCase_     )

      # get latents
      _UpperCAmelCase				:      Tuple				=  self.scheduler.add_noise(lowerCAmelCase_		,	lowerCAmelCase_		,	lowerCAmelCase_     )
      _UpperCAmelCase				:      List[Any]				=  init_latents

      return latents




   def      _lowerCAmelCase						(					self	:						Dict		,	lowerCAmelCase__	:						str     )			->       Any:







      """simple docstring"""




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




   def      _lowerCAmelCase						(					self	:						int		,	lowerCAmelCase__	:						Dict		,	lowerCAmelCase__	:						str     )			->       Union[str, Any]:







      """simple docstring"""




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




   @torch.enable_grad()
   def      _lowerCAmelCase						(					self	:						Optional[int]		,	lowerCAmelCase__	:						str		,	lowerCAmelCase__	:						Optional[Any]		,	lowerCAmelCase__	:						Optional[int]		,	lowerCAmelCase__	:						Dict		,	lowerCAmelCase__	:						str		,	lowerCAmelCase__	:						Tuple		,	lowerCAmelCase__	:						Optional[Any]		,	)			->       Optional[int]:







      """simple docstring"""




      _UpperCAmelCase				:      List[str]				=  latents.detach().requires_grad_()

      _UpperCAmelCase				:      Dict				=  self.scheduler.scale_model_input(lowerCAmelCase_		,	lowerCAmelCase_     )

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

      if isinstance(self.scheduler		,	(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)     ):
         _UpperCAmelCase				:      Tuple				=  self.scheduler.alphas_cumprod[timestep]
         _UpperCAmelCase				:      Tuple				=  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
         _UpperCAmelCase				:      Tuple				=  (latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

         _UpperCAmelCase				:      Union[str, Any]				=  torch.sqrt(lowerCAmelCase_     )
         _UpperCAmelCase				:      List[str]				=  pred_original_sample * (fac) + latents * (1 - fac)
      elif isinstance(self.scheduler		,	lowerCAmelCase_     ):
         _UpperCAmelCase				:      int				=  self.scheduler.sigmas[index]
         _UpperCAmelCase				:      int				=  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
      _UpperCAmelCase				:      int				=  1 / 0.1_8215 * sample
      _UpperCAmelCase				:      Optional[int]				=  self.vae.decode(lowerCAmelCase_     ).sample
      _UpperCAmelCase				:      Optional[Any]				=  (image / 2 + 0.5).clamp(0		,	1     )

      _UpperCAmelCase				:      str				=  transforms.Resize(self.feature_extractor_size     )(lowerCAmelCase_     )
      _UpperCAmelCase				:      int				=  self.normalize(lowerCAmelCase_     ).to(latents.dtype     )

      _UpperCAmelCase				:      Union[str, Any]				=  self.clip_model.get_image_features(lowerCAmelCase_     )
      _UpperCAmelCase				:      Dict				=  image_embeddings_clip / image_embeddings_clip.norm(p=2		,	dim=-1		,	keepdim=lowerCAmelCase_     )

      _UpperCAmelCase				:      Tuple				=  spherical_dist_loss(lowerCAmelCase_		,	lowerCAmelCase_     ).mean() * clip_guidance_scale

      _UpperCAmelCase				:      List[Any]				=  -torch.autograd.grad(lowerCAmelCase_		,	lowerCAmelCase_     )[0]

      if isinstance(self.scheduler		,	lowerCAmelCase_     ):
         _UpperCAmelCase				:      Tuple				=  latents.detach() + grads * (sigma**2)
         _UpperCAmelCase				:      int				=  noise_pred_original
      else:
         _UpperCAmelCase				:      Tuple				=  noise_pred_original - torch.sqrt(lowerCAmelCase_     ) * grads
      return noise_pred, latents




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







      """simple docstring"""




      if isinstance(lowerCAmelCase_		,	lowerCAmelCase_     ) and len(lowerCAmelCase_     ) != batch_size:
         raise ValueError(F"""You have passed {batch_size} batch_size, but only {len(lowerCAmelCase_     )} 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(lowerCAmelCase_		,	torch.Generator     ) and batch_size > 1:
         _UpperCAmelCase				:      str				=  [generator] + [None] * (batch_size - 1)

      _UpperCAmelCase				:      List[Any]				=  [
          ("""model""", self.coca_model is None),
          ("""tokenizer""", self.coca_tokenizer is None),
          ("""transform""", self.coca_transform is None),
      ]
      _UpperCAmelCase				:      Dict				=  [x[0] for x in coca_is_none if x[1]]
      _UpperCAmelCase				:      Union[str, Any]				=  """, """.join(lowerCAmelCase_     )
      # generate prompts with coca model if prompt is None
      if content_prompt is None:
         if len(lowerCAmelCase_     ):
            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."""     )
         _UpperCAmelCase				:      Any				=  self.get_image_description(lowerCAmelCase_     )
      if style_prompt is None:
         if len(lowerCAmelCase_     ):
            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."""     )
         _UpperCAmelCase				:      List[str]				=  self.get_image_description(lowerCAmelCase_     )

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

      _UpperCAmelCase				:      Optional[int]				=  self.tokenizer(
          lowerCAmelCase_		,	padding="max_length"		,	max_length=self.tokenizer.model_max_length		,	truncation=lowerCAmelCase_		,	return_tensors="pt"		,	)
      _UpperCAmelCase				:      Tuple				=  self.text_encoder(style_text_input.input_ids.to(self.device     )     )[0]

      _UpperCAmelCase				:      Any				=  slerp(lowerCAmelCase_		,	lowerCAmelCase_		,	lowerCAmelCase_     )

      # duplicate text embeddings for each generation per prompt
      _UpperCAmelCase				:      Any				=  text_embeddings.repeat_interleave(lowerCAmelCase_		,	dim=0     )

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

      self.scheduler.set_timesteps(lowerCAmelCase_		,	**lowerCAmelCase_     )
      # 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     )

      _UpperCAmelCase				:      str				=  self.get_timesteps(lowerCAmelCase_		,	lowerCAmelCase_		,	self.device     )
      _UpperCAmelCase				:      List[str]				=  timesteps[:1].repeat(lowerCAmelCase_     )

      # Preprocess image
      _UpperCAmelCase				:      Tuple				=  preprocess(lowerCAmelCase_		,	lowerCAmelCase_		,	lowerCAmelCase_     )
      _UpperCAmelCase				:      List[str]				=  self.prepare_latents(
          lowerCAmelCase_		,	lowerCAmelCase_		,	lowerCAmelCase_		,	text_embeddings.dtype		,	self.device		,	lowerCAmelCase_     )

      _UpperCAmelCase				:      Tuple				=  preprocess(lowerCAmelCase_		,	lowerCAmelCase_		,	lowerCAmelCase_     )
      _UpperCAmelCase				:      List[Any]				=  self.prepare_latents(
          lowerCAmelCase_		,	lowerCAmelCase_		,	lowerCAmelCase_		,	text_embeddings.dtype		,	self.device		,	lowerCAmelCase_     )

      _UpperCAmelCase				:      Optional[int]				=  slerp(lowerCAmelCase_		,	lowerCAmelCase_		,	lowerCAmelCase_     )

      if clip_guidance_scale > 0:
         _UpperCAmelCase				:      Any				=  self.get_clip_image_embeddings(lowerCAmelCase_		,	lowerCAmelCase_     )
         _UpperCAmelCase				:      Optional[Any]				=  self.get_clip_image_embeddings(lowerCAmelCase_		,	lowerCAmelCase_     )
         _UpperCAmelCase				:      str				=  slerp(
             lowerCAmelCase_		,	lowerCAmelCase_		,	lowerCAmelCase_     )

      # 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.
      _UpperCAmelCase				:      Any				=  guidance_scale > 1.0
      # get unconditional embeddings for classifier free guidance
      if do_classifier_free_guidance:
         _UpperCAmelCase				:      str				=  content_text_input.input_ids.shape[-1]
         _UpperCAmelCase				:      Dict				=  self.tokenizer([""]		,	padding="max_length"		,	max_length=lowerCAmelCase_		,	return_tensors="pt"     )
         _UpperCAmelCase				:      Union[str, Any]				=  self.text_encoder(uncond_input.input_ids.to(self.device     )     )[0]
         # duplicate unconditional embeddings for each generation per prompt
         _UpperCAmelCase				:      str				=  uncond_embeddings.repeat_interleave(lowerCAmelCase_		,	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
         _UpperCAmelCase				:      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`.
      _UpperCAmelCase				:      Any				=  (batch_size, self.unet.config.in_channels, height // 8, width // 8)
      _UpperCAmelCase				:      List[str]				=  text_embeddings.dtype
      if latents is None:
         if self.device.type == "mps":
            # randn does not work reproducibly on mps
            _UpperCAmelCase				:      Union[str, Any]				=  torch.randn(lowerCAmelCase_		,	generator=lowerCAmelCase_		,	device="cpu"		,	dtype=lowerCAmelCase_     ).to(
                self.device     )
         else:
            _UpperCAmelCase				:      Tuple				=  torch.randn(lowerCAmelCase_		,	generator=lowerCAmelCase_		,	device=self.device		,	dtype=lowerCAmelCase_     )
      else:
         if latents.shape != latents_shape:
            raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""     )
         _UpperCAmelCase				:      Dict				=  latents.to(self.device     )

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

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

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

            # predict the noise residual
            _UpperCAmelCase				:      Optional[int]				=  self.unet(lowerCAmelCase_		,	lowerCAmelCase_		,	encoder_hidden_states=lowerCAmelCase_     ).sample

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

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

            # compute the previous noisy sample x_t -> x_t-1
            _UpperCAmelCase				:      List[Any]				=  self.scheduler.step(lowerCAmelCase_		,	lowerCAmelCase_		,	lowerCAmelCase_		,	**lowerCAmelCase_     ).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
      _UpperCAmelCase				:      Tuple				=  1 / 0.1_8215 * latents
      _UpperCAmelCase				:      Tuple				=  self.vae.decode(lowerCAmelCase_     ).sample

      _UpperCAmelCase				:      Optional[int]				=  (image / 2 + 0.5).clamp(0		,	1     )
      _UpperCAmelCase				:      str				=  image.cpu().permute(0		,	2		,	3		,	1     ).numpy()

      if output_type == "pil":
         _UpperCAmelCase				:      Optional[int]				=  self.numpy_to_pil(lowerCAmelCase_     )

      if not return_dict:
         return (image, None)

      return StableDiffusionPipelineOutput(images=lowerCAmelCase_		,	nsfw_content_detected=lowerCAmelCase_     )







"""simple docstring"""
import warnings
from typing import Dict

import numpy as np

from ..utils import ExplicitEnum, add_end_docstrings, is_tf_available, is_torch_available
from .base import PIPELINE_INIT_ARGS, GenericTensor, Pipeline


if is_tf_available():
							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING

if is_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING






def       __UpperCamelCase						(		snake_case__       ):
	return 1.0 / (1.0 + np.exp(-_outputs       ))






def       __UpperCamelCase						(		snake_case__       ):
	A_		:    Union[str, Any]						     = np.max(_outputs , axis=-1 , keepdims=snake_case__       )
	A_		:    str						     = np.exp(_outputs - maxes       )
	return shifted_exp / shifted_exp.sum(axis=-1 , keepdims=snake_case__       )



class   SCREAMING_SNAKE_CASE	(					_SCREAMING_SNAKE_CASE    ):





	"""simple docstring"""




	_A							:					Any 			=					"""sigmoid"""
	_A							:					Any 			=					"""softmax"""
	_A							:					Union[str, Any] 			=					"""none"""






@add_end_docstrings(
    _SCREAMING_SNAKE_CASE							,							R"""
        return_all_scores (`bool`, *optional*, defaults to `False`):
            Whether to return all prediction scores or just the one of the predicted class.
        function_to_apply (`str`, *optional*, defaults to `\"default\"`):
            The function to apply to the model outputs in order to retrieve the scores. Accepts four different values:

            - `\"default\"`: if the model has a single label, will apply the sigmoid function on the output. If the model
              has several labels, will apply the softmax function on the output.
            - `\"sigmoid\"`: Applies the sigmoid function on the output.
            - `\"softmax\"`: Applies the softmax function on the output.
            - `\"none\"`: Does not apply any function on the output.
    """							,							)
class   SCREAMING_SNAKE_CASE	(					_SCREAMING_SNAKE_CASE    ):





	"""simple docstring"""




	_A							:					Optional[int] 			=					False
	_A							:					Dict 			=					ClassificationFunction.NONE





	def __init__(self  ,  **lowerCAmelCase_				):
		super().__init__(**lowerCAmelCase_				)

		self.check_model_type(
		    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING
		    if self.framework == """tf"""
		    else MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING				)





	def  lowerCamelCase(self  ,  lowerCAmelCase_=None  ,  lowerCAmelCase_=None  ,  lowerCAmelCase_=""  ,  **lowerCAmelCase_				):
		# Using "" as default argument because we're going to use `top_k=None` in user code to declare
		# "No top_k"
		A_		:    Union[str, Any]						     = tokenizer_kwargs

		A_		:    List[str]						     = {}
		if hasattr(self.model.config  ,  """return_all_scores"""				) and return_all_scores is None:
			A_		:    Optional[int]						     = self.model.config.return_all_scores

		if isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				) or top_k is None:
			A_		:    Dict						     = top_k
			A_		:    Any						     = False
		elif return_all_scores is not None:
			warnings.warn(
			    """`return_all_scores` is now deprecated,  if want a similar functionality use `top_k=None` instead of"""
			    """ `return_all_scores=True` or `top_k=1` instead of `return_all_scores=False`."""  ,  lowerCAmelCase_  ,  )
			if return_all_scores:
				A_		:    List[Any]						     = None
			else:
				A_		:    List[str]						     = 1

		if isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				):
			A_		:    str						     = ClassificationFunction[function_to_apply.upper()]

		if function_to_apply is not None:
			A_		:    List[str]						     = function_to_apply
		return preprocess_params, {}, postprocess_params





	def __call__(self  ,  *lowerCAmelCase_  ,  **lowerCAmelCase_				):
		A_		:    List[str]						     = super().__call__(*lowerCAmelCase_  ,  **lowerCAmelCase_				)
		# TODO try and retrieve it in a nicer way from _sanitize_parameters.
		A_		:    Union[str, Any]						     = """top_k""" not in kwargs
		if isinstance(args[0]  ,  lowerCAmelCase_				) and _legacy:
			# This pipeline is odd, and return a list when single item is run
			return [result]
		else:
			return result





	def  lowerCamelCase(self  ,  lowerCAmelCase_  ,  **lowerCAmelCase_				):
		A_		:    Union[str, Any]						     = self.framework
		if isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				):
			return self.tokenizer(**lowerCAmelCase_  ,  return_tensors=lowerCAmelCase_  ,  **lowerCAmelCase_				)
		elif isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				) and len(lowerCAmelCase_				) == 1 and isinstance(inputs[0]  ,  lowerCAmelCase_				) and len(inputs[0]				) == 2:
			# It used to be valid to use a list of list of list for text pairs, keeping this path for BC
			return self.tokenizer(
			    text=inputs[0][0]  ,  text_pair=inputs[0][1]  ,  return_tensors=lowerCAmelCase_  ,  **lowerCAmelCase_				)
		elif isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				):
			# This is likely an invalid usage of the pipeline attempting to pass text pairs.
			raise ValueError(
			    """The pipeline received invalid inputs, if you are trying to send text pairs, you can try to send a"""
			    """ dictionary `{\"text\": \"My text\", \"text_pair\": \"My pair\"}` in order to send a text pair."""				)
		return self.tokenizer(lowerCAmelCase_  ,  return_tensors=lowerCAmelCase_  ,  **lowerCAmelCase_				)





	def  lowerCamelCase(self  ,  lowerCAmelCase_				):
		return self.model(**lowerCAmelCase_				)





	def  lowerCamelCase(self  ,  lowerCAmelCase_  ,  lowerCAmelCase_=None  ,  lowerCAmelCase_=1  ,  lowerCAmelCase_=True				):
		# `_legacy` is used to determine if we're running the naked pipeline and in backward
		# compatibility mode, or if running the pipeline with `pipeline(..., top_k=1)` we're running
		# the more natural result containing the list.
		# Default value before `set_parameters`
		if function_to_apply is None:
			if self.model.config.problem_type == "multi_label_classification" or self.model.config.num_labels == 1:
				A_		:    Optional[Any]						     = ClassificationFunction.SIGMOID
			elif self.model.config.problem_type == "single_label_classification" or self.model.config.num_labels > 1:
				A_		:    Tuple						     = ClassificationFunction.SOFTMAX
			elif hasattr(self.model.config  ,  """function_to_apply"""				) and function_to_apply is None:
				A_		:    Any						     = self.model.config.function_to_apply
			else:
				A_		:    Dict						     = ClassificationFunction.NONE

		A_		:    Optional[Any]						     = model_outputs["""logits"""][0]
		A_		:    Tuple						     = outputs.numpy()

		if function_to_apply == ClassificationFunction.SIGMOID:
			A_		:    str						     = sigmoid(lowerCAmelCase_				)
		elif function_to_apply == ClassificationFunction.SOFTMAX:
			A_		:    Dict						     = softmax(lowerCAmelCase_				)
		elif function_to_apply == ClassificationFunction.NONE:
			A_		:    Optional[int]						     = outputs
		else:
			raise ValueError(f"""Unrecognized `function_to_apply` argument: {function_to_apply}"""				)

		if top_k == 1 and _legacy:
			return {"label": self.model.config.idalabel[scores.argmax().item()], "score": scores.max().item()}

		A_		:    Optional[Any]						     = [
		    {"""label""": self.model.config.idalabel[i], """score""": score.item()} for i, score in enumerate(lowerCAmelCase_				)
		]
		if not _legacy:
			dict_scores.sort(key=lambda lowerCAmelCase_				: x["score"]  ,  reverse=lowerCAmelCase_				)
			if top_k is not None:
				A_		:    str						     = dict_scores[:top_k]
		return dict_scores

