


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 				UpperCAmelCase					(  UpperCAmelCase					,UpperCAmelCase					,UpperCAmelCase   )->		Optional[Any]:







					'''simple docstring'''



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

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







def 				UpperCAmelCase					(  UpperCAmelCase					,UpperCAmelCase					,UpperCAmelCase					,UpperCAmelCase=0.9_9_9_5   )->		Union[str, Any]:







					'''simple docstring'''



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

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

					if inputs_are_torch:
										SCREAMING_SNAKE_CASE_						   =							torch.from_numpy(_lowerCamelCase   ).to(_lowerCamelCase   )

					return va







def 				UpperCAmelCase					(  UpperCAmelCase					,UpperCAmelCase   )->		Union[str, Any]:







					'''simple docstring'''



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







def 				UpperCAmelCase					(  UpperCAmelCase					,UpperCAmelCase   )->		int:







					'''simple docstring'''



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






class     snake_case      (							__lowerCAmelCase						):
			'''simple docstring'''

			def __init__(						self  :  Optional[int]				,				lowerCAmelCase_  :  Tuple				,				lowerCAmelCase_  :  List[Any]				,				lowerCAmelCase_  :  List[str]				,				lowerCAmelCase_  :  Any				,				lowerCAmelCase_  :  int				,				lowerCAmelCase_  :  List[str]				,				lowerCAmelCase_  :  Optional[int]				,				lowerCAmelCase_  :  Tuple=None				,				lowerCAmelCase_  :  Any=None				,				lowerCAmelCase_  :  str=None				,				)       ->       str:




								"""simple docstring"""






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

			def _lowercase			(						self  :  List[str]				,				lowerCAmelCase_  :  int = "auto"       )       ->       List[Any]:




								"""simple docstring"""






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

			def _lowercase			(						self  :  Optional[Any]       )       ->       str:




								"""simple docstring"""






								self.enable_attention_slicing(__SCREAMING_SNAKE_CASE       )

			def _lowercase			(						self  :  Dict       )       ->       Tuple:




								"""simple docstring"""






								set_requires_grad(self.vae				,				__SCREAMING_SNAKE_CASE       )

			def _lowercase			(						self  :  List[Any]       )       ->       Optional[int]:




								"""simple docstring"""






								set_requires_grad(self.vae				,				__SCREAMING_SNAKE_CASE       )

			def _lowercase			(						self  :  Union[str, Any]       )       ->       str:




								"""simple docstring"""






								set_requires_grad(self.unet				,				__SCREAMING_SNAKE_CASE       )

			def _lowercase			(						self  :  Optional[Any]       )       ->       List[str]:




								"""simple docstring"""






								set_requires_grad(self.unet				,				__SCREAMING_SNAKE_CASE       )

			def _lowercase			(						self  :  Optional[int]				,				lowerCAmelCase_  :  int				,				lowerCAmelCase_  :  List[Any]				,				lowerCAmelCase_  :  Dict       )       ->       Any:




								"""simple docstring"""






								SCREAMING_SNAKE_CASE_						   =							min(int(num_inference_steps * strength       )				,				__SCREAMING_SNAKE_CASE       )

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

								return timesteps, num_inference_steps - t_start

			def _lowercase			(						self  :  List[str]				,				lowerCAmelCase_  :  Union[str, Any]				,				lowerCAmelCase_  :  List[str]				,				lowerCAmelCase_  :  Union[str, Any]				,				lowerCAmelCase_  :  List[str]				,				lowerCAmelCase_  :  Dict				,				lowerCAmelCase_  :  Tuple=None       )       ->       str:




								"""simple docstring"""






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

								SCREAMING_SNAKE_CASE_						   =							image.to(device=__SCREAMING_SNAKE_CASE				,				dtype=__SCREAMING_SNAKE_CASE       )

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

								# Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
								SCREAMING_SNAKE_CASE_						   =							0.18_215 * init_latents
								SCREAMING_SNAKE_CASE_						   =							init_latents.repeat_interleave(__SCREAMING_SNAKE_CASE				,				dim=0       )

								SCREAMING_SNAKE_CASE_						   =							randn_tensor(init_latents.shape				,				generator=__SCREAMING_SNAKE_CASE				,				device=__SCREAMING_SNAKE_CASE				,				dtype=__SCREAMING_SNAKE_CASE       )

								# get latents
								SCREAMING_SNAKE_CASE_						   =							self.scheduler.add_noise(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE       )
								SCREAMING_SNAKE_CASE_						   =							init_latents

								return latents

			def _lowercase			(						self  :  List[str]				,				lowerCAmelCase_  :  Optional[int]       )       ->       str:




								"""simple docstring"""






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

			def _lowercase			(						self  :  Optional[int]				,				lowerCAmelCase_  :  List[str]				,				lowerCAmelCase_  :  Dict       )       ->       int:




								"""simple docstring"""






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

			@torch.enable_grad()
			def _lowercase			(						self  :  List[Any]				,				lowerCAmelCase_  :  Dict				,				lowerCAmelCase_  :  Optional[Any]				,				lowerCAmelCase_  :  Tuple				,				lowerCAmelCase_  :  Tuple				,				lowerCAmelCase_  :  Dict				,				lowerCAmelCase_  :  int				,				lowerCAmelCase_  :  Union[str, Any]				,				)       ->       int:




								"""simple docstring"""






								SCREAMING_SNAKE_CASE_						   =							latents.detach().requires_grad_()

								SCREAMING_SNAKE_CASE_						   =							self.scheduler.scale_model_input(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE       )

								# predict the noise residual
								SCREAMING_SNAKE_CASE_						   =							self.unet(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				encoder_hidden_states=__SCREAMING_SNAKE_CASE       ).sample

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

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

								SCREAMING_SNAKE_CASE_						   =							transforms.Resize(self.feature_extractor_size       )(__SCREAMING_SNAKE_CASE       )
								SCREAMING_SNAKE_CASE_						   =							self.normalize(__SCREAMING_SNAKE_CASE       ).to(latents.dtype       )

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

								SCREAMING_SNAKE_CASE_						   =							spherical_dist_loss(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE       ).mean() * clip_guidance_scale

								SCREAMING_SNAKE_CASE_						   =							-torch.autograd.grad(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE       )[0]

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




			@torch.no_grad()
			def __call__(						self  :  str				,				lowerCAmelCase_  :  Optional[int]				,				lowerCAmelCase_  :  List[Any]				,				lowerCAmelCase_  :  List[str] = None				,				lowerCAmelCase_  :  Union[str, Any] = None				,				lowerCAmelCase_  :  Union[str, Any] = 512				,				lowerCAmelCase_  :  Tuple = 512				,				lowerCAmelCase_  :  List[Any] = 0.6				,				lowerCAmelCase_  :  List[Any] = 50				,				lowerCAmelCase_  :  str = 7.5				,				lowerCAmelCase_  :  Dict = 1				,				lowerCAmelCase_  :  str = 0.0				,				lowerCAmelCase_  :  Dict = 100				,				lowerCAmelCase_  :  Dict = None				,				lowerCAmelCase_  :  List[str] = "pil"				,				lowerCAmelCase_  :  Tuple = True				,				lowerCAmelCase_  :  List[Any] = 0.8				,				lowerCAmelCase_  :  Tuple = 0.1				,				lowerCAmelCase_  :  Optional[int] = 0.1				,				)       ->       Any:




								"""simple docstring"""






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

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

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

								SCREAMING_SNAKE_CASE_						   =							self.tokenizer(
								    __SCREAMING_SNAKE_CASE				,				padding='''max_length'''				,				max_length=self.tokenizer.model_max_length				,				truncation=__SCREAMING_SNAKE_CASE				,				return_tensors='''pt'''				,				)
								SCREAMING_SNAKE_CASE_						   =							self.text_encoder(style_text_input.input_ids.to(self.device       )       )[0]

								SCREAMING_SNAKE_CASE_						   =							slerp(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE       )

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

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

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

								SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_						   =							self.get_timesteps(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				self.device       )
								SCREAMING_SNAKE_CASE_						   =							timesteps[:1].repeat(__SCREAMING_SNAKE_CASE       )

								# Preprocess image
								SCREAMING_SNAKE_CASE_						   =							preprocess(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE       )
								SCREAMING_SNAKE_CASE_						   =							self.prepare_latents(
								    __SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				text_embeddings.dtype				,				self.device				,				__SCREAMING_SNAKE_CASE       )

								SCREAMING_SNAKE_CASE_						   =							preprocess(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE       )
								SCREAMING_SNAKE_CASE_						   =							self.prepare_latents(
								    __SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				text_embeddings.dtype				,				self.device				,				__SCREAMING_SNAKE_CASE       )

								SCREAMING_SNAKE_CASE_						   =							slerp(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE       )

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

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

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

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

																		# predict the noise residual
																		SCREAMING_SNAKE_CASE_						   =							self.unet(__SCREAMING_SNAKE_CASE				,				__SCREAMING_SNAKE_CASE				,				encoder_hidden_states=__SCREAMING_SNAKE_CASE       ).sample

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

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

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

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
								SCREAMING_SNAKE_CASE_						   =							1 / 0.18_215 * latents
								SCREAMING_SNAKE_CASE_						   =							self.vae.decode(__SCREAMING_SNAKE_CASE       ).sample

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

								if output_type == "pil":
													SCREAMING_SNAKE_CASE_						   =							self.numpy_to_pil(__SCREAMING_SNAKE_CASE       )

								if not return_dict:
													return (image, None)

								return StableDiffusionPipelineOutput(images=__SCREAMING_SNAKE_CASE				,				nsfw_content_detected=__SCREAMING_SNAKE_CASE       )








