






'''simple docstring'''







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 						SCREAMING_SNAKE_CASE       :






			def __init__(			self   :      str	,    __SCREAMING_SNAKE_CASE   :      str = "cpu"	,    __SCREAMING_SNAKE_CASE   :      str = "openai/clip-vit-large-patch14"       )      ->      None:
							a_		:	Union[str, Any]									=   device
							a_		:	Union[str, Any]									=   CLIPTokenizerFast.from_pretrained(__lowerCamelCase       )
							a_		:	List[str]									=   [0.4814_5466, 0.457_8275, 0.4082_1073]
							a_		:	int									=   [0.2686_2954, 0.2613_0258, 0.2757_7711]
							a_		:	Tuple									=   torchvision.transforms.Normalize(self.image_mean	,    self.image_std       )
							a_		:	Union[str, Any]									=   torchvision.transforms.Resize(224       )
							a_		:	Optional[int]									=   torchvision.transforms.CenterCrop(224       )






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[Any]	,    __SCREAMING_SNAKE_CASE   :      Dict       )      ->      Dict:
							a_		:	Optional[Any]									=   self.resize(__lowerCamelCase       )
							a_		:	int									=   self.center_crop(__lowerCamelCase       )
							a_		:	str									=   self.normalize(__lowerCamelCase       )
							return images







			def __call__(			self   :      Dict	,    __SCREAMING_SNAKE_CASE   :      Tuple=None	,    __SCREAMING_SNAKE_CASE   :      Tuple=None	,    **__SCREAMING_SNAKE_CASE   :      int       )      ->      int:
							a_		:	List[Any]									=   self.tokenizer(text=__lowerCamelCase	,    **__lowerCamelCase       )
							a_		:	List[str]									=   self.preprocess_img(__lowerCamelCase       )
							a_		:	List[str]									=   {key: value.to(self.device       ) for (key, value) in encoding.items()}
							return encoding





