





"""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 	__snake_case  (    SCREAMING_SNAKE_CASE:					List[Any]     ,  SCREAMING_SNAKE_CASE:					Any     ,  SCREAMING_SNAKE_CASE:					Optional[int]						):






							"""simple docstring"""



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

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






def 	__snake_case  (    SCREAMING_SNAKE_CASE:					int     ,  SCREAMING_SNAKE_CASE:					Optional[Any]     ,  SCREAMING_SNAKE_CASE:					Tuple     ,  SCREAMING_SNAKE_CASE:					Union[str, Any]=0.99_95						):






							"""simple docstring"""



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

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

							if inputs_are_torch:
														_lowerCAmelCase  	=			torch.from_numpy(SCREAMING_SNAKE_CASE						).to(SCREAMING_SNAKE_CASE						)

							return va






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






							"""simple docstring"""



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






def 	__snake_case  (    SCREAMING_SNAKE_CASE:					List[str]     ,  SCREAMING_SNAKE_CASE:					int						):






							"""simple docstring"""



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

class        _SCREAMING_SNAKE_CASE  (  UpperCAmelCase							):






				'''simple docstring'''



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







											"""simple docstring"""




											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_				,     )
											_lowerCAmelCase  	=			(
											    feature_extractor.size
											    if isinstance(feature_extractor.size				,     UpperCAmelCase_  )
											    else feature_extractor.size['shortest_edge']
											)
											_lowerCAmelCase  	=			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 				__lowerCamelCase	(					self	:			Optional[int]				,     UpperCAmelCase_	:			Optional[Union[str, 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
																		_lowerCAmelCase  	=			self.unet.config.attention_head_dim // 2
											self.unet.set_attention_slice(UpperCAmelCase_  )


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







											"""simple docstring"""




											self.enable_attention_slicing(UpperCAmelCase_  )


				def 				__lowerCamelCase	(					self	:			int  )		->  Any:







											"""simple docstring"""




											set_requires_grad(self.vae				,     UpperCAmelCase_  )


				def 				__lowerCamelCase	(					self	:			Optional[int]  )		->  Union[str, Any]:







											"""simple docstring"""




											set_requires_grad(self.vae				,     UpperCAmelCase_  )


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







											"""simple docstring"""




											set_requires_grad(self.unet				,     UpperCAmelCase_  )


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







											"""simple docstring"""




											set_requires_grad(self.unet				,     UpperCAmelCase_  )


				def 				__lowerCamelCase	(					self	:			Optional[int]				,     UpperCAmelCase_	:			Union[str, Any]				,     UpperCAmelCase_	:			Tuple				,     UpperCAmelCase_	:			Tuple  )		->  Optional[Any]:







											"""simple docstring"""




											_lowerCAmelCase  	=			min(int(num_inference_steps * strength  )				,     UpperCAmelCase_  )

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

											return timesteps, num_inference_steps - t_start


				def 				__lowerCamelCase	(					self	:			Optional[Any]				,     UpperCAmelCase_	:			Dict				,     UpperCAmelCase_	:			Optional[Any]				,     UpperCAmelCase_	:			List[Any]				,     UpperCAmelCase_	:			Any				,     UpperCAmelCase_	:			str				,     UpperCAmelCase_	:			Any=None  )		->  str:







											"""simple docstring"""




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

											_lowerCAmelCase  	=			image.to(device=UpperCAmelCase_				,     dtype=UpperCAmelCase_  )

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

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

											_lowerCAmelCase  	=			randn_tensor(init_latents.shape				,     generator=UpperCAmelCase_				,     device=UpperCAmelCase_				,     dtype=UpperCAmelCase_  )

											# get latents
											_lowerCAmelCase  	=			self.scheduler.add_noise(UpperCAmelCase_				,     UpperCAmelCase_				,     UpperCAmelCase_  )
											_lowerCAmelCase  	=			init_latents

											return latents


				def 				__lowerCamelCase	(					self	:			str				,     UpperCAmelCase_	:			Tuple  )		->  Dict:







											"""simple docstring"""




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


				def 				__lowerCamelCase	(					self	:			List[str]				,     UpperCAmelCase_	:			str				,     UpperCAmelCase_	:			List[Any]  )		->  int:







											"""simple docstring"""




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


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







											"""simple docstring"""




											_lowerCAmelCase  	=			latents.detach().requires_grad_()

											_lowerCAmelCase  	=			self.scheduler.scale_model_input(UpperCAmelCase_				,     UpperCAmelCase_  )

											# predict the noise residual
											_lowerCAmelCase  	=			self.unet(UpperCAmelCase_				,     UpperCAmelCase_				,     encoder_hidden_states=UpperCAmelCase_  ).sample

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

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

											_lowerCAmelCase  	=			transforms.Resize(self.feature_extractor_size  )(UpperCAmelCase_  )
											_lowerCAmelCase  	=			self.normalize(UpperCAmelCase_  ).to(latents.dtype  )

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

											_lowerCAmelCase  	=			spherical_dist_loss(UpperCAmelCase_				,     UpperCAmelCase_  ).mean() * clip_guidance_scale

											_lowerCAmelCase  	=			-torch.autograd.grad(UpperCAmelCase_				,     UpperCAmelCase_  )[0]

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







				@torch.no_grad()
				def __call__(					self	:			Optional[Any]				,     UpperCAmelCase_	:			Union[torch.FloatTensor, PIL.Image.Image]				,     UpperCAmelCase_	:			Union[torch.FloatTensor, PIL.Image.Image]				,     UpperCAmelCase_	:			Optional[str] = None				,     UpperCAmelCase_	:			Optional[str] = None				,     UpperCAmelCase_	:			Optional[int] = 512				,     UpperCAmelCase_	:			Optional[int] = 512				,     UpperCAmelCase_	:			float = 0.6				,     UpperCAmelCase_	:			Optional[int] = 50				,     UpperCAmelCase_	:			Optional[float] = 7.5				,     UpperCAmelCase_	:			Optional[int] = 1				,     UpperCAmelCase_	:			float = 0.0				,     UpperCAmelCase_	:			Optional[float] = 100				,     UpperCAmelCase_	:			Optional[torch.Generator] = None				,     UpperCAmelCase_	:			Optional[str] = "pil"				,     UpperCAmelCase_	:			bool = True				,     UpperCAmelCase_	:			float = 0.8				,     UpperCAmelCase_	:			float = 0.1				,     UpperCAmelCase_	:			float = 0.1				,     )		->  List[str]:







											"""simple docstring"""




											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:
																		_lowerCAmelCase  	=			[generator] + [None] * (batch_size - 1)

											_lowerCAmelCase  	=			[
											    ('model', self.coca_model is None),
											    ('tokenizer', self.coca_tokenizer is None),
											    ('transform', self.coca_transform is None),
											]
											_lowerCAmelCase  	=			[x[0] for x in coca_is_none if x[1]]
											_lowerCAmelCase  	=			', '.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."""  )
																		_lowerCAmelCase  	=			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."""  )
																		_lowerCAmelCase  	=			self.get_image_description(UpperCAmelCase_  )

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

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

											_lowerCAmelCase  	=			slerp(UpperCAmelCase_				,     UpperCAmelCase_				,     UpperCAmelCase_  )

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

											# set timesteps
											_lowerCAmelCase  	=			'offset' in set(inspect.signature(self.scheduler.set_timesteps  ).parameters.keys()  )
											_lowerCAmelCase  	=			{}
											if accepts_offset:
																		_lowerCAmelCase  	=			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  )

											_lowerCAmelCase	,		_lowerCAmelCase  	=			self.get_timesteps(UpperCAmelCase_				,     UpperCAmelCase_				,     self.device  )
											_lowerCAmelCase  	=			timesteps[:1].repeat(UpperCAmelCase_  )

											# Preprocess image
											_lowerCAmelCase  	=			preprocess(UpperCAmelCase_				,     UpperCAmelCase_				,     UpperCAmelCase_  )
											_lowerCAmelCase  	=			self.prepare_latents(
											    UpperCAmelCase_				,     UpperCAmelCase_				,     UpperCAmelCase_				,     text_embeddings.dtype				,     self.device				,     UpperCAmelCase_  )

											_lowerCAmelCase  	=			preprocess(UpperCAmelCase_				,     UpperCAmelCase_				,     UpperCAmelCase_  )
											_lowerCAmelCase  	=			self.prepare_latents(
											    UpperCAmelCase_				,     UpperCAmelCase_				,     UpperCAmelCase_				,     text_embeddings.dtype				,     self.device				,     UpperCAmelCase_  )

											_lowerCAmelCase  	=			slerp(UpperCAmelCase_				,     UpperCAmelCase_				,     UpperCAmelCase_  )

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

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

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

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

																									# predict the noise residual
																									_lowerCAmelCase  	=			self.unet(UpperCAmelCase_				,     UpperCAmelCase_				,     encoder_hidden_states=UpperCAmelCase_  ).sample

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

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

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

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

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

											if output_type == "pil":
																		_lowerCAmelCase  	=			self.numpy_to_pil(UpperCAmelCase_  )

											if not return_dict:
																		return (image, None)

											return StableDiffusionPipelineOutput(images=UpperCAmelCase_				,     nsfw_content_detected=UpperCAmelCase_  )








"""simple docstring"""



import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import ClassLabel, Features, Image
from .base import TaskTemplate
@dataclass(frozen=UpperCAmelCase							)
class        _SCREAMING_SNAKE_CASE  (  UpperCAmelCase							):






				'''simple docstring'''



				SCREAMING_SNAKE_CASE_:							str							=					field(default="image-classification"      ,				metadata={"include_in_asdict_even_if_is_default": True}							)
				SCREAMING_SNAKE_CASE_:							ClassVar[Features]							=					Features({"image": Image()}							)
				SCREAMING_SNAKE_CASE_:							ClassVar[Features]							=					Features({"labels": ClassLabel}							)
				SCREAMING_SNAKE_CASE_:							str							=					"image"
				SCREAMING_SNAKE_CASE_:							str							=					"labels"


				def 				__lowerCamelCase	(					self	:			str				,     UpperCAmelCase_	:			Optional[Any]  )		->  Dict:







											"""simple docstring"""




											if self.label_column not in features:
																		raise ValueError(F"""Column {self.label_column} is not present in features."""  )
											if not isinstance(features[self.label_column]				,     UpperCAmelCase_  ):
																		raise ValueError(F"""Column {self.label_column} is not a ClassLabel."""  )
											_lowerCAmelCase  	=			copy.deepcopy(self  )
											_lowerCAmelCase  	=			self.label_schema.copy()
											_lowerCAmelCase  	=			features[self.label_column]
											_lowerCAmelCase  	=			label_schema
											return task_template







				@property
				def 				__lowerCamelCase	(					self	:			Union[str, Any]  )		->  Dict[str, str]:







											"""simple docstring"""




											return {
											    self.image_column: "image",
											    self.label_column: "labels",
											}






