

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__							(	SCREAMING_SNAKE_CASE_:							int	,       SCREAMING_SNAKE_CASE_:							List[str]	,       SCREAMING_SNAKE_CASE_:							int      )			->      Optional[int]:




				'''simple docstring'''





				if isinstance(snake_case_	,       torch.Tensor      ):
								return image
				elif isinstance(snake_case_	,       PIL.Image.Image      ):
								A__        =			[image]

				if isinstance(image[0]	,       PIL.Image.Image      ):
								A__        =			[np.array(i.resize((w, h)	,       resample=PIL_INTERPOLATION["lanczos"]      )      )[None, :] for i in image]
								A__        =			np.concatenate(snake_case_	,       axis=0      )
								A__        =			np.array(snake_case_      ).astype(np.floataa      ) / 2_5_5.0
								A__        =			image.transpose(0	,       3	,       1	,       2      )
								A__        =			2.0 * image - 1.0
								A__        =			torch.from_numpy(snake_case_      )
				elif isinstance(image[0]	,       torch.Tensor      ):
								A__        =			torch.cat(snake_case_	,       dim=0      )
				return image



def 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							Dict	,       SCREAMING_SNAKE_CASE_:							Union[str, Any]	,       SCREAMING_SNAKE_CASE_:							Union[str, Any]	,       SCREAMING_SNAKE_CASE_:							Any=0.9995      )			->      Optional[int]:




				'''simple docstring'''





				if not isinstance(snake_case_	,       np.ndarray      ):
								A__        =			True
								A__        =			va.device
								A__        =			va.cpu().numpy()
								A__        =			va.cpu().numpy()

				A__        =			np.sum(va * va / (np.linalg.norm(snake_case_      ) * np.linalg.norm(snake_case_      ))      )
				if np.abs(snake_case_      ) > DOT_THRESHOLD:
								A__        =			(1 - t) * va + t * va
				else:
								A__        =			np.arccos(snake_case_      )
								A__        =			np.sin(snake_case_      )
								A__        =			theta_a * t
								A__        =			np.sin(snake_case_      )
								A__        =			np.sin(theta_a - theta_t      ) / sin_theta_a
								A__        =			sin_theta_t / sin_theta_a
								A__        =			sa * va + sa * va

				if inputs_are_torch:
								A__        =			torch.from_numpy(snake_case_      ).to(snake_case_      )

				return va



def 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							Union[str, Any]	,       SCREAMING_SNAKE_CASE_:							Optional[Any]      )			->      str:




				'''simple docstring'''





				A__        =			F.normalize(snake_case_	,       dim=-1      )
				A__        =			F.normalize(snake_case_	,       dim=-1      )
				return (x - y).norm(dim=-1      ).div(2      ).arcsin().pow(2      ).mul(2      )



def 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							Dict	,       SCREAMING_SNAKE_CASE_:							List[Any]      )			->      int:




				'''simple docstring'''





				for param in model.parameters():
								A__        =			value





