

'''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 	_a       (				__lowerCAmelCase		:      Tuple   ,							__lowerCAmelCase		:      Optional[int]   ,							__lowerCAmelCase		:      Dict     ):

		"""simple docstring"""

		if isinstance(__lowerCAmelCase   ,							torch.Tensor     ):
				return image
		elif isinstance(__lowerCAmelCase   ,							PIL.Image.Image     ):
				snake_case__						:       str       						=  [image]

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




def 	_a       (				__lowerCAmelCase		:      List[Any]   ,							__lowerCAmelCase		:      Union[str, Any]   ,							__lowerCAmelCase		:      Any   ,							__lowerCAmelCase		:      List[Any]=0.9995     ):

		"""simple docstring"""

		if not isinstance(__lowerCAmelCase   ,							np.ndarray     ):
				snake_case__						:       Optional[int]       						=  True
				snake_case__						:       Dict       						=  va.device
				snake_case__						:       Optional[int]       						=  va.cpu().numpy()
				snake_case__						:       Dict       						=  va.cpu().numpy()

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

		if inputs_are_torch:
				snake_case__						:       List[Any]       						=  torch.from_numpy(__lowerCAmelCase     ).to(__lowerCAmelCase     )

		return va




def 	_a       (				__lowerCAmelCase		:      Tuple   ,							__lowerCAmelCase		:      Tuple     ):

		"""simple docstring"""

		snake_case__						:       Optional[Any]       						=  F.normalize(__lowerCAmelCase   ,							dim=-1     )
		snake_case__						:       Optional[int]       						=  F.normalize(__lowerCAmelCase   ,							dim=-1     )
		return (x - y).norm(dim=-1     ).div(2     ).arcsin().pow(2     ).mul(2     )




def 	_a       (				__lowerCAmelCase		:      Tuple   ,							__lowerCAmelCase		:      List[str]     ):

		"""simple docstring"""

		for param in model.parameters():
				snake_case__						:       str       						=  value





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						def __init__(							self					:      List[Any]			,			snake_case_					:      AutoencoderKL			,			snake_case_					:      CLIPTextModel			,			snake_case_					:      CLIPModel			,			snake_case_					:      CLIPTokenizer			,			snake_case_					:      UNetaDConditionModel			,			snake_case_					:      Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]			,			snake_case_					:      CLIPFeatureExtractor			,			snake_case_					:      Optional[Any]=None			,			snake_case_					:      List[str]=None			,			snake_case_					:      str=None			,			):



								'''simple docstring'''



								super().__init__()
								self.register_modules(
								    vae=snake_case_			,			text_encoder=snake_case_			,			clip_model=snake_case_			,			tokenizer=snake_case_			,			unet=snake_case_			,			scheduler=snake_case_			,			feature_extractor=snake_case_			,			coca_model=snake_case_			,			coca_tokenizer=snake_case_			,			coca_transform=snake_case_			,			)
								snake_case__						:       int       						=  (
								    feature_extractor.size
								    if isinstance(feature_extractor.size			,			snake_case_  )
								    else feature_extractor.size['''shortest_edge''']
								)
								snake_case__						:       Union[str, Any]       						=  transforms.Normalize(mean=feature_extractor.image_mean			,			std=feature_extractor.image_std  )
								set_requires_grad(self.text_encoder			,			snake_case_  )
								set_requires_grad(self.clip_model			,			snake_case_  )







						def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      Optional[Union[str, int]] = "auto"  ):



								'''simple docstring'''



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







						def       __magic_name__							(							self					:      str  ):



								'''simple docstring'''



								self.enable_attention_slicing(snake_case_  )







						def       __magic_name__							(							self					:      str  ):



								'''simple docstring'''



								set_requires_grad(self.vae			,			snake_case_  )







						def       __magic_name__							(							self					:      Any  ):



								'''simple docstring'''



								set_requires_grad(self.vae			,			snake_case_  )







						def       __magic_name__							(							self					:      Optional[Any]  ):



								'''simple docstring'''



								set_requires_grad(self.unet			,			snake_case_  )







						def       __magic_name__							(							self					:      Union[str, Any]  ):



								'''simple docstring'''



								set_requires_grad(self.unet			,			snake_case_  )







						def       __magic_name__							(							self					:      Any			,			snake_case_					:      Optional[Any]			,			snake_case_					:      Tuple			,			snake_case_					:      List[str]  ):



								'''simple docstring'''



								snake_case__						:       List[str]       						=  min(int(num_inference_steps * strength  )			,			snake_case_  )

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

								return timesteps, num_inference_steps - t_start







						def       __magic_name__							(							self					:      Any			,			snake_case_					:      Union[str, Any]			,			snake_case_					:      List[Any]			,			snake_case_					:      List[Any]			,			snake_case_					:      Tuple			,			snake_case_					:      str			,			snake_case_					:      Optional[int]=None  ):



								'''simple docstring'''



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

								snake_case__						:       Union[str, Any]       						=  image.to(device=snake_case_			,			dtype=snake_case_  )

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

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

								snake_case__						:       str       						=  randn_tensor(init_latents.shape			,			generator=snake_case_			,			device=snake_case_			,			dtype=snake_case_  )

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

								return latents







						def       __magic_name__							(							self					:      Any			,			snake_case_					:      List[Any]  ):



								'''simple docstring'''



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







						def       __magic_name__							(							self					:      int			,			snake_case_					:      Union[str, Any]			,			snake_case_					:      Any  ):



								'''simple docstring'''



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







						@torch.enable_grad()
						def       __magic_name__							(							self					:      Any			,			snake_case_					:      Tuple			,			snake_case_					:      Union[str, Any]			,			snake_case_					:      Optional[Any]			,			snake_case_					:      int			,			snake_case_					:      Optional[int]			,			snake_case_					:      Dict			,			snake_case_					:      List[str]			,			):



								'''simple docstring'''



								snake_case__						:       Optional[int]       						=  latents.detach().requires_grad_()

								snake_case__						:       int       						=  self.scheduler.scale_model_input(snake_case_			,			snake_case_  )

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

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

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

								snake_case__						:       Tuple       						=  transforms.Resize(self.feature_extractor_size  )(snake_case_  )
								snake_case__						:       Optional[Any]       						=  self.normalize(snake_case_  ).to(latents.dtype  )

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

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

								snake_case__						:       List[str]       						=  -torch.autograd.grad(snake_case_			,			snake_case_  )[0]

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







						@torch.no_grad()
						def __call__(							self					:      Dict			,			snake_case_					:      Union[torch.FloatTensor, PIL.Image.Image]			,			snake_case_					:      Union[torch.FloatTensor, PIL.Image.Image]			,			snake_case_					:      Optional[str] = None			,			snake_case_					:      Optional[str] = None			,			snake_case_					:      Optional[int] = 5_1_2			,			snake_case_					:      Optional[int] = 5_1_2			,			snake_case_					:      float = 0.6			,			snake_case_					:      Optional[int] = 5_0			,			snake_case_					:      Optional[float] = 7.5			,			snake_case_					:      Optional[int] = 1			,			snake_case_					:      float = 0.0			,			snake_case_					:      Optional[float] = 1_0_0			,			snake_case_					:      Optional[torch.Generator] = None			,			snake_case_					:      Optional[str] = "pil"			,			snake_case_					:      bool = True			,			snake_case_					:      float = 0.8			,			snake_case_					:      float = 0.1			,			snake_case_					:      float = 0.1			,			):



								'''simple docstring'''



								if isinstance(snake_case_			,			snake_case_  ) and len(snake_case_  ) != batch_size:
										raise ValueError(F"""You have passed {batch_size} batch_size, but only {len(snake_case_  )} generators."""  )

								if height % 8 != 0 or width % 8 != 0:
										raise ValueError(F"""`height` and `width` have to be divisible by 8 but are {height} and {width}."""  )

								if isinstance(snake_case_			,			torch.Generator  ) and batch_size > 1:
										snake_case__						:       int       						=  [generator] + [None] * (batch_size - 1)

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

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

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

								snake_case__						:       List[Any]       						=  slerp(snake_case_			,			snake_case_			,			snake_case_  )

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

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

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

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

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

								snake_case__						:       str       						=  preprocess(snake_case_			,			snake_case_			,			snake_case_  )
								snake_case__						:       Dict       						=  self.prepare_latents(
								    snake_case_			,			snake_case_			,			snake_case_			,			text_embeddings.dtype			,			self.device			,			snake_case_  )

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

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

								# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
								# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
								# corresponds to doing no classifier free guidance.
								snake_case__						:       List[str]       						=  guidance_scale > 1.0
								# get unconditional embeddings for classifier free guidance
								if do_classifier_free_guidance:
										snake_case__						:       List[str]       						=  content_text_input.input_ids.shape[-1]
										snake_case__						:       str       						=  self.tokenizer(['''''']			,			padding='''max_length'''			,			max_length=snake_case_			,			return_tensors='''pt'''  )
										snake_case__						:       Any       						=  self.text_encoder(uncond_input.input_ids.to(self.device  )  )[0]
										# duplicate unconditional embeddings for each generation per prompt
										snake_case__						:       Any       						=  uncond_embeddings.repeat_interleave(snake_case_			,			dim=0  )

										# For classifier free guidance, we need to do two forward passes.
										# Here we concatenate the unconditional and text embeddings into a single batch
										# to avoid doing two forward passes
										snake_case__						:       Any       						=  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`.
								snake_case__						:       Any       						=  (batch_size, self.unet.config.in_channels, height // 8, width // 8)
								snake_case__						:       Any       						=  text_embeddings.dtype
								if latents is None:
										if self.device.type == "mps":
												# randn does not work reproducibly on mps
												snake_case__						:       Optional[Any]       						=  torch.randn(snake_case_			,			generator=snake_case_			,			device='''cpu'''			,			dtype=snake_case_  ).to(
												    self.device  )
										else:
												snake_case__						:       List[str]       						=  torch.randn(snake_case_			,			generator=snake_case_			,			device=self.device			,			dtype=snake_case_  )
								else:
										if latents.shape != latents_shape:
												raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""  )
										snake_case__						:       List[str]       						=  latents.to(self.device  )

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

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

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

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

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

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

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

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

								snake_case__						:       List[Any]       						=  (image / 2 + 0.5).clamp(0			,			1  )
								snake_case__						:       Optional[Any]       						=  image.cpu().permute(0			,			2			,			3			,			1  ).numpy()

								if output_type == "pil":
										snake_case__						:       int       						=  self.numpy_to_pil(snake_case_  )

								if not return_dict:
										return (image, None)

								return StableDiffusionPipelineOutput(images=snake_case_			,			nsfw_content_detected=snake_case_  )






'''simple docstring'''


import unittest

from parameterized import parameterized

from transformers import AutoTokenizer, GPTNeoXConfig, is_torch_available, set_seed
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
					import torch

					from transformers import (
					    GPTNeoXForCausalLM,
					    GPTNeoXForQuestionAnswering,
					    GPTNeoXForSequenceClassification,
					    GPTNeoXForTokenClassification,
					    GPTNeoXModel,
					)




class     a			:






						"""simple docstring"""




						def __init__(							self					:      Any			,			snake_case_					:      str			,			snake_case_					:      Optional[Any]=1_3			,			snake_case_					:      int=7			,			snake_case_					:      int=True			,			snake_case_					:      Optional[Any]=True			,			snake_case_					:      Dict=True			,			snake_case_					:      int=True			,			snake_case_					:      Optional[Any]=9_9			,			snake_case_					:      int=6_4			,			snake_case_					:      Dict=5			,			snake_case_					:      List[Any]=4			,			snake_case_					:      Union[str, Any]=3_7			,			snake_case_					:      Dict="gelu"			,			snake_case_					:      Union[str, Any]=0.1			,			snake_case_					:      Dict=0.1			,			snake_case_					:      Any=5_1_2			,			snake_case_					:      Any=1_6			,			snake_case_					:      Any=2			,			snake_case_					:      Dict=0.0_2			,			snake_case_					:      List[str]=3			,			snake_case_					:      Optional[int]=4			,			snake_case_					:      str=None			,			):



								'''simple docstring'''



								snake_case__						:       List[Any]       						=  parent
								snake_case__						:       int       						=  batch_size
								snake_case__						:       Dict       						=  seq_length
								snake_case__						:       int       						=  is_training
								snake_case__						:       Optional[Any]       						=  use_input_mask
								snake_case__						:       Optional[Any]       						=  use_token_type_ids
								snake_case__						:       Dict       						=  use_labels
								snake_case__						:       int       						=  vocab_size
								snake_case__						:       Any       						=  hidden_size
								snake_case__						:       int       						=  num_hidden_layers
								snake_case__						:       Union[str, Any]       						=  num_attention_heads
								snake_case__						:       List[Any]       						=  intermediate_size
								snake_case__						:       int       						=  hidden_act
								snake_case__						:       int       						=  hidden_dropout_prob
								snake_case__						:       List[Any]       						=  attention_probs_dropout_prob
								snake_case__						:       Optional[int]       						=  max_position_embeddings
								snake_case__						:       Optional[int]       						=  type_vocab_size
								snake_case__						:       Any       						=  type_sequence_label_size
								snake_case__						:       str       						=  initializer_range
								snake_case__						:       List[str]       						=  num_labels
								snake_case__						:       Dict       						=  num_choices
								snake_case__						:       Union[str, Any]       						=  scope
								snake_case__						:       List[Any]       						=  vocab_size - 1







						def       __magic_name__							(							self					:      List[Any]  ):



								'''simple docstring'''



								snake_case__						:       Dict       						=  ids_tensor([self.batch_size, self.seq_length]			,			self.vocab_size  )

								snake_case__						:       Optional[int]       						=  None
								if self.use_input_mask:
										snake_case__						:       int       						=  random_attention_mask([self.batch_size, self.seq_length]  )

								snake_case__						:       Union[str, Any]       						=  None
								if self.use_labels:
										snake_case__						:       Optional[Any]       						=  ids_tensor([self.batch_size, self.seq_length]			,			self.num_labels  )

								snake_case__						:       Dict       						=  self.get_config()

								return config, input_ids, input_mask, token_labels







						def       __magic_name__							(							self					:      List[Any]  ):



								'''simple docstring'''



								return GPTNeoXConfig(
								    vocab_size=self.vocab_size			,			hidden_size=self.hidden_size			,			num_hidden_layers=self.num_hidden_layers			,			num_attention_heads=self.num_attention_heads			,			intermediate_size=self.intermediate_size			,			hidden_act=self.hidden_act			,			hidden_dropout_prob=self.hidden_dropout_prob			,			attention_probs_dropout_prob=self.attention_probs_dropout_prob			,			max_position_embeddings=self.max_position_embeddings			,			type_vocab_size=self.type_vocab_size			,			is_decoder=snake_case_			,			initializer_range=self.initializer_range			,			pad_token_id=self.pad_token_id			,			)







						def       __magic_name__							(							self					:      List[str]  ):



								'''simple docstring'''



								snake_case__						,       snake_case__						,       snake_case__						,       snake_case__						:       str       						=  self.prepare_config_and_inputs()

								snake_case__						:       List[str]       						=  True

								return config, input_ids, input_mask, token_labels







						def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      Any			,			snake_case_					:      str			,			snake_case_					:      str  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  GPTNeoXModel(config=snake_case_  )
								model.to(snake_case_  )
								model.eval()
								snake_case__						:       Union[str, Any]       						=  model(snake_case_			,			attention_mask=snake_case_  )
								snake_case__						:       int       						=  model(snake_case_  )
								self.parent.assertEqual(result.last_hidden_state.shape			,			(self.batch_size, self.seq_length, self.hidden_size)  )







						def       __magic_name__							(							self					:      List[Any]			,			snake_case_					:      Any			,			snake_case_					:      List[Any]			,			snake_case_					:      Optional[int]  ):



								'''simple docstring'''



								snake_case__						:       Union[str, Any]       						=  True
								snake_case__						:       Tuple       						=  GPTNeoXModel(snake_case_  )
								model.to(snake_case_  )
								model.eval()
								snake_case__						:       Dict       						=  model(snake_case_			,			attention_mask=snake_case_  )
								self.parent.assertEqual(result.last_hidden_state.shape			,			(self.batch_size, self.seq_length, self.hidden_size)  )







						def       __magic_name__							(							self					:      Any			,			snake_case_					:      str			,			snake_case_					:      Tuple			,			snake_case_					:      Optional[int]			,			snake_case_					:      List[Any]  ):



								'''simple docstring'''



								snake_case__						:       Union[str, Any]       						=  GPTNeoXForCausalLM(config=snake_case_  )
								model.to(snake_case_  )
								model.eval()
								snake_case__						:       Tuple       						=  model(snake_case_			,			attention_mask=snake_case_			,			labels=snake_case_  )
								self.parent.assertEqual(result.logits.shape			,			(self.batch_size, self.seq_length, self.vocab_size)  )







						def       __magic_name__							(							self					:      Optional[int]			,			snake_case_					:      str			,			snake_case_					:      Tuple			,			snake_case_					:      Optional[Any]			,			snake_case_					:      List[str]  ):



								'''simple docstring'''



								snake_case__						:       Dict       						=  self.num_labels
								snake_case__						:       List[Any]       						=  GPTNeoXForQuestionAnswering(snake_case_  )
								model.to(snake_case_  )
								model.eval()
								snake_case__						:       Any       						=  model(snake_case_			,			attention_mask=snake_case_  )
								self.parent.assertEqual(result.start_logits.shape			,			(self.batch_size, self.seq_length)  )
								self.parent.assertEqual(result.end_logits.shape			,			(self.batch_size, self.seq_length)  )







						def       __magic_name__							(							self					:      List[Any]			,			snake_case_					:      List[Any]			,			snake_case_					:      Union[str, Any]			,			snake_case_					:      List[Any]			,			snake_case_					:      Dict  ):



								'''simple docstring'''



								snake_case__						:       str       						=  self.num_labels
								snake_case__						:       List[str]       						=  GPTNeoXForSequenceClassification(snake_case_  )
								model.to(snake_case_  )
								model.eval()
								snake_case__						:       Tuple       						=  ids_tensor([self.batch_size]			,			self.type_sequence_label_size  )
								snake_case__						:       str       						=  model(snake_case_			,			attention_mask=snake_case_			,			labels=snake_case_  )
								self.parent.assertEqual(result.logits.shape			,			(self.batch_size, self.num_labels)  )







						def       __magic_name__							(							self					:      Union[str, Any]			,			snake_case_					:      Optional[Any]			,			snake_case_					:      int			,			snake_case_					:      Dict			,			snake_case_					:      Optional[int]  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  self.num_labels
								snake_case__						:       Any       						=  GPTNeoXForTokenClassification(snake_case_  )
								model.to(snake_case_  )
								model.eval()
								snake_case__						:       Optional[Any]       						=  model(snake_case_			,			attention_mask=snake_case_			,			labels=snake_case_  )
								self.parent.assertEqual(result.logits.shape			,			(self.batch_size, self.seq_length, self.num_labels)  )







						def       __magic_name__							(							self					:      Union[str, Any]			,			snake_case_					:      int			,			snake_case_					:      Union[str, Any]			,			snake_case_					:      Tuple  ):



								'''simple docstring'''



								snake_case__						:       Optional[Any]       						=  True
								snake_case__						:       Union[str, Any]       						=  GPTNeoXForCausalLM(config=snake_case_  )
								model.to(snake_case_  )
								model.eval()

								# first forward pass
								snake_case__						:       int       						=  model(snake_case_			,			attention_mask=snake_case_			,			use_cache=snake_case_  )
								snake_case__						:       Dict       						=  outputs.past_key_values

								# create hypothetical multiple next token and extent to next_input_ids
								snake_case__						:       Tuple       						=  ids_tensor((self.batch_size, 3)			,			config.vocab_size  )
								snake_case__						:       Union[str, Any]       						=  ids_tensor((self.batch_size, 3)			,			vocab_size=2  )

								# append to next input_ids and
								snake_case__						:       Dict       						=  torch.cat([input_ids, next_tokens]			,			dim=-1  )
								snake_case__						:       Optional[int]       						=  torch.cat([input_mask, next_mask]			,			dim=-1  )

								snake_case__						:       Optional[Any]       						=  model(snake_case_			,			attention_mask=snake_case_			,			output_hidden_states=snake_case_  )
								snake_case__						:       Union[str, Any]       						=  output_from_no_past['''hidden_states'''][0]
								snake_case__						:       str       						=  model(
								    snake_case_			,			attention_mask=snake_case_			,			past_key_values=snake_case_			,			output_hidden_states=snake_case_			,			)['''hidden_states'''][0]

								# select random slice
								snake_case__						:       str       						=  ids_tensor((1,)			,			output_from_past.shape[-1]  ).item()
								snake_case__						:       str       						=  output_from_no_past[:, -3:, random_slice_idx].detach()
								snake_case__						:       Union[str, Any]       						=  output_from_past[:, :, random_slice_idx].detach()

								self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]  )

								# test that outputs are equal for slice
								self.parent.assertTrue(torch.allclose(snake_case_			,			snake_case_			,			atol=1e-3  )  )







						def       __magic_name__							(							self					:      Optional[int]  ):



								'''simple docstring'''



								snake_case__						:       Dict       						=  self.prepare_config_and_inputs()
								snake_case__						,       snake_case__						,       snake_case__						,       snake_case__						:       Tuple       						=  config_and_inputs
								snake_case__						:       str       						=  {'''input_ids''': input_ids, '''attention_mask''': input_mask}
								return config, inputs_dict





@require_torch
class     a			(				SCREAMING_SNAKE_CASE			,			SCREAMING_SNAKE_CASE			,			SCREAMING_SNAKE_CASE			,			unittest.TestCase       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    (
						    (
						        GPTNeoXModel,
						        GPTNeoXForCausalLM,
						        GPTNeoXForQuestionAnswering,
						        GPTNeoXForSequenceClassification,
						        GPTNeoXForTokenClassification,
						    )
						    if is_torch_available()
						    else ()
						)
						__UpperCAmelCase			     =    (GPTNeoXForCausalLM,) if is_torch_available() else ()
						__UpperCAmelCase			     =    (
						    {
						        """feature-extraction""": GPTNeoXModel,
						        """question-answering""": GPTNeoXForQuestionAnswering,
						        """text-classification""": GPTNeoXForSequenceClassification,
						        """text-generation""": GPTNeoXForCausalLM,
						        """token-classification""": GPTNeoXForTokenClassification,
						        """zero-shot""": GPTNeoXForSequenceClassification,
						    }
						    if is_torch_available()
						    else {}
						)
						__UpperCAmelCase			     =    False
						__UpperCAmelCase			     =    False
						__UpperCAmelCase			     =    False
						__UpperCAmelCase			     =    False







						def       __magic_name__							(							self					:      int  ):



								'''simple docstring'''



								snake_case__						:       Optional[int]       						=  GPTNeoXModelTester(self  )
								snake_case__						:       List[str]       						=  ConfigTester(self			,			config_class=snake_case_			,			hidden_size=6_4			,			num_attention_heads=8  )







						def       __magic_name__							(							self					:      List[str]  ):



								'''simple docstring'''



								self.config_tester.run_common_tests()







						def       __magic_name__							(							self					:      Dict  ):



								'''simple docstring'''



								snake_case__						,       snake_case__						,       snake_case__						,       snake_case__						:       Union[str, Any]       						=  self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(snake_case_			,			snake_case_			,			snake_case_  )







						def       __magic_name__							(							self					:      Dict  ):



								'''simple docstring'''



								snake_case__						,       snake_case__						,       snake_case__						,       snake_case__						:       List[Any]       						=  self.model_tester.prepare_config_and_inputs_for_decoder()
								self.model_tester.create_and_check_model_as_decoder(snake_case_			,			snake_case_			,			snake_case_  )







						def       __magic_name__							(							self					:      Union[str, Any]  ):



								'''simple docstring'''



								snake_case__						,       snake_case__						,       snake_case__						,       snake_case__						:       Any       						=  self.model_tester.prepare_config_and_inputs_for_decoder()

								snake_case__						:       Optional[int]       						=  None

								self.model_tester.create_and_check_model_as_decoder(snake_case_			,			snake_case_			,			snake_case_  )







						def       __magic_name__							(							self					:      Optional[int]  ):



								'''simple docstring'''



								snake_case__						,       snake_case__						,       snake_case__						,       snake_case__						:       Dict       						=  self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_decoder_model_past_large_inputs(snake_case_			,			snake_case_			,			snake_case_  )







						def       __magic_name__							(							self					:      int  ):



								'''simple docstring'''



								snake_case__						:       List[str]       						=  self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_causal_lm(*snake_case_  )







						def       __magic_name__							(							self					:      List[str]  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_question_answering(*snake_case_  )







						def       __magic_name__							(							self					:      Any  ):



								'''simple docstring'''



								snake_case__						:       Union[str, Any]       						=  self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_sequence_classification(*snake_case_  )







						def       __magic_name__							(							self					:      List[Any]  ):



								'''simple docstring'''



								snake_case__						:       int       						=  self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_token_classification(*snake_case_  )







						@unittest.skip(reason='''Feed forward chunking is not implemented'''  )
						def       __magic_name__							(							self					:      Optional[Any]  ):



								'''simple docstring'''



								pass







						@parameterized.expand([('''linear''',), ('''dynamic''',)]  )
						def       __magic_name__							(							self					:      Optional[int]			,			snake_case_					:      Optional[Any]  ):



								'''simple docstring'''



								snake_case__						,       snake_case__						:       List[Any]       						=  self.model_tester.prepare_config_and_inputs_for_common()
								snake_case__						:       Dict       						=  ids_tensor([1, 1_0]			,			config.vocab_size  )
								snake_case__						:       List[Any]       						=  ids_tensor([1, int(config.max_position_embeddings * 1.5  )]			,			config.vocab_size  )

								set_seed(4_2  )  # Fixed seed at init time so the two models get the same random weights
								snake_case__						:       Tuple       						=  GPTNeoXModel(snake_case_  )
								original_model.to(snake_case_  )
								original_model.eval()
								snake_case__						:       Any       						=  original_model(snake_case_  ).last_hidden_state
								snake_case__						:       List[str]       						=  original_model(snake_case_  ).last_hidden_state

								set_seed(4_2  )  # Fixed seed at init time so the two models get the same random weights
								snake_case__						:       Optional[Any]       						=  {'''type''': scaling_type, '''factor''': 1_0.0}
								snake_case__						:       Optional[Any]       						=  GPTNeoXModel(snake_case_  )
								scaled_model.to(snake_case_  )
								scaled_model.eval()
								snake_case__						:       Optional[int]       						=  scaled_model(snake_case_  ).last_hidden_state
								snake_case__						:       List[str]       						=  scaled_model(snake_case_  ).last_hidden_state

								# Dynamic scaling does not change the RoPE embeddings until it receives an input longer than the original
								# maximum sequence length, so the outputs for the short input should match.
								if scaling_type == "dynamic":
										self.assertTrue(torch.allclose(snake_case_			,			snake_case_			,			atol=1e-5  )  )
								else:
										self.assertFalse(torch.allclose(snake_case_			,			snake_case_			,			atol=1e-5  )  )

								# The output should be different for long inputs
								self.assertFalse(torch.allclose(snake_case_			,			snake_case_			,			atol=1e-5  )  )





@require_torch
class     a			(				unittest.TestCase       ):






						"""simple docstring"""




						@slow
						def       __magic_name__							(							self					:      List[str]  ):



								'''simple docstring'''



								snake_case__						:       Dict       						=  AutoTokenizer.from_pretrained('''EleutherAI/pythia-410m-deduped'''  )
								for checkpointing in [True, False]:
										snake_case__						:       str       						=  GPTNeoXForCausalLM.from_pretrained('''EleutherAI/pythia-410m-deduped'''  )

										if checkpointing:
												model.gradient_checkpointing_enable()
										else:
												model.gradient_checkpointing_disable()
										model.to(snake_case_  )

										snake_case__						:       Tuple       						=  tokenizer('''My favorite food is'''			,			return_tensors='''pt'''  ).to(snake_case_  )
										# The hub repo. is updated on 2023-04-04, resulting in poor outputs.
										# See: https://github.com/huggingface/transformers/pull/24193
										snake_case__						:       List[str]       						=  '''My favorite food is a good old-fashioned, old-fashioned, old-fashioned.\n\nI\'m not sure'''

										snake_case__						:       Optional[int]       						=  model.generate(**snake_case_			,			do_sample=snake_case_			,			max_new_tokens=2_0  )
										snake_case__						:       Tuple       						=  tokenizer.batch_decode(snake_case_  )[0]

										self.assertEqual(snake_case_			,			snake_case_  )



