






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     :   Dict ,							SCREAMING_SNAKE_CASE_     :   str = "cpu" ,							SCREAMING_SNAKE_CASE_     :   str = "openai/clip-vit-large-patch14"	):
											lowerCAmelCase__      			=						device
											lowerCAmelCase__      			=						CLIPTokenizerFast.from_pretrained(SCREAMING_SNAKE_CASE_	)
											lowerCAmelCase__      			=						[0.48_145_466, 0.4_578_275, 0.40_821_073]
											lowerCAmelCase__      			=						[0.26_862_954, 0.26_130_258, 0.27_577_711]
											lowerCAmelCase__      			=						torchvision.transforms.Normalize(self.image_mean ,							self.image_std	)
											lowerCAmelCase__      			=						torchvision.transforms.Resize(224	)
											lowerCAmelCase__      			=						torchvision.transforms.CenterCrop(224	)




							def 						__snake_case					(   self     :   Tuple ,							SCREAMING_SNAKE_CASE_     :   int	):
											lowerCAmelCase__      			=						self.resize(SCREAMING_SNAKE_CASE_	)
											lowerCAmelCase__      			=						self.center_crop(SCREAMING_SNAKE_CASE_	)
											lowerCAmelCase__      			=						self.normalize(SCREAMING_SNAKE_CASE_	)
											return images




							def __call__(   self     :   List[Any] ,							SCREAMING_SNAKE_CASE_     :   str=None ,							SCREAMING_SNAKE_CASE_     :   List[Any]=None ,							**SCREAMING_SNAKE_CASE_     :   List[str]	):
											lowerCAmelCase__      			=						self.tokenizer(text=SCREAMING_SNAKE_CASE_ ,							**SCREAMING_SNAKE_CASE_	)
											lowerCAmelCase__      			=						self.preprocess_img(SCREAMING_SNAKE_CASE_	)
											lowerCAmelCase__      			=						{key: value.to(self.device	) for (key, value) in encoding.items()}
											return encoding



