
"""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 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase	):
 if isinstance(_UpperCamelCase ,						torch.Tensor	):
  return image
 elif isinstance(_UpperCamelCase ,						PIL.Image.Image	):
  __lowerCAmelCase    :		str		    =   [image]

 if isinstance(image[0] ,						PIL.Image.Image	):
  __lowerCAmelCase    :		Union[str, Any]		    =   [np.array(i.resize((w, h) ,						resample=PIL_INTERPOLATION['lanczos']	)	)[None, :] for i in image]
  __lowerCAmelCase    :		int		    =   np.concatenate(_UpperCamelCase ,						axis=0	)
  __lowerCAmelCase    :		Optional[int]		    =   np.array(_UpperCamelCase	).astype(np.floataa	) / 255.0
  __lowerCAmelCase    :		Optional[int]		    =   image.transpose(0 ,						3 ,						1 ,						2	)
  __lowerCAmelCase    :		Union[str, Any]		    =   2.0 * image - 1.0
  __lowerCAmelCase    :		int		    =   torch.from_numpy(_UpperCamelCase	)
 elif isinstance(image[0] ,						torch.Tensor	):
  __lowerCAmelCase    :		Optional[int]		    =   torch.cat(_UpperCamelCase ,						dim=0	)
 return image
def 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase=0.9995	):
 if not isinstance(_UpperCamelCase ,						np.ndarray	):
  __lowerCAmelCase    :		str		    =   True
  __lowerCAmelCase    :		str		    =   va.device
  __lowerCAmelCase    :		Tuple		    =   va.cpu().numpy()
  __lowerCAmelCase    :		Optional[Any]		    =   va.cpu().numpy()

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

 if inputs_are_torch:
  __lowerCAmelCase    :		Optional[Any]		    =   torch.from_numpy(_UpperCamelCase	).to(_UpperCamelCase	)

 return va
def 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase	):
 __lowerCAmelCase    :		int		    =   F.normalize(_UpperCamelCase ,						dim=-1	)
 __lowerCAmelCase    :		List[Any]		    =   F.normalize(_UpperCamelCase ,						dim=-1	)
 return (x - y).norm(dim=-1	).div(2	).arcsin().pow(2	).mul(2	)
def 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase	):
 for param in model.parameters():
  __lowerCAmelCase    :		Optional[Any]		    =   value




class 				A__ (			_lowerCamelCase):

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

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

  def 							__lowerCamelCase							(   self				):
   self.enable_attention_slicing(_SCREAMING_SNAKE_CASE				)

  def 							__lowerCamelCase							(   self				):
   set_requires_grad(self.vae ,	_SCREAMING_SNAKE_CASE				)

  def 							__lowerCamelCase							(   self				):
   set_requires_grad(self.vae ,	_SCREAMING_SNAKE_CASE				)

  def 							__lowerCamelCase							(   self				):
   set_requires_grad(self.unet ,	_SCREAMING_SNAKE_CASE				)

  def 							__lowerCamelCase							(   self				):
   set_requires_grad(self.unet ,	_SCREAMING_SNAKE_CASE				)

  def 							__lowerCamelCase							(   self ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				):
   # get the original timestep using init_timestep
   __lowerCAmelCase    :		Union[str, Any]		    =   min(int(num_inference_steps * strength				) ,	_SCREAMING_SNAKE_CASE				)

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

   return timesteps, num_inference_steps - t_start

  def 							__lowerCamelCase							(   self ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE=None				):
   if not isinstance(_SCREAMING_SNAKE_CASE ,	torch.Tensor				):
    raise ValueError(f"`image` has to be of type `torch.Tensor` but is {type(_SCREAMING_SNAKE_CASE				)}"				)

   __lowerCAmelCase    :		Any		    =   image.to(device=_SCREAMING_SNAKE_CASE ,	dtype=_SCREAMING_SNAKE_CASE				)

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

   # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
   __lowerCAmelCase    :		Union[str, Any]		    =   0.1_8215 * init_latents
   __lowerCAmelCase    :		str		    =   init_latents.repeat_interleave(_SCREAMING_SNAKE_CASE ,	dim=0				)

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

   # get latents
   __lowerCAmelCase    :		List[Any]		    =   self.scheduler.add_noise(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)
   __lowerCAmelCase    :		Any		    =   init_latents

   return latents

  def 							__lowerCamelCase							(   self ,	_SCREAMING_SNAKE_CASE				):
   __lowerCAmelCase    :		Optional[Any]		    =   self.coca_transform(_SCREAMING_SNAKE_CASE				).unsqueeze(0				)
   with torch.no_grad(), torch.cuda.amp.autocast():
    __lowerCAmelCase    :		Optional[Any]		    =   self.coca_model.generate(transformed_image.to(device=self.device ,	dtype=self.coca_model.dtype				)				)
   __lowerCAmelCase    :		Tuple		    =   self.coca_tokenizer.decode(generated[0].cpu().numpy()				)
   return generated.split('<end_of_text>'				)[0].replace('<start_of_text>' ,	''				).rstrip(' .,'				)

  def 							__lowerCamelCase							(   self ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				):
   __lowerCAmelCase    :		Union[str, Any]		    =   self.feature_extractor.preprocess(_SCREAMING_SNAKE_CASE				)
   __lowerCAmelCase    :		Any		    =   torch.from_numpy(clip_image_input['pixel_values'][0]				).unsqueeze(0				).to(self.device				).half()
   __lowerCAmelCase    :		Optional[Any]		    =   self.clip_model.get_image_features(_SCREAMING_SNAKE_CASE				)
   __lowerCAmelCase    :		Any		    =   image_embeddings_clip / image_embeddings_clip.norm(p=2 ,	dim=-1 ,	keepdim=_SCREAMING_SNAKE_CASE				)
   __lowerCAmelCase    :		Union[str, Any]		    =   image_embeddings_clip.repeat_interleave(_SCREAMING_SNAKE_CASE ,	dim=0				)
   return image_embeddings_clip

  @torch.enable_grad()
  def 							__lowerCamelCase							(   self ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	):
   __lowerCAmelCase    :		Tuple		    =   latents.detach().requires_grad_()

   __lowerCAmelCase    :		Optional[int]		    =   self.scheduler.scale_model_input(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)

   # predict the noise residual
   __lowerCAmelCase    :		Optional[Any]		    =   self.unet(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	encoder_hidden_states=_SCREAMING_SNAKE_CASE				).sample

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

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

   __lowerCAmelCase    :		Any		    =   transforms.Resize(self.feature_extractor_size				)(_SCREAMING_SNAKE_CASE				)
   __lowerCAmelCase    :		Tuple		    =   self.normalize(_SCREAMING_SNAKE_CASE				).to(latents.dtype				)

   __lowerCAmelCase    :		Optional[int]		    =   self.clip_model.get_image_features(_SCREAMING_SNAKE_CASE				)
   __lowerCAmelCase    :		Any		    =   image_embeddings_clip / image_embeddings_clip.norm(p=2 ,	dim=-1 ,	keepdim=_SCREAMING_SNAKE_CASE				)

   __lowerCAmelCase    :		int		    =   spherical_dist_loss(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				).mean() * clip_guidance_scale

   __lowerCAmelCase    :		List[str]		    =   -torch.autograd.grad(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)[0]

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







  @torch.no_grad()
  def __call__(   self ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE = None ,	_SCREAMING_SNAKE_CASE = None ,	_SCREAMING_SNAKE_CASE = 5_12 ,	_SCREAMING_SNAKE_CASE = 5_12 ,	_SCREAMING_SNAKE_CASE = 0.6 ,	_SCREAMING_SNAKE_CASE = 50 ,	_SCREAMING_SNAKE_CASE = 7.5 ,	_SCREAMING_SNAKE_CASE = 1 ,	_SCREAMING_SNAKE_CASE = 0.0 ,	_SCREAMING_SNAKE_CASE = 1_00 ,	_SCREAMING_SNAKE_CASE = None ,	_SCREAMING_SNAKE_CASE = "pil" ,	_SCREAMING_SNAKE_CASE = True ,	_SCREAMING_SNAKE_CASE = 0.8 ,	_SCREAMING_SNAKE_CASE = 0.1 ,	_SCREAMING_SNAKE_CASE = 0.1 ,	):
   if isinstance(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				) and len(_SCREAMING_SNAKE_CASE				) != batch_size:
    raise ValueError(f"You have passed {batch_size} batch_size, but only {len(_SCREAMING_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(_SCREAMING_SNAKE_CASE ,	torch.Generator				) and batch_size > 1:
    __lowerCAmelCase    :		Optional[Any]		    =   [generator] + [None] * (batch_size - 1)

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

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

   __lowerCAmelCase    :		Any		    =   self.tokenizer(
       _SCREAMING_SNAKE_CASE ,	padding='max_length' ,	max_length=self.tokenizer.model_max_length ,	truncation=_SCREAMING_SNAKE_CASE ,	return_tensors='pt' ,	)
   __lowerCAmelCase    :		List[Any]		    =   self.text_encoder(style_text_input.input_ids.to(self.device				)				)[0]

   __lowerCAmelCase    :		str		    =   slerp(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)

   # duplicate text embeddings for each generation per prompt
   __lowerCAmelCase    :		List[str]		    =   text_embeddings.repeat_interleave(_SCREAMING_SNAKE_CASE ,	dim=0				)

   # set timesteps
   __lowerCAmelCase    :		Union[str, Any]		    =   'offset' in set(inspect.signature(self.scheduler.set_timesteps				).parameters.keys()				)
   __lowerCAmelCase    :		List[Any]		    =   {}
   if accepts_offset:
    __lowerCAmelCase    :		Union[str, Any]		    =   1

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

   __lowerCAmelCase    :		int		    =   self.get_timesteps(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	self.device				)
   __lowerCAmelCase    :		Any		    =   timesteps[:1].repeat(_SCREAMING_SNAKE_CASE				)

   # Preprocess image
   __lowerCAmelCase    :		Union[str, Any]		    =   preprocess(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)
   __lowerCAmelCase    :		List[Any]		    =   self.prepare_latents(
       _SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	text_embeddings.dtype ,	self.device ,	_SCREAMING_SNAKE_CASE				)

   __lowerCAmelCase    :		Tuple		    =   preprocess(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)
   __lowerCAmelCase    :		str		    =   self.prepare_latents(
       _SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	text_embeddings.dtype ,	self.device ,	_SCREAMING_SNAKE_CASE				)

   __lowerCAmelCase    :		Tuple		    =   slerp(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE				)

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

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

   # check if the scheduler accepts generator
   __lowerCAmelCase    :		List[Any]		    =   'generator' in set(inspect.signature(self.scheduler.step				).parameters.keys()				)
   if accepts_generator:
    __lowerCAmelCase    :		Any		    =   generator

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

     # predict the noise residual
     __lowerCAmelCase    :		Union[str, Any]		    =   self.unet(_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	encoder_hidden_states=_SCREAMING_SNAKE_CASE				).sample

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

     # perform clip guidance
     if clip_guidance_scale > 0:
      __lowerCAmelCase    :		Optional[int]		    =   (
          text_embeddings.chunk(2				)[1] if do_classifier_free_guidance else text_embeddings
      )
      __lowerCAmelCase    :		Any		    =   self.cond_fn(
          _SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	_SCREAMING_SNAKE_CASE ,	)

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

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
   __lowerCAmelCase    :		int		    =   1 / 0.1_8215 * latents
   __lowerCAmelCase    :		Union[str, Any]		    =   self.vae.decode(_SCREAMING_SNAKE_CASE				).sample

   __lowerCAmelCase    :		Dict		    =   (image / 2 + 0.5).clamp(0 ,	1				)
   __lowerCAmelCase    :		int		    =   image.cpu().permute(0 ,	2 ,	3 ,	1				).numpy()

   if output_type == "pil":
    __lowerCAmelCase    :		Optional[Any]		    =   self.numpy_to_pil(_SCREAMING_SNAKE_CASE				)

   if not return_dict:
    return (image, None)

   return StableDiffusionPipelineOutput(images=_SCREAMING_SNAKE_CASE ,	nsfw_content_detected=_SCREAMING_SNAKE_CASE				)

"""simple docstring"""







import inspect
import os
import re

from transformers.configuration_utils import PretrainedConfig
from transformers.utils import direct_transformers_import


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_config_docstrings.py
lowerCamelCase__           =	"""src/transformers"""


# This is to make sure the transformers module imported is the one in the repo.
lowerCamelCase__           =	direct_transformers_import(PATH_TO_TRANSFORMERS)

lowerCamelCase__           =	transformers.models.auto.configuration_auto.CONFIG_MAPPING

lowerCamelCase__           =	{
    # used to compute the property `self.chunk_length`
    """EncodecConfig""": ["""overlap"""],
    # used as `self.bert_model = BertModel(config, ...)`
    """DPRConfig""": True,
    # not used in modeling files, but it's an important information
    """FSMTConfig""": ["""langs"""],
    # used internally in the configuration class file
    """GPTNeoConfig""": ["""attention_types"""],
    # used internally in the configuration class file
    """EsmConfig""": ["""is_folding_model"""],
    # used during training (despite we don't have training script for these models yet)
    """Mask2FormerConfig""": ["""ignore_value"""],
    # `ignore_value` used during training (despite we don't have training script for these models yet)
    # `norm` used in conversion script (despite not using in the modeling file)
    """OneFormerConfig""": ["""ignore_value""", """norm"""],
    # used during preprocessing and collation, see `collating_graphormer.py`
    """GraphormerConfig""": ["""spatial_pos_max"""],
    # used internally in the configuration class file
    """T5Config""": ["""feed_forward_proj"""],
    # used internally in the configuration class file
    # `tokenizer_class` get default value `T5Tokenizer` intentionally
    """MT5Config""": ["""feed_forward_proj""", """tokenizer_class"""],
    """UMT5Config""": ["""feed_forward_proj""", """tokenizer_class"""],
    # used internally in the configuration class file
    """LongT5Config""": ["""feed_forward_proj"""],
    # used internally in the configuration class file
    """SwitchTransformersConfig""": ["""feed_forward_proj"""],
    # having default values other than `1e-5` - we can't fix them without breaking
    """BioGptConfig""": ["""layer_norm_eps"""],
    # having default values other than `1e-5` - we can't fix them without breaking
    """GLPNConfig""": ["""layer_norm_eps"""],
    # having default values other than `1e-5` - we can't fix them without breaking
    """SegformerConfig""": ["""layer_norm_eps"""],
    # having default values other than `1e-5` - we can't fix them without breaking
    """CvtConfig""": ["""layer_norm_eps"""],
    # having default values other than `1e-5` - we can't fix them without breaking
    """PerceiverConfig""": ["""layer_norm_eps"""],
    # used internally to calculate the feature size
    """InformerConfig""": ["""num_static_real_features""", """num_time_features"""],
    # used internally to calculate the feature size
    """TimeSeriesTransformerConfig""": ["""num_static_real_features""", """num_time_features"""],
    # used internally to calculate the feature size
    """AutoformerConfig""": ["""num_static_real_features""", """num_time_features"""],
    # used internally to calculate `mlp_dim`
    """SamVisionConfig""": ["""mlp_ratio"""],
    # For (head) training, but so far not implemented
    """ClapAudioConfig""": ["""num_classes"""],
    # Not used, but providing useful information to users
    """SpeechT5HifiGanConfig""": ["""sampling_rate"""],
}


# TODO (ydshieh): Check the failing cases, try to fix them or move some cases to the above block once we are sure
SPECIAL_CASES_TO_ALLOW.update(
    {
        """CLIPSegConfig""": True,
        """DeformableDetrConfig""": True,
        """DetaConfig""": True,
        """DinatConfig""": True,
        """DonutSwinConfig""": True,
        """EfficientFormerConfig""": True,
        """FSMTConfig""": True,
        """JukeboxConfig""": True,
        """LayoutLMv2Config""": True,
        """MaskFormerSwinConfig""": True,
        """MT5Config""": True,
        """NatConfig""": True,
        """OneFormerConfig""": True,
        """PerceiverConfig""": True,
        """RagConfig""": True,
        """SpeechT5Config""": True,
        """SwinConfig""": True,
        """Swin2SRConfig""": True,
        """Swinv2Config""": True,
        """SwitchTransformersConfig""": True,
        """TableTransformerConfig""": True,
        """TapasConfig""": True,
        """TransfoXLConfig""": True,
        """UniSpeechConfig""": True,
        """UniSpeechSatConfig""": True,
        """WavLMConfig""": True,
        """WhisperConfig""": True,
        # TODO: @Arthur (for `alignment_head` and `alignment_layer`)
        """JukeboxPriorConfig""": True,
        # TODO: @Younes (for `is_decoder`)
        """Pix2StructTextConfig""": True,
    }
)
def 					__lowerCAmelCase       (_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase	):
	__lowerCAmelCase    :		Optional[int]		    =   False
	for attribute in attributes:
		for modeling_source in source_strings:
			# check if we can find `config.xxx`, `getattr(config, "xxx", ...)` or `getattr(self.config, "xxx", ...)`
			if (
			    F"config.{attribute}" in modeling_source
			    or F"getattr(config, \"{attribute}\"" in modeling_source
			    or F"getattr(self.config, \"{attribute}\"" in modeling_source
			):
				__lowerCAmelCase    :		Union[str, Any]		    =   True
			# Deal with multi-line cases
			elif (
			    re.search(
			        rF"getattr[ \t\v\n\r\f]*\([ \t\v\n\r\f]*(self\.)?config,[ \t\v\n\r\f]*\"{attribute}\"" ,						_UpperCamelCase ,						)
			    is not None
			):
				__lowerCAmelCase    :		Optional[Any]		    =   True
			# `SequenceSummary` is called with `SequenceSummary(config)`
			elif attribute in [
			    "summary_type",
			    "summary_use_proj",
			    "summary_activation",
			    "summary_last_dropout",
			    "summary_proj_to_labels",
			    "summary_first_dropout",
			]:
				if "SequenceSummary" in modeling_source:
					__lowerCAmelCase    :		int		    =   True
			if attribute_used:
				break
		if attribute_used:
			break

    # common and important attributes, even if they do not always appear in the modeling files
	__lowerCAmelCase    :		Optional[Any]		    =   [
	    'bos_index',
	    'eos_index',
	    'pad_index',
	    'unk_index',
	    'mask_index',
	    'image_size',
	    'use_cache',
	    'out_features',
	    'out_indices',
	]
	__lowerCAmelCase    :		Tuple		    =   ['encoder_no_repeat_ngram_size']

	# Special cases to be allowed
	__lowerCAmelCase    :		Dict		    =   True
	if not attribute_used:
		__lowerCAmelCase    :		int		    =   False
		for attribute in attributes:
			# Allow if the default value in the configuration class is different from the one in `PretrainedConfig`
			if attribute in ["is_encoder_decoder"] and default_value is True:
				__lowerCAmelCase    :		int		    =   True
			elif attribute in ["tie_word_embeddings"] and default_value is False:
				__lowerCAmelCase    :		Dict		    =   True

			# Allow cases without checking the default value in the configuration class
			elif attribute in attributes_to_allow + attributes_used_in_generation:
				__lowerCAmelCase    :		Tuple		    =   True
			elif attribute.endswith('_token_id'	):
				__lowerCAmelCase    :		int		    =   True

			# configuration class specific cases
			if not case_allowed:
				__lowerCAmelCase    :		Dict		    =   SPECIAL_CASES_TO_ALLOW.get(config_class.__name__ ,						[]	)
				__lowerCAmelCase    :		Any		    =   allowed_cases is True or attribute in allowed_cases

	return attribute_used or case_allowed
def 					__lowerCAmelCase       (_UpperCamelCase	):
	__lowerCAmelCase    :		List[str]		    =   dict(inspect.signature(config_class.__init__	).parameters	)
	__lowerCAmelCase    :		Tuple		    =   [x for x in list(signature.keys()	) if x not in ['self', 'kwargs']]
	__lowerCAmelCase    :		Tuple		    =   [signature[param].default for param in parameter_names]

	# If `attribute_map` exists, an attribute can have different names to be used in the modeling files, and as long
	# as one variant is used, the test should pass
	__lowerCAmelCase    :		List[Any]		    =   {}
	if len(config_class.attribute_map	) > 0:
		__lowerCAmelCase    :		Union[str, Any]		    =   {v: k for k, v in config_class.attribute_map.items()}

	# Get the path to modeling source files
	__lowerCAmelCase    :		Dict		    =   inspect.getsourcefile(_UpperCamelCase	)
	__lowerCAmelCase    :		List[str]		    =   os.path.dirname(_UpperCamelCase	)
	# Let's check against all frameworks: as long as one framework uses an attribute, we are good.
	__lowerCAmelCase    :		Tuple		    =   [os.path.join(_UpperCamelCase ,						_UpperCamelCase	) for fn in os.listdir(_UpperCamelCase	) if fn.startswith('modeling_'	)]

	# Get the source code strings
	__lowerCAmelCase    :		str		    =   []
	for path in modeling_paths:
		if os.path.isfile(_UpperCamelCase	):
			with open(_UpperCamelCase	) as fp:
				modeling_sources.append(fp.read()	)

	__lowerCAmelCase    :		Optional[int]		    =   []
	for config_param, default_value in zip(_UpperCamelCase ,						_UpperCamelCase	):
		# `attributes` here is all the variant names for `config_param`
		__lowerCAmelCase    :		Optional[int]		    =   [config_param]
		# some configuration classes have non-empty `attribute_map`, and both names could be used in the
		# corresponding modeling files. As long as one of them appears, it is fine.
		if config_param in reversed_attribute_map:
			attributes.append(reversed_attribute_map[config_param]	)

		if not check_attribute_being_used(_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase ,						_UpperCamelCase	):
			unused_attributes.append(attributes[0]	)

	return sorted(_UpperCamelCase	)
def 					__lowerCAmelCase       ():
	__lowerCAmelCase    :		List[str]		    =   {}
	for _config_class in list(CONFIG_MAPPING.values()	):
		# Skip deprecated models
		if "models.deprecated" in _config_class.__module__:
			continue
		# Some config classes are not in `CONFIG_MAPPING` (e.g. `CLIPVisionConfig`, `Blip2VisionConfig`, etc.)
		__lowerCAmelCase    :		int		    =   [
		    cls
		    for name, cls in inspect.getmembers(
		        inspect.getmodule(_config_class	) ,						lambda _UpperCamelCase	: inspect.isclass(_UpperCamelCase	)
		        and issubclass(_UpperCamelCase ,						_UpperCamelCase	)
		        and inspect.getmodule(_UpperCamelCase	) == inspect.getmodule(_config_class	) ,						)
		]
		for config_class in config_classes_in_module:
			__lowerCAmelCase    :		List[Any]		    =   check_config_attributes_being_used(_UpperCamelCase	)
			if len(_UpperCamelCase	) > 0:
				__lowerCAmelCase    :		str		    =   unused_attributes

	if len(_UpperCamelCase	) > 0:
		__lowerCAmelCase    :		int		    =   'The following configuration classes contain unused attributes in the corresponding modeling files:\n'
		for name, attributes in configs_with_unused_attributes.items():
			error += F"{name}: {attributes}\n"

		raise ValueError(_UpperCamelCase	)


if __name__ == "__main__":
							check_config_attributes()