


from dataclasses import asdict, dataclass
from typing import Optional

from ...configuration_utils import PretrainedConfig
from ...utils import logging


snake_case__			 =						logging.get_logger(__name__)

# TODO Update this
snake_case__			 =						{
    '''facebook/esm-1b''': '''https://huggingface.co/facebook/esm-1b/resolve/main/config.json''',
    # See all ESM models at https://huggingface.co/models?filter=esm
}


class 					lowerCAmelCase_ (       _a):
						lowerCamelCase_         =     'esm'




						def __init__(	self						:   Optional[Any]   ,     __A						:   Optional[int]=None   ,     __A						:   Union[str, Any]=None   ,     __A						:   str=None   ,     __A						:   List[str]=768   ,     __A						:   Tuple=12   ,     __A						:   Optional[Any]=12   ,     __A						:   int=3072   ,     __A						:   Optional[Any]=0.1   ,     __A						:   str=0.1   ,     __A						:   Optional[int]=1026   ,     __A						:   List[str]=0.02   ,     __A						:   str=1E-12   ,     __A						:   Optional[int]="absolute"   ,     __A						:   Optional[int]=True   ,     __A						:   Dict=None   ,     __A						:   Dict=False   ,     __A						:   Dict=False   ,     __A						:   int=None   ,     __A						:   Any=None   ,     **__A						:   Dict   ,     )      ->Optional[Any]:






									"""simple docstring"""






									super().__init__(pad_token_id=__A   ,     mask_token_id=__A   ,     **__A						)

									a__ :Union[str, Any]						  =    vocab_size
									a__ :Dict						  =    hidden_size
									a__ :Tuple						  =    num_hidden_layers
									a__ :Optional[int]						  =    num_attention_heads
									a__ :Optional[Any]						  =    intermediate_size
									a__ :Dict						  =    hidden_dropout_prob
									a__ :Optional[Any]						  =    attention_probs_dropout_prob
									a__ :Any						  =    max_position_embeddings
									a__ :str						  =    initializer_range
									a__ :str						  =    layer_norm_eps
									a__ :Union[str, Any]						  =    position_embedding_type
									a__ :Dict						  =    use_cache
									a__ :Optional[Any]						  =    emb_layer_norm_before
									a__ :Dict						  =    token_dropout
									a__ :str						  =    is_folding_model
									if is_folding_model:
												if esmfold_config is None:
															logger.info("No esmfold_config supplied for folding model, using default values."						)
															a__ :List[Any]						  =    EsmFoldConfig()
												elif isinstance(__A   ,     __A						):
															a__ :int						  =    EsmFoldConfig(**__A						)
												a__ :Tuple						  =    esmfold_config
												if vocab_list is None:
															logger.warning("No vocab_list supplied for folding model, assuming the ESM-2 vocabulary!"						)
															a__ :str						  =    get_default_vocab_list()
												else:
															a__ :Union[str, Any]						  =    vocab_list
									else:
												a__ :Union[str, Any]						  =    None
												a__ :Union[str, Any]						  =    None
									if self.esmfold_config is not None and getattr(self.esmfold_config   ,     "use_esm_attn_map"   ,     __A						):
												raise ValueError("The HuggingFace port of ESMFold does not support use_esm_attn_map at this time!"						)






						def  _snake_case					(	self						:   Dict						)      ->str:






									"""simple docstring"""






									a__ :Any						  =    super().to_dict()
									if isinstance(self.esmfold_config   ,     __A						):
												a__ :Tuple						  =    self.esmfold_config.to_dict()
									return output






@dataclass
class 					lowerCAmelCase_ :
						lowerCamelCase_         =     None
						lowerCamelCase_         =     True
						lowerCamelCase_         =     False
						lowerCamelCase_         =     False
						lowerCamelCase_         =     False
						lowerCamelCase_         =     0

						lowerCamelCase_         =     True
						lowerCamelCase_         =     False

						lowerCamelCase_         =     128
						lowerCamelCase_         =     None




						def  _snake_case					(	self						:   Dict						)      ->int:






									"""simple docstring"""






									if self.trunk is None:
												a__ :str						  =    TrunkConfig()
									elif isinstance(self.trunk   ,     __A						):
												a__ :Any						  =    TrunkConfig(**self.trunk						)






						def  _snake_case					(	self						:   Any						)      ->Any:






									"""simple docstring"""






									a__ :Tuple						  =    asdict(self						)
									a__ :str						  =    self.trunk.to_dict()
									return output