class 					a__						(					UpperCamelCase_			):
			"""simple docstring"""







			def __init__(    self      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase=None      ,      lowercase=None      ,      lowercase=None      ,      )  ->						str:
							'''simple docstring'''






							super().__init__()
							self.register_modules(
							    vae=lowercase      ,      text_encoder=lowercase      ,      clip_model=lowercase      ,      tokenizer=lowercase      ,      unet=lowercase      ,      scheduler=lowercase      ,      feature_extractor=lowercase      ,      coca_model=lowercase      ,      coca_tokenizer=lowercase      ,      coca_transform=lowercase      ,      )
							A__        =			(
							    feature_extractor.size
							    if isinstance(feature_extractor.size      ,      lowercase       )
							    else feature_extractor.size['''shortest_edge''']
							)
							A__        =			transforms.Normalize(mean=feature_extractor.image_mean      ,      std=feature_extractor.image_std       )
							set_requires_grad(self.text_encoder      ,      lowercase       )
							set_requires_grad(self.clip_model      ,      lowercase       )


			def      UpperCamelCase		(    self      ,      lowercase = "auto"       )  ->						Dict:
							'''simple docstring'''






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


			def      UpperCamelCase		(    self       )  ->						int:
							'''simple docstring'''






							self.enable_attention_slicing(lowercase       )


			def      UpperCamelCase		(    self       )  ->						str:
							'''simple docstring'''






							set_requires_grad(self.vae      ,      lowercase       )


			def      UpperCamelCase		(    self       )  ->						Any:
							'''simple docstring'''






							set_requires_grad(self.vae      ,      lowercase       )


			def      UpperCamelCase		(    self       )  ->						Optional[Any]:
							'''simple docstring'''






							set_requires_grad(self.unet      ,      lowercase       )


			def      UpperCamelCase		(    self       )  ->						Union[str, Any]:
							'''simple docstring'''






							set_requires_grad(self.unet      ,      lowercase       )


			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase      ,      lowercase       )  ->						Optional[int]:
							'''simple docstring'''






							A__        =			min(int(num_inference_steps * strength       )      ,      lowercase       )

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

							return timesteps, num_inference_steps - t_start


			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase=None       )  ->						Any:
							'''simple docstring'''






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

							A__        =			image.to(device=lowercase      ,      dtype=lowercase       )

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

							# Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
							A__        =			0.1_8215 * init_latents
							A__        =			init_latents.repeat_interleave(lowercase      ,      dim=0       )

							A__        =			randn_tensor(init_latents.shape      ,      generator=lowercase      ,      device=lowercase      ,      dtype=lowercase       )

							# get latents
							A__        =			self.scheduler.add_noise(lowercase      ,      lowercase      ,      lowercase       )
							A__        =			init_latents

							return latents


			def      UpperCamelCase		(    self      ,      lowercase       )  ->						Any:
							'''simple docstring'''






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


			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase       )  ->						List[str]:
							'''simple docstring'''






							A__        =			self.feature_extractor.preprocess(lowercase       )
							A__        =			torch.from_numpy(clip_image_input["pixel_values"][0]       ).unsqueeze(0       ).to(self.device       ).half()
							A__        =			self.clip_model.get_image_features(lowercase       )
							A__        =			image_embeddings_clip / image_embeddings_clip.norm(p=2      ,      dim=-1      ,      keepdim=lowercase       )
							A__        =			image_embeddings_clip.repeat_interleave(lowercase      ,      dim=0       )
							return image_embeddings_clip


			@torch.enable_grad()
			def      UpperCamelCase		(    self      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      lowercase      ,      )  ->						Tuple:
							'''simple docstring'''






							A__        =			latents.detach().requires_grad_()

							A__        =			self.scheduler.scale_model_input(lowercase      ,      lowercase       )

							# predict the noise residual
							A__        =			self.unet(lowercase      ,      lowercase      ,      encoder_hidden_states=lowercase       ).sample

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

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

							A__        =			transforms.Resize(self.feature_extractor_size       )(lowercase       )
							A__        =			self.normalize(lowercase       ).to(latents.dtype       )

							A__        =			self.clip_model.get_image_features(lowercase       )
							A__        =			image_embeddings_clip / image_embeddings_clip.norm(p=2      ,      dim=-1      ,      keepdim=lowercase       )

							A__        =			spherical_dist_loss(lowercase      ,      lowercase       ).mean() * clip_guidance_scale

							A__        =			-torch.autograd.grad(lowercase      ,      lowercase       )[0]

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


			@torch.no_grad()
			def __call__(    self      ,      lowercase      ,      lowercase      ,      lowercase = None      ,      lowercase = None      ,      lowercase = 512      ,      lowercase = 512      ,      lowercase = 0.6      ,      lowercase = 50      ,      lowercase = 7.5      ,      lowercase = 1      ,      lowercase = 0.0      ,      lowercase = 100      ,      lowercase = None      ,      lowercase = "pil"      ,      lowercase = True      ,      lowercase = 0.8      ,      lowercase = 0.1      ,      lowercase = 0.1      ,      )  ->						Optional[Any]:
							'''simple docstring'''






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

							A__        =			[
							    ('''model''', self.coca_model is None),
							    ('''tokenizer''', self.coca_tokenizer is None),
							    ('''transform''', self.coca_transform is None),
							]
							A__        =			[x[0] for x in coca_is_none if x[1]]
							A__        =			''', '''.join(lowercase       )
							# generate prompts with coca model if prompt is None
							if content_prompt is None:
											if len(lowercase       ):
															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.'       )
											A__        =			self.get_image_description(lowercase       )
							if style_prompt is None:
											if len(lowercase       ):
															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.'       )
											A__        =			self.get_image_description(lowercase       )

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

							A__        =			self.tokenizer(
							    lowercase      ,      padding="max_length"      ,      max_length=self.tokenizer.model_max_length      ,      truncation=lowercase      ,      return_tensors="pt"      ,      )
							A__        =			self.text_encoder(style_text_input.input_ids.to(self.device       )       )[0]

							A__        =			slerp(lowercase      ,      lowercase      ,      lowercase       )

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

							# set timesteps
							A__        =			'''offset''' in set(inspect.signature(self.scheduler.set_timesteps       ).parameters.keys()       )
							A__        =			{}
							if accepts_offset:
											A__        =			1

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

							A__        =			self.get_timesteps(lowercase      ,      lowercase      ,      self.device       )
							A__        =			timesteps[:1].repeat(lowercase       )

							# Preprocess image
							A__        =			preprocess(lowercase      ,      lowercase      ,      lowercase       )
							A__        =			self.prepare_latents(
							    lowercase      ,      lowercase      ,      lowercase      ,      text_embeddings.dtype      ,      self.device      ,      lowercase       )

							A__        =			preprocess(lowercase      ,      lowercase      ,      lowercase       )
							A__        =			self.prepare_latents(
							    lowercase      ,      lowercase      ,      lowercase      ,      text_embeddings.dtype      ,      self.device      ,      lowercase       )

							A__        =			slerp(lowercase      ,      lowercase      ,      lowercase       )

							if clip_guidance_scale > 0:
											A__        =			self.get_clip_image_embeddings(lowercase      ,      lowercase       )
											A__        =			self.get_clip_image_embeddings(lowercase      ,      lowercase       )
											A__        =			slerp(
											    lowercase      ,      lowercase      ,      lowercase       )

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

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

							# check if the scheduler accepts generator
							A__        =			'''generator''' in set(inspect.signature(self.scheduler.step       ).parameters.keys()       )
							if accepts_generator:
											A__        =			generator

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

															# predict the noise residual
															A__        =			self.unet(lowercase      ,      lowercase      ,      encoder_hidden_states=lowercase       ).sample

															# perform classifier free guidance
															if do_classifier_free_guidance:
																			A__        =			noise_pred.chunk(2       )
																			A__        =			noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

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

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

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

							A__        =			(image / 2 + 0.5).clamp(0      ,      1       )
							A__        =			image.cpu().permute(0      ,      2      ,      3      ,      1       ).numpy()

							if output_type == "pil":
											A__        =			self.numpy_to_pil(lowercase       )

							if not return_dict:
											return (image, None)

							return StableDiffusionPipelineOutput(images=lowercase      ,      nsfw_content_detected=lowercase       )