'''simple docstring'''



import re
import time
from typing import Optional

import IPython.display as disp

from ..trainer_callback import TrainerCallback
from ..trainer_utils import IntervalStrategy, has_length







def       _UpperCamelCase			(_lowerCamelCase				:  Tuple						)-> List[str]:




     '''simple docstring'''





     __snake_case            =	int(_lowerCamelCase						)
     __snake_case       ,     __snake_case       ,     __snake_case            =	t // 36_00, (t // 60) % 60, t % 60
     return f'''{h}:{m:02d}:{s:02d}''' if h != 0 else f'''{m:02d}:{s:02d}'''







def       _UpperCamelCase			(_lowerCamelCase				:  Any ,						_lowerCamelCase				:  Union[str, Any] ,						_lowerCamelCase				:  str ,						_lowerCamelCase				:  int ,						_lowerCamelCase				:  List[Any]=3_00						)-> int:




     '''simple docstring'''





     return f'''
    <div>
      {prefix}
      <progress value=\'{value}\' max=\'{total}\' style=\'width:{width}px; height:20px; vertical-align: middle;\'></progress>
      {label}
    </div>
    '''







def       _UpperCamelCase			(_lowerCamelCase				:  int						)-> List[Any]:




     '''simple docstring'''





     __snake_case            =	'''<table border="1" class="dataframe">\n'''
     html_code += """  <thead>\n <tr style="text-align: left;">\n"""
     for i in items[0]:
          html_code += f'''      <th>{i}</th>\n'''
     html_code += "    </tr>\n  </thead>\n  <tbody>\n"
     for line in items[1:]:
          html_code += "    <tr>\n"
          for elt in line:
               __snake_case            =	f'''{elt:.6f}''' if isinstance(_lowerCamelCase ,						_lowerCamelCase						) else str(_lowerCamelCase						)
               html_code += f'''      <td>{elt}</td>\n'''
          html_code += "    </tr>\n"
     html_code += "  </tbody>\n</table><p>"
     return html_code





