


'''simple docstring'''






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







    """simple docstring"""



    lowercase    =					IFPipeline
    lowercase    =					TEXT_TO_IMAGE_PARAMS - {'width', 'height', 'latents'}
    lowercase    =					TEXT_TO_IMAGE_BATCH_PARAMS
    lowercase    =					PipelineTesterMixin.required_optional_params - {'latents'}

    def __lowercase   (   self    :  List[Any]		)   ->  Tuple:
      return self._get_dummy_components()

    def __lowercase   (   self    :  Union[str, Any]    ,						lowerCamelCase    :  Dict    ,						lowerCamelCase    :  str=0		)   ->  Optional[int]:



      if str(a__		).startswith("""mps"""		):
        lowerCAmelCase_			:      Optional[int]			=   torch.manual_seed(a__		)
      else:
        lowerCAmelCase_			:      Optional[int]			=   torch.Generator(device=a__		).manual_seed(a__		)

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

      return inputs

    def __lowercase   (   self    :  Tuple		)   ->  str:
      self._test_save_load_optional_components()

    @unittest.skipIf(torch_device != """cuda"""    ,						reason="""float16 requires CUDA"""		)
    def __lowercase   (   self    :  Union[str, Any]		)   ->  Tuple:
      super().test_save_load_floataa(expected_max_diff=1E-1		)

    def __lowercase   (   self    :  List[str]		)   ->  List[str]:
      self._test_attention_slicing_forward_pass(expected_max_diff=1E-2		)

    def __lowercase   (   self    :  Union[str, Any]		)   ->  Union[str, Any]:
      self._test_save_load_local()

    def __lowercase   (   self    :  Any		)   ->  List[Any]:
      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 __lowercase   (   self    :  Any		)   ->  str:
      self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3		)







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







    """simple docstring"""



    def __lowercase   (   self    :  Union[str, Any]		)   ->  Union[str, Any]:
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()

    def __lowercase   (   self    :  int		)   ->  Optional[Any]:
      lowerCAmelCase_			:      List[str]			=   IFPipeline.from_pretrained("""DeepFloyd/IF-I-XL-v1.0"""    ,						variant="""fp16"""    ,						torch_dtype=torch.floataa		)

      lowerCAmelCase_			:      int			=   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_			:      Optional[int]			=   pipe_a.encode_prompt("""anime turtle"""    ,						device="""cuda"""		)

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

      lowerCAmelCase_			:      str			=   None
      lowerCAmelCase_			:      str			=   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_			:      int			=   IFImgaImgPipeline(**pipe_a.components		)
      lowerCAmelCase_			:      int			=   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_			:      Optional[Any]			=   IFInpaintingPipeline(**pipe_a.components		)
      lowerCAmelCase_			:      List[Any]			=   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 __lowercase   (   self    :  str    ,						lowerCamelCase    :  Optional[Any]    ,						lowerCamelCase    :  List[Any]    ,						lowerCamelCase    :  Tuple    ,						lowerCamelCase    :  Optional[int]		)   ->  Optional[int]:
      _start_torch_memory_measurement()

      lowerCAmelCase_			:      Optional[Any]			=   torch.Generator(device="""cpu"""		).manual_seed(0		)
      lowerCAmelCase_			:      str			=   pipe_a(
          prompt_embeds=a__    ,						negative_prompt_embeds=a__    ,						num_inference_steps=2    ,						generator=a__    ,						output_type="""np"""    ,						)

      lowerCAmelCase_			:      Optional[int]			=   output.images[0]

      assert image.shape == (64, 64, 3)

      lowerCAmelCase_			:      int			=   torch.cuda.max_memory_allocated()
      assert mem_bytes < 13 * 10**9

      lowerCAmelCase_			:      Tuple			=   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_			:      List[Any]			=   torch.Generator(device="""cpu"""		).manual_seed(0		)

      lowerCAmelCase_			:      Dict			=   floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(a__		)

      lowerCAmelCase_			:      List[Any]			=   pipe_a(
          prompt_embeds=a__    ,						negative_prompt_embeds=a__    ,						image=a__    ,						generator=a__    ,						num_inference_steps=2    ,						output_type="""np"""    ,						)

      lowerCAmelCase_			:      Any			=   output.images[0]

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

      lowerCAmelCase_			:      Any			=   torch.cuda.max_memory_allocated()
      assert mem_bytes < 4 * 10**9

      lowerCAmelCase_			:      Optional[int]			=   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 __lowercase   (   self    :  str    ,						lowerCamelCase    :  List[Any]    ,						lowerCamelCase    :  Optional[Any]    ,						lowerCamelCase    :  Tuple    ,						lowerCamelCase    :  str		)   ->  Union[str, Any]:
      _start_torch_memory_measurement()

      lowerCAmelCase_			:      str			=   floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(a__		)

      lowerCAmelCase_			:      List[str]			=   torch.Generator(device="""cpu"""		).manual_seed(0		)

      lowerCAmelCase_			:      List[str]			=   pipe_a(
          prompt_embeds=a__    ,						negative_prompt_embeds=a__    ,						image=a__    ,						num_inference_steps=2    ,						generator=a__    ,						output_type="""np"""    ,						)

      lowerCAmelCase_			:      str			=   output.images[0]

      assert image.shape == (64, 64, 3)

      lowerCAmelCase_			:      List[str]			=   torch.cuda.max_memory_allocated()
      assert mem_bytes < 10 * 10**9

      lowerCAmelCase_			:      Union[str, Any]			=   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_			:      Union[str, Any]			=   torch.Generator(device="""cpu"""		).manual_seed(0		)

      lowerCAmelCase_			:      Tuple			=   floats_tensor((1, 3, 2_56, 2_56)    ,						rng=random.Random(0		)		).to(a__		)
      lowerCAmelCase_			:      Any			=   floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(a__		)

      lowerCAmelCase_			:      int			=   pipe_a(
          prompt_embeds=a__    ,						negative_prompt_embeds=a__    ,						image=a__    ,						original_image=a__    ,						generator=a__    ,						num_inference_steps=2    ,						output_type="""np"""    ,						)

      lowerCAmelCase_			:      Tuple			=   output.images[0]

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

      lowerCAmelCase_			:      List[str]			=   torch.cuda.max_memory_allocated()
      assert mem_bytes < 4 * 10**9

      lowerCAmelCase_			:      Dict			=   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 __lowercase   (   self    :  List[str]    ,						lowerCamelCase    :  List[str]    ,						lowerCamelCase    :  int    ,						lowerCamelCase    :  Union[str, Any]    ,						lowerCamelCase    :  int		)   ->  Optional[Any]:
      _start_torch_memory_measurement()

      lowerCAmelCase_			:      int			=   floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(a__		)
      lowerCAmelCase_			:      Any			=   floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(1		)		).to(a__		)

      lowerCAmelCase_			:      List[Any]			=   torch.Generator(device="""cpu"""		).manual_seed(0		)
      lowerCAmelCase_			:      Tuple			=   pipe_a(
          prompt_embeds=a__    ,						negative_prompt_embeds=a__    ,						image=a__    ,						mask_image=a__    ,						num_inference_steps=2    ,						generator=a__    ,						output_type="""np"""    ,						)

      lowerCAmelCase_			:      Union[str, Any]			=   output.images[0]

      assert image.shape == (64, 64, 3)

      lowerCAmelCase_			:      Tuple			=   torch.cuda.max_memory_allocated()
      assert mem_bytes < 10 * 10**9

      lowerCAmelCase_			:      int			=   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_			:      Optional[Any]			=   torch.Generator(device="""cpu"""		).manual_seed(0		)

      lowerCAmelCase_			:      Dict			=   floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(a__		)
      lowerCAmelCase_			:      str			=   floats_tensor((1, 3, 2_56, 2_56)    ,						rng=random.Random(0		)		).to(a__		)
      lowerCAmelCase_			:      Tuple			=   floats_tensor((1, 3, 2_56, 2_56)    ,						rng=random.Random(1		)		).to(a__		)

      lowerCAmelCase_			:      Union[str, Any]			=   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_			:      Optional[Any]			=   output.images[0]

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

      lowerCAmelCase_			:      int			=   torch.cuda.max_memory_allocated()
      assert mem_bytes < 4 * 10**9

      lowerCAmelCase_			:      Dict			=   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 				UpperCamelCase_			(    ):
  '''simple docstring'''





  torch.cuda.empty_cache()
  torch.cuda.reset_max_memory_allocated()
  torch.cuda.reset_peak_memory_stats()










