

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 				lowerCamelCase__							(       UpperCamelCase__):




	'''simple docstring'''




	_A			=  4_2




class 				lowerCamelCase__							(       nn.Module):




	'''simple docstring'''




	def __init__(      self	:str  ,     a	:Optional[Any]=3  ,     a	:Any=3  ,     a	:Optional[int]=("DownEncoderBlock2D",)  ,     a	:Union[str, Any]=(6_4,)  ,     a	:int=2  ,     a	:Any=3_2  ,     a	:List[str]="silu"  ,     a	:str=True  ,     )	->				Any:
								super().__init__()
								__UpperCamelCase			:     Optional[int]         =      layers_per_block

								__UpperCamelCase			:     Tuple         =      torch.nn.Convad(
								    _a  ,     block_out_channels[0]  ,     kernel_size=3  ,     stride=1  ,     padding=1  ,     )

								__UpperCamelCase			:     str         =      None
								__UpperCamelCase			:     int         =      nn.ModuleList([]		)

								# down
								__UpperCamelCase			:     Tuple         =      block_out_channels[0]
								for i, down_block_type in enumerate(_a		):
															__UpperCamelCase			:     Optional[int]         =      output_channel
															__UpperCamelCase			:     List[Any]         =      block_out_channels[i]
															__UpperCamelCase			:     List[Any]         =      i == len(_a		) - 1

															__UpperCamelCase			:     Union[str, Any]         =      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
								__UpperCamelCase			:     List[str]         =      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
								__UpperCamelCase			:     Dict         =      nn.GroupNorm(num_channels=block_out_channels[-1]  ,     num_groups=_a  ,     eps=1E-6		)
								__UpperCamelCase			:     Optional[int]         =      nn.SiLU()

								__UpperCamelCase			:     List[str]         =      2 * out_channels if double_z else out_channels
								__UpperCamelCase			:     List[str]         =      nn.Convad(block_out_channels[-1]  ,     _a  ,     3  ,     padding=1		)

								__UpperCamelCase			:     int         =      False







	def      _lowerCamelCase       (      self	:Any  ,     a	:List[Any]		)	->				Optional[Any]:
								__UpperCamelCase			:     Dict         =      x
								__UpperCamelCase			:     Tuple         =      self.conv_in(_a		)

								if self.training and self.gradient_checkpointing:

															def create_custom_forward(a	:Optional[Any]		):
																						def custom_forward(*a	:Optional[int]		):
																													return module(*_a		)

																						return custom_forward

															# down
															if is_torch_version(">="  ,     "1.11.0"		):
																						for down_block in self.down_blocks:
																													__UpperCamelCase			:     Tuple         =      torch.utils.checkpoint.checkpoint(
																													    create_custom_forward(_a		)  ,     _a  ,     use_reentrant=_a		)
																						# middle
																						__UpperCamelCase			:     Optional[int]         =      torch.utils.checkpoint.checkpoint(
																						    create_custom_forward(self.mid_block		)  ,     _a  ,     use_reentrant=_a		)
															else:
																						for down_block in self.down_blocks:
																													__UpperCamelCase			:     Optional[Any]         =      torch.utils.checkpoint.checkpoint(create_custom_forward(_a		)  ,     _a		)
																						# middle
																						__UpperCamelCase			:     Union[str, Any]         =      torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block		)  ,     _a		)

								else:
															# down
															for down_block in self.down_blocks:
																						__UpperCamelCase			:     Optional[int]         =      down_block(_a		)

															# middle
															__UpperCamelCase			:     int         =      self.mid_block(_a		)

								# post-process
								__UpperCamelCase			:     Tuple         =      self.conv_norm_out(_a		)
								__UpperCamelCase			:     Optional[int]         =      self.conv_act(_a		)
								__UpperCamelCase			:     Optional[int]         =      self.conv_out(_a		)

								return sample