class      lowerCAmelCase :
       __lowercase	:   str										=      5
       __lowercase	:   Optional[Any]										=      0.2


       def __init__( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE = None     ,	__SCREAMING_SNAKE_CASE = True     ,	__SCREAMING_SNAKE_CASE = None     ,	__SCREAMING_SNAKE_CASE = 300     ,	)   ->	List[Any]:



            '''simple docstring'''




            __snake_case            =	total
            __snake_case            =	'''''' if prefix is None else prefix
            __snake_case            =	leave
            __snake_case            =	parent
            __snake_case            =	width
            __snake_case            =	None
            __snake_case            =	None
            __snake_case            =	None


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE = False     ,	__SCREAMING_SNAKE_CASE = None	)   ->	Any:



            '''simple docstring'''




            __snake_case            =	value
            if comment is not None:
                 __snake_case            =	comment
            if self.last_value is None:
                 __snake_case            =	__snake_case            =	time.time()
                 __snake_case            =	__snake_case            =	value
                 __snake_case            =	__snake_case            =	None
                 __snake_case            =	self.warmup
                 __snake_case            =	1
                 self.update_bar(__SCREAMING_SNAKE_CASE	)
            elif value <= self.last_value and not force_update:
                 return
            elif force_update or self.first_calls > 0 or value >= min(self.last_value + self.wait_for     ,	self.total	):
                 if self.first_calls > 0:
                      self.first_calls -= 1
                 __snake_case            =	time.time()
                 __snake_case            =	current_time - self.start_time
                 # We could have value = self.start_value if the update is called twixe with the same start value.
                 if value > self.start_value:
                      __snake_case            =	self.elapsed_time / (value - self.start_value)
                 else:
                      __snake_case            =	None
                 if value >= self.total:
                      __snake_case            =	self.total
                      __snake_case            =	None
                      if not self.leave:
                           self.close()
                 elif self.average_time_per_item is not None:
                      __snake_case            =	self.average_time_per_item * (self.total - value)
                 self.update_bar(__SCREAMING_SNAKE_CASE	)
                 __snake_case            =	value
                 __snake_case            =	current_time
                 if self.average_time_per_item is None:
                      __snake_case            =	1
                 else:
                      __snake_case            =	max(int(self.update_every / self.average_time_per_item	)     ,	1	)


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=None	)   ->	List[str]:



            '''simple docstring'''




            __snake_case            =	''' ''' * (len(str(self.total	)	) - len(str(__SCREAMING_SNAKE_CASE	)	)) + str(__SCREAMING_SNAKE_CASE	)
            if self.elapsed_time is None:
                 __snake_case            =	F'''[{spaced_value}/{self.total} : < :'''
            elif self.predicted_remaining is None:
                 __snake_case            =	F'''[{spaced_value}/{self.total} {format_time(self.elapsed_time	)}'''
            else:
                 __snake_case            =	(
                     F'''[{spaced_value}/{self.total} {format_time(self.elapsed_time	)} <'''
                     F''' {format_time(self.predicted_remaining	)}'''
                 )
                 self.label += F''', {1/self.average_time_per_item:.2f} it/s'''
            self.label += "]" if self.comment is None or len(self.comment	) == 0 else F''', {self.comment}]'''
            self.display()


       def    lowerCAmelCase   ( self	)   ->	List[Any]:



            '''simple docstring'''




            __snake_case            =	html_progress_bar(self.value     ,	self.total     ,	self.prefix     ,	self.label     ,	self.width	)
            if self.parent is not None:
                 # If this is a child bar, the parent will take care of the display.
                 self.parent.display()
                 return
            if self.output is None:
                 __snake_case            =	disp.display(disp.HTML(self.html_code	)     ,	display_id=__SCREAMING_SNAKE_CASE	)
            else:
                 self.output.update(disp.HTML(self.html_code	)	)




       def    lowerCAmelCase   ( self	)   ->	Dict:



            '''simple docstring'''




            if self.parent is None and self.output is not None:
                 self.output.update(disp.HTML(''''''	)	)