from datasets.utils.patching import _PatchedModuleObj, patch_submodule

from . import _test_patching






def lowerCamelCase__    (							)      ->				Any:
   import os as original_os
   from os import path as original_path
   from os import rename as original_rename
   from os.path import dirname as original_dirname
   from os.path import join as original_join

   assert _test_patching.os is original_os
   assert _test_patching.path is original_path
   assert _test_patching.join is original_join

   assert _test_patching.renamed_os is original_os
   assert _test_patching.renamed_path is original_path
   assert _test_patching.renamed_join is original_join

   __snake_case							       =   '''__test_patch_submodule_mock__'''
   with patch_submodule(_test_patching       ,      '''os.path.join'''       ,      snake_case_       ):
      # Every way to access os.path.join must be patched, and the rest must stay untouched

      # check os.path.join
      assert isinstance(_test_patching.os       ,      _PatchedModuleObj       )
      assert isinstance(_test_patching.os.path       ,      _PatchedModuleObj       )
      assert _test_patching.os.path.join is mock

      # check path.join
      assert isinstance(_test_patching.path       ,      _PatchedModuleObj       )
      assert _test_patching.path.join is mock

      # check join
      assert _test_patching.join is mock

      # check that the other attributes are untouched
      assert _test_patching.os.rename is original_rename
      assert _test_patching.path.dirname is original_dirname
      assert _test_patching.os.path.dirname is original_dirname

      # Even renamed modules or objects must be patched

      # check renamed_os.path.join
      assert isinstance(_test_patching.renamed_os       ,      _PatchedModuleObj       )
      assert isinstance(_test_patching.renamed_os.path       ,      _PatchedModuleObj       )
      assert _test_patching.renamed_os.path.join is mock

      # check renamed_path.join
      assert isinstance(_test_patching.renamed_path       ,      _PatchedModuleObj       )
      assert _test_patching.renamed_path.join is mock

      # check renamed_join
      assert _test_patching.renamed_join is mock

      # check that the other attributes are untouched
      assert _test_patching.renamed_os.rename is original_rename
      assert _test_patching.renamed_path.dirname is original_dirname
      assert _test_patching.renamed_os.path.dirname is original_dirname

   # check that everthing is back to normal when the patch is over

   assert _test_patching.os is original_os
   assert _test_patching.path is original_path
   assert _test_patching.join is original_join

   assert _test_patching.renamed_os is original_os
   assert _test_patching.renamed_path is original_path
   assert _test_patching.renamed_join is original_join






