





"""simple docstring"""




import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import (
    PIL_INTERPOLATION,
    randn_tensor,
)





def 	a_    (							_lowercase	,						_lowercase	,						_lowercase       ):
						if isinstance(_lowercase	,						torch.Tensor       ):
												return image
						elif isinstance(_lowercase	,						PIL.Image.Image       ):
												_UpperCamelCase      : Union[str, Any]         =   [image]

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





def 	a_    (							_lowercase	,						_lowercase	,						_lowercase	,						_lowercase=0.9995       ):
						if not isinstance(_lowercase	,						np.ndarray       ):
												_UpperCamelCase      : Any         =   True
												_UpperCamelCase      : Optional[int]         =   va.device
												_UpperCamelCase      : Optional[int]         =   va.cpu().numpy()
												_UpperCamelCase      : Tuple         =   va.cpu().numpy()

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

						if inputs_are_torch:
												_UpperCamelCase      : Optional[int]         =   torch.from_numpy(_lowercase       ).to(_lowercase       )

						return va





def 	a_    (							_lowercase	,						_lowercase       ):
						_UpperCamelCase      : Union[str, Any]         =   F.normalize(_lowercase	,						dim=-1       )
						_UpperCamelCase      : Tuple         =   F.normalize(_lowercase	,						dim=-1       )
						return (x - y).norm(dim=-1       ).div(2       ).arcsin().pow(2       ).mul(2       )





def 	a_    (							_lowercase	,						_lowercase       ):
						for param in model.parameters():
												_UpperCamelCase      : Tuple         =   value






