





'''simple docstring'''


import gc
import random
import tempfile
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMInverseScheduler,
    DDIMScheduler,
    DPMSolverMultistepInverseScheduler,
    DPMSolverMultistepScheduler,
    StableDiffusionDiffEditPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_image, slow
from diffusers.utils.testing_utils import enable_full_determinism, floats_tensor, require_torch_gpu, torch_device

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


enable_full_determinism()
class 				a_       (    _lowerCAmelCase     ,						_lowerCAmelCase     ,						unittest.TestCase							):
     __A								=      StableDiffusionDiffEditPipeline
     __A								=      TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {"height", "width", "image"} | {"image_latents"}
     __A								=      TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS - {"image"} | {"image_latents"}
     __A								=      frozenset(
         []							)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
     __A								=      frozenset([]							)





     def 							lowercase__				(			self		: Optional[Any]						):





        """simple docstring"""




        torch.manual_seed(0						)
        lowercase_      :str         =			UNetaDConditionModel(
            block_out_channels=(32, 64)     ,				layers_per_block=2     ,				sample_size=32     ,				in_channels=4     ,				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_      :Any         =			DDIMScheduler(
            beta_start=0.0_00_85     ,				beta_end=0.0_12     ,				beta_schedule="scaled_linear"     ,				clip_sample=__A     ,				set_alpha_to_one=__A     ,				)
        lowercase_      :Union[str, Any]         =			DDIMInverseScheduler(
            beta_start=0.0_00_85     ,				beta_end=0.0_12     ,				beta_schedule="scaled_linear"     ,				clip_sample=__A     ,				set_alpha_to_zero=__A     ,				)
        torch.manual_seed(0						)
        lowercase_      :Union[str, Any]         =			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_      :int         =			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_      :Union[str, Any]         =			CLIPTextModel(__A						)
        lowercase_      :Optional[int]         =			CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"						)

        lowercase_      :int         =			{
            "unet": unet,
            "scheduler": scheduler,
            "inverse_scheduler": inverse_scheduler,
            "vae": vae,
            "text_encoder": text_encoder,
            "tokenizer": tokenizer,
            "safety_checker": None,
            "feature_extractor": None,
        }

        return components





     def 							lowercase__				(			self		: Any     ,				lowercase		: Optional[int]     ,				lowercase		: Union[str, Any]=0						):





        """simple docstring"""




        lowercase_      :int         =			floats_tensor((1, 16, 16)     ,				rng=random.Random(__A						)						).to(__A						)
        lowercase_      :List[str]         =			floats_tensor((1, 2, 4, 16, 16)     ,				rng=random.Random(__A						)						).to(__A						)
        if str(__A						).startswith("mps"						):
           lowercase_      :Any         =			torch.manual_seed(__A						)
        else:
           lowercase_      :Optional[int]         =			torch.Generator(device=__A						).manual_seed(__A						)
        lowercase_      :List[Any]         =			{
            "prompt": "a dog and a newt",
            "mask_image": mask,
            "image_latents": latents,
            "generator": generator,
            "num_inference_steps": 2,
            "inpaint_strength": 1.0,
            "guidance_scale": 6.0,
            "output_type": "numpy",
        }

        return inputs





     def 							lowercase__				(			self		: Union[str, Any]     ,				lowercase		: int     ,				lowercase		: List[str]=0						):





        """simple docstring"""




        lowercase_      :List[Any]         =			floats_tensor((1, 3, 32, 32)     ,				rng=random.Random(__A						)						).to(__A						)
        lowercase_      :Optional[Any]         =			image.cpu().permute(0     ,				2     ,				3     ,				1						)[0]
        lowercase_      :int         =			Image.fromarray(np.uinta(__A						)						).convert("RGB"						)
        if str(__A						).startswith("mps"						):
           lowercase_      :List[str]         =			torch.manual_seed(__A						)
        else:
           lowercase_      :List[Any]         =			torch.Generator(device=__A						).manual_seed(__A						)
        lowercase_      :Union[str, Any]         =			{
            "image": image,
            "source_prompt": "a cat and a frog",
            "target_prompt": "a dog and a newt",
            "generator": generator,
            "num_inference_steps": 2,
            "num_maps_per_mask": 2,
            "mask_encode_strength": 1.0,
            "guidance_scale": 6.0,
            "output_type": "numpy",
        }

        return inputs





     def 							lowercase__				(			self		: Optional[int]     ,				lowercase		: Any     ,				lowercase		: int=0						):





        """simple docstring"""




        lowercase_      :Tuple         =			floats_tensor((1, 3, 32, 32)     ,				rng=random.Random(__A						)						).to(__A						)
        lowercase_      :Union[str, Any]         =			image.cpu().permute(0     ,				2     ,				3     ,				1						)[0]
        lowercase_      :Union[str, Any]         =			Image.fromarray(np.uinta(__A						)						).convert("RGB"						)
        if str(__A						).startswith("mps"						):
           lowercase_      :List[str]         =			torch.manual_seed(__A						)
        else:
           lowercase_      :Dict         =			torch.Generator(device=__A						).manual_seed(__A						)
        lowercase_      :Union[str, Any]         =			{
            "image": image,
            "prompt": "a cat and a frog",
            "generator": generator,
            "num_inference_steps": 2,
            "inpaint_strength": 1.0,
            "guidance_scale": 6.0,
            "decode_latents": True,
            "output_type": "numpy",
        }
        return inputs





     def 							lowercase__				(			self		: Optional[int]						):





        """simple docstring"""




        if not hasattr(self.pipeline_class     ,				"_optional_components"						):
           return

        lowercase_      :Tuple         =			self.get_dummy_components()
        lowercase_      :Dict         =			self.pipeline_class(**__A						)
        pipe.to(__A						)
        pipe.set_progress_bar_config(disable=__A						)

        # set all optional components to None and update pipeline config accordingly
        for optional_component in pipe._optional_components:
           setattr(__A     ,				__A     ,				__A						)
        pipe.register_modules(**{optional_component: None for optional_component in pipe._optional_components}						)

        lowercase_      :Any         =			self.get_dummy_inputs(__A						)
        lowercase_      :Dict         =			pipe(**__A						)[0]

        with tempfile.TemporaryDirectory() as tmpdir:
           pipe.save_pretrained(__A						)
           lowercase_      :Optional[int]         =			self.pipeline_class.from_pretrained(__A						)
           pipe_loaded.to(__A						)
           pipe_loaded.set_progress_bar_config(disable=__A						)

        for optional_component in pipe._optional_components:
           self.assertTrue(
               getattr(__A     ,				__A						) is None     ,				F'`{optional_component}` did not stay set to None after loading.'     ,				)

        lowercase_      :Union[str, Any]         =			self.get_dummy_inputs(__A						)
        lowercase_      :Optional[Any]         =			pipe_loaded(**__A						)[0]

        lowercase_      :List[str]         =			np.abs(output - output_loaded						).max()
        self.assertLess(__A     ,				1e-4						)





     def 							lowercase__				(			self		: Any						):





        """simple docstring"""




        lowercase_      :Tuple         =			"cpu"

        lowercase_      :Dict         =			self.get_dummy_components()
        lowercase_      :Tuple         =			self.pipeline_class(**__A						)
        pipe.to(__A						)
        pipe.set_progress_bar_config(disable=__A						)

        lowercase_      :Union[str, Any]         =			self.get_dummy_mask_inputs(__A						)
        lowercase_      :int         =			pipe.generate_mask(**__A						)
        lowercase_      :Tuple         =			mask[0, -3:, -3:]

        self.assertEqual(mask.shape     ,				(1, 16, 16)						)
        lowercase_      :str         =			np.array([0] * 9						)
        lowercase_      :Optional[int]         =			np.abs(mask_slice.flatten() - expected_slice						).max()
        self.assertLessEqual(__A     ,				1e-3						)
        self.assertEqual(mask[0, -3, -4]     ,				0						)





     def 							lowercase__				(			self		: str						):





        """simple docstring"""




        lowercase_      :Union[str, Any]         =			"cpu"

        lowercase_      :Optional[int]         =			self.get_dummy_components()
        lowercase_      :Optional[Any]         =			self.pipeline_class(**__A						)
        pipe.to(__A						)
        pipe.set_progress_bar_config(disable=__A						)

        lowercase_      :List[str]         =			self.get_dummy_inversion_inputs(__A						)
        lowercase_      :Union[str, Any]         =			pipe.invert(**__A						).images
        lowercase_      :Dict         =			image[0, -1, -3:, -3:]

        self.assertEqual(image.shape     ,				(2, 32, 32, 3)						)
        lowercase_      :Dict         =			np.array(
            [0.51_50, 0.51_34, 0.50_43, 0.53_76, 0.46_94, 0.5_10_50, 0.50_15, 0.44_07, 0.47_99]     ,				)
        lowercase_      :List[str]         =			np.abs(image_slice.flatten() - expected_slice						).max()
        self.assertLessEqual(__A     ,				1e-3						)





     def 							lowercase__				(			self		: Dict						):





        """simple docstring"""




        super().test_inference_batch_single_identical(expected_max_diff=5e-3						)







     def 							lowercase__				(			self		: Optional[Any]						):





        """simple docstring"""




        lowercase_      :Union[str, Any]         =			"cpu"

        lowercase_      :Union[str, Any]         =			self.get_dummy_components()

        lowercase_      :Optional[int]         =			{"beta_start": 0.0_00_85, "beta_end": 0.0_12, "beta_schedule": "scaled_linear"}
        lowercase_      :List[Any]         =			DPMSolverMultistepScheduler(**__A						)
        lowercase_      :List[Any]         =			DPMSolverMultistepInverseScheduler(**__A						)

        lowercase_      :List[Any]         =			self.pipeline_class(**__A						)
        pipe.to(__A						)
        pipe.set_progress_bar_config(disable=__A						)

        lowercase_      :Dict         =			self.get_dummy_inversion_inputs(__A						)
        lowercase_      :str         =			pipe.invert(**__A						).images
        lowercase_      :Optional[Any]         =			image[0, -1, -3:, -3:]

        self.assertEqual(image.shape     ,				(2, 32, 32, 3)						)
        lowercase_      :Union[str, Any]         =			np.array(
            [0.51_50, 0.51_34, 0.50_43, 0.53_76, 0.46_94, 0.5_10_50, 0.50_15, 0.44_07, 0.47_99]     ,				)
        lowercase_      :Dict         =			np.abs(image_slice.flatten() - expected_slice						).max()
        self.assertLessEqual(__A     ,				1e-3						)







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





     def 							lowercase__				(			self		: str						):





        """simple docstring"""




        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()





     @classmethod
     def 							lowercase__				(			cls		: Optional[int]						):





        """simple docstring"""




        lowercase_      :List[Any]         =			load_image(
            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/diffedit/fruit.png"						)

        lowercase_      :Any         =			raw_image.convert("RGB"						).resize((768, 768)						)

        lowercase_      :Any         =			raw_image





     def 							lowercase__				(			self		: Dict						):





        """simple docstring"""




        lowercase_      :int         =			torch.manual_seed(0						)

        lowercase_      :Dict         =			StableDiffusionDiffEditPipeline.from_pretrained(
            "stabilityai/stable-diffusion-2-1"     ,				safety_checker=__A     ,				torch_dtype=torch.floataa						)
        lowercase_      :Any         =			DDIMScheduler.from_config(pipe.scheduler.config						)
        lowercase_      :Union[str, Any]         =			DDIMInverseScheduler.from_config(pipe.scheduler.config						)
        pipe.enable_model_cpu_offload()
        pipe.set_progress_bar_config(disable=__A						)

        lowercase_      :str         =			"a bowl of fruit"
        lowercase_      :int         =			"a bowl of pears"

        lowercase_      :List[str]         =			pipe.generate_mask(
            image=self.raw_image     ,				source_prompt=__A     ,				target_prompt=__A     ,				generator=__A     ,				)

        lowercase_      :Optional[Any]         =			pipe.invert(
            prompt=__A     ,				image=self.raw_image     ,				inpaint_strength=0.7     ,				generator=__A						).latents

        lowercase_      :List[str]         =			pipe(
            prompt=__A     ,				mask_image=__A     ,				image_latents=__A     ,				generator=__A     ,				negative_prompt=__A     ,				inpaint_strength=0.7     ,				output_type="numpy"     ,				).images[0]

        lowercase_      :Any         =			(
            np.array(
                load_image(
                    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
                    "/diffedit/pears.png"						).resize((768, 768)						)						)
            / 255
        )
        assert np.abs((expected_image - image).max()						) < 5e-1







     def 							lowercase__				(			self		: str						):





        """simple docstring"""




        lowercase_      :Optional[int]         =			torch.manual_seed(0						)

        lowercase_      :Dict         =			StableDiffusionDiffEditPipeline.from_pretrained(
            "stabilityai/stable-diffusion-2-1"     ,				safety_checker=__A     ,				torch_dtype=torch.floataa						)
        lowercase_      :Optional[int]         =			DPMSolverMultistepScheduler.from_config(pipe.scheduler.config						)
        lowercase_      :Any         =			DPMSolverMultistepInverseScheduler.from_config(pipe.scheduler.config						)
        pipe.enable_model_cpu_offload()
        pipe.set_progress_bar_config(disable=__A						)

        lowercase_      :Union[str, Any]         =			"a bowl of fruit"
        lowercase_      :List[Any]         =			"a bowl of pears"

        lowercase_      :Any         =			pipe.generate_mask(
            image=self.raw_image     ,				source_prompt=__A     ,				target_prompt=__A     ,				generator=__A     ,				)

        lowercase_      :Tuple         =			pipe.invert(
            prompt=__A     ,				image=self.raw_image     ,				inpaint_strength=0.7     ,				generator=__A     ,				num_inference_steps=25     ,				).latents

        lowercase_      :Any         =			pipe(
            prompt=__A     ,				mask_image=__A     ,				image_latents=__A     ,				generator=__A     ,				negative_prompt=__A     ,				inpaint_strength=0.7     ,				num_inference_steps=25     ,				output_type="numpy"     ,				).images[0]

        lowercase_      :Optional[int]         =			(
            np.array(
                load_image(
                    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
                    "/diffedit/pears.png"						).resize((768, 768)						)						)
            / 255
        )
        assert np.abs((expected_image - image).max()						) < 5e-1









'''simple docstring'''


from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
    is_vision_available,
)


lowerCAmelCase :					int 							={
    '''configuration_clip''': [
        '''CLIP_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''CLIPConfig''',
        '''CLIPOnnxConfig''',
        '''CLIPTextConfig''',
        '''CLIPVisionConfig''',
    ],
    '''processing_clip''': ['''CLIPProcessor'''],
    '''tokenization_clip''': ['''CLIPTokenizer'''],
}

try:
 if not is_tokenizers_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 lowerCAmelCase :					Any 							=['''CLIPTokenizerFast''']

try:
 if not is_vision_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 lowerCAmelCase :					str 							=['''CLIPFeatureExtractor''']
 lowerCAmelCase :					Optional[int] 							=['''CLIPImageProcessor''']

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 lowerCAmelCase :					Any 							=[
     '''CLIP_PRETRAINED_MODEL_ARCHIVE_LIST''',
     '''CLIPModel''',
     '''CLIPPreTrainedModel''',
     '''CLIPTextModel''',
     '''CLIPTextModelWithProjection''',
     '''CLIPVisionModel''',
     '''CLIPVisionModelWithProjection''',
 ]

try:
 if not is_tf_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 lowerCAmelCase :					Union[str, Any] 							=[
     '''TF_CLIP_PRETRAINED_MODEL_ARCHIVE_LIST''',
     '''TFCLIPModel''',
     '''TFCLIPPreTrainedModel''',
     '''TFCLIPTextModel''',
     '''TFCLIPVisionModel''',
 ]

try:
 if not is_flax_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 lowerCAmelCase :					List[Any] 							=[
     '''FlaxCLIPModel''',
     '''FlaxCLIPPreTrainedModel''',
     '''FlaxCLIPTextModel''',
     '''FlaxCLIPTextPreTrainedModel''',
     '''FlaxCLIPVisionModel''',
     '''FlaxCLIPVisionPreTrainedModel''',
 ]


if TYPE_CHECKING:
 from .configuration_clip import (
     CLIP_PRETRAINED_CONFIG_ARCHIVE_MAP,
     CLIPConfig,
     CLIPOnnxConfig,
     CLIPTextConfig,
     CLIPVisionConfig,
 )
 from .processing_clip import CLIPProcessor
 from .tokenization_clip import CLIPTokenizer

 try:
  if not is_tokenizers_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .tokenization_clip_fast import CLIPTokenizerFast

 try:
  if not is_vision_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .feature_extraction_clip import CLIPFeatureExtractor
  from .image_processing_clip import CLIPImageProcessor

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_clip import (
      CLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
      CLIPModel,
      CLIPPreTrainedModel,
      CLIPTextModel,
      CLIPTextModelWithProjection,
      CLIPVisionModel,
      CLIPVisionModelWithProjection,
  )

 try:
  if not is_tf_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_tf_clip import (
      TF_CLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
      TFCLIPModel,
      TFCLIPPreTrainedModel,
      TFCLIPTextModel,
      TFCLIPVisionModel,
  )

 try:
  if not is_flax_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_flax_clip import (
      FlaxCLIPModel,
      FlaxCLIPPreTrainedModel,
      FlaxCLIPTextModel,
      FlaxCLIPTextPreTrainedModel,
      FlaxCLIPVisionModel,
      FlaxCLIPVisionPreTrainedModel,
  )


else:
 import sys

 lowerCAmelCase :					int 							=_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)


