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 		_lowerCamelCase		(   UpperCamelCase_    ,     UpperCamelCase_    ,     unittest.TestCase    ):
    __a  =      IFPipeline
    __a  =      TEXT_TO_IMAGE_PARAMS - {"width", "height", "latents"}
    __a  =      TEXT_TO_IMAGE_BATCH_PARAMS
    __a  =      PipelineTesterMixin.required_optional_params - {"latents"}
    def 						UpperCamelCase_				(  self   )       ->					int:
     return self._get_dummy_components()
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase=0   )       ->					Dict:
     if str(lowerCAmelCase   ).startswith('''mps'''   ):
      SCREAMING_SNAKE_CASE__:   List[str]=      torch.manual_seed(lowerCAmelCase   )
     else:
      SCREAMING_SNAKE_CASE__:   str=      torch.Generator(device=lowerCAmelCase   ).manual_seed(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      {
         '''prompt''': '''A painting of a squirrel eating a burger''',
         '''generator''': generator,
         '''num_inference_steps''': 2,
         '''output_type''': '''numpy''',
     }

     return inputs
    def 						UpperCamelCase_				(  self   )       ->					List[Any]:
     self._test_save_load_optional_components()
    @unittest.skipIf(torch_device != '''cuda''' , reason='''float16 requires CUDA'''   )
    def 						UpperCamelCase_				(  self   )       ->					int:
     # Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
     super().test_save_load_floataa(expected_max_diff=1e-1   )
    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     self._test_attention_slicing_forward_pass(expected_max_diff=1e-2   )
    def 						UpperCamelCase_				(  self   )       ->					Any:
     self._test_save_load_local()
    def 						UpperCamelCase_				(  self   )       ->					str:
     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 						UpperCamelCase_				(  self   )       ->					Union[str, Any]:
     self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3   )






