




import gc
import random
import unittest

import torch

from diffusers import (
    IFImgaImgPipeline,
    IFImgaImgSuperResolutionPipeline,
    IFInpaintingPipeline,
    IFInpaintingSuperResolutionPipeline,
    IFPipeline,
    IFSuperResolutionPipeline,
)
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import floats_tensor, load_numpy, require_torch_gpu, skip_mps, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference
from . import IFPipelineTesterMixin
@skip_mps
class 			a__    (					snake_case__ ,				snake_case__ ,				unittest.TestCase      ):
						_a							:    str					=    IFPipeline
						_a							:    List[Any]					=    TEXT_TO_IMAGE_PARAMS - {"""width""", """height""", """latents"""}
						_a							:    Any					=    TEXT_TO_IMAGE_BATCH_PARAMS
						_a							:    Dict					=    PipelineTesterMixin.required_optional_params - {"""latents"""}






						def 				__SCREAMING_SNAKE_CASE(   self			):




											"""simple docstring"""



											return self._get_dummy_components()






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A		,			_A=0			):




											"""simple docstring"""



											if str(_A			).startswith("mps"			):
																__lowerCAmelCase				   =       torch.manual_seed(_A			)
											else:
																__lowerCAmelCase				   =       torch.Generator(device=_A			).manual_seed(_A			)

											__lowerCAmelCase				   =       {
											    "prompt": "A painting of a squirrel eating a burger",
											    "generator": generator,
											    "num_inference_steps": 2,
											    "output_type": "numpy",
											}

											return inputs






						def 				__SCREAMING_SNAKE_CASE(   self			):




											"""simple docstring"""



											self._test_save_load_optional_components()






						@unittest.skipIf(torch_device != "cuda"		,			reason="float16 requires CUDA"			)
						def 				__SCREAMING_SNAKE_CASE(   self			):




											"""simple docstring"""



											super().test_save_load_floataa(expected_max_diff=1E-1			)






						def 				__SCREAMING_SNAKE_CASE(   self			):




											"""simple docstring"""



											self._test_attention_slicing_forward_pass(expected_max_diff=1E-2			)






						def 				__SCREAMING_SNAKE_CASE(   self			):




											"""simple docstring"""



											self._test_save_load_local()






						def 				__SCREAMING_SNAKE_CASE(   self			):




											"""simple docstring"""



											self._test_inference_batch_single_identical(
											    expected_max_diff=1E-2		,			)






						@unittest.skipIf(
						    torch_device != "cuda" or not is_xformers_available()		,			reason="XFormers attention is only available with CUDA and `xformers` installed"		,			)
						def 				__SCREAMING_SNAKE_CASE(   self			):




											"""simple docstring"""



											self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3			)






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






						def 				__SCREAMING_SNAKE_CASE(   self			):




											"""simple docstring"""



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






						def 				__SCREAMING_SNAKE_CASE(   self			):




											"""simple docstring"""



											__lowerCAmelCase				   =       IFPipeline.from_pretrained("DeepFloyd/IF-I-XL-v1.0"		,			variant="fp16"		,			torch_dtype=torch.floataa			)

											__lowerCAmelCase				   =       IFSuperResolutionPipeline.from_pretrained(
											    "DeepFloyd/IF-II-L-v1.0"		,			variant="fp16"		,			torch_dtype=torch.floataa		,			text_encoder=_A		,			tokenizer=_A			)

											# pre compute text embeddings and remove T5 to save memory

											pipe_a.text_encoder.to("cuda"			)

											__lowerCAmelCase   ,							__lowerCAmelCase				   =       pipe_a.encode_prompt("anime turtle"		,			device="cuda"			)

											del pipe_a.tokenizer
											del pipe_a.text_encoder
											gc.collect()

											__lowerCAmelCase				   =       None
											__lowerCAmelCase				   =       None

											pipe_a.enable_model_cpu_offload()
											pipe_a.enable_model_cpu_offload()

											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()			)
											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()			)

											self._test_if(_A		,			_A		,			_A		,			_A			)

											pipe_a.remove_all_hooks()
											pipe_a.remove_all_hooks()

											# img2img

											__lowerCAmelCase				   =       IFImgaImgPipeline(**pipe_a.components			)
											__lowerCAmelCase				   =       IFImgaImgSuperResolutionPipeline(**pipe_a.components			)

											pipe_a.enable_model_cpu_offload()
											pipe_a.enable_model_cpu_offload()

											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()			)
											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()			)

											self._test_if_imgaimg(_A		,			_A		,			_A		,			_A			)

											pipe_a.remove_all_hooks()
											pipe_a.remove_all_hooks()

											# inpainting

											__lowerCAmelCase				   =       IFInpaintingPipeline(**pipe_a.components			)
											__lowerCAmelCase				   =       IFInpaintingSuperResolutionPipeline(**pipe_a.components			)

											pipe_a.enable_model_cpu_offload()
											pipe_a.enable_model_cpu_offload()

											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()			)
											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()			)

											self._test_if_inpainting(_A		,			_A		,			_A		,			_A			)






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A		,			_A		,			_A		,			_A			):




											"""simple docstring"""



											_start_torch_memory_measurement()

											__lowerCAmelCase				   =       torch.Generator(device="cpu"			).manual_seed(0			)
											__lowerCAmelCase				   =       pipe_a(
											    prompt_embeds=_A		,			negative_prompt_embeds=_A		,			num_inference_steps=2		,			generator=_A		,			output_type="np"		,			)

											__lowerCAmelCase				   =       output.images[0]

											assert image.shape == (6_4, 6_4, 3)

											__lowerCAmelCase				   =       torch.cuda.max_memory_allocated()
											assert mem_bytes < 1_3 * 1_0**9

											__lowerCAmelCase				   =       load_numpy(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if.npy"			)
											assert_mean_pixel_difference(_A		,			_A			)

											# pipeline 2

											_start_torch_memory_measurement()

											__lowerCAmelCase				   =       torch.Generator(device="cpu"			).manual_seed(0			)

											__lowerCAmelCase				   =       floats_tensor((1, 3, 6_4, 6_4)		,			rng=random.Random(0			)			).to(_A			)

											__lowerCAmelCase				   =       pipe_a(
											    prompt_embeds=_A		,			negative_prompt_embeds=_A		,			image=_A		,			generator=_A		,			num_inference_steps=2		,			output_type="np"		,			)

											__lowerCAmelCase				   =       output.images[0]

											assert image.shape == (2_5_6, 2_5_6, 3)

											__lowerCAmelCase				   =       torch.cuda.max_memory_allocated()
											assert mem_bytes < 4 * 1_0**9

											__lowerCAmelCase				   =       load_numpy(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_superresolution_stage_II.npy"			)
											assert_mean_pixel_difference(_A		,			_A			)






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A		,			_A		,			_A		,			_A			):




											"""simple docstring"""



											_start_torch_memory_measurement()

											__lowerCAmelCase				   =       floats_tensor((1, 3, 6_4, 6_4)		,			rng=random.Random(0			)			).to(_A			)

											__lowerCAmelCase				   =       torch.Generator(device="cpu"			).manual_seed(0			)

											__lowerCAmelCase				   =       pipe_a(
											    prompt_embeds=_A		,			negative_prompt_embeds=_A		,			image=_A		,			num_inference_steps=2		,			generator=_A		,			output_type="np"		,			)

											__lowerCAmelCase				   =       output.images[0]

											assert image.shape == (6_4, 6_4, 3)

											__lowerCAmelCase				   =       torch.cuda.max_memory_allocated()
											assert mem_bytes < 1_0 * 1_0**9

											__lowerCAmelCase				   =       load_numpy(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img.npy"			)
											assert_mean_pixel_difference(_A		,			_A			)

											# pipeline 2

											_start_torch_memory_measurement()

											__lowerCAmelCase				   =       torch.Generator(device="cpu"			).manual_seed(0			)

											__lowerCAmelCase				   =       floats_tensor((1, 3, 2_5_6, 2_5_6)		,			rng=random.Random(0			)			).to(_A			)
											__lowerCAmelCase				   =       floats_tensor((1, 3, 6_4, 6_4)		,			rng=random.Random(0			)			).to(_A			)

											__lowerCAmelCase				   =       pipe_a(
											    prompt_embeds=_A		,			negative_prompt_embeds=_A		,			image=_A		,			original_image=_A		,			generator=_A		,			num_inference_steps=2		,			output_type="np"		,			)

											__lowerCAmelCase				   =       output.images[0]

											assert image.shape == (2_5_6, 2_5_6, 3)

											__lowerCAmelCase				   =       torch.cuda.max_memory_allocated()
											assert mem_bytes < 4 * 1_0**9

											__lowerCAmelCase				   =       load_numpy(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img_superresolution_stage_II.npy"			)
											assert_mean_pixel_difference(_A		,			_A			)






						def 				__SCREAMING_SNAKE_CASE(   self		,			_A		,			_A		,			_A		,			_A			):




											"""simple docstring"""



											_start_torch_memory_measurement()

											__lowerCAmelCase				   =       floats_tensor((1, 3, 6_4, 6_4)		,			rng=random.Random(0			)			).to(_A			)
											__lowerCAmelCase				   =       floats_tensor((1, 3, 6_4, 6_4)		,			rng=random.Random(1			)			).to(_A			)

											__lowerCAmelCase				   =       torch.Generator(device="cpu"			).manual_seed(0			)
											__lowerCAmelCase				   =       pipe_a(
											    prompt_embeds=_A		,			negative_prompt_embeds=_A		,			image=_A		,			mask_image=_A		,			num_inference_steps=2		,			generator=_A		,			output_type="np"		,			)

											__lowerCAmelCase				   =       output.images[0]

											assert image.shape == (6_4, 6_4, 3)

											__lowerCAmelCase				   =       torch.cuda.max_memory_allocated()
											assert mem_bytes < 1_0 * 1_0**9

											__lowerCAmelCase				   =       load_numpy(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting.npy"			)
											assert_mean_pixel_difference(_A		,			_A			)

											# pipeline 2

											_start_torch_memory_measurement()

											__lowerCAmelCase				   =       torch.Generator(device="cpu"			).manual_seed(0			)

											__lowerCAmelCase				   =       floats_tensor((1, 3, 6_4, 6_4)		,			rng=random.Random(0			)			).to(_A			)
											__lowerCAmelCase				   =       floats_tensor((1, 3, 2_5_6, 2_5_6)		,			rng=random.Random(0			)			).to(_A			)
											__lowerCAmelCase				   =       floats_tensor((1, 3, 2_5_6, 2_5_6)		,			rng=random.Random(1			)			).to(_A			)

											__lowerCAmelCase				   =       pipe_a(
											    prompt_embeds=_A		,			negative_prompt_embeds=_A		,			image=_A		,			mask_image=_A		,			original_image=_A		,			generator=_A		,			num_inference_steps=2		,			output_type="np"		,			)

											__lowerCAmelCase				   =       output.images[0]

											assert image.shape == (2_5_6, 2_5_6, 3)

											__lowerCAmelCase				   =       torch.cuda.max_memory_allocated()
											assert mem_bytes < 4 * 1_0**9

											__lowerCAmelCase				   =       load_numpy(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting_superresolution_stage_II.npy"			)
											assert_mean_pixel_difference(_A		,			_A			)







def 	_a      (						):
					torch.cuda.empty_cache()
					torch.cuda.reset_max_memory_allocated()
					torch.cuda.reset_peak_memory_stats()












"""simple docstring"""






import copy
import tempfile
import unittest

from huggingface_hub import HfFolder, delete_repo
from parameterized import parameterized
from requests.exceptions import HTTPError

from transformers import AutoConfig, GenerationConfig
from transformers.testing_utils import TOKEN, USER, is_staging_test




class 							lowerCAmelCase__				(			unittest.TestCase ):






      '''simple docstring'''

      @parameterized.expand([(None,), ('''foo.json''',)])
      def _SCREAMING_SNAKE_CASE (						self     :  Dict						,	lowercase_     :  Union[str, Any]):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  List[str]    						=		GenerationConfig(
              do_sample=lowercase_						,	temperature=0.7						,	length_penalty=1.0						,	bad_words_ids=[[1, 2, 3], [4, 5]]						,	)
          with tempfile.TemporaryDirectory() as tmp_dir:
              config.save_pretrained(lowercase_						,	config_name=lowercase_)
              SCREAMING_SNAKE_CASE_      :  Tuple    						=		GenerationConfig.from_pretrained(lowercase_						,	config_name=lowercase_)

          # Checks parameters that were specified
          self.assertEqual(loaded_config.do_sample						,	lowercase_)
          self.assertEqual(loaded_config.temperature						,	0.7)
          self.assertEqual(loaded_config.length_penalty						,	1.0)
          self.assertEqual(loaded_config.bad_words_ids						,	[[1, 2, 3], [4, 5]])

          # Checks parameters that were not specified (defaults)
          self.assertEqual(loaded_config.top_k						,	50)
          self.assertEqual(loaded_config.max_length						,	20)
          self.assertEqual(loaded_config.max_time						,	lowercase_)
      def _SCREAMING_SNAKE_CASE (						self     :  Optional[Any]):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  Optional[Any]    						=		AutoConfig.from_pretrained('''gpt2''')
          SCREAMING_SNAKE_CASE_      :  int    						=		GenerationConfig.from_model_config(lowercase_)
          SCREAMING_SNAKE_CASE_      :  int    						=		GenerationConfig()

          # The generation config has loaded a few non-default parameters from the model config
          self.assertNotEqual(lowercase_						,	lowercase_)

          # One of those parameters is eos_token_id -- check if it matches
          self.assertNotEqual(generation_config_from_model.eos_token_id						,	default_generation_config.eos_token_id)
          self.assertEqual(generation_config_from_model.eos_token_id						,	model_config.eos_token_id)
      def _SCREAMING_SNAKE_CASE (						self     :  Dict):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  str    						=		GenerationConfig()
          SCREAMING_SNAKE_CASE_      :  Any    						=		{
              '''max_new_tokens''': 1024,
              '''foo''': '''bar''',
          }
          SCREAMING_SNAKE_CASE_      :  str    						=		copy.deepcopy(lowercase_)
          SCREAMING_SNAKE_CASE_      :  Tuple    						=		generation_config.update(**lowercase_)

          # update_kwargs was not modified (no side effects)
          self.assertEqual(lowercase_						,	lowercase_)

          # update_kwargs was used to update the config on valid attributes
          self.assertEqual(generation_config.max_new_tokens						,	1024)

          # `.update()` returns a dictionary of unused kwargs
          self.assertEqual(lowercase_						,	{'''foo''': '''bar'''})
      def _SCREAMING_SNAKE_CASE (						self     :  Any):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  Optional[Any]    						=		GenerationConfig()
          SCREAMING_SNAKE_CASE_      :  List[str]    						=		'''bar'''

          with tempfile.TemporaryDirectory('''test-generation-config''') as tmp_dir:
              generation_config.save_pretrained(lowercase_)

              SCREAMING_SNAKE_CASE_      :  List[Any]    						=		GenerationConfig.from_pretrained(lowercase_)
          # update_kwargs was used to update the config on valid attributes
          self.assertEqual(new_config.foo						,	'''bar''')

          SCREAMING_SNAKE_CASE_      :  Optional[Any]    						=		GenerationConfig.from_model_config(lowercase_)
          assert not hasattr(lowercase_						,	'''foo''')  # no new kwargs should be initialized if from config



      def _SCREAMING_SNAKE_CASE (						self     :  Tuple):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  Tuple    						=		GenerationConfig()
          self.assertEqual(default_config.temperature						,	1.0)
          self.assertEqual(default_config.do_sample						,	lowercase_)
          self.assertEqual(default_config.num_beams						,	1)

          SCREAMING_SNAKE_CASE_      :  Dict    						=		GenerationConfig(
              do_sample=lowercase_						,	temperature=0.7						,	length_penalty=1.0						,	bad_words_ids=[[1, 2, 3], [4, 5]]						,	)
          self.assertEqual(config.temperature						,	0.7)
          self.assertEqual(config.do_sample						,	lowercase_)
          self.assertEqual(config.num_beams						,	1)

          with tempfile.TemporaryDirectory() as tmp_dir:
              config.save_pretrained(lowercase_)
              SCREAMING_SNAKE_CASE_      :  Tuple    						=		GenerationConfig.from_pretrained(lowercase_						,	temperature=1.0)

          self.assertEqual(loaded_config.temperature						,	1.0)
          self.assertEqual(loaded_config.do_sample						,	lowercase_)
          self.assertEqual(loaded_config.num_beams						,	1)  # default value




@is_staging_test
class 							lowerCAmelCase__				(			unittest.TestCase ):






      '''simple docstring'''

      @classmethod
      def _SCREAMING_SNAKE_CASE (						cls     :  Any):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  str    						=		TOKEN
          HfFolder.save_token(lowercase_)
      @classmethod
      def _SCREAMING_SNAKE_CASE (						cls     :  List[str]):




          '''simple docstring'''


          try:
              delete_repo(token=cls._token						,	repo_id='''test-generation-config''')
          except HTTPError:
              pass

          try:
              delete_repo(token=cls._token						,	repo_id='''valid_org/test-generation-config-org''')
          except HTTPError:
              pass
      def _SCREAMING_SNAKE_CASE (						self     :  Dict):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  Tuple    						=		GenerationConfig(
              do_sample=lowercase_						,	temperature=0.7						,	length_penalty=1.0						,	)
          config.push_to_hub('''test-generation-config'''						,	use_auth_token=self._token)

          SCREAMING_SNAKE_CASE_      :  int    						=		GenerationConfig.from_pretrained(F'{USER}/test-generation-config')
          for k, v in config.to_dict().items():
              if k != "transformers_version":
                  self.assertEqual(lowercase_						,	getattr(lowercase_						,	lowercase_))

        # Reset repo
          delete_repo(token=self._token						,	repo_id='''test-generation-config''')

          # Push to hub via save_pretrained
          with tempfile.TemporaryDirectory() as tmp_dir:
              config.save_pretrained(
                  lowercase_						,	repo_id='''test-generation-config'''						,	push_to_hub=lowercase_						,	use_auth_token=self._token)

          SCREAMING_SNAKE_CASE_      :  Optional[int]    						=		GenerationConfig.from_pretrained(F'{USER}/test-generation-config')
          for k, v in config.to_dict().items():
              if k != "transformers_version":
                  self.assertEqual(lowercase_						,	getattr(lowercase_						,	lowercase_))



      def _SCREAMING_SNAKE_CASE (						self     :  Union[str, Any]):




          '''simple docstring'''


          SCREAMING_SNAKE_CASE_      :  Dict    						=		GenerationConfig(
              do_sample=lowercase_						,	temperature=0.7						,	length_penalty=1.0						,	)
          config.push_to_hub('''valid_org/test-generation-config-org'''						,	use_auth_token=self._token)

          SCREAMING_SNAKE_CASE_      :  Any    						=		GenerationConfig.from_pretrained('''valid_org/test-generation-config-org''')
          for k, v in config.to_dict().items():
              if k != "transformers_version":
                  self.assertEqual(lowercase_						,	getattr(lowercase_						,	lowercase_))

        # Reset repo
          delete_repo(token=self._token						,	repo_id='''valid_org/test-generation-config-org''')

          # Push to hub via save_pretrained
          with tempfile.TemporaryDirectory() as tmp_dir:
              config.save_pretrained(
                  lowercase_						,	repo_id='''valid_org/test-generation-config-org'''						,	push_to_hub=lowercase_						,	use_auth_token=self._token)

          SCREAMING_SNAKE_CASE_      :  Any    						=		GenerationConfig.from_pretrained('''valid_org/test-generation-config-org''')
          for k, v in config.to_dict().items():
              if k != "transformers_version":
                  self.assertEqual(lowercase_						,	getattr(lowercase_						,	lowercase_))


