






from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block




@dataclass
class 					UpperCamelCase_ (     snake_case_							):



	'''simple docstring'''




	lowerCAmelCase						=					42






class 					UpperCamelCase_ (     nn.Module							):



	'''simple docstring'''




	def __init__(			self			,     a=3			,     a=3			,     a=("DownEncoderBlock2D",)			,     a=(64,)			,     a=2			,     a=32			,     a="silu"			,     a=True			,     )			->       str:
								super().__init__()
								snake_case_           =				layers_per_block

								snake_case_           =				torch.nn.Convad(
								    a			,     block_out_channels[0]			,     kernel_size=3			,     stride=1			,     padding=1			,     )

								snake_case_           =				None
								snake_case_           =				nn.ModuleList([]   )

								# down
								snake_case_           =				block_out_channels[0]
								for i, down_block_type in enumerate(a   ):
															snake_case_           =				output_channel
															snake_case_           =				block_out_channels[i]
															snake_case_           =				i == len(a   ) - 1

															snake_case_           =				get_down_block(
															    a			,     num_layers=self.layers_per_block			,     in_channels=a			,     out_channels=a			,     add_downsample=not is_final_block			,     resnet_eps=1E-6			,     downsample_padding=0			,     resnet_act_fn=a			,     resnet_groups=a			,     attention_head_dim=a			,     temb_channels=a			,     )
															self.down_blocks.append(a   )

								# mid
								snake_case_           =				UNetMidBlockaD(
								    in_channels=block_out_channels[-1]			,     resnet_eps=1E-6			,     resnet_act_fn=a			,     output_scale_factor=1			,     resnet_time_scale_shift='default'			,     attention_head_dim=block_out_channels[-1]			,     resnet_groups=a			,     temb_channels=a			,     )

								# out
								snake_case_           =				nn.GroupNorm(num_channels=block_out_channels[-1]			,     num_groups=a			,     eps=1E-6   )
								snake_case_           =				nn.SiLU()

								snake_case_           =				2 * out_channels if double_z else out_channels
								snake_case_           =				nn.Convad(block_out_channels[-1]			,     a			,     3			,     padding=1   )

								snake_case_           =				False






	def 				_UpperCamelCase     (			self			,     a   )			->       int:
								snake_case_           =				x
								snake_case_           =				self.conv_in(a   )

								if self.training and self.gradient_checkpointing:

															def create_custom_forward(a   ):
																						def custom_forward(*a   ):
																													return module(*a   )

																						return custom_forward

															# down
															if is_torch_version('>='			,     '1.11.0'   ):
																						for down_block in self.down_blocks:
																													snake_case_           =				torch.utils.checkpoint.checkpoint(
																													    create_custom_forward(a   )			,     a			,     use_reentrant=a   )
																						# middle
																						snake_case_           =				torch.utils.checkpoint.checkpoint(
																						    create_custom_forward(self.mid_block   )			,     a			,     use_reentrant=a   )
															else:
																						for down_block in self.down_blocks:
																													snake_case_           =				torch.utils.checkpoint.checkpoint(create_custom_forward(a   )			,     a   )
																						# middle
																						snake_case_           =				torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block   )			,     a   )

								else:
															# down
															for down_block in self.down_blocks:
																						snake_case_           =				down_block(a   )

															# middle
															snake_case_           =				self.mid_block(a   )

								# post-process
								snake_case_           =				self.conv_norm_out(a   )
								snake_case_           =				self.conv_act(a   )
								snake_case_           =				self.conv_out(a   )

								return sample






