





'''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 			UpperCAmelCase					:




      def __init__(      self      :			Optional[int]      ,      __snake_case      :			str = "cpu"      ,      __snake_case      :			str = "openai/clip-vit-large-patch14"  )    ->				None:
            _lowerCAmelCase												=							device
            _lowerCAmelCase												=							CLIPTokenizerFast.from_pretrained(__snake_case  )
            _lowerCAmelCase												=							[0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73]
            _lowerCAmelCase												=							[0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11]
            _lowerCAmelCase												=							torchvision.transforms.Normalize(self.image_mean      ,      self.image_std  )
            _lowerCAmelCase												=							torchvision.transforms.Resize(2_24  )
            _lowerCAmelCase												=							torchvision.transforms.CenterCrop(2_24  )




      def     lowercase__   (      self      :			Union[str, Any]      ,      __snake_case      :			Tuple  )    ->				Optional[Any]:
            _lowerCAmelCase												=							self.resize(__snake_case  )
            _lowerCAmelCase												=							self.center_crop(__snake_case  )
            _lowerCAmelCase												=							self.normalize(__snake_case  )
            return images




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




class 			UpperCAmelCase					(      nn.Module       ):




      def __init__(      self      :			Dict      ,      __snake_case      :			int=10      ,      __snake_case      :			Optional[int]=0.01      ,      __snake_case      :			Tuple=None      ,      __snake_case      :			Optional[Any]=None      ,      __snake_case      :			Optional[Any]=None      ,      __snake_case      :			List[str]=None      ,      __snake_case      :			Dict=None      ,      __snake_case      :			Optional[int]=None      ,      __snake_case      :			Optional[Any]=False      ,      __snake_case      :			Optional[int]=True      ,      __snake_case      :			Union[str, Any]="image"      ,      __snake_case      :			List[str]=True      ,      __snake_case      :			Any=False      ,      __snake_case      :			Any=False      ,      __snake_case      :			List[str]=False      ,      )    ->				None:
            super().__init__()
            _lowerCAmelCase												=							None
            _lowerCAmelCase												=							device if device else get_device()
            if vqgan:
                  _lowerCAmelCase												=							vqgan
            else:
                  _lowerCAmelCase												=							load_vqgan(self.device      ,      conf_path=__snake_case      ,      ckpt_path=__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     lowercase__   (      self      :			Optional[Any]      ,      __snake_case      :			str=None      ,      __snake_case      :			Tuple=None      ,      __snake_case      :			str=5      ,      __snake_case      :			Optional[Any]=True  )    ->				str:
            _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(__snake_case  ):
                  raise ValueError(
                      """No images found in save path, aborting (did you pass save_intermediate=True to the generate"""
                      """ function?)"""  )
            if len(__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(__snake_case  )
            _lowerCAmelCase												=							[frame_duration] * len(__snake_case  )
            if extend_frames:
                  _lowerCAmelCase												=							1.5
                  _lowerCAmelCase												=							3
            for file_name in paths:
                  if file_name.endswith(""".png"""  ):
                        images.append(imageio.imread(__snake_case  )  )
            imageio.mimsave(__snake_case      ,      __snake_case      ,      duration=__snake_case  )
            print(f"gif saved to {output_path}"  )




      def     lowercase__   (      self      :			int      ,      __snake_case      :			Union[str, Any]=None      ,      __snake_case      :			Union[str, Any]=None  )    ->				Dict:
            if not (path or img):
                  raise ValueError("""Input either path or tensor"""  )
            if img is not None:
                  raise NotImplementedError
            _lowerCAmelCase												=							preprocess(Image.open(__snake_case  )      ,      target_image_size=2_56  ).to(self.device  )
            _lowerCAmelCase												=							preprocess_vqgan(__snake_case  )
            _lowerCAmelCase					,							*_lowerCAmelCase												=							self.vqgan.encode(__snake_case  )
            return z




      def     lowercase__   (      self      :			Optional[Any]      ,      __snake_case      :			str  )    ->				int:
            _lowerCAmelCase												=							self.latent.detach().requires_grad_()
            _lowerCAmelCase												=							base_latent + transform_vector
            if self.quantize:
                  _lowerCAmelCase					,							*_lowerCAmelCase												=							self.vqgan.quantize(__snake_case  )
            else:
                  _lowerCAmelCase												=							trans_latent
            return self.vqgan.decode(__snake_case  )




      def     lowercase__   (      self      :			str      ,      __snake_case      :			Optional[Any]      ,      __snake_case      :			Union[str, Any]      ,      __snake_case      :			List[str]=None  )    ->				Tuple:
            _lowerCAmelCase												=							self.clip_preprocessor(text=__snake_case      ,      images=__snake_case      ,      return_tensors="""pt"""      ,      padding=__snake_case  )
            _lowerCAmelCase												=							self.clip(**__snake_case  )
            _lowerCAmelCase												=							clip_outputs.logits_per_image
            if weights is not None:
                  _lowerCAmelCase												=							similarity_logits * weights
            return similarity_logits.sum()




      def     lowercase__   (      self      :			int      ,      __snake_case      :			List[str]      ,      __snake_case      :			List[str]      ,      __snake_case      :			str  )    ->				Tuple:
            _lowerCAmelCase												=							self._get_clip_similarity(pos_prompts["""prompts"""]      ,      __snake_case      ,      weights=(1 / pos_prompts["""weights"""])  )
            if neg_prompts:
                  _lowerCAmelCase												=							self._get_clip_similarity(neg_prompts["""prompts"""]      ,      __snake_case      ,      weights=neg_prompts["""weights"""]  )
            else:
                  _lowerCAmelCase												=							torch.tensor([1]      ,      device=self.device  )
            _lowerCAmelCase												=							-torch.log(__snake_case  ) + torch.log(__snake_case  )
            return loss




      def     lowercase__   (      self      :			List[str]      ,      __snake_case      :			List[str]      ,      __snake_case      :			str      ,      __snake_case      :			Dict  )    ->				int:
            _lowerCAmelCase												=							torch.randn_like(self.latent      ,      requires_grad=__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(__snake_case  )
                  _lowerCAmelCase												=							loop_post_process(__snake_case  )
                  _lowerCAmelCase												=							self._get_CLIP_loss(__snake_case      ,      __snake_case      ,      __snake_case  )
                  print("""CLIP loss"""      ,      __snake_case  )
                  if self.log:
                        wandb.log({"""CLIP Loss""": clip_loss}  )
                  clip_loss.backward(retain_graph=__snake_case  )
                  optim.step()
                  if self.return_val == "image":
                        yield custom_to_pil(transformed_img[0]  )
                  else:
                        yield vector




      def     lowercase__   (      self      :			List[Any]      ,      __snake_case      :			str      ,      __snake_case      :			Optional[int]      ,      __snake_case      :			int  )    ->				List[Any]:
            wandb.init(reinit=__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(__snake_case  )
                  _lowerCAmelCase												=							image.resize((2_56, 2_56)  )
                  wandb.log("""Original Image"""      ,      wandb.Image(__snake_case  )  )




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




      def     lowercase__   (      self      :			int      ,      __snake_case      :			Dict      ,      __snake_case      :			Optional[int]=None      ,      __snake_case      :			Optional[Any]=None      ,      __snake_case      :			Union[str, Any]=True      ,      __snake_case      :			str=False      ,      __snake_case      :			int=True      ,      __snake_case      :			List[str]=True      ,      __snake_case      :			Tuple=None      ,      )    ->				Tuple:
            if image_path:
                  _lowerCAmelCase												=							self._get_latent(__snake_case  )
            else:
                  _lowerCAmelCase												=							torch.randn(self.latent_dim      ,      device=self.device  )
            if self.log:
                  self._init_logging(__snake_case      ,      __snake_case      ,      __snake_case  )

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

            _lowerCAmelCase												=							self.vqgan.decode(self.latent  )[0]
            if show_intermediate:
                  print("""Original Image"""  )
                  show_pil(custom_to_pil(__snake_case  )  )

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













'''simple docstring'''






import unittest

from transformers import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING, is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
      from PIL import Image
else:




      class 			UpperCAmelCase					:




            @staticmethod
            def     lowercase__   (      *__snake_case      :			Optional[Any]      ,      **__snake_case      :			Any  )    ->				Tuple:
                  pass




@is_pipeline_test
@require_torch
@require_vision
class 			UpperCAmelCase					(      unittest.TestCase       ):
      _lowercase:      Union[str, Any]											=       MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING




      def     lowercase__   (      self      :			List[str]      ,      __snake_case      :			Optional[Any]      ,      __snake_case      :			str      ,      __snake_case      :			List[str]  )    ->				int:
            _lowerCAmelCase												=							pipeline("""visual-question-answering"""      ,      model="""hf-internal-testing/tiny-vilt-random-vqa"""  )
            _lowerCAmelCase												=							[
                {
                    """image""": Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""  ),
                    """question""": """How many cats are there?""",
                },
                {
                    """image""": """./tests/fixtures/tests_samples/COCO/000000039769.png""",
                    """question""": """How many cats are there?""",
                },
            ]
            return vqa_pipeline, examples




      def     lowercase__   (      self      :			Any      ,      __snake_case      :			List[Any]      ,      __snake_case      :			List[Any]  )    ->				Union[str, Any]:
            _lowerCAmelCase												=							vqa_pipeline(__snake_case      ,      top_k=1  )
            self.assertEqual(
                __snake_case      ,      [
                    [{"""score""": ANY(__snake_case  ), """answer""": ANY(__snake_case  )}],
                    [{"""score""": ANY(__snake_case  ), """answer""": ANY(__snake_case  )}],
                ]      ,      )




      @require_torch
      def     lowercase__   (      self      :			str  )    ->				int:
            _lowerCAmelCase												=							pipeline("""visual-question-answering"""      ,      model="""hf-internal-testing/tiny-vilt-random-vqa"""  )
            _lowerCAmelCase												=							"""./tests/fixtures/tests_samples/COCO/000000039769.png"""
            _lowerCAmelCase												=							"""How many cats are there?"""

            _lowerCAmelCase												=							vqa_pipeline(image=__snake_case      ,      question="""How many cats are there?"""      ,      top_k=2  )
            self.assertEqual(
                __snake_case      ,      [{"""score""": ANY(__snake_case  ), """answer""": ANY(__snake_case  )}, {"""score""": ANY(__snake_case  ), """answer""": ANY(__snake_case  )}]  )

            _lowerCAmelCase												=							vqa_pipeline({"""image""": image, """question""": question}      ,      top_k=2  )
            self.assertEqual(
                __snake_case      ,      [{"""score""": ANY(__snake_case  ), """answer""": ANY(__snake_case  )}, {"""score""": ANY(__snake_case  ), """answer""": ANY(__snake_case  )}]  )




      @slow
      @require_torch
      def     lowercase__   (      self      :			List[Any]  )    ->				List[str]:
            _lowerCAmelCase												=							pipeline("""visual-question-answering"""      ,      model="""dandelin/vilt-b32-finetuned-vqa"""  )
            _lowerCAmelCase												=							"""./tests/fixtures/tests_samples/COCO/000000039769.png"""
            _lowerCAmelCase												=							"""How many cats are there?"""

            _lowerCAmelCase												=							vqa_pipeline(image=__snake_case      ,      question=__snake_case      ,      top_k=2  )
            self.assertEqual(
                nested_simplify(__snake_case      ,      decimals=4  )      ,      [{"""score""": 0.87_99, """answer""": """2"""}, {"""score""": 0.2_96, """answer""": """1"""}]  )

            _lowerCAmelCase												=							vqa_pipeline({"""image""": image, """question""": question}      ,      top_k=2  )
            self.assertEqual(
                nested_simplify(__snake_case      ,      decimals=4  )      ,      [{"""score""": 0.87_99, """answer""": """2"""}, {"""score""": 0.2_96, """answer""": """1"""}]  )

            _lowerCAmelCase												=							vqa_pipeline(
                [{"""image""": image, """question""": question}, {"""image""": image, """question""": question}]      ,      top_k=2  )
            self.assertEqual(
                nested_simplify(__snake_case      ,      decimals=4  )      ,      [[{"""score""": 0.87_99, """answer""": """2"""}, {"""score""": 0.2_96, """answer""": """1"""}]] * 2      ,      )




      @require_tf
      @unittest.skip("""Visual question answering not implemented in TF"""  )
      def     lowercase__   (      self      :			List[str]  )    ->				Union[str, Any]:
            pass






