






"""simple docstring"""






import dataclasses
import json
import warnings
from dataclasses import dataclass, field
from time import time
from typing import List

from ..utils import logging


SCREAMING_SNAKE_CASE						:						List[Any]					   =       logging.get_logger(__name__)



def  __lowerCamelCase					(					lowerCAmelCase__=None       ,lowerCAmelCase__=None				):
				return field(default_factory=lambda: default       ,metadata=lowerCAmelCase__				)







@dataclass
class       snake_case_       :



	"""simple docstring"""

	SCREAMING_SNAKE_CASE_:							List[str]   			=					list_field(
	    default=[]    ,							metadata={
	        """help""": (
	            """Model checkpoints to be provided to the AutoModel classes. Leave blank to benchmark the base version"""
	            """ of all available models"""
	        )
	    }    ,							)

	SCREAMING_SNAKE_CASE_:							List[int]   			=					list_field(
	    default=[8]    ,							metadata={"""help""": """List of batch sizes for which memory and time performance will be evaluated"""}       )

	SCREAMING_SNAKE_CASE_:							List[int]   			=					list_field(
	    default=[8, 32, 1_28, 5_12]    ,							metadata={"""help""": """List of sequence lengths for which memory and time performance will be evaluated"""}    ,							)

	SCREAMING_SNAKE_CASE_:							bool   			=					field(
	    default=_lowerCamelCase    ,							metadata={"""help""": """Whether to benchmark inference of model. Inference can be disabled via --no-inference."""}    ,							)
	SCREAMING_SNAKE_CASE_:							bool   			=					field(
	    default=_lowerCamelCase    ,							metadata={"""help""": """Whether to run on available cuda devices. Cuda can be disabled via --no-cuda."""}    ,							)
	SCREAMING_SNAKE_CASE_:							bool   			=					field(
	    default=_lowerCamelCase    ,							metadata={"""help""": """Whether to run on available tpu devices. TPU can be disabled via --no-tpu."""}       )
	SCREAMING_SNAKE_CASE_:							bool   			=					field(default=_lowerCamelCase    ,							metadata={"""help""": """Use FP16 to accelerate inference."""}       )
	SCREAMING_SNAKE_CASE_:							bool   			=					field(default=_lowerCamelCase    ,							metadata={"""help""": """Benchmark training of model"""}       )
	SCREAMING_SNAKE_CASE_:							bool   			=					field(default=_lowerCamelCase    ,							metadata={"""help""": """Verbose memory tracing"""}       )
	SCREAMING_SNAKE_CASE_:							bool   			=					field(
	    default=_lowerCamelCase    ,							metadata={"""help""": """Whether to perform speed measurements. Speed measurements can be disabled via --no-speed."""}    ,							)
	SCREAMING_SNAKE_CASE_:							bool   			=					field(
	    default=_lowerCamelCase    ,							metadata={
	        """help""": """Whether to perform memory measurements. Memory measurements can be disabled via --no-memory"""
	    }    ,							)
	SCREAMING_SNAKE_CASE_:							bool   			=					field(default=_lowerCamelCase    ,							metadata={"""help""": """Trace memory line by line"""}       )
	SCREAMING_SNAKE_CASE_:							bool   			=					field(default=_lowerCamelCase    ,							metadata={"""help""": """Save result to a CSV file"""}       )
	SCREAMING_SNAKE_CASE_:							bool   			=					field(default=_lowerCamelCase    ,							metadata={"""help""": """Save all print statements in a log file"""}       )
	SCREAMING_SNAKE_CASE_:							bool   			=					field(default=_lowerCamelCase    ,							metadata={"""help""": """Whether to print environment information"""}       )
	SCREAMING_SNAKE_CASE_:							bool   			=					field(
	    default=_lowerCamelCase    ,							metadata={
	        """help""": (
	            """Whether to use multiprocessing for memory and speed measurement. It is highly recommended to use"""
	            """ multiprocessing for accurate CPU and GPU memory measurements. This option should only be disabled"""
	            """ for debugging / testing and on TPU."""
	        )
	    }    ,							)
	SCREAMING_SNAKE_CASE_:							str   			=					field(
	    default=F"""inference_time_{round(time()       )}.csv"""    ,							metadata={"""help""": """CSV filename used if saving time results to csv."""}    ,							)
	SCREAMING_SNAKE_CASE_:							str   			=					field(
	    default=F"""inference_memory_{round(time()       )}.csv"""    ,							metadata={"""help""": """CSV filename used if saving memory results to csv."""}    ,							)
	SCREAMING_SNAKE_CASE_:							str   			=					field(
	    default=F"""train_time_{round(time()       )}.csv"""    ,							metadata={"""help""": """CSV filename used if saving time results to csv for training."""}    ,							)
	SCREAMING_SNAKE_CASE_:							str   			=					field(
	    default=F"""train_memory_{round(time()       )}.csv"""    ,							metadata={"""help""": """CSV filename used if saving memory results to csv for training."""}    ,							)
	SCREAMING_SNAKE_CASE_:							str   			=					field(
	    default=F"""env_info_{round(time()       )}.csv"""    ,							metadata={"""help""": """CSV filename used if saving environment information."""}    ,							)
	SCREAMING_SNAKE_CASE_:							str   			=					field(
	    default=F"""log_{round(time()       )}.csv"""    ,							metadata={"""help""": """Log filename used if print statements are saved in log."""}    ,							)
	SCREAMING_SNAKE_CASE_:							int   			=					field(default=3    ,							metadata={"""help""": """Times an experiment will be run."""}       )
	SCREAMING_SNAKE_CASE_:							bool   			=					field(
	    default=_lowerCamelCase    ,							metadata={
	        """help""": (
	            """Instead of loading the model as defined in `config.architectures` if exists, just load the pretrain"""
	            """ model weights."""
	        )
	    }    ,							)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					warnings.warn(
					    f'''The class {self.__class__} is deprecated. Hugging Face Benchmarking utils'''
					    ' are deprecated in general and it is advised to use external Benchmarking libraries '
					    ' to benchmark Transformer models.'  ,					__a  ,					)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					return json.dumps(dataclasses.asdict(self						)  ,					indent=2						)


	@property
	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					if len(self.models						) <= 0:
									raise ValueError(
									    'Please make sure you provide at least one model name / model identifier, *e.g.* `--models'
									    ' bert-base-cased` or `args.models = [\'bert-base-cased\'].'						)
					return self.models






	@property
	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					if not self.multi_process:
									return False
					elif self.is_tpu:
									logger.info('Multiprocessing is currently not possible on TPU.'						)
									return False
					else:
									return True










