





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       :      List[str]				,			a__       :      str = "cpu"				,			a__       :      str = "openai/clip-vit-large-patch14"      ):





         """simple docstring"""



         __snake_case							       =   device
         __snake_case							       =   CLIPTokenizerFast.from_pretrained(a__      )
         __snake_case							       =   [0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3]
         __snake_case							       =   [0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1]
         __snake_case							       =   torchvision.transforms.Normalize(self.image_mean				,			self.image_std      )
         __snake_case							       =   torchvision.transforms.Resize(224      )
         __snake_case							       =   torchvision.transforms.CenterCrop(224      )


      def    a (self       :      Optional[Any]				,			a__       :      str      ):





         """simple docstring"""



         __snake_case							       =   self.resize(a__      )
         __snake_case							       =   self.center_crop(a__      )
         __snake_case							       =   self.normalize(a__      )
         return images


      def __call__(self       :      Union[str, Any]				,			a__       :      Tuple=None				,			a__       :      Tuple=None				,			**a__       :      int      ):





         """simple docstring"""



         __snake_case							       =   self.tokenizer(text=a__				,			**a__      )
         __snake_case							       =   self.preprocess_img(a__      )
         __snake_case							       =   {key: value.to(self.device      ) for (key, value) in encoding.items()}
         return encoding