@dataclass
class 					lowerCAmelCase_ :
						lowerCamelCase_         =     48
						lowerCamelCase_         =     1_024
						lowerCamelCase_         =     128
						lowerCamelCase_         =     32
						lowerCamelCase_         =     32
						lowerCamelCase_         =     32
						lowerCamelCase_         =     0
						lowerCamelCase_         =     0
						lowerCamelCase_         =     False
						lowerCamelCase_         =     4
						lowerCamelCase_         =     128
						lowerCamelCase_         =     None




						def  _snake_case					(	self						:   Optional[Any]						)      ->Dict:






									"""simple docstring"""






									if self.structure_module is None:
												a__ :int						  =    StructureModuleConfig()
									elif isinstance(self.structure_module   ,     __A						):
												a__ :Any						  =    StructureModuleConfig(**self.structure_module						)

									if self.max_recycles <= 0:
												raise ValueError(F'''`max_recycles` should be positive, got {self.max_recycles}.'''						)
									if self.sequence_state_dim % self.sequence_state_dim != 0:
												raise ValueError(
												    "`sequence_state_dim` should be a round multiple of `sequence_state_dim`, got"
												    F''' {self.sequence_state_dim} and {self.sequence_state_dim}.'''						)
									if self.pairwise_state_dim % self.pairwise_state_dim != 0:
												raise ValueError(
												    "`pairwise_state_dim` should be a round multiple of `pairwise_state_dim`, got"
												    F''' {self.pairwise_state_dim} and {self.pairwise_state_dim}.'''						)

									a__ :Dict						  =    self.sequence_state_dim // self.sequence_head_width
									a__ :str						  =    self.pairwise_state_dim // self.pairwise_head_width

									if self.sequence_state_dim != sequence_num_heads * self.sequence_head_width:
												raise ValueError(
												    "`sequence_state_dim` should be equal to `sequence_num_heads * sequence_head_width, got"
												    F''' {self.sequence_state_dim} != {sequence_num_heads} * {self.sequence_head_width}.'''						)
									if self.pairwise_state_dim != pairwise_num_heads * self.pairwise_head_width:
												raise ValueError(
												    "`pairwise_state_dim` should be equal to `pairwise_num_heads * pairwise_head_width, got"
												    F''' {self.pairwise_state_dim} != {pairwise_num_heads} * {self.pairwise_head_width}.'''						)
									if self.pairwise_state_dim % 2 != 0:
												raise ValueError(F'''`pairwise_state_dim` should be even, got {self.pairwise_state_dim}.'''						)

									if self.dropout >= 0.4:
												raise ValueError(F'''`dropout` should not be greater than 0.4, got {self.dropout}.'''						)






						def  _snake_case					(	self						:   int						)      ->Dict:






									"""simple docstring"""






									a__ :Union[str, Any]						  =    asdict(self						)
									a__ :List[str]						  =    self.structure_module.to_dict()
									return output






@dataclass
class 					lowerCAmelCase_ :
						lowerCamelCase_         =     384
						lowerCamelCase_         =     128
						lowerCamelCase_         =     16
						lowerCamelCase_         =     128
						lowerCamelCase_         =     12
						lowerCamelCase_         =     4
						lowerCamelCase_         =     8
						lowerCamelCase_         =     0.1
						lowerCamelCase_         =     8
						lowerCamelCase_         =     1
						lowerCamelCase_         =     2
						lowerCamelCase_         =     7
						lowerCamelCase_         =     10
						lowerCamelCase_         =     1e-8
						lowerCamelCase_         =     1e5




						def  _snake_case					(	self						:   Union[str, Any]						)      ->int:






									"""simple docstring"""






									return asdict(self						)