def lowerCamelCase__    (							)      ->				Any:
   assert _test_patching.open is open

   __snake_case							       =   '''__test_patch_submodule_builtin_mock__'''
   # _test_patching has "open" in its globals
   assert _test_patching.open is open
   with patch_submodule(_test_patching       ,      '''open'''       ,      snake_case_       ):
      assert _test_patching.open is mock

   # check that everthing is back to normal when the patch is over

   assert _test_patching.open is open






def lowerCamelCase__    (							)      ->				List[str]:
   # pandas.read_csv is not present in _test_patching
   __snake_case							       =   '''__test_patch_submodule_missing_mock__'''
   with patch_submodule(_test_patching       ,      '''pandas.read_csv'''       ,      snake_case_       ):
      pass






def lowerCamelCase__    (							)      ->				Union[str, Any]:
   # builtin should always be mocked even if they're not in the globals
   # in case they're loaded at one point
   __snake_case							       =   '''__test_patch_submodule_missing_builtin_mock__'''
   # _test_patching doesn't have "len" in its globals
   assert getattr(_test_patching       ,      '''len'''       ,      snake_case_       ) is None
   with patch_submodule(_test_patching       ,      '''len'''       ,      snake_case_       ):
      assert _test_patching.len is mock
   assert _test_patching.len is len






def lowerCamelCase__    (							)      ->				Union[str, Any]:
   __snake_case							       =   '''__test_patch_submodule_start_and_stop_mock__'''
   __snake_case							       =   patch_submodule(_test_patching       ,      '''open'''       ,      snake_case_       )
   assert _test_patching.open is open
   patch.start()
   assert _test_patching.open is mock
   patch.stop()
   assert _test_patching.open is open






def lowerCamelCase__    (							)      ->				Optional[int]:
   from os import rename as original_rename
   from os.path import dirname as original_dirname
   from os.path import join as original_join

   __snake_case							       =   '''__test_patch_submodule_successive_join__'''
   __snake_case							       =   '''__test_patch_submodule_successive_dirname__'''
   __snake_case							       =   '''__test_patch_submodule_successive_rename__'''
   assert _test_patching.os.path.join is original_join
   assert _test_patching.os.path.dirname is original_dirname
   assert _test_patching.os.rename is original_rename

   with patch_submodule(_test_patching       ,      '''os.path.join'''       ,      snake_case_       ):
      with patch_submodule(_test_patching       ,      '''os.rename'''       ,      snake_case_       ):
         with patch_submodule(_test_patching       ,      '''os.path.dirname'''       ,      snake_case_       ):
            assert _test_patching.os.path.join is mock_join
            assert _test_patching.os.path.dirname is mock_dirname
            assert _test_patching.os.rename is mock_rename

    # try another order
   with patch_submodule(_test_patching       ,      '''os.rename'''       ,      snake_case_       ):
      with patch_submodule(_test_patching       ,      '''os.path.join'''       ,      snake_case_       ):
         with patch_submodule(_test_patching       ,      '''os.path.dirname'''       ,      snake_case_       ):
            assert _test_patching.os.path.join is mock_join
            assert _test_patching.os.path.dirname is mock_dirname
            assert _test_patching.os.rename is mock_rename

   assert _test_patching.os.path.join is original_join
   assert _test_patching.os.path.dirname is original_dirname
   assert _test_patching.os.rename is original_rename







def lowerCamelCase__    (							)      ->				Tuple:
   __snake_case							       =   '''__test_patch_submodule_doesnt_exist_mock__'''
   with patch_submodule(_test_patching       ,      '''__module_that_doesn_exist__.__attribute_that_doesn_exist__'''       ,      snake_case_       ):
      pass
   with patch_submodule(_test_patching       ,      '''os.__attribute_that_doesn_exist__'''       ,      snake_case_       ):
      pass