class      lowerCAmelCase (			__lowerCAmelCase):


       def __init__( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=None	)   ->	Any:



            '''simple docstring'''




            super().__init__(__SCREAMING_SNAKE_CASE	)
            __snake_case            =	None if column_names is None else [column_names]
            __snake_case            =	None


       def    lowerCAmelCase   ( self	)   ->	Any:



            '''simple docstring'''




            __snake_case            =	html_progress_bar(self.value     ,	self.total     ,	self.prefix     ,	self.label     ,	self.width	)
            if self.inner_table is not None:
                 self.html_code += text_to_html_table(self.inner_table	)
            if self.child_bar is not None:
                 self.html_code += self.child_bar.html_code
            if self.output is None:
                 __snake_case            =	disp.display(disp.HTML(self.html_code	)     ,	display_id=__SCREAMING_SNAKE_CASE	)
            else:
                 self.output.update(disp.HTML(self.html_code	)	)


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE	)   ->	Dict:



            '''simple docstring'''




            if self.inner_table is None:
                 __snake_case            =	[list(values.keys()	), list(values.values()	)]
            else:
                 __snake_case            =	self.inner_table[0]
                 if len(self.inner_table	) == 1:
                      # We give a chance to update the column names at the first iteration
                      for key in values.keys():
                           if key not in columns:
                                columns.append(__SCREAMING_SNAKE_CASE	)
                      __snake_case            =	columns
                 self.inner_table.append([values[c] for c in columns]	)


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=None     ,	__SCREAMING_SNAKE_CASE=300	)   ->	List[str]:



            '''simple docstring'''




            __snake_case            =	NotebookProgressBar(__SCREAMING_SNAKE_CASE     ,	prefix=__SCREAMING_SNAKE_CASE     ,	parent=self     ,	width=__SCREAMING_SNAKE_CASE	)
            return self.child_bar




       def    lowerCAmelCase   ( self	)   ->	Optional[int]:



            '''simple docstring'''




            __snake_case            =	None
            self.display()