class 				lowerCamelCase__							(       nn.Module):




	'''simple docstring'''




	def __init__(      self	:Tuple  ,     a	:List[str]=3  ,     a	:List[Any]=3  ,     a	:List[Any]=("UpDecoderBlock2D",)  ,     a	:Union[str, Any]=(6_4,)  ,     a	:int=2  ,     a	:Optional[int]=3_2  ,     a	:List[Any]="silu"  ,     a	:str="group"  ,     )	->				Optional[int]:
								super().__init__()
								__UpperCamelCase			:     str         =      layers_per_block

								__UpperCamelCase			:     str         =      nn.Convad(
								    _a  ,     block_out_channels[-1]  ,     kernel_size=3  ,     stride=1  ,     padding=1  ,     )

								__UpperCamelCase			:     Optional[int]         =      None
								__UpperCamelCase			:     int         =      nn.ModuleList([]		)

								__UpperCamelCase			:     Dict         =      in_channels if norm_type == """spatial""" else None

								# mid
								__UpperCamelCase			:     Tuple         =      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
								__UpperCamelCase			:     Union[str, Any]         =      list(reversed(_a		)		)
								__UpperCamelCase			:     int         =      reversed_block_out_channels[0]
								for i, up_block_type in enumerate(_a		):
															__UpperCamelCase			:     List[Any]         =      output_channel
															__UpperCamelCase			:     Tuple         =      reversed_block_out_channels[i]

															__UpperCamelCase			:     Optional[Any]         =      i == len(_a		) - 1

															__UpperCamelCase			:     Optional[int]         =      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		)
															__UpperCamelCase			:     str         =      output_channel

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

								__UpperCamelCase			:     int         =      False







	def      _lowerCamelCase       (      self	:Any  ,     a	:Dict  ,     a	:Optional[Any]=None		)	->				Tuple:
								__UpperCamelCase			:     Union[str, Any]         =      z
								__UpperCamelCase			:     int         =      self.conv_in(_a		)

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

															def create_custom_forward(a	:str		):
																						def custom_forward(*a	:Any		):
																													return module(*_a		)

																						return custom_forward

															if is_torch_version(">="  ,     "1.11.0"		):
																						# middle
																						__UpperCamelCase			:     List[str]         =      torch.utils.checkpoint.checkpoint(
																						    create_custom_forward(self.mid_block		)  ,     _a  ,     _a  ,     use_reentrant=_a		)
																						__UpperCamelCase			:     List[Any]         =      sample.to(_a		)

																						# up
																						for up_block in self.up_blocks:
																													__UpperCamelCase			:     Optional[int]         =      torch.utils.checkpoint.checkpoint(
																													    create_custom_forward(_a		)  ,     _a  ,     _a  ,     use_reentrant=_a		)
															else:
																						# middle
																						__UpperCamelCase			:     int         =      torch.utils.checkpoint.checkpoint(
																						    create_custom_forward(self.mid_block		)  ,     _a  ,     _a		)
																						__UpperCamelCase			:     int         =      sample.to(_a		)

																						# up
																						for up_block in self.up_blocks:
																													__UpperCamelCase			:     int         =      torch.utils.checkpoint.checkpoint(create_custom_forward(_a		)  ,     _a  ,     _a		)
								else:
															# middle
															__UpperCamelCase			:     Any         =      self.mid_block(_a  ,     _a		)
															__UpperCamelCase			:     Union[str, Any]         =      sample.to(_a		)

															# up
															for up_block in self.up_blocks:
																						__UpperCamelCase			:     Union[str, Any]         =      up_block(_a  ,     _a		)

        # post-process
								if latent_embeds is None:
															__UpperCamelCase			:     Any         =      self.conv_norm_out(_a		)
								else:
															__UpperCamelCase			:     Union[str, Any]         =      self.conv_norm_out(_a  ,     _a		)
								__UpperCamelCase			:     str         =      self.conv_act(_a		)
								__UpperCamelCase			:     Tuple         =      self.conv_out(_a		)

								return sample