class 		SCREAMING_SNAKE_CASE__				(       nn.Module  ):


      def __init__(self       :      Optional[int]				,			a__       :      Dict=10				,			a__       :      Dict=0.0_1				,			a__       :      Union[str, Any]=None				,			a__       :      Any=None				,			a__       :      Optional[Any]=None				,			a__       :      Dict=None				,			a__       :      Union[str, Any]=None				,			a__       :      List[str]=None				,			a__       :      Union[str, Any]=False				,			a__       :      Dict=True				,			a__       :      Dict="image"				,			a__       :      Optional[Any]=True				,			a__       :      List[str]=False				,			a__       :      Optional[Any]=False				,			a__       :      List[str]=False				,			):





         """simple docstring"""



         super().__init__()
         __snake_case							       =   None
         __snake_case							       =   device if device else get_device()
         if vqgan:
            __snake_case							       =   vqgan
         else:
            __snake_case							       =   load_vqgan(self.device				,			conf_path=a__				,			ckpt_path=a__      )
         self.vqgan.eval()
         if clip:
            __snake_case							       =   clip
         else:
            __snake_case							       =   CLIPModel.from_pretrained('''openai/clip-vit-base-patch32'''      )
         self.clip.to(self.device      )
         __snake_case							       =   ProcessorGradientFlow(device=self.device      )

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


      def    a (self       :      str				,			a__       :      Optional[int]=None				,			a__       :      Tuple=None				,			a__       :      str=5				,			a__       :      int=True      ):





         """simple docstring"""



         __snake_case							       =   []
         if output_path is None:
            __snake_case							       =   '''./animation.gif'''
         if input_path is None:
            __snake_case							       =   self.save_path
         __snake_case							       =   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?)'''      )
         __snake_case							       =   total_duration / len(a__      )
         __snake_case							       =   [frame_duration] * len(a__      )
         if extend_frames:
            __snake_case							       =   1.5
            __snake_case							       =   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    a (self       :      List[Any]				,			a__       :      List[str]=None				,			a__       :      Optional[int]=None      ):





         """simple docstring"""



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


      def    a (self       :      List[str]				,			a__       :      int      ):





         """simple docstring"""



         __snake_case							       =   self.latent.detach().requires_grad_()
         __snake_case							       =   base_latent + transform_vector
         if self.quantize:
            __snake_case       ,	*__snake_case							       =   self.vqgan.quantize(a__      )
         else:
            __snake_case							       =   trans_latent
         return self.vqgan.decode(a__      )


      def    a (self       :      int				,			a__       :      Dict				,			a__       :      str				,			a__       :      str=None      ):





         """simple docstring"""



         __snake_case							       =   self.clip_preprocessor(text=a__				,			images=a__				,			return_tensors='''pt'''				,			padding=a__      )
         __snake_case							       =   self.clip(**a__      )
         __snake_case							       =   clip_outputs.logits_per_image
         if weights is not None:
            __snake_case							       =   similarity_logits * weights
         return similarity_logits.sum()


      def    a (self       :      Dict				,			a__       :      Union[str, Any]				,			a__       :      Union[str, Any]				,			a__       :      int      ):





         """simple docstring"""



         __snake_case							       =   self._get_clip_similarity(pos_prompts['''prompts''']				,			a__				,			weights=(1 / pos_prompts['''weights'''])      )
         if neg_prompts:
            __snake_case							       =   self._get_clip_similarity(neg_prompts['''prompts''']				,			a__				,			weights=neg_prompts['''weights''']      )
         else:
            __snake_case							       =   torch.tensor([1]				,			device=self.device      )
         __snake_case							       =   -torch.log(a__      ) + torch.log(a__      )
         return loss


      def    a (self       :      Any				,			a__       :      Optional[Any]				,			a__       :      List[str]				,			a__       :      Tuple      ):





         """simple docstring"""



         __snake_case							       =   torch.randn_like(self.latent				,			requires_grad=a__				,			device=self.device      )
         __snake_case							       =   torch.optim.Adam([vector]				,			lr=self.lr      )

         for i in range(self.iterations      ):
            optim.zero_grad()
            __snake_case							       =   self._add_vector(a__      )
            __snake_case							       =   loop_post_process(a__      )
            __snake_case							       =   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    a (self       :      int				,			a__       :      Optional[Any]				,			a__       :      Optional[int]				,			a__       :      List[Any]      ):





         """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:
            __snake_case							       =   Image.open(a__      )
            __snake_case							       =   image.resize((256, 256)      )
            wandb.log('''Original Image'''				,			wandb.Image(a__      )      )


      def    a (self       :      Optional[int]				,			a__       :      int      ):





         """simple docstring"""



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


      def    a (self       :      Optional[Any]				,			a__       :      Union[str, Any]				,			a__       :      Any=None				,			a__       :      List[Any]=None				,			a__       :      Optional[int]=True				,			a__       :      Any=False				,			a__       :      Any=True				,			a__       :      int=True				,			a__       :      List[str]=None				,			):





         """simple docstring"""



         if image_path:
            __snake_case							       =   self._get_latent(a__      )
         else:
            __snake_case							       =   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."
         __snake_case							       =   self.process_prompts(a__      )
         __snake_case							       =   self.process_prompts(a__      )
         if save_final and save_path is None:
            __snake_case							       =   os.path.join('''./outputs/'''				,			'''_'''.join(pos_prompts['''prompts''']      )      )
         if not os.path.exists(a__      ):
            os.makedirs(a__      )
         else:
            __snake_case							       =   save_path + '''_''' + get_timestamp()
            os.makedirs(a__      )
         __snake_case							       =   save_path

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

         __snake_case							       =   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"""      )      )


"""simple docstring"""


import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()


class 				a (						lowerCAmelCase_		,							lowerCAmelCase_		,							lowerCAmelCase_		,							unittest.TestCase      ):
    _snake_case   :     int   							=							StableUnCLIPPipeline
    _snake_case   :     str   							=							TEXT_TO_IMAGE_PARAMS
    _snake_case   :     Any   							=							TEXT_TO_IMAGE_BATCH_PARAMS
    _snake_case   :     Optional[Any]   							=							TEXT_TO_IMAGE_IMAGE_PARAMS
    _snake_case   :     str   							=							TEXT_TO_IMAGE_IMAGE_PARAMS

    # TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
    _snake_case   :     str   							=							False


    def    lowerCAmelCase_     (    self		:     Optional[int]    ):
      _UpperCAmelCase         =					32
      _UpperCAmelCase         =					embedder_hidden_size

      # prior components

      torch.manual_seed(0    )
      _UpperCAmelCase         =					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )

      torch.manual_seed(0    )
      _UpperCAmelCase         =					CLIPTextModelWithProjection(
          CLIPTextConfig(
              bos_token_id=0      , eos_token_id=2      , hidden_size=__lowerCAmelCase      , projection_dim=__lowerCAmelCase      , intermediate_size=37      , layer_norm_eps=1e-0_5      , num_attention_heads=4      , num_hidden_layers=5      , pad_token_id=1      , vocab_size=1000      , )    )

      torch.manual_seed(0    )
      _UpperCAmelCase         =					PriorTransformer(
          num_attention_heads=2      , attention_head_dim=12      , embedding_dim=__lowerCAmelCase      , num_layers=1      , )

      torch.manual_seed(0    )
      _UpperCAmelCase         =					DDPMScheduler(
          variance_type="""fixed_small_log"""      , prediction_type="""sample"""      , num_train_timesteps=1000      , clip_sample=__lowerCAmelCase      , clip_sample_range=5.0      , beta_schedule="""squaredcos_cap_v2"""      , )

      # regular denoising components

      torch.manual_seed(0    )
      _UpperCAmelCase         =					StableUnCLIPImageNormalizer(embedding_dim=__lowerCAmelCase    )
      _UpperCAmelCase         =					DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""    )

      torch.manual_seed(0    )
      _UpperCAmelCase         =					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )

      torch.manual_seed(0    )
      _UpperCAmelCase         =					CLIPTextModel(
          CLIPTextConfig(
              bos_token_id=0      , eos_token_id=2      , hidden_size=__lowerCAmelCase      , projection_dim=32      , intermediate_size=37      , layer_norm_eps=1e-0_5      , num_attention_heads=4      , num_hidden_layers=5      , pad_token_id=1      , vocab_size=1000      , )    )

      torch.manual_seed(0    )
      _UpperCAmelCase         =					UNetaDConditionModel(
          sample_size=32      , in_channels=4      , out_channels=4      , down_block_types=("""CrossAttnDownBlock2D""", """DownBlock2D""")      , up_block_types=("""UpBlock2D""", """CrossAttnUpBlock2D""")      , block_out_channels=(32, 64)      , attention_head_dim=(2, 4)      , class_embed_type="""projection"""      , projection_class_embeddings_input_dim=embedder_projection_dim * 2      , cross_attention_dim=__lowerCAmelCase      , layers_per_block=1      , upcast_attention=__lowerCAmelCase      , use_linear_projection=__lowerCAmelCase      , )

      torch.manual_seed(0    )
      _UpperCAmelCase         =					DDIMScheduler(
          beta_schedule="""scaled_linear"""      , beta_start=0.00_085      , beta_end=0.012      , prediction_type="""v_prediction"""      , set_alpha_to_one=__lowerCAmelCase      , steps_offset=1      , )

      torch.manual_seed(0    )
      _UpperCAmelCase         =					AutoencoderKL()

      _UpperCAmelCase         =					{
          # prior components
          """prior_tokenizer""": prior_tokenizer,
          """prior_text_encoder""": prior_text_encoder,
          """prior""": prior,
          """prior_scheduler""": prior_scheduler,
          # image noising components
          """image_normalizer""": image_normalizer,
          """image_noising_scheduler""": image_noising_scheduler,
          # regular denoising components
          """tokenizer""": tokenizer,
          """text_encoder""": text_encoder,
          """unet""": unet,
          """scheduler""": scheduler,
          """vae""": vae,
      }

      return components


    def    lowerCAmelCase_     (    self		:     Optional[int]      , __lowerCAmelCase		:     List[Any]      , __lowerCAmelCase		:     str=0    ):
      if str(__lowerCAmelCase    ).startswith("""mps"""    ):
        _UpperCAmelCase         =					torch.manual_seed(__lowerCAmelCase    )
      else:
        _UpperCAmelCase         =					torch.Generator(device=__lowerCAmelCase    ).manual_seed(__lowerCAmelCase    )
      _UpperCAmelCase         =					{
          """prompt""": """A painting of a squirrel eating a burger""",
          """generator""": generator,
          """num_inference_steps""": 2,
          """prior_num_inference_steps""": 2,
          """output_type""": """numpy""",
      }
      return inputs


    def    lowerCAmelCase_     (    self		:     Optional[int]    ):
      _UpperCAmelCase         =					torch_device == """cpu"""

      self._test_attention_slicing_forward_pass(test_max_difference=__lowerCAmelCase    )


    def    lowerCAmelCase_     (    self		:     List[str]    ):
      _UpperCAmelCase         =					torch_device in ["""cpu""", """mps"""]

      self._test_inference_batch_single_identical(test_max_difference=__lowerCAmelCase    )



@slow
@require_torch_gpu
class 				a (						unittest.TestCase      ):


    def    lowerCAmelCase_     (    self		:     str    ):
      # clean up the VRAM after each test
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()


    def    lowerCAmelCase_     (    self		:     List[Any]    ):
      _UpperCAmelCase         =					load_numpy(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""    )

      _UpperCAmelCase         =					StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""      , torch_dtype=torch.floataa    )
      pipe.to(__lowerCAmelCase    )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase    )
      # stable unclip will oom when integration tests are run on a V100,
      # so turn on memory savings
      pipe.enable_attention_slicing()
      pipe.enable_sequential_cpu_offload()

      _UpperCAmelCase         =					torch.Generator(device="""cpu"""    ).manual_seed(0    )
      _UpperCAmelCase         =					pipe("""anime turle"""      , generator=__lowerCAmelCase      , output_type="""np"""    )

      _UpperCAmelCase         =					output.images[0]

      assert image.shape == (768, 768, 3)

      assert_mean_pixel_difference(__lowerCAmelCase      , __lowerCAmelCase    )


    def    lowerCAmelCase_     (    self		:     Any    ):
      torch.cuda.empty_cache()
      torch.cuda.reset_max_memory_allocated()
      torch.cuda.reset_peak_memory_stats()

      _UpperCAmelCase         =					StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""      , torch_dtype=torch.floataa    )
      _UpperCAmelCase         =					pipe.to(__lowerCAmelCase    )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase    )
      pipe.enable_attention_slicing()
      pipe.enable_sequential_cpu_offload()

      _UpperCAmelCase         =					pipe(
          """anime turtle"""      , prior_num_inference_steps=2      , num_inference_steps=2      , output_type="""np"""      , )

      _UpperCAmelCase         =					torch.cuda.max_memory_allocated()
      # make sure that less than 7 GB is allocated
      assert mem_bytes < 7 * 10**9
