

"""simple docstring"""






import inspect
from typing import List, Optional, Union

import torch
from torch import nn
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPImageProcessor, CLIPModel, CLIPTextModel, CLIPTokenizer

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






class       _UpperCAmelCase     (						nn.Module):
			def __init__(						self ,						_A ,						_A=1.0       )      ->      Union[str, Any]:






				'''simple docstring'''







				super().__init__()

				_UpperCAmelCase      :							Optional[int]       =   cut_size
				_UpperCAmelCase      :							Tuple       =   cut_power
			def 							__snake_case	(						self ,						_A ,						_A       )      ->      Tuple:






				'''simple docstring'''







				_UpperCAmelCase  ,				_UpperCAmelCase      :							Union[str, Any]       =   pixel_values.shape[2:4]
				_UpperCAmelCase      :							Union[str, Any]       =   min(_A ,						_A       )
				_UpperCAmelCase      :							str       =   min(_A ,						_A ,						self.cut_size       )
				_UpperCAmelCase      :							Union[str, Any]       =   []
				for _ in range(_A       ):
					_UpperCAmelCase      :							Dict       =   int(torch.rand([]       ) ** self.cut_power * (max_size - min_size) + min_size       )
					_UpperCAmelCase      :							List[str]       =   torch.randint(0 ,						sideX - size + 1 ,						()       )
					_UpperCAmelCase      :							List[str]       =   torch.randint(0 ,						sideY - size + 1 ,						()       )
					_UpperCAmelCase      :							str       =   pixel_values[:, :, offsety : offsety + size, offsetx : offsetx + size]
					cutouts.append(F.adaptive_avg_poolad(_A ,						self.cut_size       )       )
				return torch.cat(_A       )
def  UpperCamelCase   (     _lowerCAmelCase       :  int,					_lowerCAmelCase       :  Any	)			->     int:
	_UpperCAmelCase      :							str       =   F.normalize(_lowerCAmelCase,					dim=-1	)
	_UpperCAmelCase      :							List[str]       =   F.normalize(_lowerCAmelCase,					dim=-1	)
	return (x - y).norm(dim=-1	).div(2	).arcsin().pow(2	).mul(2	)
def  UpperCamelCase   (     _lowerCAmelCase       :  Any,					_lowerCAmelCase       :  Optional[int]	)			->     List[str]:
	for param in model.parameters():
		_UpperCAmelCase      :							str       =   value