class  lowerCAmelCase_       (				nn.Module					):




							def __init__(   self     :   Union[str, Any] ,							SCREAMING_SNAKE_CASE_     :   Tuple=10 ,							SCREAMING_SNAKE_CASE_     :   Any=0.01 ,							SCREAMING_SNAKE_CASE_     :   List[str]=None ,							SCREAMING_SNAKE_CASE_     :   List[Any]=None ,							SCREAMING_SNAKE_CASE_     :   List[str]=None ,							SCREAMING_SNAKE_CASE_     :   Optional[Any]=None ,							SCREAMING_SNAKE_CASE_     :   Tuple=None ,							SCREAMING_SNAKE_CASE_     :   Optional[Any]=None ,							SCREAMING_SNAKE_CASE_     :   str=False ,							SCREAMING_SNAKE_CASE_     :   Optional[int]=True ,							SCREAMING_SNAKE_CASE_     :   Any="image" ,							SCREAMING_SNAKE_CASE_     :   Optional[int]=True ,							SCREAMING_SNAKE_CASE_     :   Optional[Any]=False ,							SCREAMING_SNAKE_CASE_     :   Any=False ,							SCREAMING_SNAKE_CASE_     :   Union[str, Any]=False ,							):
											super().__init__()
											lowerCAmelCase__      			=						None
											lowerCAmelCase__      			=						device if device else get_device()
											if vqgan:
															lowerCAmelCase__      			=						vqgan
											else:
															lowerCAmelCase__      			=						load_vqgan(self.device ,							conf_path=SCREAMING_SNAKE_CASE_ ,							ckpt_path=SCREAMING_SNAKE_CASE_	)
											self.vqgan.eval()
											if clip:
															lowerCAmelCase__      			=						clip
											else:
															lowerCAmelCase__      			=						CLIPModel.from_pretrained('''openai/clip-vit-base-patch32'''	)
											self.clip.to(self.device	)
											lowerCAmelCase__      			=						ProcessorGradientFlow(device=self.device	)

											lowerCAmelCase__      			=						iterations
											lowerCAmelCase__      			=						lr
											lowerCAmelCase__      			=						log
											lowerCAmelCase__      			=						make_grid
											lowerCAmelCase__      			=						return_val
											lowerCAmelCase__      			=						quantize
											lowerCAmelCase__      			=						self.vqgan.decoder.z_shape




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




							def 						__snake_case					(   self     :   str ,							SCREAMING_SNAKE_CASE_     :   Dict=None ,							SCREAMING_SNAKE_CASE_     :   Any=None	):
											if not (path or img):
															raise ValueError('''Input either path or tensor'''	)
											if img is not None:
															raise NotImplementedError
											lowerCAmelCase__      			=						preprocess(Image.open(SCREAMING_SNAKE_CASE_	) ,							target_image_size=256	).to(self.device	)
											lowerCAmelCase__      			=						preprocess_vqgan(SCREAMING_SNAKE_CASE_	)
											lowerCAmelCase__			,		*lowerCAmelCase__      			=						self.vqgan.encode(SCREAMING_SNAKE_CASE_	)
											return z




							def 						__snake_case					(   self     :   List[Any] ,							SCREAMING_SNAKE_CASE_     :   Dict	):
											lowerCAmelCase__      			=						self.latent.detach().requires_grad_()
											lowerCAmelCase__      			=						base_latent + transform_vector
											if self.quantize:
															lowerCAmelCase__			,		*lowerCAmelCase__      			=						self.vqgan.quantize(SCREAMING_SNAKE_CASE_	)
											else:
															lowerCAmelCase__      			=						trans_latent
											return self.vqgan.decode(SCREAMING_SNAKE_CASE_	)




							def 						__snake_case					(   self     :   List[Any] ,							SCREAMING_SNAKE_CASE_     :   Any ,							SCREAMING_SNAKE_CASE_     :   str ,							SCREAMING_SNAKE_CASE_     :   List[Any]=None	):
											lowerCAmelCase__      			=						self.clip_preprocessor(text=SCREAMING_SNAKE_CASE_ ,							images=SCREAMING_SNAKE_CASE_ ,							return_tensors='''pt''' ,							padding=SCREAMING_SNAKE_CASE_	)
											lowerCAmelCase__      			=						self.clip(**SCREAMING_SNAKE_CASE_	)
											lowerCAmelCase__      			=						clip_outputs.logits_per_image
											if weights is not None:
															lowerCAmelCase__      			=						similarity_logits * weights
											return similarity_logits.sum()




							def 						__snake_case					(   self     :   Tuple ,							SCREAMING_SNAKE_CASE_     :   Optional[int] ,							SCREAMING_SNAKE_CASE_     :   Union[str, Any] ,							SCREAMING_SNAKE_CASE_     :   Any	):
											lowerCAmelCase__      			=						self._get_clip_similarity(pos_prompts['''prompts'''] ,							SCREAMING_SNAKE_CASE_ ,							weights=(1 / pos_prompts['''weights'''])	)
											if neg_prompts:
															lowerCAmelCase__      			=						self._get_clip_similarity(neg_prompts['''prompts'''] ,							SCREAMING_SNAKE_CASE_ ,							weights=neg_prompts['''weights''']	)
											else:
															lowerCAmelCase__      			=						torch.tensor([1] ,							device=self.device	)
											lowerCAmelCase__      			=						-torch.log(SCREAMING_SNAKE_CASE_	) + torch.log(SCREAMING_SNAKE_CASE_	)
											return loss




							def 						__snake_case					(   self     :   Any ,							SCREAMING_SNAKE_CASE_     :   Tuple ,							SCREAMING_SNAKE_CASE_     :   List[str] ,							SCREAMING_SNAKE_CASE_     :   Optional[int]	):
											lowerCAmelCase__      			=						torch.randn_like(self.latent ,							requires_grad=SCREAMING_SNAKE_CASE_ ,							device=self.device	)
											lowerCAmelCase__      			=						torch.optim.Adam([vector] ,							lr=self.lr	)

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




							def 						__snake_case					(   self     :   List[Any] ,							SCREAMING_SNAKE_CASE_     :   int ,							SCREAMING_SNAKE_CASE_     :   List[str] ,							SCREAMING_SNAKE_CASE_     :   List[Any]	):
											wandb.init(reinit=SCREAMING_SNAKE_CASE_ ,							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:
															lowerCAmelCase__      			=						Image.open(SCREAMING_SNAKE_CASE_	)
															lowerCAmelCase__      			=						image.resize((256, 256)	)
															wandb.log('''Original Image''' ,							wandb.Image(SCREAMING_SNAKE_CASE_	)	)




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




							def 						__snake_case					(   self     :   Optional[int] ,							SCREAMING_SNAKE_CASE_     :   Dict ,							SCREAMING_SNAKE_CASE_     :   int=None ,							SCREAMING_SNAKE_CASE_     :   str=None ,							SCREAMING_SNAKE_CASE_     :   Dict=True ,							SCREAMING_SNAKE_CASE_     :   Optional[Any]=False ,							SCREAMING_SNAKE_CASE_     :   str=True ,							SCREAMING_SNAKE_CASE_     :   Optional[int]=True ,							SCREAMING_SNAKE_CASE_     :   int=None ,							):
											if image_path:
															lowerCAmelCase__      			=						self._get_latent(SCREAMING_SNAKE_CASE_	)
											else:
															lowerCAmelCase__      			=						torch.randn(self.latent_dim ,							device=self.device	)
											if self.log:
															self._init_logging(SCREAMING_SNAKE_CASE_ ,							SCREAMING_SNAKE_CASE_ ,							SCREAMING_SNAKE_CASE_	)

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

											lowerCAmelCase__      			=						self.vqgan.decode(self.latent	)[0]
											if show_intermediate:
															print('''Original Image'''	)
															show_pil(custom_to_pil(SCREAMING_SNAKE_CASE_	)	)

											lowerCAmelCase__      			=						loop_post_process(SCREAMING_SNAKE_CASE_	)
											for iter, transformed_img in enumerate(self._optimize_CLIP(SCREAMING_SNAKE_CASE_ ,							SCREAMING_SNAKE_CASE_ ,							SCREAMING_SNAKE_CASE_	)	):
															if show_intermediate:
																			show_pil(SCREAMING_SNAKE_CASE_	)
															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(SCREAMING_SNAKE_CASE_	)}	)
											if show_final:
															show_pil(SCREAMING_SNAKE_CASE_	)
											if save_final:
															transformed_img.save(os.path.join(self.save_path ,							f'iter_{iter:03d}_final.png'	)	)