class      lowerCAmelCase (			__lowerCAmelCase):


       def __init__( self	)   ->	str:



            '''simple docstring'''




            __snake_case            =	None
            __snake_case            =	None
            __snake_case            =	False


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	**__SCREAMING_SNAKE_CASE	)   ->	Optional[int]:



            '''simple docstring'''




            __snake_case            =	'''Epoch''' if args.evaluation_strategy == IntervalStrategy.EPOCH else '''Step'''
            __snake_case            =	0
            __snake_case            =	0
            __snake_case            =	[self.first_column] + ['''Training Loss''']
            if args.evaluation_strategy != IntervalStrategy.NO:
                 column_names.append('''Validation Loss'''	)
            __snake_case            =	NotebookTrainingTracker(state.max_steps     ,	__SCREAMING_SNAKE_CASE	)


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	**__SCREAMING_SNAKE_CASE	)   ->	List[Any]:



            '''simple docstring'''




            __snake_case            =	int(state.epoch	) if int(state.epoch	) == state.epoch else F'''{state.epoch:.2f}'''
            self.training_tracker.update(
                state.global_step + 1     ,	comment=F'''Epoch {epoch}/{state.num_train_epochs}'''     ,	force_update=self._force_next_update     ,	)
            __snake_case            =	False


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=None     ,	**__SCREAMING_SNAKE_CASE	)   ->	Optional[int]:



            '''simple docstring'''




            if not has_length(__SCREAMING_SNAKE_CASE	):
                 return
            if self.prediction_bar is None:
                 if self.training_tracker is not None:
                      __snake_case            =	self.training_tracker.add_child(len(__SCREAMING_SNAKE_CASE	)	)
                 else:
                      __snake_case            =	NotebookProgressBar(len(__SCREAMING_SNAKE_CASE	)	)
                 self.prediction_bar.update(1	)
            else:
                 self.prediction_bar.update(self.prediction_bar.value + 1	)


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	**__SCREAMING_SNAKE_CASE	)   ->	Optional[int]:



            '''simple docstring'''




            if self.prediction_bar is not None:
                 self.prediction_bar.close()
            __snake_case            =	None


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=None     ,	**__SCREAMING_SNAKE_CASE	)   ->	List[Any]:



            '''simple docstring'''




            if args.evaluation_strategy == IntervalStrategy.NO and "loss" in logs:
                 __snake_case            =	{'''Training Loss''': logs['''loss''']}
                 # First column is necessarily Step sine we're not in epoch eval strategy
                 __snake_case            =	state.global_step
                 self.training_tracker.write_line(__SCREAMING_SNAKE_CASE	)


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=None     ,	**__SCREAMING_SNAKE_CASE	)   ->	Optional[Any]:



            '''simple docstring'''




            if self.training_tracker is not None:
                 __snake_case            =	{'''Training Loss''': '''No log''', '''Validation Loss''': '''No log'''}
                 for log in reversed(state.log_history	):
                      if "loss" in log:
                           __snake_case            =	log['''loss''']
                           break

                 if self.first_column == "Epoch":
                      __snake_case            =	int(state.epoch	)
                 else:
                      __snake_case            =	state.global_step
                 __snake_case            =	'''eval'''
                 for k in metrics:
                      if k.endswith('''_loss'''	):
                           __snake_case            =	re.sub(r'''\_loss$'''     ,	''''''     ,	__SCREAMING_SNAKE_CASE	)
                 __snake_case            =	metrics.pop('''total_flos'''     ,	__SCREAMING_SNAKE_CASE	)
                 __snake_case            =	metrics.pop('''epoch'''     ,	__SCREAMING_SNAKE_CASE	)
                 __snake_case            =	metrics.pop(F'''{metric_key_prefix}_runtime'''     ,	__SCREAMING_SNAKE_CASE	)
                 __snake_case            =	metrics.pop(F'''{metric_key_prefix}_samples_per_second'''     ,	__SCREAMING_SNAKE_CASE	)
                 __snake_case            =	metrics.pop(F'''{metric_key_prefix}_steps_per_second'''     ,	__SCREAMING_SNAKE_CASE	)
                 __snake_case            =	metrics.pop(F'''{metric_key_prefix}_jit_compilation_time'''     ,	__SCREAMING_SNAKE_CASE	)
                 for k, v in metrics.items():
                      if k == F'''{metric_key_prefix}_loss''':
                           __snake_case            =	v
                      else:
                           __snake_case            =	k.split('''_'''	)
                           __snake_case            =	''' '''.join([part.capitalize() for part in splits[1:]]	)
                           __snake_case            =	v
                 self.training_tracker.write_line(__SCREAMING_SNAKE_CASE	)
                 self.training_tracker.remove_child()
                 __snake_case            =	None
                 # Evaluation takes a long time so we should force the next update.
                 __snake_case            =	True




       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE     ,	**__SCREAMING_SNAKE_CASE	)   ->	Optional[int]:



            '''simple docstring'''




            self.training_tracker.update(
                state.global_step     ,	comment=F'''Epoch {int(state.epoch	)}/{state.num_train_epochs}'''     ,	force_update=__SCREAMING_SNAKE_CASE	)
            __snake_case            =	None