"""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       snake_case_       (    _lowerCamelCase    ,							_lowerCamelCase    ,							unittest.TestCase       ):



	"""simple docstring"""

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


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					torch.manual_seed(0						)
					A__							 =       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  ,					)
					A__							 =       DDIMScheduler(
					    beta_start=0.0_0085  ,					beta_end=0.012  ,					beta_schedule='scaled_linear'  ,					clip_sample=__a  ,					set_alpha_to_one=__a  ,					)
					A__							 =       DDIMInverseScheduler(
					    beta_start=0.0_0085  ,					beta_end=0.012  ,					beta_schedule='scaled_linear'  ,					clip_sample=__a  ,					set_alpha_to_zero=__a  ,					)
					torch.manual_seed(0						)
					A__							 =       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						)
					A__							 =       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  ,					)
					A__							 =       CLIPTextModel(__a						)
					A__							 =       CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'						)

					A__							 =       {
					    '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  ,					__a  ,					__a=0						):


					"""simple docstring"""

					A__							 =       floats_tensor((1, 16, 16)  ,					rng=random.Random(__a						)						).to(__a						)
					A__							 =       floats_tensor((1, 2, 4, 16, 16)  ,					rng=random.Random(__a						)						).to(__a						)
					if str(__a						).startswith('mps'						):
									A__							 =       torch.manual_seed(__a						)
					else:
									A__							 =       torch.Generator(device=__a						).manual_seed(__a						)
					A__							 =       {
					    '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  ,					__a  ,					__a=0						):


					"""simple docstring"""

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


					"""simple docstring"""

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

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

					A__							 =       self.get_dummy_inputs(__a						)
					A__							 =       pipe(**__a						)[0]

					with tempfile.TemporaryDirectory() as tmpdir:
									pipe.save_pretrained(__a						)
									A__							 =       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.'''  ,					)

					A__							 =       self.get_dummy_inputs(__a						)
					A__							 =       pipe_loaded(**__a						)[0]

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


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       'cpu'

					A__							 =       self.get_dummy_components()
					A__							 =       self.pipeline_class(**__a						)
					pipe.to(__a						)
					pipe.set_progress_bar_config(disable=__a						)

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

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


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       'cpu'

					A__							 =       self.get_dummy_components()
					A__							 =       self.pipeline_class(**__a						)
					pipe.to(__a						)
					pipe.set_progress_bar_config(disable=__a						)

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

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


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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






	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       'cpu'

					A__							 =       self.get_dummy_components()

					A__							 =       {'beta_start': 0.0_0085, 'beta_end': 0.012, 'beta_schedule': 'scaled_linear'}
					A__							 =       DPMSolverMultistepScheduler(**__a						)
					A__							 =       DPMSolverMultistepInverseScheduler(**__a						)

					A__							 =       self.pipeline_class(**__a						)
					pipe.to(__a						)
					pipe.set_progress_bar_config(disable=__a						)

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

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





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



	"""simple docstring"""

	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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


	@classmethod
	def       _UpperCAmelCase						(      cls						):


					"""simple docstring"""

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

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

					A__							 =       raw_image


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       torch.manual_seed(0						)

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

					A__							 =       'a bowl of fruit'
					A__							 =       'a bowl of pears'

					A__							 =       pipe.generate_mask(
					    image=self.raw_image  ,					source_prompt=__a  ,					target_prompt=__a  ,					generator=__a  ,					)

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

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

					A__							 =       (
					    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"""

					A__							 =       torch.manual_seed(0						)

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

					A__							 =       'a bowl of fruit'
					A__							 =       'a bowl of pears'

					A__							 =       pipe.generate_mask(
					    image=self.raw_image  ,					source_prompt=__a  ,					target_prompt=__a  ,					generator=__a  ,					)

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

					A__							 =       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]

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