def 				lowerCamelCase__	(      )    ->	Union[str, Any]:





			"""simple docstring"""
			return (
			    "<cls>",
			    "<pad>",
			    "<eos>",
			    "<unk>",
			    "L",
			    "A",
			    "G",
			    "V",
			    "S",
			    "E",
			    "R",
			    "T",
			    "I",
			    "D",
			    "P",
			    "K",
			    "Q",
			    "N",
			    "F",
			    "Y",
			    "M",
			    "H",
			    "W",
			    "C",
			    "X",
			    "B",
			    "U",
			    "Z",
			    "O",
			    ".",
			    "-",
			    "<null_1>",
			    "<mask>",
			)






import os
from glob import glob

import imageio
import torch
import torchvision
import wandb
from img_processing import custom_to_pil, loop_post_process, preprocess, preprocess_vqgan
from loaders import load_vqgan
from PIL import Image
from torch import nn

from transformers import CLIPModel, CLIPTokenizerFast
from utils import get_device, get_timestamp, show_pil


class 					lowerCAmelCase_ :




						def __init__(	self						:   str   ,     __A						:   str = "cpu"   ,     __A						:   str = "openai/clip-vit-large-patch14"						)      ->None:






									"""simple docstring"""






									a__ :List[str]						  =    device
									a__ :Optional[Any]						  =    CLIPTokenizerFast.from_pretrained(__A						)
									a__ :str						  =    [0.48_145_466, 0.4_578_275, 0.40_821_073]
									a__ :Any						  =    [0.26_862_954, 0.26_130_258, 0.27_577_711]
									a__ :Union[str, Any]						  =    torchvision.transforms.Normalize(self.image_mean   ,     self.image_std						)
									a__ :List[Any]						  =    torchvision.transforms.Resize(224						)
									a__ :Optional[Any]						  =    torchvision.transforms.CenterCrop(224						)




						def  _snake_case					(	self						:   int   ,     __A						:   int						)      ->List[Any]:






									"""simple docstring"""






									a__ :List[str]						  =    self.resize(__A						)
									a__ :Union[str, Any]						  =    self.center_crop(__A						)
									a__ :List[str]						  =    self.normalize(__A						)
									return images






						def __call__(	self						:   Union[str, Any]   ,     __A						:   Dict=None   ,     __A						:   List[Any]=None   ,     **__A						:   Union[str, Any]						)      ->Tuple:






									"""simple docstring"""






									a__ :Optional[int]						  =    self.tokenizer(text=__A   ,     **__A						)
									a__ :str						  =    self.preprocess_img(__A						)
									a__ :Any						  =    {key: value.to(self.device						) for (key, value) in encoding.items()}
									return encoding