class 			_a						(				_lowerCAmelCase    ):






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




										'''simple docstring'''



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






				def     snake_case     (			self		:   Union[str, Any],			lowerCAmelCase__		:   Optional[Union[str, int]] = "auto"							)     ->				Dict:




										'''simple docstring'''



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






				def     snake_case     (			self		:   Tuple							)     ->				Tuple:




										'''simple docstring'''



										self.enable_attention_slicing(lowerCAmelCase__							)






				def     snake_case     (			self		:   int							)     ->				Tuple:




										'''simple docstring'''



										set_requires_grad(self.vae,			lowerCAmelCase__							)






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




										'''simple docstring'''



										set_requires_grad(self.vae,			lowerCAmelCase__							)






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




										'''simple docstring'''



										set_requires_grad(self.unet,			lowerCAmelCase__							)






				def     snake_case     (			self		:   Any							)     ->				Tuple:




										'''simple docstring'''



										set_requires_grad(self.unet,			lowerCAmelCase__							)






				def     snake_case     (			self		:   Optional[int],			lowerCAmelCase__		:   Optional[int],			lowerCAmelCase__		:   Optional[int],			lowerCAmelCase__		:   Any							)     ->				Optional[int]:




										'''simple docstring'''



										_UpperCamelCase      : List[str]         =   min(int(num_inference_steps * strength							),			lowerCAmelCase__							)

										_UpperCamelCase      : Tuple         =   max(num_inference_steps - init_timestep,			0							)
										_UpperCamelCase      : List[Any]         =   self.scheduler.timesteps[t_start:]

										return timesteps, num_inference_steps - t_start






				def     snake_case     (			self		:   str,			lowerCAmelCase__		:   Optional[Any],			lowerCAmelCase__		:   Dict,			lowerCAmelCase__		:   int,			lowerCAmelCase__		:   Tuple,			lowerCAmelCase__		:   Optional[int],			lowerCAmelCase__		:   int=None							)     ->				List[str]:




										'''simple docstring'''



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

										_UpperCamelCase      : Dict         =   image.to(device=lowerCAmelCase__,			dtype=lowerCAmelCase__							)

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

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

										_UpperCamelCase      : Optional[int]         =   randn_tensor(init_latents.shape,			generator=lowerCAmelCase__,			device=lowerCAmelCase__,			dtype=lowerCAmelCase__							)

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

										return latents






				def     snake_case     (			self		:   str,			lowerCAmelCase__		:   Tuple							)     ->				Optional[Any]:




										'''simple docstring'''



										_UpperCamelCase      : int         =   self.coca_transform(lowerCAmelCase__							).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      : List[Any]         =   self.coca_tokenizer.decode(generated[0].cpu().numpy()							)
										return generated.split('''<end_of_text>'''							)[0].replace('''<start_of_text>''',			''''''							).rstrip(''' .,'''							)






				def     snake_case     (			self		:   str,			lowerCAmelCase__		:   List[str],			lowerCAmelCase__		:   Optional[Any]							)     ->				Optional[Any]:




										'''simple docstring'''



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






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




										'''simple docstring'''



										_UpperCamelCase      : List[Any]         =   latents.detach().requires_grad_()

										_UpperCamelCase      : Any         =   self.scheduler.scale_model_input(lowerCAmelCase__,			lowerCAmelCase__							)

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

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

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

										_UpperCamelCase      : str         =   transforms.Resize(self.feature_extractor_size							)(lowerCAmelCase__							)
										_UpperCamelCase      : Dict         =   self.normalize(lowerCAmelCase__							).to(latents.dtype							)

										_UpperCamelCase      : Any         =   self.clip_model.get_image_features(lowerCAmelCase__							)
										_UpperCamelCase      : Dict         =   image_embeddings_clip / image_embeddings_clip.norm(p=2,			dim=-1,			keepdim=lowerCAmelCase__							)

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

										_UpperCamelCase      : Union[str, Any]         =   -torch.autograd.grad(lowerCAmelCase__,			lowerCAmelCase__							)[0]

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






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




										'''simple docstring'''



										if isinstance(lowerCAmelCase__,			lowerCAmelCase__							) and len(lowerCAmelCase__							) != batch_size:
																raise ValueError(f"""You have passed {batch_size} batch_size, but only {len(lowerCAmelCase__							)} 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(lowerCAmelCase__,			torch.Generator							) and batch_size > 1:
																_UpperCamelCase      : Dict         =   [generator] + [None] * (batch_size - 1)

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

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

										_UpperCamelCase      : Optional[int]         =   self.tokenizer(
										    lowerCAmelCase__,			padding='''max_length''',			max_length=self.tokenizer.model_max_length,			truncation=lowerCAmelCase__,			return_tensors='''pt''',			)
										_UpperCamelCase      : Tuple         =   self.text_encoder(style_text_input.input_ids.to(self.device							)							)[0]

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

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

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

										self.scheduler.set_timesteps(lowerCAmelCase__,			**lowerCAmelCase__							)
										# 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      : Any         =   self.get_timesteps(lowerCAmelCase__,			lowerCAmelCase__,			self.device							)
										_UpperCamelCase      : Dict         =   timesteps[:1].repeat(lowerCAmelCase__							)

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

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

										_UpperCamelCase      : Tuple         =   slerp(lowerCAmelCase__,			lowerCAmelCase__,			lowerCAmelCase__							)

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

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

										# get the initial random noise unless the user supplied it

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

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

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

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

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

																						# perform classifier free guidance
																						if do_classifier_free_guidance:
																												_UpperCamelCase       ,							_UpperCamelCase      : Union[str, Any]         =   noise_pred.chunk(2							)
																												_UpperCamelCase      : Dict         =   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      : Tuple         =   self.cond_fn(
																												    lowerCAmelCase__,			lowerCAmelCase__,			lowerCAmelCase__,			lowerCAmelCase__,			lowerCAmelCase__,			lowerCAmelCase__,			lowerCAmelCase__,			)

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

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
										_UpperCamelCase      : Dict         =   1 / 0.18_215 * latents
										_UpperCamelCase      : Union[str, Any]         =   self.vae.decode(lowerCAmelCase__							).sample

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

										if output_type == "pil":
																_UpperCamelCase      : Tuple         =   self.numpy_to_pil(lowerCAmelCase__							)

										if not return_dict:
																return (image, None)

										return StableDiffusionPipelineOutput(images=lowerCAmelCase__,			nsfw_content_detected=lowerCAmelCase__							)










"""simple docstring"""




from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType




class 			_a						(				_lowerCAmelCase    ):
				UpperCamelCase								=   ['''image_processor''', '''tokenizer''']
				UpperCamelCase								=   '''BlipImageProcessor'''
				UpperCamelCase								=   '''AutoTokenizer'''






				def __init__(			self		:   List[str],			lowerCAmelCase__		:   Optional[int],			lowerCAmelCase__		:   Optional[int]							)     ->				int:




										'''simple docstring'''



										_UpperCamelCase      : Any         =   False
										super().__init__(lowerCAmelCase__,			lowerCAmelCase__							)
										_UpperCamelCase      : Tuple         =   self.image_processor






				def __call__(			self		:   str,			lowerCAmelCase__		:   ImageInput = None,			lowerCAmelCase__		:   Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None,			lowerCAmelCase__		:   bool = True,			lowerCAmelCase__		:   Union[bool, str, PaddingStrategy] = False,			lowerCAmelCase__		:   Union[bool, str, TruncationStrategy] = None,			lowerCAmelCase__		:   Optional[int] = None,			lowerCAmelCase__		:   int = 0,			lowerCAmelCase__		:   Optional[int] = None,			lowerCAmelCase__		:   Optional[bool] = None,			lowerCAmelCase__		:   bool = False,			lowerCAmelCase__		:   bool = False,			lowerCAmelCase__		:   bool = False,			lowerCAmelCase__		:   bool = False,			lowerCAmelCase__		:   bool = False,			lowerCAmelCase__		:   bool = True,			lowerCAmelCase__		:   Optional[Union[str, TensorType]] = None,			**lowerCAmelCase__		:   Optional[Any],			)     ->				BatchEncoding:




										'''simple docstring'''



										if images is None and text is None:
																raise ValueError('''You have to specify either images or text.'''							)

										# Get only text
										if images is None:
																_UpperCamelCase      : int         =   self.tokenizer
																_UpperCamelCase      : List[str]         =   self.tokenizer(
																    text=lowerCAmelCase__,			add_special_tokens=lowerCAmelCase__,			padding=lowerCAmelCase__,			truncation=lowerCAmelCase__,			max_length=lowerCAmelCase__,			stride=lowerCAmelCase__,			pad_to_multiple_of=lowerCAmelCase__,			return_attention_mask=lowerCAmelCase__,			return_overflowing_tokens=lowerCAmelCase__,			return_special_tokens_mask=lowerCAmelCase__,			return_offsets_mapping=lowerCAmelCase__,			return_token_type_ids=lowerCAmelCase__,			return_length=lowerCAmelCase__,			verbose=lowerCAmelCase__,			return_tensors=lowerCAmelCase__,			**lowerCAmelCase__,			)
																return text_encoding

										# add pixel_values
										_UpperCamelCase      : List[str]         =   self.image_processor(lowerCAmelCase__,			return_tensors=lowerCAmelCase__							)

										if text is not None:
																_UpperCamelCase      : Any         =   self.tokenizer(
																    text=lowerCAmelCase__,			add_special_tokens=lowerCAmelCase__,			padding=lowerCAmelCase__,			truncation=lowerCAmelCase__,			max_length=lowerCAmelCase__,			stride=lowerCAmelCase__,			pad_to_multiple_of=lowerCAmelCase__,			return_attention_mask=lowerCAmelCase__,			return_overflowing_tokens=lowerCAmelCase__,			return_special_tokens_mask=lowerCAmelCase__,			return_offsets_mapping=lowerCAmelCase__,			return_token_type_ids=lowerCAmelCase__,			return_length=lowerCAmelCase__,			verbose=lowerCAmelCase__,			return_tensors=lowerCAmelCase__,			**lowerCAmelCase__,			)
										else:
																_UpperCamelCase      : List[Any]         =   None

										if text_encoding is not None:
																encoding_image_processor.update(lowerCAmelCase__							)

										return encoding_image_processor






				def     snake_case     (			self		:   List[Any],			*lowerCAmelCase__		:   List[str],			**lowerCAmelCase__		:   str							)     ->				Optional[int]:




										'''simple docstring'''



										return self.tokenizer.batch_decode(*lowerCAmelCase__,			**lowerCAmelCase__							)






				def     snake_case     (			self		:   List[Any],			*lowerCAmelCase__		:   Dict,			**lowerCAmelCase__		:   Any							)     ->				List[str]:




										'''simple docstring'''



										return self.tokenizer.decode(*lowerCAmelCase__,			**lowerCAmelCase__							)






				@property
				# Copied from transformers.models.blip.processing_blip.BlipProcessor.model_input_names
				def     snake_case     (			self		:   Any							)     ->				str:




										'''simple docstring'''



										_UpperCamelCase      : List[str]         =   self.tokenizer.model_input_names
										_UpperCamelCase      : Union[str, Any]         =   self.image_processor.model_input_names
										return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names							)							)