class       _UpperCAmelCase     (						__a):
			def __init__(						self ,						_A ,						_A ,						_A ,						_A ,						_A ,						_A ,						_A ,						)      ->      List[str]:






				'''simple docstring'''







				super().__init__()
				self.register_modules(
				    vae=_A ,						text_encoder=_A ,						clip_model=_A ,						tokenizer=_A ,						unet=_A ,						scheduler=_A ,						feature_extractor=_A ,						)

				_UpperCAmelCase      :							Tuple       =   transforms.Normalize(mean=feature_extractor.image_mean ,						std=feature_extractor.image_std       )
				_UpperCAmelCase      :							str       =   (
				    feature_extractor.size
				    if isinstance(feature_extractor.size ,						_A       )
				    else feature_extractor.size["""shortest_edge"""]
				)
				_UpperCAmelCase      :							int       =   MakeCutouts(self.cut_out_size       )

				set_requires_grad(self.text_encoder ,						_A       )
				set_requires_grad(self.clip_model ,						_A       )
			def 							__snake_case	(						self ,						_A = "auto"       )      ->      str:






				'''simple docstring'''







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






				'''simple docstring'''







				self.enable_attention_slicing(_A       )
			def 							__snake_case	(						self       )      ->      Optional[Any]:






				'''simple docstring'''







				set_requires_grad(self.vae ,						_A       )
			def 							__snake_case	(						self       )      ->      Optional[Any]:






				'''simple docstring'''







				set_requires_grad(self.vae ,						_A       )
			def 							__snake_case	(						self       )      ->      Optional[Any]:






				'''simple docstring'''







				set_requires_grad(self.unet ,						_A       )
			def 							__snake_case	(						self       )      ->      Tuple:






				'''simple docstring'''







				set_requires_grad(self.unet ,						_A       )
			@torch.enable_grad()
			def 							__snake_case	(						self ,						_A ,						_A ,						_A ,						_A ,						_A ,						_A ,						_A ,						_A ,						_A=True ,						)      ->      Union[str, Any]:






				'''simple docstring'''







				_UpperCAmelCase      :							Optional[Any]       =   latents.detach().requires_grad_()

				_UpperCAmelCase      :							List[Any]       =   self.scheduler.scale_model_input(_A ,						_A       )

				# predict the noise residual
				_UpperCAmelCase      :							str       =   self.unet(_A ,						_A ,						encoder_hidden_states=_A       ).sample

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

					_UpperCAmelCase      :							str       =   torch.sqrt(_A       )
					_UpperCAmelCase      :							List[str]       =   pred_original_sample * (fac) + latents * (1 - fac)
				elif isinstance(self.scheduler ,						_A       ):
					_UpperCAmelCase      :							List[str]       =   self.scheduler.sigmas[index]
					_UpperCAmelCase      :							Optional[Any]       =   latents - sigma * noise_pred
				else:
					raise ValueError(f'''scheduler type {type(self.scheduler       )} not supported'''       )

				_UpperCAmelCase      :							List[Any]       =   1 / self.vae.config.scaling_factor * sample
				_UpperCAmelCase      :							Optional[int]       =   self.vae.decode(_A       ).sample
				_UpperCAmelCase      :							Union[str, Any]       =   (image / 2 + 0.5).clamp(0 ,						1       )

				if use_cutouts:
					_UpperCAmelCase      :							Optional[Any]       =   self.make_cutouts(_A ,						_A       )
				else:
					_UpperCAmelCase      :							Optional[Any]       =   transforms.Resize(self.cut_out_size       )(_A       )
				_UpperCAmelCase      :							Optional[Any]       =   self.normalize(_A       ).to(latents.dtype       )

				_UpperCAmelCase      :							Dict       =   self.clip_model.get_image_features(_A       )
				_UpperCAmelCase      :							Optional[int]       =   image_embeddings_clip / image_embeddings_clip.norm(p=2 ,						dim=-1 ,						keepdim=_A       )

				if use_cutouts:
					_UpperCAmelCase      :							Any       =   spherical_dist_loss(_A ,						_A       )
					_UpperCAmelCase      :							int       =   dists.view([num_cutouts, sample.shape[0], -1]       )
					_UpperCAmelCase      :							Any       =   dists.sum(2       ).mean(0       ).sum() * clip_guidance_scale
				else:
					_UpperCAmelCase      :							int       =   spherical_dist_loss(_A ,						_A       ).mean() * clip_guidance_scale

				_UpperCAmelCase      :							int       =   -torch.autograd.grad(_A ,						_A       )[0]

				if isinstance(self.scheduler ,						_A       ):
					_UpperCAmelCase      :							Any       =   latents.detach() + grads * (sigma**2)
					_UpperCAmelCase      :							List[Any]       =   noise_pred_original
				else:
					_UpperCAmelCase      :							List[str]       =   noise_pred_original - torch.sqrt(_A       ) * grads
				return noise_pred, latents
			@torch.no_grad()
			def __call__(						self ,						_A ,						_A = 5_12 ,						_A = 5_12 ,						_A = 50 ,						_A = 7.5 ,						_A = 1 ,						_A = 0.0 ,						_A = 1_00 ,						_A = None ,						_A = 4 ,						_A = True ,						_A = None ,						_A = None ,						_A = "pil" ,						_A = True ,						)      ->      Any:






				'''simple docstring'''







				if isinstance(_A ,						_A       ):
					_UpperCAmelCase      :							List[str]       =   1
				elif isinstance(_A ,						_A       ):
					_UpperCAmelCase      :							Union[str, Any]       =   len(_A       )
				else:
					raise ValueError(f'''`prompt` has to be of type `str` or `list` but is {type(_A       )}'''       )

				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}.'''       )

				# get prompt text embeddings
				_UpperCAmelCase      :							int       =   self.tokenizer(
				    _A ,						padding="""max_length""" ,						max_length=self.tokenizer.model_max_length ,						truncation=_A ,						return_tensors="""pt""" ,						)
				_UpperCAmelCase      :							int       =   self.text_encoder(text_input.input_ids.to(self.device       )       )[0]
				# duplicate text embeddings for each generation per prompt
				_UpperCAmelCase      :							Optional[Any]       =   text_embeddings.repeat_interleave(_A ,						dim=0       )

				if clip_guidance_scale > 0:
					if clip_prompt is not None:
						_UpperCAmelCase      :							Optional[Any]       =   self.tokenizer(
						    _A ,						padding="""max_length""" ,						max_length=self.tokenizer.model_max_length ,						truncation=_A ,						return_tensors="""pt""" ,						).input_ids.to(self.device       )
					else:
						_UpperCAmelCase      :							Dict       =   text_input.input_ids.to(self.device       )
					_UpperCAmelCase      :							List[str]       =   self.clip_model.get_text_features(_A       )
					_UpperCAmelCase      :							Union[str, Any]       =   text_embeddings_clip / text_embeddings_clip.norm(p=2 ,						dim=-1 ,						keepdim=_A       )
					# duplicate text embeddings clip for each generation per prompt
					_UpperCAmelCase      :							Any       =   text_embeddings_clip.repeat_interleave(_A ,						dim=0       )

				# 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      :							Any       =   guidance_scale > 1.0
				# get unconditional embeddings for classifier free guidance
				if do_classifier_free_guidance:
					_UpperCAmelCase      :							str       =   text_input.input_ids.shape[-1]
					_UpperCAmelCase      :							List[Any]       =   self.tokenizer([""""""] ,						padding="""max_length""" ,						max_length=_A ,						return_tensors="""pt"""       )
					_UpperCAmelCase      :							Union[str, Any]       =   self.text_encoder(uncond_input.input_ids.to(self.device       )       )[0]
					# duplicate unconditional embeddings for each generation per prompt
					_UpperCAmelCase      :							Any       =   uncond_embeddings.repeat_interleave(_A ,						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      :							Optional[int]       =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
				_UpperCAmelCase      :							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(_A ,						generator=_A ,						device="""cpu""" ,						dtype=_A       ).to(
						    self.device       )
					else:
						_UpperCAmelCase      :							str       =   torch.randn(_A ,						generator=_A ,						device=self.device ,						dtype=_A       )
				else:
					if latents.shape != latents_shape:
						raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''       )
					_UpperCAmelCase      :							Any       =   latents.to(self.device       )

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

				self.scheduler.set_timesteps(_A ,						**_A       )

				# Some schedulers like PNDM have timesteps as arrays
				# It's more optimized to move all timesteps to correct device beforehand
				_UpperCAmelCase      :							int       =   self.scheduler.timesteps.to(self.device       )

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

				# check if the scheduler accepts generator
				_UpperCAmelCase      :							Optional[Any]       =   """generator""" in set(inspect.signature(self.scheduler.step       ).parameters.keys()       )
				if accepts_generator:
					_UpperCAmelCase      :							List[str]       =   generator

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

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

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

					# perform clip guidance
					if clip_guidance_scale > 0:
						_UpperCAmelCase      :							List[Any]       =   (
						    text_embeddings.chunk(2       )[1] if do_classifier_free_guidance else text_embeddings
						)
						_UpperCAmelCase  ,				_UpperCAmelCase      :							Dict       =   self.cond_fn(
						    _A ,						_A ,						_A ,						_A ,						_A ,						_A ,						_A ,						_A ,						_A ,						)

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

				# scale and decode the image latents with vae
				_UpperCAmelCase      :							Union[str, Any]       =   1 / self.vae.config.scaling_factor * latents
				_UpperCAmelCase      :							Dict       =   self.vae.decode(_A       ).sample

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

				if output_type == "pil":
					_UpperCAmelCase      :							Optional[Any]       =   self.numpy_to_pil(_A       )

				if not return_dict:
					return (image, None)

				return StableDiffusionPipelineOutput(images=_A ,						nsfw_content_detected=_A       )








"""simple docstring"""






from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


lowerCamelCase__			:					Union[str, Any]							  =					{
    '''configuration_unispeech_sat''': ['''UNISPEECH_SAT_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''UniSpeechSatConfig'''],
}

try:
						if not is_torch_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						lowerCamelCase__			:					List[str]							  =					[
						    '''UNISPEECH_SAT_PRETRAINED_MODEL_ARCHIVE_LIST''',
						    '''UniSpeechSatForAudioFrameClassification''',
						    '''UniSpeechSatForCTC''',
						    '''UniSpeechSatForPreTraining''',
						    '''UniSpeechSatForSequenceClassification''',
						    '''UniSpeechSatForXVector''',
						    '''UniSpeechSatModel''',
						    '''UniSpeechSatPreTrainedModel''',
						]

if TYPE_CHECKING:
						from .configuration_unispeech_sat import UNISPEECH_SAT_PRETRAINED_CONFIG_ARCHIVE_MAP, UniSpeechSatConfig

						try:
												if not is_torch_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_unispeech_sat import (
												    UNISPEECH_SAT_PRETRAINED_MODEL_ARCHIVE_LIST,
												    UniSpeechSatForAudioFrameClassification,
												    UniSpeechSatForCTC,
												    UniSpeechSatForPreTraining,
												    UniSpeechSatForSequenceClassification,
												    UniSpeechSatForXVector,
												    UniSpeechSatModel,
												    UniSpeechSatPreTrainedModel,
												)

else:
						import sys

						lowerCamelCase__			:					Any							  =					_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)