from collections import OrderedDict
from typing import Any, List, Mapping, Optional

from ... import PreTrainedTokenizer, TensorType, is_torch_available
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfigWithPast, PatchingSpec
from ...utils import logging


_UpperCAmelCase			:		Tuple   				=    logging.get_logger(__name__)


_UpperCAmelCase			:		int   				=    {
    "Salesforce/codegen-350M-nl": "https://huggingface.co/Salesforce/codegen-350M-nl/resolve/main/config.json",
    "Salesforce/codegen-350M-multi": "https://huggingface.co/Salesforce/codegen-350M-multi/resolve/main/config.json",
    "Salesforce/codegen-350M-mono": "https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/config.json",
    "Salesforce/codegen-2B-nl": "https://huggingface.co/Salesforce/codegen-2B-nl/resolve/main/config.json",
    "Salesforce/codegen-2B-multi": "https://huggingface.co/Salesforce/codegen-2B-multi/resolve/main/config.json",
    "Salesforce/codegen-2B-mono": "https://huggingface.co/Salesforce/codegen-2B-mono/resolve/main/config.json",
    "Salesforce/codegen-6B-nl": "https://huggingface.co/Salesforce/codegen-6B-nl/resolve/main/config.json",
    "Salesforce/codegen-6B-multi": "https://huggingface.co/Salesforce/codegen-6B-multi/resolve/main/config.json",
    "Salesforce/codegen-6B-mono": "https://huggingface.co/Salesforce/codegen-6B-mono/resolve/main/config.json",
    "Salesforce/codegen-16B-nl": "https://huggingface.co/Salesforce/codegen-16B-nl/resolve/main/config.json",
    "Salesforce/codegen-16B-multi": "https://huggingface.co/Salesforce/codegen-16B-multi/resolve/main/config.json",
    "Salesforce/codegen-16B-mono": "https://huggingface.co/Salesforce/codegen-16B-mono/resolve/main/config.json",
}
class  lowerCAmelCase_       (				snake_case__					):
							UpperCamelCase_		:int		      =    'codegen'
							UpperCamelCase_		:int		      =    {
							    'max_position_embeddings': 'n_positions',
							    'hidden_size': 'n_embd',
							    'num_attention_heads': 'n_head',
							    'num_hidden_layers': 'n_layer',
							}




							def __init__(   self     :   Optional[int] ,							SCREAMING_SNAKE_CASE_     :   str=50_400 ,							SCREAMING_SNAKE_CASE_     :   str=2_048 ,							SCREAMING_SNAKE_CASE_     :   int=2_048 ,							SCREAMING_SNAKE_CASE_     :   Any=4_096 ,							SCREAMING_SNAKE_CASE_     :   List[Any]=28 ,							SCREAMING_SNAKE_CASE_     :   str=16 ,							SCREAMING_SNAKE_CASE_     :   str=64 ,							SCREAMING_SNAKE_CASE_     :   Tuple=None ,							SCREAMING_SNAKE_CASE_     :   Dict="gelu_new" ,							SCREAMING_SNAKE_CASE_     :   str=0.0 ,							SCREAMING_SNAKE_CASE_     :   Any=0.0 ,							SCREAMING_SNAKE_CASE_     :   Any=0.0 ,							SCREAMING_SNAKE_CASE_     :   Any=1e-5 ,							SCREAMING_SNAKE_CASE_     :   Any=0.02 ,							SCREAMING_SNAKE_CASE_     :   Union[str, Any]=True ,							SCREAMING_SNAKE_CASE_     :   Optional[int]=50_256 ,							SCREAMING_SNAKE_CASE_     :   Any=50_256 ,							SCREAMING_SNAKE_CASE_     :   List[str]=False ,							**SCREAMING_SNAKE_CASE_     :   Union[str, Any] ,							):
											lowerCAmelCase__      			=						vocab_size
											lowerCAmelCase__      			=						n_ctx
											lowerCAmelCase__      			=						n_positions
											lowerCAmelCase__      			=						n_embd
											lowerCAmelCase__      			=						n_layer
											lowerCAmelCase__      			=						n_head
											lowerCAmelCase__      			=						n_inner
											lowerCAmelCase__      			=						rotary_dim
											lowerCAmelCase__      			=						activation_function
											lowerCAmelCase__      			=						resid_pdrop
											lowerCAmelCase__      			=						embd_pdrop
											lowerCAmelCase__      			=						attn_pdrop
											lowerCAmelCase__      			=						layer_norm_epsilon
											lowerCAmelCase__      			=						initializer_range
											lowerCAmelCase__      			=						use_cache

											lowerCAmelCase__      			=						bos_token_id
											lowerCAmelCase__      			=						eos_token_id

											super().__init__(
											    bos_token_id=SCREAMING_SNAKE_CASE_ ,							eos_token_id=SCREAMING_SNAKE_CASE_ ,							tie_word_embeddings=SCREAMING_SNAKE_CASE_ ,							**SCREAMING_SNAKE_CASE_	)



