






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      lowercase__      (     __snake_case    :							str     ,       __snake_case    :							List[Any]     ,       __snake_case    :							Dict  ):


					'''simple docstring'''


					if isinstance(__snake_case     ,       torch.Tensor  ):
										return image
					elif isinstance(__snake_case     ,       PIL.Image.Image  ):
										UpperCAmelCase_    :      Optional[int]            =						[image]

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




def      lowercase__      (     __snake_case    :							List[Any]     ,       __snake_case    :							int     ,       __snake_case    :							Union[str, Any]     ,       __snake_case    :							Optional[Any]=0.9995  ):


					'''simple docstring'''


					if not isinstance(__snake_case     ,       np.ndarray  ):
										UpperCAmelCase_    :      Optional[Any]            =						True
										UpperCAmelCase_    :      Dict            =						va.device
										UpperCAmelCase_    :      Any            =						va.cpu().numpy()
										UpperCAmelCase_    :      Dict            =						va.cpu().numpy()

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

					if inputs_are_torch:
										UpperCAmelCase_    :      Union[str, Any]            =						torch.from_numpy(__snake_case  ).to(__snake_case  )

					return va




def      lowercase__      (     __snake_case    :							int     ,       __snake_case    :							List[Any]  ):


					'''simple docstring'''


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




def      lowercase__      (     __snake_case    :							Dict     ,       __snake_case    :							Tuple  ):


					'''simple docstring'''


					for param in model.parameters():
										UpperCAmelCase_    :      Optional[Any]            =						value






