



import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, PNDMScheduler, StableDiffusionInpaintPipeline, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, slow

from ..pipeline_params import TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS, TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


class 				snake_case (__lowercase		,      __lowercase		,      __lowercase		,      unittest.TestCase   ):
       lowerCAmelCase__  :str						   =					StableDiffusionInpaintPipeline
       lowerCAmelCase__  :Dict						   =					TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
       lowerCAmelCase__  :str						   =					TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
       lowerCAmelCase__  :Optional[int]						   =					frozenset(
           []   )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
       lowerCAmelCase__  :Any						   =					frozenset([]   )



       def 							_a      (				self      )  ->							List[str]:
           torch.manual_seed(0      )
           lowercase__		       =						UNetaDConditionModel(
               block_out_channels=(32, 64)     ,layers_per_block=2     ,sample_size=32     ,in_channels=9     ,out_channels=4     ,down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")     ,up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")     ,cross_attention_dim=32     ,attention_head_dim=(2, 4)     ,use_linear_projection=__A     ,)
           lowercase__		       =						PNDMScheduler(skip_prk_steps=__A      )
           torch.manual_seed(0      )
           lowercase__		       =						AutoencoderKL(
               block_out_channels=[32, 64]     ,in_channels=3     ,out_channels=3     ,down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]     ,up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]     ,latent_channels=4     ,sample_size=128     ,)
           torch.manual_seed(0      )
           lowercase__		       =						CLIPTextConfig(
               bos_token_id=0     ,eos_token_id=2     ,hidden_size=32     ,intermediate_size=37     ,layer_norm_eps=1E-0_5     ,num_attention_heads=4     ,num_hidden_layers=5     ,pad_token_id=1     ,vocab_size=1_000     ,hidden_act="gelu"     ,projection_dim=512     ,)
           lowercase__		       =						CLIPTextModel(__A      )
           lowercase__		       =						CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"      )

           lowercase__		       =						{
               "unet": unet,
               "scheduler": scheduler,
               "vae": vae,
               "text_encoder": text_encoder,
               "tokenizer": tokenizer,
               "safety_checker": None,
               "feature_extractor": None,
           }
           return components



       def 							_a      (				self     ,UpperCAmelCase_     ,UpperCAmelCase_=0      )  ->							Tuple:
           # TODO: use tensor inputs instead of PIL, this is here just to leave the old expected_slices untouched
           lowercase__		       =						floats_tensor((1, 3, 32, 32)     ,rng=random.Random(__A      )      ).to(__A      )
           lowercase__		       =						image.cpu().permute(0     ,2     ,3     ,1      )[0]
           lowercase__		       =						Image.fromarray(np.uinta(__A      )      ).convert("RGB"      ).resize((64, 64)      )
           lowercase__		       =						Image.fromarray(np.uinta(image + 4      )      ).convert("RGB"      ).resize((64, 64)      )
           if str(__A      ).startswith("mps"      ):
               lowercase__		       =						torch.manual_seed(__A      )
           else:
               lowercase__		       =						torch.Generator(device=__A      ).manual_seed(__A      )
           lowercase__		       =						{
               "prompt": "A painting of a squirrel eating a burger",
               "image": init_image,
               "mask_image": mask_image,
               "generator": generator,
               "num_inference_steps": 2,
               "guidance_scale": 6.0,
               "output_type": "numpy",
           }
           return inputs



       def 							_a      (				self      )  ->							Optional[Any]:
           lowercase__		       =						"cpu"  # ensure determinism for the device-dependent torch.Generator
           lowercase__		       =						self.get_dummy_components()
           lowercase__		       =						StableDiffusionInpaintPipeline(**__A      )
           lowercase__		       =						sd_pipe.to(__A      )
           sd_pipe.set_progress_bar_config(disable=__A      )

           lowercase__		       =						self.get_dummy_inputs(__A      )
           lowercase__		       =						sd_pipe(**__A      ).images
           lowercase__		       =						image[0, -3:, -3:, -1]

           assert image.shape == (1, 64, 64, 3)
           lowercase__		       =						np.array([0.47_27, 0.57_35, 0.39_41, 0.54_46, 0.59_26, 0.43_94, 0.50_62, 0.46_54, 0.44_76]      )

           assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1E-2







       def 							_a      (				self      )  ->							Optional[Any]:
           super().test_inference_batch_single_identical(expected_max_diff=3E-3      )





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



       def 							_a      (				self      )  ->							Tuple:
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()



       def 							_a      (				self      )  ->							Optional[int]:
           lowercase__		       =						load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
               "/sd2-inpaint/init_image.png"      )
           lowercase__		       =						load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint/mask.png"      )
           lowercase__		       =						load_numpy(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint"
               "/yellow_cat_sitting_on_a_park_bench.npy"      )

           lowercase__		       =						"stabilityai/stable-diffusion-2-inpainting"
           lowercase__		       =						StableDiffusionInpaintPipeline.from_pretrained(__A     ,safety_checker=__A      )
           pipe.to(__A      )
           pipe.set_progress_bar_config(disable=__A      )
           pipe.enable_attention_slicing()

           lowercase__		       =						"Face of a yellow cat, high resolution, sitting on a park bench"

           lowercase__		       =						torch.manual_seed(0      )
           lowercase__		       =						pipe(
               prompt=__A     ,image=__A     ,mask_image=__A     ,generator=__A     ,output_type="np"     ,)
           lowercase__		       =						output.images[0]

           assert image.shape == (512, 512, 3)
           assert np.abs(expected_image - image      ).max() < 9E-3



       def 							_a      (				self      )  ->							Optional[Any]:
           lowercase__		       =						load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
               "/sd2-inpaint/init_image.png"      )
           lowercase__		       =						load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint/mask.png"      )
           lowercase__		       =						load_numpy(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint"
               "/yellow_cat_sitting_on_a_park_bench_fp16.npy"      )

           lowercase__		       =						"stabilityai/stable-diffusion-2-inpainting"
           lowercase__		       =						StableDiffusionInpaintPipeline.from_pretrained(
               __A     ,torch_dtype=torch.floataa     ,safety_checker=__A     ,)
           pipe.to(__A      )
           pipe.set_progress_bar_config(disable=__A      )
           pipe.enable_attention_slicing()

           lowercase__		       =						"Face of a yellow cat, high resolution, sitting on a park bench"

           lowercase__		       =						torch.manual_seed(0      )
           lowercase__		       =						pipe(
               prompt=__A     ,image=__A     ,mask_image=__A     ,generator=__A     ,output_type="np"     ,)
           lowercase__		       =						output.images[0]

           assert image.shape == (512, 512, 3)
           assert np.abs(expected_image - image      ).max() < 5E-1







       def 							_a      (				self      )  ->							List[Any]:
           torch.cuda.empty_cache()
           torch.cuda.reset_max_memory_allocated()
           torch.cuda.reset_peak_memory_stats()

           lowercase__		       =						load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
               "/sd2-inpaint/init_image.png"      )
           lowercase__		       =						load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint/mask.png"      )

           lowercase__		       =						"stabilityai/stable-diffusion-2-inpainting"
           lowercase__		       =						PNDMScheduler.from_pretrained(__A     ,subfolder="scheduler"      )
           lowercase__		       =						StableDiffusionInpaintPipeline.from_pretrained(
               __A     ,safety_checker=__A     ,scheduler=__A     ,torch_dtype=torch.floataa     ,)
           pipe.to(__A      )
           pipe.set_progress_bar_config(disable=__A      )
           pipe.enable_attention_slicing(1      )
           pipe.enable_sequential_cpu_offload()

           lowercase__		       =						"Face of a yellow cat, high resolution, sitting on a park bench"

           lowercase__		       =						torch.manual_seed(0      )
           lowercase__		       =						pipe(
               prompt=__A     ,image=__A     ,mask_image=__A     ,generator=__A     ,num_inference_steps=2     ,output_type="np"     ,)

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