class 				lowerCamelCase__							(       nn.Module):




	'''simple docstring'''




	def __init__(      self	:List[Any]  ,     a	:Union[str, Any]  ,     a	:Optional[int]  ,     a	:int  ,     a	:Any=None  ,     a	:Tuple="random"  ,     a	:Tuple=False  ,     a	:List[str]=True		)	->				Any:
								super().__init__()
								__UpperCamelCase			:     Tuple         =      n_e
								__UpperCamelCase			:     Optional[Any]         =      vq_embed_dim
								__UpperCamelCase			:     Tuple         =      beta
								__UpperCamelCase			:     Dict         =      legacy

								__UpperCamelCase			:     List[str]         =      nn.Embedding(self.n_e  ,     self.vq_embed_dim		)
								self.embedding.weight.data.uniform_(-1.0 / self.n_e  ,     1.0 / self.n_e		)

								__UpperCamelCase			:     Tuple         =      remap
								if self.remap is not None:
															self.register_buffer("used"  ,     torch.tensor(np.load(self.remap		)		)		)
															__UpperCamelCase			:     int         =      self.used.shape[0]
															__UpperCamelCase			:     Optional[int]         =      unknown_index  # "random" or "extra" or integer
															if self.unknown_index == "extra":
																						__UpperCamelCase			:     Optional[int]         =      self.re_embed
																						__UpperCamelCase			:     Optional[Any]         =      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:
															__UpperCamelCase			:     Dict         =      n_e

								__UpperCamelCase			:     Dict         =      sane_index_shape







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







	def      _lowerCamelCase       (      self	:str  ,     a	:Union[str, Any]		)	->				Dict:
								__UpperCamelCase			:     Tuple         =      inds.shape
								assert len(_a		) > 1
								__UpperCamelCase			:     List[Any]         =      inds.reshape(ishape[0]  ,     -1		)
								__UpperCamelCase			:     Union[str, Any]         =      self.used.to(_a		)
								if self.re_embed > self.used.shape[0]:  # extra token
															__UpperCamelCase			:     str         =      0  # simply set to zero
								__UpperCamelCase			:     Any         =      torch.gather(used[None, :][inds.shape[0] * [0], :]  ,     1  ,     _a		)
								return back.reshape(_a		)







	def      _lowerCamelCase       (      self	:List[Any]  ,     a	:List[str]		)	->				str:
								# reshape z -> (batch, height, width, channel) and flatten
								__UpperCamelCase			:     Any         =      z.permute(0  ,     2  ,     3  ,     1		).contiguous()
								__UpperCamelCase			:     int         =      z.view(-1  ,     self.vq_embed_dim		)

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

								__UpperCamelCase			:     Any         =      self.embedding(_a		).view(z.shape		)
								__UpperCamelCase			:     Optional[Any]         =      None
								__UpperCamelCase			:     Any         =      None

								# compute loss for embedding
								if not self.legacy:
															__UpperCamelCase			:     List[str]         =      self.beta * torch.mean((z_q.detach() - z) ** 2		) + torch.mean((z_q - z.detach()) ** 2		)
								else:
															__UpperCamelCase			:     Any         =      torch.mean((z_q.detach() - z) ** 2		) + self.beta * torch.mean((z_q - z.detach()) ** 2		)

								# preserve gradients
								__UpperCamelCase			:     List[str]         =      z + (z_q - z).detach()

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

								if self.remap is not None:
															__UpperCamelCase			:     str         =      min_encoding_indices.reshape(z.shape[0]  ,     -1		)  # add batch axis
															__UpperCamelCase			:     int         =      self.remap_to_used(_a		)
															__UpperCamelCase			:     List[Any]         =      min_encoding_indices.reshape(-1  ,     1		)  # flatten

								if self.sane_index_shape:
															__UpperCamelCase			:     int         =      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      _lowerCamelCase       (      self	:Any  ,     a	:Union[str, Any]  ,     a	:Optional[Any]		)	->				List[str]:
								# shape specifying (batch, height, width, channel)
								if self.remap is not None:
															__UpperCamelCase			:     List[str]         =      indices.reshape(shape[0]  ,     -1		)  # add batch axis
															__UpperCamelCase			:     Optional[int]         =      self.unmap_to_all(_a		)
															__UpperCamelCase			:     Any         =      indices.reshape(-1		)  # flatten again

								# get quantized latent vectors
								__UpperCamelCase			:     str         =      self.embedding(_a		)

								if shape is not None:
															__UpperCamelCase			:     int         =      z_q.view(_a		)
															# reshape back to match original input shape
															__UpperCamelCase			:     Optional[Any]         =      z_q.permute(0  ,     3  ,     1  ,     2		).contiguous()

								return z_q