class 					lowerCAmelCase_ (       nn.Module):




						def __init__(	self						:   List[str]   ,     __A						:   List[str]=10   ,     __A						:   List[Any]=0.01   ,     __A						:   Optional[int]=None   ,     __A						:   List[str]=None   ,     __A						:   Dict=None   ,     __A						:   List[str]=None   ,     __A						:   Tuple=None   ,     __A						:   str=None   ,     __A						:   Any=False   ,     __A						:   Union[str, Any]=True   ,     __A						:   Tuple="image"   ,     __A						:   Tuple=True   ,     __A						:   int=False   ,     __A						:   Tuple=False   ,     __A						:   Union[str, Any]=False   ,     )      ->None:






									"""simple docstring"""






									super().__init__()
									a__ :List[str]						  =    None
									a__ :int						  =    device if device else get_device()
									if vqgan:
												a__ :Any						  =    vqgan
									else:
												a__ :Optional[int]						  =    load_vqgan(self.device   ,     conf_path=__A   ,     ckpt_path=__A						)
									self.vqgan.eval()
									if clip:
												a__ :Dict						  =    clip
									else:
												a__ :Any						  =    CLIPModel.from_pretrained("openai/clip-vit-base-patch32"						)
									self.clip.to(self.device						)
									a__ :Optional[int]						  =    ProcessorGradientFlow(device=self.device						)

									a__ :Any						  =    iterations
									a__ :Any						  =    lr
									a__ :List[str]						  =    log
									a__ :List[Any]						  =    make_grid
									a__ :Optional[int]						  =    return_val
									a__ :Optional[Any]						  =    quantize
									a__ :Any						  =    self.vqgan.decoder.z_shape




						def  _snake_case					(	self						:   str   ,     __A						:   str=None   ,     __A						:   Optional[int]=None   ,     __A						:   int=5   ,     __A						:   Any=True						)      ->Optional[int]:






									"""simple docstring"""






									a__ :Optional[int]						  =    []
									if output_path is None:
												a__ :Any						  =    "./animation.gif"
									if input_path is None:
												a__ :Dict						  =    self.save_path
									a__ :Dict						  =    sorted(glob(input_path + "/*"						)						)
									if not len(__A						):
												raise ValueError(
												    "No images found in save path, aborting (did you pass save_intermediate=True to the generate"
												    " function?)"						)
									if len(__A						) == 1:
												print("Only one image found in save path, (did you pass save_intermediate=True to the generate function?)"						)
									a__ :str						  =    total_duration / len(__A						)
									a__ :str						  =    [frame_duration] * len(__A						)
									if extend_frames:
												a__ :Optional[Any]						  =    1.5
												a__ :Optional[Any]						  =    3
									for file_name in paths:
												if file_name.endswith(".png"						):
															images.append(imageio.imread(__A						)						)
									imageio.mimsave(__A   ,     __A   ,     duration=__A						)
									print(F'''gif saved to {output_path}'''						)




						def  _snake_case					(	self						:   str   ,     __A						:   Tuple=None   ,     __A						:   Optional[int]=None						)      ->Optional[int]:






									"""simple docstring"""






									if not (path or img):
												raise ValueError("Input either path or tensor"						)
									if img is not None:
												raise NotImplementedError
									a__ :List[Any]						  =    preprocess(Image.open(__A						)   ,     target_image_size=256						).to(self.device						)
									a__ :List[str]						  =    preprocess_vqgan(__A						)
									a__  ,  *a__ :Union[str, Any]						  =    self.vqgan.encode(__A						)
									return z




						def  _snake_case					(	self						:   List[str]   ,     __A						:   Any						)      ->Optional[int]:






									"""simple docstring"""






									a__ :Dict						  =    self.latent.detach().requires_grad_()
									a__ :int						  =    base_latent + transform_vector
									if self.quantize:
												a__  ,  *a__ :str						  =    self.vqgan.quantize(__A						)
									else:
												a__ :List[str]						  =    trans_latent
									return self.vqgan.decode(__A						)




						def  _snake_case					(	self						:   List[str]   ,     __A						:   Optional[Any]   ,     __A						:   Optional[Any]   ,     __A						:   List[Any]=None						)      ->str:






									"""simple docstring"""






									a__ :Dict						  =    self.clip_preprocessor(text=__A   ,     images=__A   ,     return_tensors="pt"   ,     padding=__A						)
									a__ :List[str]						  =    self.clip(**__A						)
									a__ :Optional[int]						  =    clip_outputs.logits_per_image
									if weights is not None:
												a__ :Tuple						  =    similarity_logits * weights
									return similarity_logits.sum()




						def  _snake_case					(	self						:   List[str]   ,     __A						:   List[Any]   ,     __A						:   Optional[int]   ,     __A						:   Optional[int]						)      ->int:






									"""simple docstring"""






									a__ :str						  =    self._get_clip_similarity(pos_prompts["prompts"]   ,     __A   ,     weights=(1 / pos_prompts["weights"])						)
									if neg_prompts:
												a__ :Optional[int]						  =    self._get_clip_similarity(neg_prompts["prompts"]   ,     __A   ,     weights=neg_prompts["weights"]						)
									else:
												a__ :Tuple						  =    torch.tensor([1]   ,     device=self.device						)
									a__ :Dict						  =    -torch.log(__A						) + torch.log(__A						)
									return loss




						def  _snake_case					(	self						:   int   ,     __A						:   Dict   ,     __A						:   Union[str, Any]   ,     __A						:   List[str]						)      ->List[str]:






									"""simple docstring"""






									a__ :List[Any]						  =    torch.randn_like(self.latent   ,     requires_grad=__A   ,     device=self.device						)
									a__ :Optional[Any]						  =    torch.optim.Adam([vector]   ,     lr=self.lr						)

									for i in range(self.iterations						):
												optim.zero_grad()
												a__ :Tuple						  =    self._add_vector(__A						)
												a__ :Dict						  =    loop_post_process(__A						)
												a__ :str						  =    self._get_CLIP_loss(__A   ,     __A   ,     __A						)
												print("CLIP loss"   ,     __A						)
												if self.log:
															wandb.log({"CLIP Loss": clip_loss}						)
												clip_loss.backward(retain_graph=__A						)
												optim.step()
												if self.return_val == "image":
															yield custom_to_pil(transformed_img[0]						)
												else:
															yield vector




						def  _snake_case					(	self						:   Union[str, Any]   ,     __A						:   Optional[Any]   ,     __A						:   Union[str, Any]   ,     __A						:   Optional[Any]						)      ->Optional[int]:






									"""simple docstring"""






									wandb.init(reinit=__A   ,     project="face-editor"						)
									wandb.config.update({"Positive Prompts": positive_prompts}						)
									wandb.config.update({"Negative Prompts": negative_prompts}						)
									wandb.config.update({"lr": self.lr, "iterations": self.iterations}						)
									if image_path:
												a__ :List[str]						  =    Image.open(__A						)
												a__ :Any						  =    image.resize((256, 256)						)
												wandb.log("Original Image"   ,     wandb.Image(__A						)						)




						def  _snake_case					(	self						:   Union[str, Any]   ,     __A						:   Union[str, Any]						)      ->int:






									"""simple docstring"""






									if not prompts:
												return []
									a__ :List[str]						  =    []
									a__ :Tuple						  =    []
									if isinstance(__A   ,     __A						):
												a__ :Tuple						  =    [prompt.strip() for prompt in prompts.split("|"						)]
									for prompt in prompts:
												if isinstance(__A   ,     (tuple, list)						):
															a__ :str						  =    prompt[0]
															a__ :List[Any]						  =    float(prompt[1]						)
												elif ":" in prompt:
															a__  ,  a__ :Tuple						  =    prompt.split(":"						)
															a__ :Dict						  =    float(__A						)
												else:
															a__ :int						  =    prompt
															a__ :Any						  =    1.0
												processed_prompts.append(__A						)
												weights.append(__A						)
									return {
									    "prompts": processed_prompts,
									    "weights": torch.tensor(__A   ,     device=self.device						),
									}






						def  _snake_case					(	self						:   Optional[Any]   ,     __A						:   Any   ,     __A						:   int=None   ,     __A						:   int=None   ,     __A						:   str=True   ,     __A						:   List[Any]=False   ,     __A						:   Optional[int]=True   ,     __A						:   Any=True   ,     __A						:   Optional[Any]=None   ,     )      ->Optional[Any]:






									"""simple docstring"""






									if image_path:
												a__ :Optional[int]						  =    self._get_latent(__A						)
									else:
												a__ :Dict						  =    torch.randn(self.latent_dim   ,     device=self.device						)
									if self.log:
												self._init_logging(__A   ,     __A   ,     __A						)

									assert pos_prompts, "You must provide at least one positive prompt."
									a__ :int						  =    self.process_prompts(__A						)
									a__ :Any						  =    self.process_prompts(__A						)
									if save_final and save_path is None:
												a__ :Dict						  =    os.path.join("./outputs/"   ,     "_".join(pos_prompts["prompts"]						)						)
									if not os.path.exists(__A						):
												os.makedirs(__A						)
									else:
												a__ :str						  =    save_path + "_" + get_timestamp()
												os.makedirs(__A						)
									a__ :Any						  =    save_path

									a__ :Optional[int]						  =    self.vqgan.decode(self.latent						)[0]
									if show_intermediate:
												print("Original Image"						)
												show_pil(custom_to_pil(__A						)						)

									a__ :List[str]						  =    loop_post_process(__A						)
									for iter, transformed_img in enumerate(self._optimize_CLIP(__A   ,     __A   ,     __A						)						):
												if show_intermediate:
															show_pil(__A						)
												if save_intermediate:
															transformed_img.save(os.path.join(self.save_path   ,     F'''iter_{iter:03d}.png'''						)						)
												if self.log:
															wandb.log({"Image": wandb.Image(__A						)}						)
									if show_final:
												show_pil(__A						)
									if save_final:
												transformed_img.save(os.path.join(self.save_path   ,     F'''iter_{iter:03d}_final.png'''						)						)