class  lowerCAmelCase_       (				snake_case__					):




							def __init__(   self     :   str ,							SCREAMING_SNAKE_CASE_     :   PretrainedConfig ,							SCREAMING_SNAKE_CASE_     :   str = "default" ,							SCREAMING_SNAKE_CASE_     :   List[PatchingSpec] = None ,							SCREAMING_SNAKE_CASE_     :   bool = False ,							):
											super().__init__(SCREAMING_SNAKE_CASE_ ,							task=SCREAMING_SNAKE_CASE_ ,							patching_specs=SCREAMING_SNAKE_CASE_ ,							use_past=SCREAMING_SNAKE_CASE_	)
											if not getattr(self._config ,							'''pad_token_id''' ,							SCREAMING_SNAKE_CASE_	):
															# TODO: how to do that better?
															lowerCAmelCase__      			=						0




							@property
							def 						__snake_case					(   self     :   str	):
											lowerCAmelCase__      			=						OrderedDict({'''input_ids''': {0: '''batch''', 1: '''sequence'''}}	)
											if self.use_past:
															self.fill_with_past_key_values_(SCREAMING_SNAKE_CASE_ ,							direction='''inputs'''	)
															lowerCAmelCase__      			=						{0: '''batch''', 1: '''past_sequence + sequence'''}
											else:
															lowerCAmelCase__      			=						{0: '''batch''', 1: '''sequence'''}

											return common_inputs




							@property
							def 						__snake_case					(   self     :   Dict	):
											return self._config.n_layer




							@property
							def 						__snake_case					(   self     :   Union[str, Any]	):
											return self._config.n_head




							def 						__snake_case					(   self     :   Optional[Any] ,							SCREAMING_SNAKE_CASE_     :   PreTrainedTokenizer ,							SCREAMING_SNAKE_CASE_     :   int = -1 ,							SCREAMING_SNAKE_CASE_     :   int = -1 ,							SCREAMING_SNAKE_CASE_     :   bool = False ,							SCREAMING_SNAKE_CASE_     :   Optional[TensorType] = None ,							):
											lowerCAmelCase__      			=						super(SCREAMING_SNAKE_CASE_ ,							self	).generate_dummy_inputs(
											    SCREAMING_SNAKE_CASE_ ,							batch_size=SCREAMING_SNAKE_CASE_ ,							seq_length=SCREAMING_SNAKE_CASE_ ,							is_pair=SCREAMING_SNAKE_CASE_ ,							framework=SCREAMING_SNAKE_CASE_	)

											# We need to order the input in the way they appears in the forward()
											lowerCAmelCase__      			=						OrderedDict({'''input_ids''': common_inputs['''input_ids''']}	)

											# Need to add the past_keys
											if self.use_past:
															if not is_torch_available():
																			raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''	)
															else:
																			import torch

																			lowerCAmelCase__			,		lowerCAmelCase__      			=						common_inputs['''input_ids'''].shape
																			# Not using the same length for past_key_values
																			lowerCAmelCase__      			=						seqlen + 2
																			lowerCAmelCase__      			=						(
																			    batch,
																			    self.num_attention_heads,
																			    past_key_values_length,
																			    self._config.hidden_size // self.num_attention_heads,
																			)
																			lowerCAmelCase__      			=						[
																			    (torch.zeros(SCREAMING_SNAKE_CASE_	), torch.zeros(SCREAMING_SNAKE_CASE_	)) for _ in range(self.num_layers	)
																			]

											lowerCAmelCase__      			=						common_inputs['''attention_mask''']
											if self.use_past:
															lowerCAmelCase__      			=						ordered_inputs['''attention_mask'''].dtype
															lowerCAmelCase__      			=						torch.cat(
															    [ordered_inputs['''attention_mask'''], torch.ones(SCREAMING_SNAKE_CASE_ ,							SCREAMING_SNAKE_CASE_ ,							dtype=SCREAMING_SNAKE_CASE_	)] ,							dim=1	)

											return ordered_inputs




							@property
							def 						__snake_case					(   self     :   Optional[int]	):
											return 13





