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 							lowerCAmelCase(     __lowerCamelCase				,    __lowerCamelCase				,    __lowerCamelCase						):
					if isinstance(__lowerCamelCase				,    torch.Tensor						):
										return image
					elif isinstance(__lowerCamelCase				,    PIL.Image.Image						):
										__a		 =       [image]

					if isinstance(image[0]				,    PIL.Image.Image						):
										__a		 =       [np.array(i.resize((w, h)				,    resample=PIL_INTERPOLATION['lanczos']						)						)[None, :] for i in image]
										__a		 =       np.concatenate(__lowerCamelCase				,    axis=0						)
										__a		 =       np.array(__lowerCamelCase						).astype(np.floataa						) / 2_55.0
										__a		 =       image.transpose(0				,    3				,    1				,    2						)
										__a		 =       2.0 * image - 1.0
										__a		 =       torch.from_numpy(__lowerCamelCase						)
					elif isinstance(image[0]				,    torch.Tensor						):
										__a		 =       torch.cat(__lowerCamelCase				,    dim=0						)
					return image
def 							lowerCAmelCase(     __lowerCamelCase				,    __lowerCamelCase				,    __lowerCamelCase				,    __lowerCamelCase=0.99_95						):
					if not isinstance(__lowerCamelCase				,    np.ndarray						):
										__a		 =       True
										__a		 =       va.device
										__a		 =       va.cpu().numpy()
										__a		 =       va.cpu().numpy()

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

					if inputs_are_torch:
										__a		 =       torch.from_numpy(__lowerCamelCase						).to(__lowerCamelCase						)

					return va
def 							lowerCAmelCase(     __lowerCamelCase				,    __lowerCamelCase						):
					__a		 =       F.normalize(__lowerCamelCase				,    dim=-1						)
					__a		 =       F.normalize(__lowerCamelCase				,    dim=-1						)
					return (x - y).norm(dim=-1						).div(2						).arcsin().pow(2						).mul(2						)
def 							lowerCAmelCase(     __lowerCamelCase				,    __lowerCamelCase						):
					for param in model.parameters():
										__a		 =       value



