






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





	"""simple docstring"""






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

	def        UpperCamelCase    (							self    ):
							"""simple docstring"""







							torch.manual_seed(0    )
							_UpperCAmelCase									=			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=UpperCAmelCase       ,    )
							_UpperCAmelCase									=			DDIMScheduler(
							    beta_start=0.0_00_85       ,    beta_end=0.0_12       ,    beta_schedule='scaled_linear'       ,    clip_sample=UpperCAmelCase       ,    set_alpha_to_one=UpperCAmelCase       ,    )
							_UpperCAmelCase									=			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    )
							_UpperCAmelCase									=			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    )
							_UpperCAmelCase									=			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       ,    )
							_UpperCAmelCase									=			CLIPTextModel(UpperCAmelCase    )
							_UpperCAmelCase									=			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'    )

							_UpperCAmelCase									=			{
							    '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        UpperCamelCase    (							self       ,    UpperCAmelCase       ,    UpperCAmelCase=0    ):
							"""simple docstring"""







							_UpperCAmelCase									=			floats_tensor((1, 16, 16)       ,    rng=random.Random(UpperCAmelCase    )    ).to(UpperCAmelCase    )
							_UpperCAmelCase									=			floats_tensor((1, 2, 4, 16, 16)       ,    rng=random.Random(UpperCAmelCase    )    ).to(UpperCAmelCase    )
							if str(UpperCAmelCase    ).startswith('mps'    ):
													_UpperCAmelCase									=			torch.manual_seed(UpperCAmelCase    )
							else:
													_UpperCAmelCase									=			torch.Generator(device=UpperCAmelCase    ).manual_seed(UpperCAmelCase    )
							_UpperCAmelCase									=			{
							    '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        UpperCamelCase    (							self       ,    UpperCAmelCase       ,    UpperCAmelCase=0    ):
							"""simple docstring"""







							_UpperCAmelCase									=			floats_tensor((1, 3, 32, 32)       ,    rng=random.Random(UpperCAmelCase    )    ).to(UpperCAmelCase    )
							_UpperCAmelCase									=			image.cpu().permute(0       ,    2       ,    3       ,    1    )[0]
							_UpperCAmelCase									=			Image.fromarray(np.uinta(UpperCAmelCase    )    ).convert('RGB'    )
							if str(UpperCAmelCase    ).startswith('mps'    ):
													_UpperCAmelCase									=			torch.manual_seed(UpperCAmelCase    )
							else:
													_UpperCAmelCase									=			torch.Generator(device=UpperCAmelCase    ).manual_seed(UpperCAmelCase    )
							_UpperCAmelCase									=			{
							    '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        UpperCamelCase    (							self       ,    UpperCAmelCase       ,    UpperCAmelCase=0    ):
							"""simple docstring"""







							_UpperCAmelCase									=			floats_tensor((1, 3, 32, 32)       ,    rng=random.Random(UpperCAmelCase    )    ).to(UpperCAmelCase    )
							_UpperCAmelCase									=			image.cpu().permute(0       ,    2       ,    3       ,    1    )[0]
							_UpperCAmelCase									=			Image.fromarray(np.uinta(UpperCAmelCase    )    ).convert('RGB'    )
							if str(UpperCAmelCase    ).startswith('mps'    ):
													_UpperCAmelCase									=			torch.manual_seed(UpperCAmelCase    )
							else:
													_UpperCAmelCase									=			torch.Generator(device=UpperCAmelCase    ).manual_seed(UpperCAmelCase    )
							_UpperCAmelCase									=			{
							    '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        UpperCamelCase    (							self    ):
							"""simple docstring"""







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

							_UpperCAmelCase									=			self.get_dummy_components()
							_UpperCAmelCase									=			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}    )

							_UpperCAmelCase									=			self.get_dummy_inputs(UpperCAmelCase    )
							_UpperCAmelCase									=			pipe(**UpperCAmelCase    )[0]

							with tempfile.TemporaryDirectory() as tmpdir:
													pipe.save_pretrained(UpperCAmelCase    )
													_UpperCAmelCase									=			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."""       ,    )

							_UpperCAmelCase									=			self.get_dummy_inputs(UpperCAmelCase    )
							_UpperCAmelCase									=			pipe_loaded(**UpperCAmelCase    )[0]

							_UpperCAmelCase									=			np.abs(output - output_loaded    ).max()
							self.assertLess(UpperCAmelCase       ,    1e-4    )

	def        UpperCamelCase    (							self    ):
							"""simple docstring"""







							_UpperCAmelCase									=			'cpu'

							_UpperCAmelCase									=			self.get_dummy_components()
							_UpperCAmelCase									=			self.pipeline_class(**UpperCAmelCase    )
							pipe.to(UpperCAmelCase    )
							pipe.set_progress_bar_config(disable=UpperCAmelCase    )

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

							self.assertEqual(mask.shape       ,    (1, 16, 16)    )
							_UpperCAmelCase									=			np.array([0] * 9    )
							_UpperCAmelCase									=			np.abs(mask_slice.flatten() - expected_slice    ).max()
							self.assertLessEqual(UpperCAmelCase       ,    1e-3    )
							self.assertEqual(mask[0, -3, -4]       ,    0    )

	def        UpperCamelCase    (							self    ):
							"""simple docstring"""







							_UpperCAmelCase									=			'cpu'

							_UpperCAmelCase									=			self.get_dummy_components()
							_UpperCAmelCase									=			self.pipeline_class(**UpperCAmelCase    )
							pipe.to(UpperCAmelCase    )
							pipe.set_progress_bar_config(disable=UpperCAmelCase    )

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

							self.assertEqual(image.shape       ,    (2, 32, 32, 3)    )
							_UpperCAmelCase									=			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]       ,    )
							_UpperCAmelCase									=			np.abs(image_slice.flatten() - expected_slice    ).max()
							self.assertLessEqual(UpperCAmelCase       ,    1e-3    )

	def        UpperCamelCase    (							self    ):
							"""simple docstring"""







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

	def        UpperCamelCase    (							self    ):
							"""simple docstring"""







							_UpperCAmelCase									=			'cpu'

							_UpperCAmelCase									=			self.get_dummy_components()

							_UpperCAmelCase									=			{'beta_start': 0.0_00_85, 'beta_end': 0.0_12, 'beta_schedule': 'scaled_linear'}
							_UpperCAmelCase									=			DPMSolverMultistepScheduler(**UpperCAmelCase    )
							_UpperCAmelCase									=			DPMSolverMultistepInverseScheduler(**UpperCAmelCase    )

							_UpperCAmelCase									=			self.pipeline_class(**UpperCAmelCase    )
							pipe.to(UpperCAmelCase    )
							pipe.set_progress_bar_config(disable=UpperCAmelCase    )

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

							self.assertEqual(image.shape       ,    (2, 32, 32, 3)    )
							_UpperCAmelCase									=			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]       ,    )
							_UpperCAmelCase									=			np.abs(image_slice.flatten() - expected_slice    ).max()
							self.assertLessEqual(UpperCAmelCase       ,    1e-3    )





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





	"""simple docstring"""






	def        UpperCamelCase    (							self    ):
							"""simple docstring"""







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

	@classmethod
	def        UpperCamelCase    (							cls    ):
							"""simple docstring"""







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

							_UpperCAmelCase									=			raw_image.convert('RGB'    ).resize((768, 768)    )

							_UpperCAmelCase									=			raw_image

	def        UpperCamelCase    (							self    ):
							"""simple docstring"""







							_UpperCAmelCase									=			torch.manual_seed(0    )

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

							_UpperCAmelCase									=			'a bowl of fruit'
							_UpperCAmelCase									=			'a bowl of pears'

							_UpperCAmelCase									=			pipe.generate_mask(
							    image=self.raw_image       ,    source_prompt=UpperCAmelCase       ,    target_prompt=UpperCAmelCase       ,    generator=UpperCAmelCase       ,    )

							_UpperCAmelCase									=			pipe.invert(
							    prompt=UpperCAmelCase       ,    image=self.raw_image       ,    inpaint_strength=0.7       ,    generator=UpperCAmelCase    ).latents

							_UpperCAmelCase									=			pipe(
							    prompt=UpperCAmelCase       ,    mask_image=UpperCAmelCase       ,    image_latents=UpperCAmelCase       ,    generator=UpperCAmelCase       ,    negative_prompt=UpperCAmelCase       ,    inpaint_strength=0.7       ,    output_type='numpy'       ,    ).images[0]

							_UpperCAmelCase									=			(
							    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        UpperCamelCase    (							self    ):
							"""simple docstring"""







							_UpperCAmelCase									=			torch.manual_seed(0    )

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

							_UpperCAmelCase									=			'a bowl of fruit'
							_UpperCAmelCase									=			'a bowl of pears'

							_UpperCAmelCase									=			pipe.generate_mask(
							    image=self.raw_image       ,    source_prompt=UpperCAmelCase       ,    target_prompt=UpperCAmelCase       ,    generator=UpperCAmelCase       ,    )

							_UpperCAmelCase									=			pipe.invert(
							    prompt=UpperCAmelCase       ,    image=self.raw_image       ,    inpaint_strength=0.7       ,    generator=UpperCAmelCase       ,    num_inference_steps=25       ,    ).latents

							_UpperCAmelCase									=			pipe(
							    prompt=UpperCAmelCase       ,    mask_image=UpperCAmelCase       ,    image_latents=UpperCAmelCase       ,    generator=UpperCAmelCase       ,    negative_prompt=UpperCAmelCase       ,    inpaint_strength=0.7       ,    num_inference_steps=25       ,    output_type='numpy'       ,    ).images[0]

							_UpperCAmelCase									=			(
							    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




import inspect
from typing import List, Optional, Tuple, Union

import numpy as np
import PIL
import torch
import torch.utils.checkpoint

from ...models import UNetaDModel, VQModel
from ...schedulers import (
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
)
from ...utils import PIL_INTERPOLATION, randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput
def 							_UpperCAmelCase				(UpperCamelCase__     :  Union[str, Any]						):
       _A    ,			_A      :       Any				    = image.size
       _A    ,			_A      :       str				    = (x - x % 32 for x in (w, h))  # resize to integer multiple of 32
       _A      :       List[str]				    = image.resize((w, h)				,							resample=PIL_INTERPOLATION["lanczos"]						)
       _A      :       Any				    = np.array(UpperCamelCase__						).astype(np.floataa						) / 2_55.0
       _A      :       Optional[Any]				    = image[None].transpose(0				,							3				,							1				,							2						)
       _A      :       Union[str, Any]				    = torch.from_numpy(UpperCamelCase__						)
       return 2.0 * image - 1.0







class      lowerCAmelCase__       (				a):




     '''simple docstring'''


     def __init__(						self ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						)			->				Optional[int]:
            super().__init__()
            self.register_modules(vqvae=__lowerCamelCase ,						unet=__lowerCamelCase ,						scheduler=__lowerCamelCase)


     @torch.no_grad()
     def __call__(						self ,						__lowerCamelCase = None ,						__lowerCamelCase = 1 ,						__lowerCamelCase = 1_0_0 ,						__lowerCamelCase = 0.0 ,						__lowerCamelCase = None ,						__lowerCamelCase = "pil" ,						__lowerCamelCase = True ,						)			->				Union[Tuple, ImagePipelineOutput]:
            if isinstance(__lowerCamelCase ,						PIL.Image.Image):
                   _A      :       Tuple				    = 1
            elif isinstance(__lowerCamelCase ,						torch.Tensor):
                   _A      :       Union[str, Any]				    = image.shape[0]
            else:
                   raise ValueError(F"`image` has to be of type `PIL.Image.Image` or `torch.Tensor` but is {type(__lowerCamelCase)}")

            if isinstance(__lowerCamelCase ,						PIL.Image.Image):
                   _A      :       Union[str, Any]				    = preprocess(__lowerCamelCase)

            _A    ,			_A      :       Union[str, Any]				    = image.shape[-2:]

            # in_channels should be 6: 3 for latents, 3 for low resolution image
            _A      :       Optional[Any]				    = (batch_size, self.unet.config.in_channels // 2, height, width)
            _A      :       str				    = next(self.unet.parameters()).dtype

            _A      :       Union[str, Any]				    = randn_tensor(__lowerCamelCase ,						generator=__lowerCamelCase ,						device=self.device ,						dtype=__lowerCamelCase)

            _A      :       List[Any]				    = image.to(device=self.device ,						dtype=__lowerCamelCase)

            # set timesteps and move to the correct device
            self.scheduler.set_timesteps(__lowerCamelCase ,						device=self.device)
            _A      :       Any				    = self.scheduler.timesteps

            # scale the initial noise by the standard deviation required by the scheduler
            _A      :       List[str]				    = latents * self.scheduler.init_noise_sigma

            # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature.
            # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
            # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
            # and should be between [0, 1]
            _A      :       str				    = "eta" in set(inspect.signature(self.scheduler.step).parameters.keys())
            _A      :       Optional[int]				    = {}
            if accepts_eta:
                   _A      :       List[Any]				    = eta

            for t in self.progress_bar(__lowerCamelCase):
                   # concat latents and low resolution image in the channel dimension.
                   _A      :       List[Any]				    = torch.cat([latents, image] ,						dim=1)
                   _A      :       str				    = self.scheduler.scale_model_input(__lowerCamelCase ,						__lowerCamelCase)
                   # predict the noise residual
                   _A      :       Any				    = self.unet(__lowerCamelCase ,						__lowerCamelCase).sample
                   # compute the previous noisy sample x_t -> x_t-1
                   _A      :       Optional[int]				    = self.scheduler.step(__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						**__lowerCamelCase).prev_sample

            # decode the image latents with the VQVAE
            _A      :       Union[str, Any]				    = self.vqvae.decode(__lowerCamelCase).sample
            _A      :       Dict				    = torch.clamp(__lowerCamelCase ,						-1.0 ,						1.0)
            _A      :       Tuple				    = image / 2 + 0.5
            _A      :       int				    = image.cpu().permute(0 ,						2 ,						3 ,						1).numpy()

            if output_type == "pil":
                   _A      :       Optional[int]				    = self.numpy_to_pil(__lowerCamelCase)

            if not return_dict:
                   return (image,)

            return ImagePipelineOutput(images=__lowerCamelCase)