class 				lowerCamelCase__							(       UpperCamelCase__):




	'''simple docstring'''




	def __init__(      self	:Union[str, Any]  ,     a	:List[Any]  ,     a	:Optional[int]=False		)	->				Any:
								__UpperCamelCase			:     Dict         =      parameters
								__UpperCamelCase			:     Tuple         =      torch.chunk(_a  ,     2  ,     dim=1		)
								__UpperCamelCase			:     Optional[Any]         =      torch.clamp(self.logvar  ,     -30.0  ,     20.0		)
								__UpperCamelCase			:     Tuple         =      deterministic
								__UpperCamelCase			:     Any         =      torch.exp(0.5 * self.logvar		)
								__UpperCamelCase			:     Optional[int]         =      torch.exp(self.logvar		)
								if self.deterministic:
															__UpperCamelCase			:     List[Any]         =      torch.zeros_like(
															    self.mean  ,     device=self.parameters.device  ,     dtype=self.parameters.dtype		)







	def      _lowerCamelCase       (      self	:Any  ,     a	:int = None		)	->				torch.FloatTensor:
								# make sure sample is on the same device as the parameters and has same dtype
								__UpperCamelCase			:     Union[str, Any]         =      randn_tensor(
								    self.mean.shape  ,     generator=_a  ,     device=self.parameters.device  ,     dtype=self.parameters.dtype		)
								__UpperCamelCase			:     int         =      self.mean + self.std * sample
								return x







	def      _lowerCamelCase       (      self	:List[str]  ,     a	:str=None		)	->				Tuple:
								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      _lowerCamelCase       (      self	:List[Any]  ,     a	:Tuple  ,     a	:str=[1, 2, 3]		)	->				Optional[Any]:
								if self.deterministic:
															return torch.Tensor([0.0]		)
								__UpperCamelCase			:     List[str]         =      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      _lowerCamelCase       (      self	:List[Any]		)	->				str:
								return self.mean

import argparse

from transformers import CLIPImageProcessor, CLIPVisionModelWithProjection

from diffusers import UnCLIPImageVariationPipeline, UnCLIPPipeline


if __name__ == "__main__":
		lowercase						:      List[str]									= argparse.ArgumentParser()

		parser.add_argument('--dump_path', default=None, type=str, required=True, help='Path to the output model.')

		parser.add_argument(
		    '--txt2img_unclip',
		    default='kakaobrain/karlo-v1-alpha',
		    type=str,
		    required=False,
		    help='The pretrained txt2img unclip.',
		)

		lowercase						:      int									= parser.parse_args()

		lowercase						:      str									= UnCLIPPipeline.from_pretrained(args.txtaimg_unclip)

		lowercase						:      int									= CLIPImageProcessor()
		lowercase						:      Optional[int]									= CLIPVisionModelWithProjection.from_pretrained('openai/clip-vit-large-patch14')

		lowercase						:      int									= UnCLIPImageVariationPipeline(
		    decoder=txtaimg.decoder,
		    text_encoder=txtaimg.text_encoder,
		    tokenizer=txtaimg.tokenizer,
		    text_proj=txtaimg.text_proj,
		    feature_extractor=feature_extractor,
		    image_encoder=image_encoder,
		    super_res_first=txtaimg.super_res_first,
		    super_res_last=txtaimg.super_res_last,
		    decoder_scheduler=txtaimg.decoder_scheduler,
		    super_res_scheduler=txtaimg.super_res_scheduler,
		)

		imgaimg.save_pretrained(args.dump_path)