class 					UpperCamelCase_ (     nn.Module							):



	'''simple docstring'''




	def __init__(			self			,     a=3			,     a=3			,     a=("UpDecoderBlock2D",)			,     a=(64,)			,     a=2			,     a=32			,     a="silu"			,     a="group"			,     )			->       List[str]:
								super().__init__()
								snake_case_           =				layers_per_block

								snake_case_           =				nn.Convad(
								    a			,     block_out_channels[-1]			,     kernel_size=3			,     stride=1			,     padding=1			,     )

								snake_case_           =				None
								snake_case_           =				nn.ModuleList([]   )

								snake_case_           =				in_channels if norm_type == 'spatial' else None

								# mid
								snake_case_           =				UNetMidBlockaD(
								    in_channels=block_out_channels[-1]			,     resnet_eps=1E-6			,     resnet_act_fn=a			,     output_scale_factor=1			,     resnet_time_scale_shift='default' if norm_type == 'group' else norm_type			,     attention_head_dim=block_out_channels[-1]			,     resnet_groups=a			,     temb_channels=a			,     )

								# up
								snake_case_           =				list(reversed(a   )   )
								snake_case_           =				reversed_block_out_channels[0]
								for i, up_block_type in enumerate(a   ):
															snake_case_           =				output_channel
															snake_case_           =				reversed_block_out_channels[i]

															snake_case_           =				i == len(a   ) - 1

															snake_case_           =				get_up_block(
															    a			,     num_layers=self.layers_per_block + 1			,     in_channels=a			,     out_channels=a			,     prev_output_channel=a			,     add_upsample=not is_final_block			,     resnet_eps=1E-6			,     resnet_act_fn=a			,     resnet_groups=a			,     attention_head_dim=a			,     temb_channels=a			,     resnet_time_scale_shift=a			,     )
															self.up_blocks.append(a   )
															snake_case_           =				output_channel

								# out
								if norm_type == "spatial":
															snake_case_           =				SpatialNorm(block_out_channels[0]			,     a   )
								else:
															snake_case_           =				nn.GroupNorm(num_channels=block_out_channels[0]			,     num_groups=a			,     eps=1E-6   )
								snake_case_           =				nn.SiLU()
								snake_case_           =				nn.Convad(block_out_channels[0]			,     a			,     3			,     padding=1   )

								snake_case_           =				False






	def 				_UpperCamelCase     (			self			,     a			,     a=None   )			->       Any:
								snake_case_           =				z
								snake_case_           =				self.conv_in(a   )

								snake_case_           =				next(iter(self.up_blocks.parameters()   )   ).dtype
								if self.training and self.gradient_checkpointing:

															def create_custom_forward(a   ):
																						def custom_forward(*a   ):
																													return module(*a   )

																						return custom_forward

															if is_torch_version('>='			,     '1.11.0'   ):
																						# middle
																						snake_case_           =				torch.utils.checkpoint.checkpoint(
																						    create_custom_forward(self.mid_block   )			,     a			,     a			,     use_reentrant=a   )
																						snake_case_           =				sample.to(a   )

																						# up
																						for up_block in self.up_blocks:
																													snake_case_           =				torch.utils.checkpoint.checkpoint(
																													    create_custom_forward(a   )			,     a			,     a			,     use_reentrant=a   )
															else:
																						# middle
																						snake_case_           =				torch.utils.checkpoint.checkpoint(
																						    create_custom_forward(self.mid_block   )			,     a			,     a   )
																						snake_case_           =				sample.to(a   )

																						# up
																						for up_block in self.up_blocks:
																													snake_case_           =				torch.utils.checkpoint.checkpoint(create_custom_forward(a   )			,     a			,     a   )
								else:
															# middle
															snake_case_           =				self.mid_block(a			,     a   )
															snake_case_           =				sample.to(a   )

															# up
															for up_block in self.up_blocks:
																						snake_case_           =				up_block(a			,     a   )

        # post-process
								if latent_embeds is None:
															snake_case_           =				self.conv_norm_out(a   )
								else:
															snake_case_           =				self.conv_norm_out(a			,     a   )
								snake_case_           =				self.conv_act(a   )
								snake_case_           =				self.conv_out(a   )

								return sample






