

'''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__						:       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__						:       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__						:       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 copy
import json
import os
import tempfile

from transformers import is_torch_available

from .test_configuration_utils import config_common_kwargs




class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						def __init__(							self					:      Dict			,			snake_case_					:      str			,			snake_case_					:      Optional[int]=None			,			snake_case_					:      List[Any]=True			,			snake_case_					:      Any=None			,			**snake_case_					:      List[str]  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  parent
								snake_case__						:       List[Any]       						=  config_class
								snake_case__						:       Optional[int]       						=  has_text_modality
								snake_case__						:       str       						=  kwargs
								snake_case__						:       Dict       						=  common_properties







						def       __magic_name__							(							self					:      str  ):



								'''simple docstring'''



								snake_case__						:       Tuple       						=  self.config_class(**self.inputs_dict  )
								snake_case__						:       List[Any]       						=  (
								    ['''hidden_size''', '''num_attention_heads''', '''num_hidden_layers''']
								    if self.common_properties is None
								    else self.common_properties
								)

								# Add common fields for text models
								if self.has_text_modality:
										common_properties.extend(['''vocab_size''']  )

								# Test that config has the common properties as getters
								for prop in common_properties:
										self.parent.assertTrue(hasattr(snake_case_			,			snake_case_  )			,			msg=F"""`{prop}` does not exist"""  )

								# Test that config has the common properties as setter
								for idx, name in enumerate(snake_case_  ):
										try:
												setattr(snake_case_			,			snake_case_			,			snake_case_  )
												self.parent.assertEqual(
												    getattr(snake_case_			,			snake_case_  )			,			snake_case_			,			msg=F"""`{name} value {idx} expected, but was {getattr(snake_case_			,			snake_case_  )}"""  )
										except NotImplementedError:
												# Some models might not be able to implement setters for common_properties
												# In that case, a NotImplementedError is raised
												pass

        # Test if config class can be called with Config(prop_name=..)
								for idx, name in enumerate(snake_case_  ):
										try:
												snake_case__						:       Dict       						=  self.config_class(**{name: idx}  )
												self.parent.assertEqual(
												    getattr(snake_case_			,			snake_case_  )			,			snake_case_			,			msg=F"""`{name} value {idx} expected, but was {getattr(snake_case_			,			snake_case_  )}"""  )
										except NotImplementedError:
												# Some models might not be able to implement setters for common_properties
												# In that case, a NotImplementedError is raised
												pass







						def       __magic_name__							(							self					:      int  ):



								'''simple docstring'''



								snake_case__						:       Tuple       						=  self.config_class(**self.inputs_dict  )
								snake_case__						:       str       						=  json.loads(config.to_json_string()  )
								for key, value in self.inputs_dict.items():
										self.parent.assertEqual(obj[key]			,			snake_case_  )







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



								'''simple docstring'''



								snake_case__						:       Optional[int]       						=  self.config_class(**self.inputs_dict  )

								with tempfile.TemporaryDirectory() as tmpdirname:
										snake_case__						:       List[Any]       						=  os.path.join(snake_case_			,			'''config.json'''  )
										config_first.to_json_file(snake_case_  )
										snake_case__						:       Dict       						=  self.config_class.from_json_file(snake_case_  )

								self.parent.assertEqual(config_second.to_dict()			,			config_first.to_dict()  )







						def       __magic_name__							(							self					:      Dict  ):



								'''simple docstring'''



								snake_case__						:       int       						=  self.config_class(**self.inputs_dict  )

								with tempfile.TemporaryDirectory() as tmpdirname:
										config_first.save_pretrained(snake_case_  )
										snake_case__						:       Optional[Any]       						=  self.config_class.from_pretrained(snake_case_  )

								self.parent.assertEqual(config_second.to_dict()			,			config_first.to_dict()  )







						def       __magic_name__							(							self					:      str  ):



								'''simple docstring'''



								snake_case__						:       int       						=  self.config_class(**self.inputs_dict  )

								snake_case__						:       str       						=  '''test'''
								with tempfile.TemporaryDirectory() as tmpdirname:
										snake_case__						:       List[str]       						=  os.path.join(snake_case_			,			snake_case_  )
										config_first.save_pretrained(snake_case_  )
										snake_case__						:       List[str]       						=  self.config_class.from_pretrained(snake_case_			,			subfolder=snake_case_  )

								self.parent.assertEqual(config_second.to_dict()			,			config_first.to_dict()  )







						def       __magic_name__							(							self					:      Dict  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  self.config_class(**self.inputs_dict			,			num_labels=5  )
								self.parent.assertEqual(len(config.idalabel  )			,			5  )
								self.parent.assertEqual(len(config.labelaid  )			,			5  )

								snake_case__						:       List[Any]       						=  3
								self.parent.assertEqual(len(config.idalabel  )			,			3  )
								self.parent.assertEqual(len(config.labelaid  )			,			3  )







						def       __magic_name__							(							self					:      Optional[int]  ):



								'''simple docstring'''



								if self.config_class.is_composition:
										return
								snake_case__						:       Any       						=  self.config_class()
								self.parent.assertIsNotNone(snake_case_  )







						def       __magic_name__							(							self					:      int  ):



								'''simple docstring'''



								snake_case__						:       Optional[Any]       						=  copy.deepcopy(snake_case_  )
								snake_case__						:       Union[str, Any]       						=  self.config_class(**snake_case_  )
								snake_case__						:       List[str]       						=  []
								for key, value in config_common_kwargs.items():
										if key == "torch_dtype":
												if not is_torch_available():
														continue
												else:
														import torch

														if config.torch_dtype != torch.floataa:
																wrong_values.append(('''torch_dtype''', config.torch_dtype, torch.floataa)  )
										elif getattr(snake_case_			,			snake_case_  ) != value:
												wrong_values.append((key, getattr(snake_case_			,			snake_case_  ), value)  )

								if len(snake_case_  ) > 0:
										snake_case__						:       Dict       						=  '''\n'''.join([F"""- {v[0]}: got {v[1]} instead of {v[2]}""" for v in wrong_values]  )
										raise ValueError(F"""The following keys were not properly set in the config:\n{errors}"""  )







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



								'''simple docstring'''



								self.create_and_test_config_common_properties()
								self.create_and_test_config_to_json_string()
								self.create_and_test_config_to_json_file()
								self.create_and_test_config_from_and_save_pretrained()
								self.create_and_test_config_from_and_save_pretrained_subfolder()
								self.create_and_test_config_with_num_labels()
								self.check_config_can_be_init_without_params()
								self.check_config_arguments_init()