'''simple docstring'''





import copy
import random

from transformers import CLIPTokenizer


class 				snake_case (UpperCamelCase   ):



       def __init__(				self     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )  ->							List[str]:
           super().__init__(*UpperCAmelCase_     ,**UpperCAmelCase_      )
           lowercase__		       =						{}



       def 							_a      (				self     ,UpperCAmelCase_     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )  ->							str:
           lowercase__		       =						super().add_tokens(UpperCAmelCase_     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )
           if num_added_tokens == 0:
               raise ValueError(
                   F'''The tokenizer already contains the token {placeholder_token}. Please pass a different'''
                   " `placeholder_token` that is not already in the tokenizer."      )



       def 							_a      (				self     ,UpperCAmelCase_     ,*UpperCAmelCase_     ,UpperCAmelCase_=1     ,**UpperCAmelCase_      )  ->							List[Any]:
           lowercase__		       =						[]
           if num_vec_per_token == 1:
               self.try_adding_tokens(UpperCAmelCase_     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )
               output.append(UpperCAmelCase_      )
           else:
               lowercase__		       =						[]
               for i in range(UpperCAmelCase_      ):
                   lowercase__		       =						placeholder_token + F'''_{i}'''
                   self.try_adding_tokens(UpperCAmelCase_     ,*UpperCAmelCase_     ,**UpperCAmelCase_      )
                   output.append(UpperCAmelCase_      )
        # handle cases where there is a new placeholder token that contains the current placeholder token but is larger
           for token in self.token_map:
               if token in placeholder_token:
                   raise ValueError(
                       F'''The tokenizer already has placeholder token {token} that can get confused with'''
                       F''' {placeholder_token}keep placeholder tokens independent'''      )
           lowercase__		       =						output



       def 							_a      (				self     ,UpperCAmelCase_     ,UpperCAmelCase_=False     ,UpperCAmelCase_=1.0      )  ->							Optional[int]:
           if isinstance(UpperCAmelCase_     ,UpperCAmelCase_      ):
               lowercase__		       =						[]
               for i in range(len(UpperCAmelCase_      )      ):
                   output.append(self.replace_placeholder_tokens_in_text(text[i]     ,vector_shuffle=UpperCAmelCase_      )      )
               return output
           for placeholder_token in self.token_map:
               if placeholder_token in text:
                   lowercase__		       =						self.token_map[placeholder_token]
                   lowercase__		       =						tokens[: 1 + int(len(UpperCAmelCase_      ) * prop_tokens_to_load      )]
                   if vector_shuffle:
                       lowercase__		       =						copy.copy(UpperCAmelCase_      )
                       random.shuffle(UpperCAmelCase_      )
                   lowercase__		       =						text.replace(UpperCAmelCase_     ," ".join(UpperCAmelCase_      )      )
           return text



       def __call__(				self     ,UpperCAmelCase_     ,*UpperCAmelCase_     ,UpperCAmelCase_=False     ,UpperCAmelCase_=1.0     ,**UpperCAmelCase_      )  ->							int:
           return super().__call__(
               self.replace_placeholder_tokens_in_text(
                   UpperCAmelCase_     ,vector_shuffle=UpperCAmelCase_     ,prop_tokens_to_load=UpperCAmelCase_      )     ,*UpperCAmelCase_     ,**UpperCAmelCase_     ,)







       def 							_a      (				self     ,UpperCAmelCase_     ,*UpperCAmelCase_     ,UpperCAmelCase_=False     ,UpperCAmelCase_=1.0     ,**UpperCAmelCase_      )  ->							Optional[int]:
           return super().encode(
               self.replace_placeholder_tokens_in_text(
                   UpperCAmelCase_     ,vector_shuffle=UpperCAmelCase_     ,prop_tokens_to_load=UpperCAmelCase_      )     ,*UpperCAmelCase_     ,**UpperCAmelCase_     ,)