class 					UpperCamelCase_ (     nn.Module							):



	'''simple docstring'''




	def __init__(			self			,     a			,     a			,     a			,     a=None			,     a="random"			,     a=False			,     a=True   )			->       Optional[int]:
								super().__init__()
								snake_case_           =				n_e
								snake_case_           =				vq_embed_dim
								snake_case_           =				beta
								snake_case_           =				legacy

								snake_case_           =				nn.Embedding(self.n_e			,     self.vq_embed_dim   )
								self.embedding.weight.data.uniform_(-1.0 / self.n_e			,     1.0 / self.n_e   )

								snake_case_           =				remap
								if self.remap is not None:
															self.register_buffer('used'			,     torch.tensor(np.load(self.remap   )   )   )
															snake_case_           =				self.used.shape[0]
															snake_case_           =				unknown_index  # "random" or "extra" or integer
															if self.unknown_index == "extra":
																						snake_case_           =				self.re_embed
																						snake_case_           =				self.re_embed + 1
															print(
															    F'''Remapping {self.n_e} indices to {self.re_embed} indices. '''
															    F'''Using {self.unknown_index} for unknown indices.'''   )
								else:
															snake_case_           =				n_e

								snake_case_           =				sane_index_shape




	def 				_UpperCamelCase     (			self			,     a   )			->       Tuple:
								snake_case_           =				inds.shape
								assert len(a   ) > 1
								snake_case_           =				inds.reshape(ishape[0]			,     -1   )
								snake_case_           =				self.used.to(a   )
								snake_case_           =				(inds[:, :, None] == used[None, None, ...]).long()
								snake_case_           =				match.argmax(-1   )
								snake_case_           =				match.sum(2   ) < 1
								if self.unknown_index == "random":
															snake_case_           =				torch.randint(0			,     self.re_embed			,     size=new[unknown].shape   ).to(device=new.device   )
								else:
															snake_case_           =				self.unknown_index
								return new.reshape(a   )




	def 				_UpperCamelCase     (			self			,     a   )			->       int:
								snake_case_           =				inds.shape
								assert len(a   ) > 1
								snake_case_           =				inds.reshape(ishape[0]			,     -1   )
								snake_case_           =				self.used.to(a   )
								if self.re_embed > self.used.shape[0]:  # extra token
															snake_case_           =				0  # simply set to zero
								snake_case_           =				torch.gather(used[None, :][inds.shape[0] * [0], :]			,     1			,     a   )
								return back.reshape(a   )




	def 				_UpperCamelCase     (			self			,     a   )			->       Dict:
								# reshape z -> (batch, height, width, channel) and flatten
								snake_case_           =				z.permute(0			,     2			,     3			,     1   ).contiguous()
								snake_case_           =				z.view(-1			,     self.vq_embed_dim   )

								# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
								snake_case_           =				torch.argmin(torch.cdist(a			,     self.embedding.weight   )			,     dim=1   )

								snake_case_           =				self.embedding(a   ).view(z.shape   )
								snake_case_           =				None
								snake_case_           =				None

								# compute loss for embedding
								if not self.legacy:
															snake_case_           =				self.beta * torch.mean((z_q.detach() - z) ** 2   ) + torch.mean((z_q - z.detach()) ** 2   )
								else:
															snake_case_           =				torch.mean((z_q.detach() - z) ** 2   ) + self.beta * torch.mean((z_q - z.detach()) ** 2   )

								# preserve gradients
								snake_case_           =				z + (z_q - z).detach()

								# reshape back to match original input shape
								snake_case_           =				z_q.permute(0			,     3			,     1			,     2   ).contiguous()

								if self.remap is not None:
															snake_case_           =				min_encoding_indices.reshape(z.shape[0]			,     -1   )  # add batch axis
															snake_case_           =				self.remap_to_used(a   )
															snake_case_           =				min_encoding_indices.reshape(-1			,     1   )  # flatten

								if self.sane_index_shape:
															snake_case_           =				min_encoding_indices.reshape(z_q.shape[0]			,     z_q.shape[2]			,     z_q.shape[3]   )

								return z_q, loss, (perplexity, min_encodings, min_encoding_indices)






	def 				_UpperCamelCase     (			self			,     a			,     a   )			->       str:
								# shape specifying (batch, height, width, channel)
								if self.remap is not None:
															snake_case_           =				indices.reshape(shape[0]			,     -1   )  # add batch axis
															snake_case_           =				self.unmap_to_all(a   )
															snake_case_           =				indices.reshape(-1   )  # flatten again

								# get quantized latent vectors
								snake_case_           =				self.embedding(a   )

								if shape is not None:
															snake_case_           =				z_q.view(a   )
															# reshape back to match original input shape
															snake_case_           =				z_q.permute(0			,     3			,     1			,     2   ).contiguous()

								return z_q






