




'''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(							SCREAMING_SNAKE_CASE_       ,SCREAMING_SNAKE_CASE_       ,SCREAMING_SNAKE_CASE_     )		->		Union[str, Any]:
			if isinstance(SCREAMING_SNAKE_CASE_       ,torch.Tensor     ):
						return image
			elif isinstance(SCREAMING_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(SCREAMING_SNAKE_CASE_       ,axis=0     )
						A_									= np.array(SCREAMING_SNAKE_CASE_     ).astype(np.floataa     ) / 255.0
						A_									= image.transpose(0       ,3       ,1       ,2     )
						A_									= 2.0 * image - 1.0
						A_									= torch.from_numpy(SCREAMING_SNAKE_CASE_     )
			elif isinstance(image[0]       ,torch.Tensor     ):
						A_									= torch.cat(SCREAMING_SNAKE_CASE_       ,dim=0     )
			return image





def 			lowerCamelCase(							SCREAMING_SNAKE_CASE_       ,SCREAMING_SNAKE_CASE_       ,SCREAMING_SNAKE_CASE_       ,SCREAMING_SNAKE_CASE_=0.9_995     )		->		List[str]:
			if not isinstance(SCREAMING_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(SCREAMING_SNAKE_CASE_     ) * np.linalg.norm(SCREAMING_SNAKE_CASE_     ))     )
			if np.abs(SCREAMING_SNAKE_CASE_     ) > DOT_THRESHOLD:
						A_									= (1 - t) * va + t * va
			else:
						A_									= np.arccos(SCREAMING_SNAKE_CASE_     )
						A_									= np.sin(SCREAMING_SNAKE_CASE_     )
						A_									= theta_a * t
						A_									= np.sin(SCREAMING_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(SCREAMING_SNAKE_CASE_     ).to(SCREAMING_SNAKE_CASE_     )

			return va





def 			lowerCamelCase(							SCREAMING_SNAKE_CASE_       ,SCREAMING_SNAKE_CASE_     )		->		Dict:
			A_									= F.normalize(SCREAMING_SNAKE_CASE_       ,dim=-1     )
			A_									= F.normalize(SCREAMING_SNAKE_CASE_       ,dim=-1     )
			return (x - y).norm(dim=-1     ).div(2     ).arcsin().pow(2     ).mul(2     )





def 			lowerCamelCase(							SCREAMING_SNAKE_CASE_       ,SCREAMING_SNAKE_CASE_     )		->		str:
			for param in model.parameters():
						A_									= value






class   __UpperCAmelCase				(     lowerCAmelCase				):



					'''simple docstring'''





					def __init__(	self     :						Optional[int]      ,       _lowercase     :						AutoencoderKL      ,       _lowercase     :						CLIPTextModel      ,       _lowercase     :						CLIPModel      ,       _lowercase     :						CLIPTokenizer      ,       _lowercase     :						UNetaDConditionModel      ,       _lowercase     :						Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]      ,       _lowercase     :						CLIPFeatureExtractor      ,       _lowercase     :						str=None      ,       _lowercase     :						Any=None      ,       _lowercase     :						Any=None      ,       )    ->							str:
								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     __snake_case						(	self     :						Any      ,       _lowercase     :						Optional[Union[str, int]] = "auto")    ->							Union[str, Any]:
								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     __snake_case						(	self     :						Dict)    ->							List[str]:
								self.enable_attention_slicing(_lowercase)



					def     __snake_case						(	self     :						Union[str, Any])    ->							int:
								set_requires_grad(self.vae      ,       _lowercase)



					def     __snake_case						(	self     :						int)    ->							Any:
								set_requires_grad(self.vae      ,       _lowercase)



					def     __snake_case						(	self     :						Union[str, Any])    ->							Dict:
								set_requires_grad(self.unet      ,       _lowercase)



					def     __snake_case						(	self     :						List[Any])    ->							str:
								set_requires_grad(self.unet      ,       _lowercase)



					def     __snake_case						(	self     :						int      ,       _lowercase     :						Tuple      ,       _lowercase     :						List[str]      ,       _lowercase     :						Any)    ->							Dict:
								# get the original timestep using init_timestep
								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     __snake_case						(	self     :						int      ,       _lowercase     :						Dict      ,       _lowercase     :						Union[str, Any]      ,       _lowercase     :						Optional[Any]      ,       _lowercase     :						Optional[Any]      ,       _lowercase     :						List[Any]      ,       _lowercase     :						Union[str, Any]=None)    ->							str:
								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_82_15 * 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     __snake_case						(	self     :						List[Any]      ,       _lowercase     :						str)    ->							Union[str, Any]:
								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     __snake_case						(	self     :						Tuple      ,       _lowercase     :						List[Any]      ,       _lowercase     :						Optional[Any])    ->							str:
								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     __snake_case						(	self     :						List[str]      ,       _lowercase     :						int      ,       _lowercase     :						int      ,       _lowercase     :						Optional[Any]      ,       _lowercase     :						List[Any]      ,       _lowercase     :						Any      ,       _lowercase     :						Union[str, Any]      ,       _lowercase     :						Union[str, Any]      ,       )    ->							str:
								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_82_15 * 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     :						Optional[Any]      ,       _lowercase     :						Union[torch.FloatTensor, PIL.Image.Image]      ,       _lowercase     :						Union[torch.FloatTensor, PIL.Image.Image]      ,       _lowercase     :						Optional[str] = None      ,       _lowercase     :						Optional[str] = None      ,       _lowercase     :						Optional[int] = 512      ,       _lowercase     :						Optional[int] = 512      ,       _lowercase     :						float = 0.6      ,       _lowercase     :						Optional[int] = 50      ,       _lowercase     :						Optional[float] = 7.5      ,       _lowercase     :						Optional[int] = 1      ,       _lowercase     :						float = 0.0      ,       _lowercase     :						Optional[float] = 100      ,       _lowercase     :						Optional[torch.Generator] = None      ,       _lowercase     :						Optional[str] = "pil"      ,       _lowercase     :						bool = True      ,       _lowercase     :						float = 0.8      ,       _lowercase     :						float = 0.1      ,       _lowercase     :						float = 0.1      ,       )    ->							int:
								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_		, 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_		, 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_		, 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_82_15 * 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)










'''simple docstring'''
from ...configuration_utils import PretrainedConfig
from ...utils import logging


_SCREAMING_SNAKE_CASE			=						logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE			=						{"openai-gpt": "https://huggingface.co/openai-gpt/resolve/main/config.json"}
class   __UpperCAmelCase				(     lowerCAmelCase				):



					'''simple docstring'''





					_UpperCamelCase							=		"""openai-gpt"""
					_UpperCamelCase							=		{
					    """max_position_embeddings""": """n_positions""",
					    """hidden_size""": """n_embd""",
					    """num_attention_heads""": """n_head""",
					    """num_hidden_layers""": """n_layer""",
					}



					def __init__(	self     :						Any      ,       _lowercase     :						List[Any]=40_478      ,       _lowercase     :						Tuple=512      ,       _lowercase     :						Dict=768      ,       _lowercase     :						List[str]=12      ,       _lowercase     :						Optional[int]=12      ,       _lowercase     :						Tuple="gelu"      ,       _lowercase     :						int=0.1      ,       _lowercase     :						Tuple=0.1      ,       _lowercase     :						Any=0.1      ,       _lowercase     :						List[str]=1E-5      ,       _lowercase     :						str=0.02      ,       _lowercase     :						Optional[Any]="cls_index"      ,       _lowercase     :						int=True      ,       _lowercase     :						str=None      ,       _lowercase     :						Tuple=True      ,       _lowercase     :						List[Any]=0.1      ,       **_lowercase     :						int      ,       )    ->							Optional[int]:
								A_									= vocab_size
								A_									= n_positions
								A_									= n_embd
								A_									= n_layer
								A_									= n_head
								A_									= afn
								A_									= resid_pdrop
								A_									= embd_pdrop
								A_									= attn_pdrop
								A_									= layer_norm_epsilon
								A_									= initializer_range
								A_									= summary_type
								A_									= summary_use_proj
								A_									= summary_activation
								A_									= summary_first_dropout
								A_									= summary_proj_to_labels
								super().__init__(**_lowercase)