@slow
@require_torch_gpu
class 		_lowerCamelCase		(   unittest.TestCase    ):
    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     # clean up the VRAM after each test
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()
    def 						UpperCamelCase_				(  self   )       ->					List[Any]:
     # if

     SCREAMING_SNAKE_CASE__:   Tuple=      IFPipeline.from_pretrained('''DeepFloyd/IF-I-XL-v1.0''' , variant='''fp16''' , torch_dtype=torch.floataa   )

     SCREAMING_SNAKE_CASE__:   List[str]=      IFSuperResolutionPipeline.from_pretrained(
         '''DeepFloyd/IF-II-L-v1.0''' , variant='''fp16''' , torch_dtype=torch.floataa , text_encoder=lowerCAmelCase , tokenizer=lowerCAmelCase   )

     # pre compute text embeddings and remove T5 to save memory

     pipe_a.text_encoder.to('''cuda'''   )

     SCREAMING_SNAKE_CASE__,	SCREAMING_SNAKE_CASE__:   List[Any]=      pipe_a.encode_prompt('''anime turtle''' , device='''cuda'''   )

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

     SCREAMING_SNAKE_CASE__:   Any=      None
     SCREAMING_SNAKE_CASE__:   Tuple=      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(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )

     pipe_a.remove_all_hooks()
     pipe_a.remove_all_hooks()

     # img2img

     SCREAMING_SNAKE_CASE__:   int=      IFImgaImgPipeline(**pipe_a.components   )
     SCREAMING_SNAKE_CASE__:   Any=      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(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )

     pipe_a.remove_all_hooks()
     pipe_a.remove_all_hooks()

     # inpainting

     SCREAMING_SNAKE_CASE__:   Optional[int]=      IFInpaintingPipeline(**pipe_a.components   )
     SCREAMING_SNAKE_CASE__:   Union[str, 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(lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )       ->					Optional[int]:
     # pipeline 1

     _start_torch_memory_measurement()

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      torch.Generator(device='''cpu'''   ).manual_seed(0   )
     SCREAMING_SNAKE_CASE__:   Tuple=      pipe_a(
         prompt_embeds=lowerCAmelCase , negative_prompt_embeds=lowerCAmelCase , num_inference_steps=2 , generator=lowerCAmelCase , output_type='''np''' , )

     SCREAMING_SNAKE_CASE__:   List[Any]=      output.images[0]

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

     SCREAMING_SNAKE_CASE__:   Optional[int]=      torch.cuda.max_memory_allocated()
     assert mem_bytes < 13 * 10**9

     SCREAMING_SNAKE_CASE__:   int=      load_numpy(
         '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if.npy'''   )
     assert_mean_pixel_difference(lowerCAmelCase , lowerCAmelCase   )

     # pipeline 2

     _start_torch_memory_measurement()

     SCREAMING_SNAKE_CASE__:   Tuple=      torch.Generator(device='''cpu'''   ).manual_seed(0   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      floats_tensor((1, 3, 64, 64) , rng=random.Random(0   )   ).to(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      pipe_a(
         prompt_embeds=lowerCAmelCase , negative_prompt_embeds=lowerCAmelCase , image=lowerCAmelCase , generator=lowerCAmelCase , num_inference_steps=2 , output_type='''np''' , )

     SCREAMING_SNAKE_CASE__:   Dict=      output.images[0]

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

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      torch.cuda.max_memory_allocated()
     assert mem_bytes < 4 * 10**9

     SCREAMING_SNAKE_CASE__:   List[Any]=      load_numpy(
         '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_superresolution_stage_II.npy'''   )
     assert_mean_pixel_difference(lowerCAmelCase , lowerCAmelCase   )
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )       ->					Union[str, Any]:
     # pipeline 1

     _start_torch_memory_measurement()

     SCREAMING_SNAKE_CASE__:   Any=      floats_tensor((1, 3, 64, 64) , rng=random.Random(0   )   ).to(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      torch.Generator(device='''cpu'''   ).manual_seed(0   )

     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      pipe_a(
         prompt_embeds=lowerCAmelCase , negative_prompt_embeds=lowerCAmelCase , image=lowerCAmelCase , num_inference_steps=2 , generator=lowerCAmelCase , output_type='''np''' , )

     SCREAMING_SNAKE_CASE__:   Tuple=      output.images[0]

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

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

     SCREAMING_SNAKE_CASE__:   Any=      load_numpy(
         '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img.npy'''   )
     assert_mean_pixel_difference(lowerCAmelCase , lowerCAmelCase   )

     # pipeline 2

     _start_torch_memory_measurement()

     SCREAMING_SNAKE_CASE__:   Dict=      torch.Generator(device='''cpu'''   ).manual_seed(0   )

     SCREAMING_SNAKE_CASE__:   Tuple=      floats_tensor((1, 3, 256, 256) , rng=random.Random(0   )   ).to(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      floats_tensor((1, 3, 64, 64) , rng=random.Random(0   )   ).to(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   List[Any]=      pipe_a(
         prompt_embeds=lowerCAmelCase , negative_prompt_embeds=lowerCAmelCase , image=lowerCAmelCase , original_image=lowerCAmelCase , generator=lowerCAmelCase , num_inference_steps=2 , output_type='''np''' , )

     SCREAMING_SNAKE_CASE__:   Dict=      output.images[0]

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

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

     SCREAMING_SNAKE_CASE__:   str=      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(lowerCAmelCase , lowerCAmelCase   )



    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )       ->					str:
     # pipeline 1

     _start_torch_memory_measurement()

     SCREAMING_SNAKE_CASE__:   str=      floats_tensor((1, 3, 64, 64) , rng=random.Random(0   )   ).to(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Tuple=      floats_tensor((1, 3, 64, 64) , rng=random.Random(1   )   ).to(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   List[str]=      torch.Generator(device='''cpu'''   ).manual_seed(0   )
     SCREAMING_SNAKE_CASE__:   List[str]=      pipe_a(
         prompt_embeds=lowerCAmelCase , negative_prompt_embeds=lowerCAmelCase , image=lowerCAmelCase , mask_image=lowerCAmelCase , num_inference_steps=2 , generator=lowerCAmelCase , output_type='''np''' , )

     SCREAMING_SNAKE_CASE__:   Any=      output.images[0]

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

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

     SCREAMING_SNAKE_CASE__:   Any=      load_numpy(
         '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting.npy'''   )
     assert_mean_pixel_difference(lowerCAmelCase , lowerCAmelCase   )

     # pipeline 2

     _start_torch_memory_measurement()

     SCREAMING_SNAKE_CASE__:   int=      torch.Generator(device='''cpu'''   ).manual_seed(0   )

     SCREAMING_SNAKE_CASE__:   str=      floats_tensor((1, 3, 64, 64) , rng=random.Random(0   )   ).to(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   List[str]=      floats_tensor((1, 3, 256, 256) , rng=random.Random(0   )   ).to(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Tuple=      floats_tensor((1, 3, 256, 256) , rng=random.Random(1   )   ).to(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Tuple=      pipe_a(
         prompt_embeds=lowerCAmelCase , negative_prompt_embeds=lowerCAmelCase , image=lowerCAmelCase , mask_image=lowerCAmelCase , original_image=lowerCAmelCase , generator=lowerCAmelCase , num_inference_steps=2 , output_type='''np''' , )

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

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

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

     SCREAMING_SNAKE_CASE__:   int=      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(lowerCAmelCase , lowerCAmelCase   )







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


# tests directory-specific settings - this file is run automatically
# by pytest before any tests are run

import sys
import warnings
from os.path import abspath, dirname, join


# allow having multiple repository checkouts and not needing to remember to rerun
# 'pip install -e .[dev]' when switching between checkouts and running tests.
lowercase_     :   List[str] 				=				abspath(join(dirname(dirname(__file__)), 'src'))
sys.path.insert(1, git_repo_path)

# silence FutureWarning warnings in tests since often we can't act on them until
# they become normal warnings - i.e. the tests still need to test the current functionality
warnings.simplefilter(action='ignore', category=FutureWarning)







def 			A__      (	snake_case_ :		Optional[int]      ):
 from diffusers.utils.testing_utils import pytest_addoption_shared

 pytest_addoption_shared(snake_case_      )







def 			A__      (	snake_case_ :		Optional[int]      ):
 from diffusers.utils.testing_utils import pytest_terminal_summary_main

 SCREAMING_SNAKE_CASE__:   Optional[int]=      terminalreporter.config.getoption('''--make-reports'''      )
 if make_reports:
  pytest_terminal_summary_main(snake_case_       ,				id=snake_case_      )