class 					UpperCamelCase_ (     snake_case_							):



	'''simple docstring'''




	def __init__(			self			,     a			,     a=False   )			->       Any:
								snake_case_           =				parameters
								snake_case_     ,						snake_case_           =				torch.chunk(a			,     2			,     dim=1   )
								snake_case_           =				torch.clamp(self.logvar			,     -30.0			,     20.0   )
								snake_case_           =				deterministic
								snake_case_           =				torch.exp(0.5 * self.logvar   )
								snake_case_           =				torch.exp(self.logvar   )
								if self.deterministic:
															snake_case_           =				snake_case_           =				torch.zeros_like(
															    self.mean			,     device=self.parameters.device			,     dtype=self.parameters.dtype   )




	def 				_UpperCamelCase     (			self			,     a = None   )			->       torch.FloatTensor:
								# make sure sample is on the same device as the parameters and has same dtype
								snake_case_           =				randn_tensor(
								    self.mean.shape			,     generator=a			,     device=self.parameters.device			,     dtype=self.parameters.dtype   )
								snake_case_           =				self.mean + self.std * sample
								return x




	def 				_UpperCamelCase     (			self			,     a=None   )			->       Dict:
								if self.deterministic:
															return torch.Tensor([0.0]   )
								else:
															if other is None:
																						return 0.5 * torch.sum(torch.pow(self.mean			,     2   ) + self.var - 1.0 - self.logvar			,     dim=[1, 2, 3]   )
															else:
																						return 0.5 * torch.sum(
																						    torch.pow(self.mean - other.mean			,     2   ) / other.var
																						    + self.var / other.var
																						    - 1.0
																						    - self.logvar
																						    + other.logvar			,     dim=[1, 2, 3]			,     )




	def 				_UpperCamelCase     (			self			,     a			,     a=[1, 2, 3]   )			->       int:
								if self.deterministic:
															return torch.Tensor([0.0]   )
								snake_case_           =				np.log(2.0 * np.pi   )
								return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean			,     2   ) / self.var			,     dim=a   )






	def 				_UpperCamelCase     (			self   )			->       Tuple:
								return self.mean













from typing import Any, Callable, Dict, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker


lowercase    				=   "CompVis/stable-diffusion-v1-1"
lowercase    				=   "CompVis/stable-diffusion-v1-2"
lowercase    				=   "CompVis/stable-diffusion-v1-3"
lowercase    				=   "CompVis/stable-diffusion-v1-4"




class 					UpperCamelCase_ (     snake_case_							):



	'''simple docstring'''




	def __init__(			self			,     a			,     a			,     a			,     a			,     a			,     a			,     a			,     a = True			,     )			->       int:
								super()._init_()

								snake_case_           =				StableDiffusionPipeline.from_pretrained(a   )
								snake_case_           =				StableDiffusionPipeline.from_pretrained(a   )
								snake_case_           =				StableDiffusionPipeline.from_pretrained(a   )
								snake_case_           =				StableDiffusionPipeline(
								    vae=a			,     text_encoder=a			,     tokenizer=a			,     unet=a			,     scheduler=a			,     safety_checker=a			,     feature_extractor=a			,     requires_safety_checker=a			,     )

								self.register_modules(pipelinea=self.pipea			,     pipelinea=self.pipea			,     pipelinea=self.pipea			,     pipelinea=self.pipea   )




	@property
	def 				_UpperCamelCase     (			self   )			->       Dict[str, Any]:
								return {k: getattr(self			,     a   ) for k in self.config.keys() if not k.startswith('_'   )}




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




	def 				_UpperCamelCase     (			self   )			->       List[Any]:
								self.enable_attention_slicing(a   )




	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       Optional[Any]:
								return self.pipea(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )




	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       int:
								return self.pipea(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )




	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       Optional[int]:
								return self.pipea(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )




	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       Any:
								return self.pipea(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )






	@torch.no_grad()
	def 				_UpperCamelCase     (			self			,     a			,     a = 5_12			,     a = 5_12			,     a = 50			,     a = 7.5			,     a = None			,     a = 1			,     a = 0.0			,     a = None			,     a = None			,     a = "pil"			,     a = True			,     a = None			,     a = 1			,     **a			,     )			->       Optional[Any]:
								snake_case_           =				'cuda' if torch.cuda.is_available() else 'cpu'
								self.to(a   )

								# Checks if the height and width are divisible by 8 or not
								if height % 8 != 0 or width % 8 != 0:
															raise ValueError(F'''`height` and `width` must be divisible by 8 but are {height} and {width}.'''   )

								# Get first result from Stable Diffusion Checkpoint v1.1
								snake_case_           =				self.textaimg_sda_a(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )

								# Get first result from Stable Diffusion Checkpoint v1.2
								snake_case_           =				self.textaimg_sda_a(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )

								# Get first result from Stable Diffusion Checkpoint v1.3
								snake_case_           =				self.textaimg_sda_a(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )

								# Get first result from Stable Diffusion Checkpoint v1.4
								snake_case_           =				self.textaimg_sda_a(
								    prompt=a			,     height=a			,     width=a			,     num_inference_steps=a			,     guidance_scale=a			,     negative_prompt=a			,     num_images_per_prompt=a			,     eta=a			,     generator=a			,     latents=a			,     output_type=a			,     return_dict=a			,     callback=a			,     callback_steps=a			,     **a			,     )

								# Get all result images into a single list and pass it via StableDiffusionPipelineOutput for final result
								return StableDiffusionPipelineOutput([resa[0], resa[0], resa[0], resa[0]]   )





