






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    _lowerCAmelCase			(		lowercase_     ,      lowercase_     ,      unittest.TestCase      ):



				"""simple docstring"""



				_lowercase				:   str      =  StableDiffusionDiffEditPipeline
				_lowercase				:   int      =  TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {'''height''', '''width''', '''image'''} | {'''image_latents'''}
				_lowercase				:   List[Any]      =  TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS - {'''image'''} | {'''image_latents'''}
				_lowercase				:   str      =  frozenset(
				    []      )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
				_lowercase				:   Optional[int]      =  frozenset([]      )



				def 					__magic_name__							(      self    :    int):






											'''simple docstring'''







											torch.manual_seed(0)
											snake_case__	  =       UNetaDConditionModel(
											    block_out_channels=(3_2, 6_4)			,		layers_per_block=2			,		sample_size=3_2			,		in_channels=4			,		out_channels=4			,		down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")			,		up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")			,		cross_attention_dim=3_2			,		attention_head_dim=(2, 4)			,		use_linear_projection=UpperCamelCase__			,		)
											snake_case__	  =       DDIMScheduler(
											    beta_start=0.0_00_85			,		beta_end=0.0_12			,		beta_schedule="""scaled_linear"""			,		clip_sample=UpperCamelCase__			,		set_alpha_to_one=UpperCamelCase__			,		)
											snake_case__	  =       DDIMInverseScheduler(
											    beta_start=0.0_00_85			,		beta_end=0.0_12			,		beta_schedule="""scaled_linear"""			,		clip_sample=UpperCamelCase__			,		set_alpha_to_zero=UpperCamelCase__			,		)
											torch.manual_seed(0)
											snake_case__	  =       AutoencoderKL(
											    block_out_channels=[3_2, 6_4]			,		in_channels=3			,		out_channels=3			,		down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]			,		up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]			,		latent_channels=4			,		sample_size=1_2_8			,		)
											torch.manual_seed(0)
											snake_case__	  =       CLIPTextConfig(
											    bos_token_id=0			,		eos_token_id=2			,		hidden_size=3_2			,		intermediate_size=3_7			,		layer_norm_eps=1E-05			,		num_attention_heads=4			,		num_hidden_layers=5			,		pad_token_id=1			,		vocab_size=1_0_0_0			,		hidden_act="""gelu"""			,		projection_dim=5_1_2			,		)
											snake_case__	  =       CLIPTextModel(UpperCamelCase__)
											snake_case__	  =       CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

											snake_case__	  =       {
											    """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 					__magic_name__							(      self    :    Union[str, Any]			,		UpperCamelCase__    :    int			,		UpperCamelCase__    :    Union[str, Any]=0):






											'''simple docstring'''







											snake_case__	  =       floats_tensor((1, 1_6, 1_6)			,		rng=random.Random(UpperCamelCase__)).to(UpperCamelCase__)
											snake_case__	  =       floats_tensor((1, 2, 4, 1_6, 1_6)			,		rng=random.Random(UpperCamelCase__)).to(UpperCamelCase__)
											if str(UpperCamelCase__).startswith("""mps"""):
																		snake_case__	  =       torch.manual_seed(UpperCamelCase__)
											else:
																		snake_case__	  =       torch.Generator(device=UpperCamelCase__).manual_seed(UpperCamelCase__)
											snake_case__	  =       {
											    """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 					__magic_name__							(      self    :    int			,		UpperCamelCase__    :    Union[str, Any]			,		UpperCamelCase__    :    int=0):






											'''simple docstring'''







											snake_case__	  =       floats_tensor((1, 3, 3_2, 3_2)			,		rng=random.Random(UpperCamelCase__)).to(UpperCamelCase__)
											snake_case__	  =       image.cpu().permute(0			,		2			,		3			,		1)[0]
											snake_case__	  =       Image.fromarray(np.uinta(UpperCamelCase__)).convert("""RGB""")
											if str(UpperCamelCase__).startswith("""mps"""):
																		snake_case__	  =       torch.manual_seed(UpperCamelCase__)
											else:
																		snake_case__	  =       torch.Generator(device=UpperCamelCase__).manual_seed(UpperCamelCase__)
											snake_case__	  =       {
											    """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 					__magic_name__							(      self    :    Any			,		UpperCamelCase__    :    Union[str, Any]			,		UpperCamelCase__    :    str=0):






											'''simple docstring'''







											snake_case__	  =       floats_tensor((1, 3, 3_2, 3_2)			,		rng=random.Random(UpperCamelCase__)).to(UpperCamelCase__)
											snake_case__	  =       image.cpu().permute(0			,		2			,		3			,		1)[0]
											snake_case__	  =       Image.fromarray(np.uinta(UpperCamelCase__)).convert("""RGB""")
											if str(UpperCamelCase__).startswith("""mps"""):
																		snake_case__	  =       torch.manual_seed(UpperCamelCase__)
											else:
																		snake_case__	  =       torch.Generator(device=UpperCamelCase__).manual_seed(UpperCamelCase__)
											snake_case__	  =       {
											    """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 					__magic_name__							(      self    :    Optional[Any]):






											'''simple docstring'''







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

											snake_case__	  =       self.get_dummy_components()
											snake_case__	  =       self.pipeline_class(**UpperCamelCase__)
											pipe.to(UpperCamelCase__)
											pipe.set_progress_bar_config(disable=UpperCamelCase__)

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

											snake_case__	  =       self.get_dummy_inputs(UpperCamelCase__)
											snake_case__	  =       pipe(**UpperCamelCase__)[0]

											with tempfile.TemporaryDirectory() as tmpdir:
																		pipe.save_pretrained(UpperCamelCase__)
																		snake_case__	  =       self.pipeline_class.from_pretrained(UpperCamelCase__)
																		pipe_loaded.to(UpperCamelCase__)
																		pipe_loaded.set_progress_bar_config(disable=UpperCamelCase__)

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

											snake_case__	  =       self.get_dummy_inputs(UpperCamelCase__)
											snake_case__	  =       pipe_loaded(**UpperCamelCase__)[0]

											snake_case__	  =       np.abs(output - output_loaded).max()
											self.assertLess(UpperCamelCase__			,		1E-4)



				def 					__magic_name__							(      self    :    Optional[int]):






											'''simple docstring'''







											snake_case__	  =       """cpu"""

											snake_case__	  =       self.get_dummy_components()
											snake_case__	  =       self.pipeline_class(**UpperCamelCase__)
											pipe.to(UpperCamelCase__)
											pipe.set_progress_bar_config(disable=UpperCamelCase__)

											snake_case__	  =       self.get_dummy_mask_inputs(UpperCamelCase__)
											snake_case__	  =       pipe.generate_mask(**UpperCamelCase__)
											snake_case__	  =       mask[0, -3:, -3:]

											self.assertEqual(mask.shape			,		(1, 1_6, 1_6))
											snake_case__	  =       np.array([0] * 9)
											snake_case__	  =       np.abs(mask_slice.flatten() - expected_slice).max()
											self.assertLessEqual(UpperCamelCase__			,		1E-3)
											self.assertEqual(mask[0, -3, -4]			,		0)



				def 					__magic_name__							(      self    :    Optional[Any]):






											'''simple docstring'''







											snake_case__	  =       """cpu"""

											snake_case__	  =       self.get_dummy_components()
											snake_case__	  =       self.pipeline_class(**UpperCamelCase__)
											pipe.to(UpperCamelCase__)
											pipe.set_progress_bar_config(disable=UpperCamelCase__)

											snake_case__	  =       self.get_dummy_inversion_inputs(UpperCamelCase__)
											snake_case__	  =       pipe.invert(**UpperCamelCase__).images
											snake_case__	  =       image[0, -1, -3:, -3:]

											self.assertEqual(image.shape			,		(2, 3_2, 3_2, 3))
											snake_case__	  =       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]			,		)
											snake_case__	  =       np.abs(image_slice.flatten() - expected_slice).max()
											self.assertLessEqual(UpperCamelCase__			,		1E-3)



				def 					__magic_name__							(      self    :    Dict):






											'''simple docstring'''







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



				def 					__magic_name__							(      self    :    List[str]):






											'''simple docstring'''







											snake_case__	  =       """cpu"""

											snake_case__	  =       self.get_dummy_components()

											snake_case__	  =       {"""beta_start""": 0.0_00_85, """beta_end""": 0.0_12, """beta_schedule""": """scaled_linear"""}
											snake_case__	  =       DPMSolverMultistepScheduler(**UpperCamelCase__)
											snake_case__	  =       DPMSolverMultistepInverseScheduler(**UpperCamelCase__)

											snake_case__	  =       self.pipeline_class(**UpperCamelCase__)
											pipe.to(UpperCamelCase__)
											pipe.set_progress_bar_config(disable=UpperCamelCase__)

											snake_case__	  =       self.get_dummy_inversion_inputs(UpperCamelCase__)
											snake_case__	  =       pipe.invert(**UpperCamelCase__).images
											snake_case__	  =       image[0, -1, -3:, -3:]

											self.assertEqual(image.shape			,		(2, 3_2, 3_2, 3))
											snake_case__	  =       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]			,		)
											snake_case__	  =       np.abs(image_slice.flatten() - expected_slice).max()
											self.assertLessEqual(UpperCamelCase__			,		1E-3)


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



				"""simple docstring"""



				def 					__magic_name__							(      self    :    Dict):






											'''simple docstring'''







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



				@classmethod
				def 					__magic_name__							(      cls    :    List[Any]):






											'''simple docstring'''







											snake_case__	  =       load_image(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/diffedit/fruit.png""")

											snake_case__	  =       raw_image.convert("""RGB""").resize((7_6_8, 7_6_8))

											snake_case__	  =       raw_image



				def 					__magic_name__							(      self    :    Dict):






											'''simple docstring'''







											snake_case__	  =       torch.manual_seed(0)

											snake_case__	  =       StableDiffusionDiffEditPipeline.from_pretrained(
											    """stabilityai/stable-diffusion-2-1"""			,		safety_checker=UpperCamelCase__			,		torch_dtype=torch.floataa)
											snake_case__	  =       DDIMScheduler.from_config(pipe.scheduler.config)
											snake_case__	  =       DDIMInverseScheduler.from_config(pipe.scheduler.config)
											pipe.enable_model_cpu_offload()
											pipe.set_progress_bar_config(disable=UpperCamelCase__)

											snake_case__	  =       """a bowl of fruit"""
											snake_case__	  =       """a bowl of pears"""

											snake_case__	  =       pipe.generate_mask(
											    image=self.raw_image			,		source_prompt=UpperCamelCase__			,		target_prompt=UpperCamelCase__			,		generator=UpperCamelCase__			,		)

											snake_case__	  =       pipe.invert(
											    prompt=UpperCamelCase__			,		image=self.raw_image			,		inpaint_strength=0.7			,		generator=UpperCamelCase__).latents

											snake_case__	  =       pipe(
											    prompt=UpperCamelCase__			,		mask_image=UpperCamelCase__			,		image_latents=UpperCamelCase__			,		generator=UpperCamelCase__			,		negative_prompt=UpperCamelCase__			,		inpaint_strength=0.7			,		output_type="""numpy"""			,		).images[0]

											snake_case__	  =       (
											    np.array(
											        load_image(
											            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
											            """/diffedit/pears.png""").resize((7_6_8, 7_6_8)))
											    / 2_5_5
											)
											assert np.abs((expected_image - image).max()) < 5E-1



				def 					__magic_name__							(      self    :    Tuple):






											'''simple docstring'''







											snake_case__	  =       torch.manual_seed(0)

											snake_case__	  =       StableDiffusionDiffEditPipeline.from_pretrained(
											    """stabilityai/stable-diffusion-2-1"""			,		safety_checker=UpperCamelCase__			,		torch_dtype=torch.floataa)
											snake_case__	  =       DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
											snake_case__	  =       DPMSolverMultistepInverseScheduler.from_config(pipe.scheduler.config)
											pipe.enable_model_cpu_offload()
											pipe.set_progress_bar_config(disable=UpperCamelCase__)

											snake_case__	  =       """a bowl of fruit"""
											snake_case__	  =       """a bowl of pears"""

											snake_case__	  =       pipe.generate_mask(
											    image=self.raw_image			,		source_prompt=UpperCamelCase__			,		target_prompt=UpperCamelCase__			,		generator=UpperCamelCase__			,		)

											snake_case__	  =       pipe.invert(
											    prompt=UpperCamelCase__			,		image=self.raw_image			,		inpaint_strength=0.7			,		generator=UpperCamelCase__			,		num_inference_steps=2_5			,		).latents

											snake_case__	  =       pipe(
											    prompt=UpperCamelCase__			,		mask_image=UpperCamelCase__			,		image_latents=UpperCamelCase__			,		generator=UpperCamelCase__			,		negative_prompt=UpperCamelCase__			,		inpaint_strength=0.7			,		num_inference_steps=2_5			,		output_type="""numpy"""			,		).images[0]

											snake_case__	  =       (
											    np.array(
											        load_image(
											            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
											            """/diffedit/pears.png""").resize((7_6_8, 7_6_8)))
											    / 2_5_5
											)
											assert np.abs((expected_image - image).max()) < 5E-1








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,
)


a__       =     {
    """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:
	a__       =     ["""CLIPTokenizerFast"""]

try:
	if not is_vision_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	a__       =     ["""CLIPFeatureExtractor"""]
	a__       =     ["""CLIPImageProcessor"""]

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

try:
	if not is_tf_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	a__       =     [
	    """TF_CLIP_PRETRAINED_MODEL_ARCHIVE_LIST""",
	    """TFCLIPModel""",
	    """TFCLIPPreTrainedModel""",
	    """TFCLIPTextModel""",
	    """TFCLIPVisionModel""",
	]

try:
	if not is_flax_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	a__       =     [
	    """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

	a__       =     _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)