import inspect
import re

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

# Regex pattern used to find the checkpoint mentioned in the docstring of `config_class`.
# For example, `[bert-base-uncased](https://huggingface.co/bert-base-uncased)`
lowerCAmelCase__	   = re.compile(R"""\[(.+?)\]\((https://huggingface\.co/.+?)\)""")


lowerCAmelCase__	   = {
    """DecisionTransformerConfig""",
    """EncoderDecoderConfig""",
    """MusicgenConfig""",
    """RagConfig""",
    """SpeechEncoderDecoderConfig""",
    """TimmBackboneConfig""",
    """VisionEncoderDecoderConfig""",
    """VisionTextDualEncoderConfig""",
    """LlamaConfig""",
}



def 					lowerCAmelCase__							(	SCREAMING_SNAKE_CASE_:							Union[str, Any]      )			->      Dict:




				'''simple docstring'''





				A__        =			None

				# source code of `config_class`
				A__        =			inspect.getsource(SCREAMING_SNAKE_CASE_      )
				A__        =			_re_checkpoint.findall(SCREAMING_SNAKE_CASE_      )

				# Each `checkpoint` is a tuple of a checkpoint name and a checkpoint link.
				# For example, `('bert-base-uncased', 'https://huggingface.co/bert-base-uncased')`
				for ckpt_name, ckpt_link in checkpoints:
								# allow the link to end with `/`
								if ckpt_link.endswith("/"      ):
												A__        =			ckpt_link[:-1]

								# verify the checkpoint name corresponds to the checkpoint link
								A__        =			F'https://huggingface.co/{ckpt_name}'
								if ckpt_link == ckpt_link_from_name:
												A__        =			ckpt_name
												break

				return checkpoint



def 					lowerCAmelCase__							(	)			->      List[str]:




				'''simple docstring'''





				A__        =			[]

				for config_class in list(CONFIG_MAPPING.values()      ):
								# Skip deprecated models
								if "models.deprecated" in config_class.__module__:
												continue
								A__        =			get_checkpoint_from_config_class(SCREAMING_SNAKE_CASE_      )

								A__        =			config_class.__name__
								if checkpoint is None and name not in CONFIG_CLASSES_TO_IGNORE_FOR_DOCSTRING_CHECKPOINT_CHECK:
												configs_without_checkpoint.append(SCREAMING_SNAKE_CASE_      )

				if len(SCREAMING_SNAKE_CASE_      ) > 0:
								A__        =			"\n".join(sorted(SCREAMING_SNAKE_CASE_      )      )
								raise ValueError(F'The following configurations don\'t contain any valid checkpoint:\n{message}'      )


if __name__ == "__main__":
	check_config_docstrings_have_checkpoints()