class 						a__							(       __snake_case		):



			def __init__(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase=None				,						UpperCAmelCase=None				,						UpperCAmelCase=None				,						)			->   Optional[Any]:
								super().__init__()
								self.register_modules(
								    vae=UpperCAmelCase				,						text_encoder=UpperCAmelCase				,						clip_model=UpperCAmelCase				,						tokenizer=UpperCAmelCase				,						unet=UpperCAmelCase				,						scheduler=UpperCAmelCase				,						feature_extractor=UpperCAmelCase				,						coca_model=UpperCAmelCase				,						coca_tokenizer=UpperCAmelCase				,						coca_transform=UpperCAmelCase				,						)
								__a		 =       (
								    feature_extractor.size
								    if isinstance(feature_extractor.size				,						UpperCAmelCase   )
								    else feature_extractor.size['shortest_edge']
								)
								__a		 =       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 __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase = "auto"   )			->   Optional[Any]:
								if slice_size == "auto":
													# half the attention head size is usually a good trade-off between
													# speed and memory
													__a		 =       self.unet.config.attention_head_dim // 2
								self.unet.set_attention_slice(UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   Optional[int]:
								self.enable_attention_slicing(UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   List[Any]:
								set_requires_grad(self.vae				,						UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   List[Any]:
								set_requires_grad(self.vae				,						UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   Union[str, Any]:
								set_requires_grad(self.unet				,						UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self   )			->   List[Any]:
								set_requires_grad(self.unet				,						UpperCAmelCase   )



			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )			->   Dict:
								# get the original timestep using init_timestep
								__a		 =       min(int(num_inference_steps * strength   )				,						UpperCAmelCase   )

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

								return timesteps, num_inference_steps - t_start



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

								__a		 =       image.to(device=UpperCAmelCase				,						dtype=UpperCAmelCase   )

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

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

								__a		 =       randn_tensor(init_latents.shape				,						generator=UpperCAmelCase				,						device=UpperCAmelCase				,						dtype=UpperCAmelCase   )

								# get latents
								__a		 =       self.scheduler.add_noise(UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )
								__a		 =       init_latents

								return latents



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



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



			@torch.enable_grad()
			def __SCREAMING_SNAKE_CASE							(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						)			->   Tuple:
								__a		 =       latents.detach().requires_grad_()

								__a		 =       self.scheduler.scale_model_input(UpperCAmelCase				,						UpperCAmelCase   )

								# predict the noise residual
								__a		 =       self.unet(UpperCAmelCase				,						UpperCAmelCase				,						encoder_hidden_states=UpperCAmelCase   ).sample

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

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

								__a		 =       transforms.Resize(self.feature_extractor_size   )(UpperCAmelCase   )
								__a		 =       self.normalize(UpperCAmelCase   ).to(latents.dtype   )

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

								__a		 =       spherical_dist_loss(UpperCAmelCase				,						UpperCAmelCase   ).mean() * clip_guidance_scale

								__a		 =       -torch.autograd.grad(UpperCAmelCase				,						UpperCAmelCase   )[0]

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







			@torch.no_grad()
			def __call__(						self				,						UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase = None				,						UpperCAmelCase = None				,						UpperCAmelCase = 5_1_2				,						UpperCAmelCase = 5_1_2				,						UpperCAmelCase = 0.6				,						UpperCAmelCase = 5_0				,						UpperCAmelCase = 7.5				,						UpperCAmelCase = 1				,						UpperCAmelCase = 0.0				,						UpperCAmelCase = 1_0_0				,						UpperCAmelCase = None				,						UpperCAmelCase = "pil"				,						UpperCAmelCase = True				,						UpperCAmelCase = 0.8				,						UpperCAmelCase = 0.1				,						UpperCAmelCase = 0.1				,						)			->   Union[str, Any]:
								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:
													__a		 =       [generator] + [None] * (batch_size - 1)

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

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

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

								__a		 =       slerp(UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )

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

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

								__a ,				__a		 =       self.get_timesteps(UpperCAmelCase				,						UpperCAmelCase				,						self.device   )
								__a		 =       timesteps[:1].repeat(UpperCAmelCase   )

								# Preprocess image
								__a		 =       preprocess(UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )
								__a		 =       self.prepare_latents(
								    UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						text_embeddings.dtype				,						self.device				,						UpperCAmelCase   )

								__a		 =       preprocess(UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )
								__a		 =       self.prepare_latents(
								    UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase				,						text_embeddings.dtype				,						self.device				,						UpperCAmelCase   )

								__a		 =       slerp(UpperCAmelCase				,						UpperCAmelCase				,						UpperCAmelCase   )

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

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

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

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

																		# predict the noise residual
																		__a		 =       self.unet(UpperCAmelCase				,						UpperCAmelCase				,						encoder_hidden_states=UpperCAmelCase   ).sample

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

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

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

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

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

								if output_type == "pil":
													__a		 =       self.numpy_to_pil(UpperCAmelCase   )

								if not return_dict:
													return (image, None)

								return StableDiffusionPipelineOutput(images=UpperCAmelCase				,						nsfw_content_detected=UpperCAmelCase   )





from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tf_available, is_torch_available


lowerCamelCase_    :		Dict							=	{"""configuration_swin""": ["""SWIN_PRETRAINED_CONFIG_ARCHIVE_MAP""", """SwinConfig""", """SwinOnnxConfig"""]}


try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	lowerCamelCase_    :		int							=	[
	    """SWIN_PRETRAINED_MODEL_ARCHIVE_LIST""",
	    """SwinForImageClassification""",
	    """SwinForMaskedImageModeling""",
	    """SwinModel""",
	    """SwinPreTrainedModel""",
	    """SwinBackbone""",
	]

try:
	if not is_tf_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	lowerCamelCase_    :		Any							=	[
	    """TF_SWIN_PRETRAINED_MODEL_ARCHIVE_LIST""",
	    """TFSwinForImageClassification""",
	    """TFSwinForMaskedImageModeling""",
	    """TFSwinModel""",
	    """TFSwinPreTrainedModel""",
	]

if TYPE_CHECKING:
	from .configuration_swin import SWIN_PRETRAINED_CONFIG_ARCHIVE_MAP, SwinConfig, SwinOnnxConfig

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_swin import (
		    SWIN_PRETRAINED_MODEL_ARCHIVE_LIST,
		    SwinBackbone,
		    SwinForImageClassification,
		    SwinForMaskedImageModeling,
		    SwinModel,
		    SwinPreTrainedModel,
		)

	try:
		if not is_tf_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_tf_swin import (
		    TF_SWIN_PRETRAINED_MODEL_ARCHIVE_LIST,
		    TFSwinForImageClassification,
		    TFSwinForMaskedImageModeling,
		    TFSwinModel,
		    TFSwinPreTrainedModel,
		)

else:
	import sys

	lowerCamelCase_    :		List[Any]							=	_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)




