






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

					'''simple docstring'''

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



					def 					lowerCAmelCase (     self :		Union[str, Any]							)	->     Any:


												"""simple docstring"""

												torch.manual_seed(0							)
												__lowercase					:							Any			=   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.00085						,   beta_end=0.012						,   beta_schedule="""scaled_linear"""						,   clip_sample=__a						,   set_alpha_to_one=__a						,   )
												__lowercase					:							Optional[Any]			=   DDIMInverseScheduler(
												    beta_start=0.00085						,   beta_end=0.012						,   beta_schedule="""scaled_linear"""						,   clip_sample=__a						,   set_alpha_to_zero=__a						,   )
												torch.manual_seed(0							)
												__lowercase					:							Tuple			=   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					:							List[str]			=   CLIPTextConfig(
												    bos_token_id=0						,   eos_token_id=2						,   hidden_size=32						,   intermediate_size=37						,   layer_norm_eps=1E-05						,   num_attention_heads=4						,   num_hidden_layers=5						,   pad_token_id=1						,   vocab_size=1000						,   hidden_act="""gelu"""						,   projection_dim=512						,   )
												__lowercase					:							Dict			=   CLIPTextModel(__a							)
												__lowercase					:							Dict			=   CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""							)

												__lowercase					:							Optional[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 					lowerCAmelCase (     self :		Optional[Any]						,   __a :		str						,   __a :		Dict=0							)	->     int:


												"""simple docstring"""

												__lowercase					:							List[Any]			=   floats_tensor((1, 16, 16)						,   rng=random.Random(__a							)							).to(__a							)
												__lowercase					:							Tuple			=   floats_tensor((1, 2, 4, 16, 16)						,   rng=random.Random(__a							)							).to(__a							)
												if str(__a							).startswith("""mps"""							):
																			__lowercase					:							Union[str, Any]			=   torch.manual_seed(__a							)
												else:
																			__lowercase					:							Optional[Any]			=   torch.Generator(device=__a							).manual_seed(__a							)
												__lowercase					:							Optional[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 					lowerCAmelCase (     self :		Union[str, Any]						,   __a :		int						,   __a :		Optional[int]=0							)	->     Optional[Any]:


												"""simple docstring"""

												__lowercase					:							Union[str, Any]			=   floats_tensor((1, 3, 32, 32)						,   rng=random.Random(__a							)							).to(__a							)
												__lowercase					:							Any			=   image.cpu().permute(0						,   2						,   3						,   1							)[0]
												__lowercase					:							Optional[int]			=   Image.fromarray(np.uinta(__a							)							).convert("""RGB"""							)
												if str(__a							).startswith("""mps"""							):
																			__lowercase					:							Any			=   torch.manual_seed(__a							)
												else:
																			__lowercase					:							List[Any]			=   torch.Generator(device=__a							).manual_seed(__a							)
												__lowercase					:							List[str]			=   {
												    """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 					lowerCAmelCase (     self :		Dict						,   __a :		Tuple						,   __a :		Union[str, Any]=0							)	->     List[Any]:


												"""simple docstring"""

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


												"""simple docstring"""

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

												__lowercase					:							str			=   self.get_dummy_components()
												__lowercase					:							Optional[int]			=   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					:							Dict			=   self.get_dummy_inputs(__a							)
												__lowercase					:							Dict			=   pipe(**__a							)[0]

												with tempfile.TemporaryDirectory() as tmpdir:
																			pipe.save_pretrained(__a							)
																			__lowercase					:							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					:							Dict			=   self.get_dummy_inputs(__a							)
												__lowercase					:							Tuple			=   pipe_loaded(**__a							)[0]

												__lowercase					:							Optional[Any]			=   np.abs(output - output_loaded							).max()
												self.assertLess(__a						,   1E-4							)



					def 					lowerCAmelCase (     self :		List[Any]							)	->     int:


												"""simple docstring"""

												__lowercase					:							List[Any]			=   """cpu"""

												__lowercase					:							Dict			=   self.get_dummy_components()
												__lowercase					:							Optional[Any]			=   self.pipeline_class(**__a							)
												pipe.to(__a							)
												pipe.set_progress_bar_config(disable=__a							)

												__lowercase					:							int			=   self.get_dummy_mask_inputs(__a							)
												__lowercase					:							Union[str, Any]			=   pipe.generate_mask(**__a							)
												__lowercase					:							List[Any]			=   mask[0, -3:, -3:]

												self.assertEqual(mask.shape						,   (1, 16, 16)							)
												__lowercase					:							Any			=   np.array([0] * 9							)
												__lowercase					:							Any			=   np.abs(mask_slice.flatten() - expected_slice							).max()
												self.assertLessEqual(__a						,   1E-3							)
												self.assertEqual(mask[0, -3, -4]						,   0							)



					def 					lowerCAmelCase (     self :		Optional[Any]							)	->     Dict:


												"""simple docstring"""

												__lowercase					:							Any			=   """cpu"""

												__lowercase					:							Any			=   self.get_dummy_components()
												__lowercase					:							List[str]			=   self.pipeline_class(**__a							)
												pipe.to(__a							)
												pipe.set_progress_bar_config(disable=__a							)

												__lowercase					:							List[Any]			=   self.get_dummy_inversion_inputs(__a							)
												__lowercase					:							Any			=   pipe.invert(**__a							).images
												__lowercase					:							Tuple			=   image[0, -1, -3:, -3:]

												self.assertEqual(image.shape						,   (2, 32, 32, 3)							)
												__lowercase					:							List[str]			=   np.array(
												    [0.5150, 0.5134, 0.5043, 0.5376, 0.4694, 0.51050, 0.5015, 0.4407, 0.4799]						,   )
												__lowercase					:							Tuple			=   np.abs(image_slice.flatten() - expected_slice							).max()
												self.assertLessEqual(__a						,   1E-3							)



					def 					lowerCAmelCase (     self :		Any							)	->     List[Any]:


												"""simple docstring"""

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



					def 					lowerCAmelCase (     self :		Union[str, Any]							)	->     Optional[Any]:


												"""simple docstring"""

												__lowercase					:							Optional[Any]			=   """cpu"""

												__lowercase					:							str			=   self.get_dummy_components()

												__lowercase					:							List[str]			=   {"""beta_start""": 0.00085, """beta_end""": 0.012, """beta_schedule""": """scaled_linear"""}
												__lowercase					:							int			=   DPMSolverMultistepScheduler(**__a							)
												__lowercase					:							List[str]			=   DPMSolverMultistepInverseScheduler(**__a							)

												__lowercase					:							str			=   self.pipeline_class(**__a							)
												pipe.to(__a							)
												pipe.set_progress_bar_config(disable=__a							)

												__lowercase					:							Optional[Any]			=   self.get_dummy_inversion_inputs(__a							)
												__lowercase					:							Union[str, Any]			=   pipe.invert(**__a							).images
												__lowercase					:							Optional[int]			=   image[0, -1, -3:, -3:]

												self.assertEqual(image.shape						,   (2, 32, 32, 3)							)
												__lowercase					:							List[Any]			=   np.array(
												    [0.5150, 0.5134, 0.5043, 0.5376, 0.4694, 0.51050, 0.5015, 0.4407, 0.4799]						,   )
												__lowercase					:							List[Any]			=   np.abs(image_slice.flatten() - expected_slice							).max()
												self.assertLessEqual(__a						,   1E-3							)







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

					'''simple docstring'''

					def 					lowerCAmelCase (     self :		List[str]							)	->     List[Any]:


												"""simple docstring"""

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



					@classmethod
					def 					lowerCAmelCase (     cls :		str							)	->     Tuple:


												"""simple docstring"""

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

												__lowercase					:							Tuple			=   raw_image.convert("""RGB"""							).resize((768, 768)							)

												__lowercase					:							int			=   raw_image



					def 					lowerCAmelCase (     self :		Dict							)	->     Tuple:


												"""simple docstring"""

												__lowercase					:							int			=   torch.manual_seed(0							)

												__lowercase					:							List[str]			=   StableDiffusionDiffEditPipeline.from_pretrained(
												    """stabilityai/stable-diffusion-2-1"""						,   safety_checker=__a						,   torch_dtype=torch.floataa							)
												__lowercase					:							List[str]			=   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					:							Optional[Any]			=   """a bowl of pears"""

												__lowercase					:							Union[str, Any]			=   pipe.generate_mask(
												    image=self.raw_image						,   source_prompt=__a						,   target_prompt=__a						,   generator=__a						,   )

												__lowercase					:							int			=   pipe.invert(
												    prompt=__a						,   image=self.raw_image						,   inpaint_strength=0.7						,   generator=__a							).latents

												__lowercase					:							Optional[int]			=   pipe(
												    prompt=__a						,   mask_image=__a						,   image_latents=__a						,   generator=__a						,   negative_prompt=__a						,   inpaint_strength=0.7						,   output_type="""numpy"""						,   ).images[0]

												__lowercase					:							str			=   (
												    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 					lowerCAmelCase (     self :		List[str]							)	->     List[Any]:


												"""simple docstring"""

												__lowercase					:							Optional[Any]			=   torch.manual_seed(0							)

												__lowercase					:							List[Any]			=   StableDiffusionDiffEditPipeline.from_pretrained(
												    """stabilityai/stable-diffusion-2-1"""						,   safety_checker=__a						,   torch_dtype=torch.floataa							)
												__lowercase					:							int			=   DPMSolverMultistepScheduler.from_config(pipe.scheduler.config							)
												__lowercase					:							Optional[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					:							Tuple			=   pipe.generate_mask(
												    image=self.raw_image						,   source_prompt=__a						,   target_prompt=__a						,   generator=__a						,   )

												__lowercase					:							str			=   pipe.invert(
												    prompt=__a						,   image=self.raw_image						,   inpaint_strength=0.7						,   generator=__a						,   num_inference_steps=25						,   ).latents

												__lowercase					:							Union[str, 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					:							List[str]			=   (
												    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







from maths.prime_check import is_prime
def 					snake_case_						(   lowerCAmelCase_				:  int					):
							if not isinstance(lowerCAmelCase_				,				lowerCAmelCase_					):
														__lowercase					:							Dict			=   F"Input value of [number={number}] must be an integer"
														raise TypeError(lowerCAmelCase_					)
							if is_prime(lowerCAmelCase_					) and is_prime(number + 2					):
														return number + 2
							else:
														return -1


if __name__ == "__main__":
	import doctest

	doctest.testmod()