class 						SCREAMING_SNAKE_CASE       (       nn.Module							):






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

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






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






			def       SCREAMING_SNAKE_CASE   (			self   :      Dict	,    __SCREAMING_SNAKE_CASE   :      List[Any]=None	,    __SCREAMING_SNAKE_CASE   :      List[Any]=None       )      ->      Dict:







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






			def       SCREAMING_SNAKE_CASE   (			self   :      Any	,    __SCREAMING_SNAKE_CASE   :      Dict       )      ->      Optional[Any]:
							a_		:	Optional[Any]									=   self.latent.detach().requires_grad_()
							a_		:	Any									=   base_latent + transform_vector
							if self.quantize:
											a_					,      *a_		:	Union[str, Any]									=   self.vqgan.quantize(__lowerCamelCase       )
							else:
											a_		:	Any									=   trans_latent
							return self.vqgan.decode(__lowerCamelCase       )






			def       SCREAMING_SNAKE_CASE   (			self   :      List[Any]	,    __SCREAMING_SNAKE_CASE   :      Tuple	,    __SCREAMING_SNAKE_CASE   :      Optional[Any]	,    __SCREAMING_SNAKE_CASE   :      str=None       )      ->      Union[str, Any]:
							a_		:	Optional[int]									=   self.clip_preprocessor(text=__lowerCamelCase	,    images=__lowerCamelCase	,    return_tensors='''pt'''	,    padding=__lowerCamelCase       )
							a_		:	int									=   self.clip(**__lowerCamelCase       )
							a_		:	List[Any]									=   clip_outputs.logits_per_image
							if weights is not None:
											a_		:	Dict									=   similarity_logits * weights
							return similarity_logits.sum()






			def       SCREAMING_SNAKE_CASE   (			self   :      int	,    __SCREAMING_SNAKE_CASE   :      Dict	,    __SCREAMING_SNAKE_CASE   :      int	,    __SCREAMING_SNAKE_CASE   :      List[Any]       )      ->      Optional[int]:
							a_		:	Optional[Any]									=   self._get_clip_similarity(pos_prompts['''prompts''']	,    __lowerCamelCase	,    weights=(1 / pos_prompts['''weights'''])       )
							if neg_prompts:
											a_		:	Optional[Any]									=   self._get_clip_similarity(neg_prompts['''prompts''']	,    __lowerCamelCase	,    weights=neg_prompts['''weights''']       )
							else:
											a_		:	List[Any]									=   torch.tensor([1]	,    device=self.device       )
							a_		:	Tuple									=   -torch.log(__lowerCamelCase       ) + torch.log(__lowerCamelCase       )
							return loss






			def       SCREAMING_SNAKE_CASE   (			self   :      List[str]	,    __SCREAMING_SNAKE_CASE   :      Optional[Any]	,    __SCREAMING_SNAKE_CASE   :      List[Any]	,    __SCREAMING_SNAKE_CASE   :      Union[str, Any]       )      ->      Union[str, Any]:
							a_		:	Union[str, Any]									=   torch.randn_like(self.latent	,    requires_grad=__lowerCamelCase	,    device=self.device       )
							a_		:	int									=   torch.optim.Adam([vector]	,    lr=self.lr       )

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






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[Any]	,    __SCREAMING_SNAKE_CASE   :      Any	,    __SCREAMING_SNAKE_CASE   :      Union[str, Any]	,    __SCREAMING_SNAKE_CASE   :      Union[str, Any]       )      ->      List[Any]:
							wandb.init(reinit=__lowerCamelCase	,    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_		:	Union[str, Any]									=   Image.open(__lowerCamelCase       )
											a_		:	Tuple									=   image.resize((256, 256)       )
											wandb.log('''Original Image'''	,    wandb.Image(__lowerCamelCase       )       )






			def       SCREAMING_SNAKE_CASE   (			self   :      List[str]	,    __SCREAMING_SNAKE_CASE   :      Any       )      ->      int:







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







			def       SCREAMING_SNAKE_CASE   (			self   :      Any	,    __SCREAMING_SNAKE_CASE   :      Tuple	,    __SCREAMING_SNAKE_CASE   :      Optional[int]=None	,    __SCREAMING_SNAKE_CASE   :      Optional[int]=None	,    __SCREAMING_SNAKE_CASE   :      Any=True	,    __SCREAMING_SNAKE_CASE   :      Optional[Any]=False	,    __SCREAMING_SNAKE_CASE   :      List[str]=True	,    __SCREAMING_SNAKE_CASE   :      Union[str, Any]=True	,    __SCREAMING_SNAKE_CASE   :      Optional[Any]=None	,    )      ->      Optional[Any]:







							if image_path:
											a_		:	Dict									=   self._get_latent(__lowerCamelCase       )
							else:
											a_		:	Tuple									=   torch.randn(self.latent_dim	,    device=self.device       )
							if self.log:
											self._init_logging(__lowerCamelCase	,    __lowerCamelCase	,    __lowerCamelCase       )

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

							a_		:	List[Any]									=   self.vqgan.decode(self.latent       )[0]
							if show_intermediate:
											print('''Original Image'''       )
											show_pil(custom_to_pil(__lowerCamelCase       )       )

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








import math
import time

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
 import torch_xla.core.xla_model as xm
 import torch_xla.debug.metrics as met






class 	__lowercase					(	__snake_case	):





      def __init__(		self    :       Dict       ,							*__lowerCamelCase    :       Dict       ,							__lowerCamelCase    :       Union[str, Any]=None       ,							__lowerCamelCase    :       str=None       ,							**__lowerCamelCase    :       Optional[int]							)     -> Optional[int]:






         """simple docstring"""







         super().__init__(*__lowerCamelCase       ,							**__lowerCamelCase							)
         UpperCAmelCase      					=  eval_examples
         UpperCAmelCase      					=  post_process_function





      def 							_lowercase					(		self    :       Any       ,							__lowerCamelCase    :       int=None       ,							__lowerCamelCase    :       int=None       ,							__lowerCamelCase    :       Tuple=None       ,							__lowerCamelCase    :       str = "eval"							)     -> List[str]:






         """simple docstring"""







         UpperCAmelCase      					=  self.eval_dataset if eval_dataset is None else eval_dataset
         UpperCAmelCase      					=  self.get_eval_dataloader(__lowerCamelCase							)
         UpperCAmelCase      					=  self.eval_examples if eval_examples is None else eval_examples

         # Temporarily disable metric computation, we will do it in the loop here.
         UpperCAmelCase      					=  self.compute_metrics
         UpperCAmelCase      					=  None
         UpperCAmelCase      					=  self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
         UpperCAmelCase      					=  time.time()
         try:
            UpperCAmelCase      					=  eval_loop(
                __lowerCamelCase       ,							description="""Evaluation"""       ,							prediction_loss_only=True if compute_metrics is None else None       ,							ignore_keys=__lowerCamelCase       ,							metric_key_prefix=__lowerCamelCase       ,							)
         finally:
            UpperCAmelCase      					=  compute_metrics
         UpperCAmelCase      					=  self.args.eval_batch_size * self.args.world_size
         if F"""{metric_key_prefix}_jit_compilation_time""" in output.metrics:
            start_time += output.metrics[F"""{metric_key_prefix}_jit_compilation_time"""]
         output.metrics.update(
             speed_metrics(
                 __lowerCamelCase       ,							__lowerCamelCase       ,							num_samples=output.num_samples       ,							num_steps=math.ceil(output.num_samples / total_batch_size							)       ,							)							)
         if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
            # Only the main node write the results by default
            UpperCAmelCase      					=  self.post_process_function(__lowerCamelCase       ,							__lowerCamelCase       ,							output.predictions							)
            UpperCAmelCase      					=  self.compute_metrics(__lowerCamelCase							)

            # Prefix all keys with metric_key_prefix + '_'
            for key in list(metrics.keys()							):
               if not key.startswith(F"""{metric_key_prefix}_"""							):
                  UpperCAmelCase      					=  metrics.pop(__lowerCamelCase							)
            metrics.update(output.metrics							)
         else:
            UpperCAmelCase      					=  output.metrics

         if self.args.should_log:
            # Only the main node log the results by default
            self.log(__lowerCamelCase							)

         if self.args.tpu_metrics_debug or self.args.debug:
            # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
            xm.master_print(met.metrics_report()							)

         UpperCAmelCase      					=  self.callback_handler.on_evaluate(self.args       ,							self.state       ,							self.control       ,							__lowerCamelCase							)
         return metrics





      def 							_lowercase					(		self    :       Optional[Any]       ,							__lowerCamelCase    :       Union[str, Any]       ,							__lowerCamelCase    :       Any       ,							__lowerCamelCase    :       Dict=None       ,							__lowerCamelCase    :       str = "test"							)     -> Dict:






         """simple docstring"""







         UpperCAmelCase      					=  self.get_test_dataloader(__lowerCamelCase							)

         # Temporarily disable metric computation, we will do it in the loop here.
         UpperCAmelCase      					=  self.compute_metrics
         UpperCAmelCase      					=  None
         UpperCAmelCase      					=  self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
         UpperCAmelCase      					=  time.time()
         try:
            UpperCAmelCase      					=  eval_loop(
                __lowerCamelCase       ,							description="""Prediction"""       ,							prediction_loss_only=True if compute_metrics is None else None       ,							ignore_keys=__lowerCamelCase       ,							metric_key_prefix=__lowerCamelCase       ,							)
         finally:
            UpperCAmelCase      					=  compute_metrics
         UpperCAmelCase      					=  self.args.eval_batch_size * self.args.world_size
         if F"""{metric_key_prefix}_jit_compilation_time""" in output.metrics:
            start_time += output.metrics[F"""{metric_key_prefix}_jit_compilation_time"""]
         output.metrics.update(
             speed_metrics(
                 __lowerCamelCase       ,							__lowerCamelCase       ,							num_samples=output.num_samples       ,							num_steps=math.ceil(output.num_samples / total_batch_size							)       ,							)							)

         if self.post_process_function is None or self.compute_metrics is None:
            return output

         UpperCAmelCase      					=  self.post_process_function(__lowerCamelCase       ,							__lowerCamelCase       ,							output.predictions       ,							"""predict"""							)
         UpperCAmelCase      					=  self.compute_metrics(__lowerCamelCase							)

         # Prefix all keys with metric_key_prefix + '_'
         for key in list(metrics.keys()							):
            if not key.startswith(F"""{metric_key_prefix}_"""							):
               UpperCAmelCase      					=  metrics.pop(__lowerCamelCase							)
         metrics.update(output.metrics							)
         return PredictionOutput(predictions=predictions.predictions       ,							label_ids=predictions.label_ids       ,							metrics=__lowerCamelCase							)