class   lowerCamelCase      (_snake_case			):







		'''simple docstring'''



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



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase = "auto"   )	->							Dict:
							if slice_size == "auto":
												# half the attention head size is usually a good trade-off between
												# speed and memory
												UpperCAmelCase_    :      Optional[Any]            =						self.unet.config.attention_head_dim // 2
							self.unet.set_attention_slice(_UpperCamelCase   )



		def 					__UpperCAmelCase			(							self   )	->							List[Any]:
							self.enable_attention_slicing(_UpperCamelCase   )



		def 					__UpperCAmelCase			(							self   )	->							int:
							set_requires_grad(self.vae				,		_UpperCamelCase   )



		def 					__UpperCAmelCase			(							self   )	->							Optional[Any]:
							set_requires_grad(self.vae				,		_UpperCamelCase   )



		def 					__UpperCAmelCase			(							self   )	->							Dict:
							set_requires_grad(self.unet				,		_UpperCamelCase   )



		def 					__UpperCAmelCase			(							self   )	->							List[Any]:
							set_requires_grad(self.unet				,		_UpperCamelCase   )



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase   )	->							Tuple:
							# get the original timestep using init_timestep
							UpperCAmelCase_    :      Union[str, Any]            =						min(int(num_inference_steps * strength   )				,		_UpperCamelCase   )

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

							return timesteps, num_inference_steps - t_start



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase=None   )	->							Optional[int]:
							if not isinstance(_UpperCamelCase				,		torch.Tensor   ):
												raise ValueError(f"`image` has to be of type `torch.Tensor` but is {type(_UpperCamelCase   )}"   )

							UpperCAmelCase_    :      Union[str, Any]            =						image.to(device=_UpperCamelCase				,		dtype=_UpperCamelCase   )

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

							# Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
							UpperCAmelCase_    :      List[Any]            =						0.1_82_15 * init_latents
							UpperCAmelCase_    :      str            =						init_latents.repeat_interleave(_UpperCamelCase				,		dim=0   )

							UpperCAmelCase_    :      Dict            =						randn_tensor(init_latents.shape				,		generator=_UpperCamelCase				,		device=_UpperCamelCase				,		dtype=_UpperCamelCase   )

							# get latents
							UpperCAmelCase_    :      List[str]            =						self.scheduler.add_noise(_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase   )
							UpperCAmelCase_    :      Optional[int]            =						init_latents

							return latents



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



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		_UpperCamelCase   )	->							int:
							UpperCAmelCase_    :      str            =						self.feature_extractor.preprocess(_UpperCamelCase   )
							UpperCAmelCase_    :      List[Any]            =						torch.from_numpy(clip_image_input['pixel_values'][0]   ).unsqueeze(0   ).to(self.device   ).half()
							UpperCAmelCase_    :      Tuple            =						self.clip_model.get_image_features(_UpperCamelCase   )
							UpperCAmelCase_    :      Optional[int]            =						image_embeddings_clip / image_embeddings_clip.norm(p=2				,		dim=-1				,		keepdim=_UpperCamelCase   )
							UpperCAmelCase_    :      Optional[Any]            =						image_embeddings_clip.repeat_interleave(_UpperCamelCase				,		dim=0   )
							return image_embeddings_clip



		@torch.enable_grad()
		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		)	->							Tuple:
							UpperCAmelCase_    :      List[str]            =						latents.detach().requires_grad_()

							UpperCAmelCase_    :      str            =						self.scheduler.scale_model_input(_UpperCamelCase				,		_UpperCamelCase   )

							# predict the noise residual
							UpperCAmelCase_    :      Any            =						self.unet(_UpperCamelCase				,		_UpperCamelCase				,		encoder_hidden_states=_UpperCamelCase   ).sample

							if isinstance(self.scheduler				,		(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)   ):
												UpperCAmelCase_    :      int            =						self.scheduler.alphas_cumprod[timestep]
												UpperCAmelCase_    :      int            =						1 - alpha_prod_t
												# compute predicted original sample from predicted noise also called
												# "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
												UpperCAmelCase_    :      List[Any]            =						(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

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

							UpperCAmelCase_    :      Dict            =						transforms.Resize(self.feature_extractor_size   )(_UpperCamelCase   )
							UpperCAmelCase_    :      Any            =						self.normalize(_UpperCamelCase   ).to(latents.dtype   )

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

							UpperCAmelCase_    :      int            =						spherical_dist_loss(_UpperCamelCase				,		_UpperCamelCase   ).mean() * clip_guidance_scale

							UpperCAmelCase_    :      List[Any]            =						-torch.autograd.grad(_UpperCamelCase				,		_UpperCamelCase   )[0]

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



		@torch.no_grad()
		def __call__(							self				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		_UpperCamelCase = 5_1_2				,		_UpperCamelCase = 5_1_2				,		_UpperCamelCase = 0.6				,		_UpperCamelCase = 5_0				,		_UpperCamelCase = 7.5				,		_UpperCamelCase = 1				,		_UpperCamelCase = 0.0				,		_UpperCamelCase = 1_0_0				,		_UpperCamelCase = None				,		_UpperCamelCase = "pil"				,		_UpperCamelCase = True				,		_UpperCamelCase = 0.8				,		_UpperCamelCase = 0.1				,		_UpperCamelCase = 0.1				,		)	->							Optional[int]:
							if isinstance(_UpperCamelCase				,		_UpperCamelCase   ) and len(_UpperCamelCase   ) != batch_size:
												raise ValueError(f"You have passed {batch_size} batch_size, but only {len(_UpperCamelCase   )} 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(_UpperCamelCase				,		torch.Generator   ) and batch_size > 1:
												UpperCAmelCase_    :      str            =						[generator] + [None] * (batch_size - 1)

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

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

							UpperCAmelCase_    :      Any            =						self.tokenizer(
							    _UpperCamelCase				,		padding='max_length'				,		max_length=self.tokenizer.model_max_length				,		truncation=_UpperCamelCase				,		return_tensors='pt'				,		)
							UpperCAmelCase_    :      Union[str, Any]            =						self.text_encoder(style_text_input.input_ids.to(self.device   )   )[0]

							UpperCAmelCase_    :      str            =						slerp(_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase   )

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

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

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

							UpperCAmelCase_       ,  UpperCAmelCase_    :      Optional[int]            =						self.get_timesteps(_UpperCamelCase				,		_UpperCamelCase				,		self.device   )
							UpperCAmelCase_    :      List[str]            =						timesteps[:1].repeat(_UpperCamelCase   )

							# Preprocess image
							UpperCAmelCase_    :      Optional[int]            =						preprocess(_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase   )
							UpperCAmelCase_    :      Any            =						self.prepare_latents(
							    _UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		text_embeddings.dtype				,		self.device				,		_UpperCamelCase   )

							UpperCAmelCase_    :      Union[str, Any]            =						preprocess(_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase   )
							UpperCAmelCase_    :      List[str]            =						self.prepare_latents(
							    _UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		text_embeddings.dtype				,		self.device				,		_UpperCamelCase   )

							UpperCAmelCase_    :      List[str]            =						slerp(_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase   )

							if clip_guidance_scale > 0:
												UpperCAmelCase_    :      List[Any]            =						self.get_clip_image_embeddings(_UpperCamelCase				,		_UpperCamelCase   )
												UpperCAmelCase_    :      List[Any]            =						self.get_clip_image_embeddings(_UpperCamelCase				,		_UpperCamelCase   )
												UpperCAmelCase_    :      str            =						slerp(
												    _UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase   )

							# 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.
							UpperCAmelCase_    :      int            =						guidance_scale > 1.0
							# get unconditional embeddings for classifier free guidance
							if do_classifier_free_guidance:
												UpperCAmelCase_    :      str            =						content_text_input.input_ids.shape[-1]
												UpperCAmelCase_    :      Tuple            =						self.tokenizer(['']				,		padding='max_length'				,		max_length=_UpperCamelCase				,		return_tensors='pt'   )
												UpperCAmelCase_    :      Dict            =						self.text_encoder(uncond_input.input_ids.to(self.device   )   )[0]
												# duplicate unconditional embeddings for each generation per prompt
												UpperCAmelCase_    :      Tuple            =						uncond_embeddings.repeat_interleave(_UpperCamelCase				,		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
												UpperCAmelCase_    :      Tuple            =						torch.cat([uncond_embeddings, text_embeddings]   )

							# get the initial random noise unless the user supplied it

							# Unlike in other pipelines, latents need to be generated in the target device
							# for 1-to-1 results reproducibility with the CompVis implementation.
							# However this currently doesn't work in `mps`.
							UpperCAmelCase_    :      List[Any]            =						(batch_size, self.unet.config.in_channels, height // 8, width // 8)
							UpperCAmelCase_    :      Tuple            =						text_embeddings.dtype
							if latents is None:
												if self.device.type == "mps":
																	# randn does not work reproducibly on mps
																	UpperCAmelCase_    :      List[Any]            =						torch.randn(_UpperCamelCase				,		generator=_UpperCamelCase				,		device='cpu'				,		dtype=_UpperCamelCase   ).to(
																	    self.device   )
												else:
																	UpperCAmelCase_    :      str            =						torch.randn(_UpperCamelCase				,		generator=_UpperCamelCase				,		device=self.device				,		dtype=_UpperCamelCase   )
							else:
												if latents.shape != latents_shape:
																	raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}"   )
												UpperCAmelCase_    :      Any            =						latents.to(self.device   )

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

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

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

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

																	# perform classifier free guidance
																	if do_classifier_free_guidance:
																						UpperCAmelCase_       ,  UpperCAmelCase_    :      Optional[int]            =						noise_pred.chunk(2   )
																						UpperCAmelCase_    :      str            =						noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

																	# perform clip guidance
																	if clip_guidance_scale > 0:
																						UpperCAmelCase_    :      Optional[Any]            =						(
																						    text_embeddings.chunk(2   )[1] if do_classifier_free_guidance else text_embeddings
																						)
																						UpperCAmelCase_       ,  UpperCAmelCase_    :      Union[str, Any]            =						self.cond_fn(
																						    _UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		)

																	# compute the previous noisy sample x_t -> x_t-1
																	UpperCAmelCase_    :      Optional[Any]            =						self.scheduler.step(_UpperCamelCase				,		_UpperCamelCase				,		_UpperCamelCase				,		**_UpperCamelCase   ).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
							UpperCAmelCase_    :      Optional[int]            =						1 / 0.1_82_15 * latents
							UpperCAmelCase_    :      List[Any]            =						self.vae.decode(_UpperCamelCase   ).sample

							UpperCAmelCase_    :      str            =						(image / 2 + 0.5).clamp(0				,		1   )
							UpperCAmelCase_    :      Any            =						image.cpu().permute(0				,		2				,		3				,		1   ).numpy()

							if output_type == "pil":
												UpperCAmelCase_    :      List[str]            =						self.numpy_to_pil(_UpperCamelCase   )

							if not return_dict:
												return (image, None)

							return StableDiffusionPipelineOutput(images=_UpperCamelCase				,		nsfw_content_detected=_UpperCamelCase   )














import os
import shutil
from pathlib import Path
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ..utils import ONNX_EXTERNAL_WEIGHTS_NAME, ONNX_WEIGHTS_NAME, is_onnx_available, logging


if is_onnx_available():
		import onnxruntime as ort


__UpperCAmelCase								=						logging.get_logger(__name__)

__UpperCAmelCase								=						{
    'tensor(bool)': np.bool_,
    'tensor(int8)': np.inta,
    'tensor(uint8)': np.uinta,
    'tensor(int16)': np.intaa,
    'tensor(uint16)': np.uintaa,
    'tensor(int32)': np.intaa,
    'tensor(uint32)': np.uintaa,
    'tensor(int64)': np.intaa,
    'tensor(uint64)': np.uintaa,
    'tensor(float16)': np.floataa,
    'tensor(float)': np.floataa,
    'tensor(double)': np.floataa,
}



class   lowerCamelCase      :







		'''simple docstring'''



		def __init__(							self				,		_UpperCamelCase=None				,		**_UpperCamelCase   )	->							Dict:
							logger.info('`diffusers.OnnxRuntimeModel` is experimental and might change in the future.'   )
							UpperCAmelCase_    :      Any            =						model
							UpperCAmelCase_    :      int            =						kwargs.get('model_save_dir'				,		_UpperCamelCase   )
							UpperCAmelCase_    :      List[Any]            =						kwargs.get('latest_model_name'				,		_UpperCamelCase   )



		def __call__(							self				,		**_UpperCamelCase   )	->							str:
							UpperCAmelCase_    :      Optional[int]            =						{k: np.array(_UpperCamelCase   ) for k, v in kwargs.items()}
							return self.model.run(_UpperCamelCase				,		_UpperCamelCase   )



		@staticmethod
		def 					__UpperCAmelCase			(							_UpperCamelCase				,		_UpperCamelCase=None				,		_UpperCamelCase=None   )	->							List[Any]:
							if provider is None:
												logger.info('No onnxruntime provider specified, using CPUExecutionProvider'   )
												UpperCAmelCase_    :      List[str]            =						'CPUExecutionProvider'

							return ort.InferenceSession(_UpperCamelCase				,		providers=[provider]				,		sess_options=_UpperCamelCase   )



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		_UpperCamelCase = None				,		**_UpperCamelCase   )	->							Dict:
							UpperCAmelCase_    :      Any            =						file_name if file_name is not None else ONNX_WEIGHTS_NAME

							UpperCAmelCase_    :      Optional[Any]            =						self.model_save_dir.joinpath(self.latest_model_name   )
							UpperCAmelCase_    :      str            =						Path(_UpperCamelCase   ).joinpath(_UpperCamelCase   )
							try:
												shutil.copyfile(_UpperCamelCase				,		_UpperCamelCase   )
							except shutil.SameFileError:
												pass

							# copy external weights (for models >2GB)
							UpperCAmelCase_    :      Optional[Any]            =						self.model_save_dir.joinpath(_UpperCamelCase   )
							if src_path.exists():
												UpperCAmelCase_    :      List[Any]            =						Path(_UpperCamelCase   ).joinpath(_UpperCamelCase   )
												try:
																	shutil.copyfile(_UpperCamelCase				,		_UpperCamelCase   )
												except shutil.SameFileError:
																	pass



		def 					__UpperCAmelCase			(							self				,		_UpperCamelCase				,		**_UpperCamelCase				,		)	->							List[str]:
							if os.path.isfile(_UpperCamelCase   ):
												logger.error(f"Provided path ({save_directory}) should be a directory, not a file"   )
												return

							os.makedirs(_UpperCamelCase				,		exist_ok=_UpperCamelCase   )

							# saving model weights/files
							self._save_pretrained(_UpperCamelCase				,		**_UpperCamelCase   )



		@classmethod
		def 					__UpperCAmelCase			(							cls				,		_UpperCamelCase				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		_UpperCamelCase = False				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		**_UpperCamelCase				,		)	->							List[str]:
							UpperCAmelCase_    :      List[str]            =						file_name if file_name is not None else ONNX_WEIGHTS_NAME
							# load model from local directory
							if os.path.isdir(_UpperCamelCase   ):
												UpperCAmelCase_    :      Union[str, Any]            =						OnnxRuntimeModel.load_model(
												    os.path.join(_UpperCamelCase				,		_UpperCamelCase   )				,		provider=_UpperCamelCase				,		sess_options=_UpperCamelCase   )
												UpperCAmelCase_    :      Tuple            =						Path(_UpperCamelCase   )
							# load model from hub
							else:
												# download model
												UpperCAmelCase_    :      List[str]            =						hf_hub_download(
												    repo_id=_UpperCamelCase				,		filename=_UpperCamelCase				,		use_auth_token=_UpperCamelCase				,		revision=_UpperCamelCase				,		cache_dir=_UpperCamelCase				,		force_download=_UpperCamelCase				,		)
												UpperCAmelCase_    :      Union[str, Any]            =						Path(_UpperCamelCase   ).parent
												UpperCAmelCase_    :      List[str]            =						Path(_UpperCamelCase   ).name
												UpperCAmelCase_    :      Union[str, Any]            =						OnnxRuntimeModel.load_model(_UpperCamelCase				,		provider=_UpperCamelCase				,		sess_options=_UpperCamelCase   )
							return cls(model=_UpperCamelCase				,		**_UpperCamelCase   )



		@classmethod
		def 					__UpperCAmelCase			(							cls				,		_UpperCamelCase				,		_UpperCamelCase = True				,		_UpperCamelCase = None				,		_UpperCamelCase = None				,		**_UpperCamelCase				,		)	->							Optional[int]:
							UpperCAmelCase_    :      List[str]            =						None
							if len(str(_UpperCamelCase   ).split('@'   )   ) == 2:
												UpperCAmelCase_       ,  UpperCAmelCase_    :      Tuple            =						model_id.split('@'   )

							return cls._from_pretrained(
							    model_id=_UpperCamelCase				,		revision=_UpperCamelCase				,		cache_dir=_UpperCamelCase				,		force_download=_UpperCamelCase				,		use_auth_token=_UpperCamelCase				,		**_UpperCamelCase				,		)






