

"""simple docstring"""




import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment



def 	UpperCAmelCase__   (				A__	,    A__=()	,    A__=None	,    A__="no"	,    A__="29500"   )      ->      Union[str, Any]:




   """simple docstring"""
   lowerCamelCase__						=			False
   lowerCamelCase__						=			False
   if any(key.startswith("KAGGLE"   ) for key in os.environ.keys()   ):
      lowerCamelCase__						=			True
   elif "IPython" in sys.modules:
      lowerCamelCase__						=			"google.colab" in str(sys.modules["IPython"].get_ipython()   )

   try:
      lowerCamelCase__						=			PrecisionType(mixed_precision.lower()   )
   except ValueError:
      raise ValueError(
          f'Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.'   )

   if (in_colab or in_kaggle) and (os.environ.get("TPU_NAME"	,    A__   ) is not None):
      # TPU launch
      import torch_xla.distributed.xla_multiprocessing as xmp

      if len(AcceleratorState._shared_state   ) > 0:
         raise ValueError(
             "To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside "
             "your training function. Restart your notebook and make sure no cells initializes an "
             "`Accelerator`."   )
      if num_processes is None:
         lowerCamelCase__						=			8

      lowerCamelCase__						=			PrepareForLaunch(A__	,    distributed_type="TPU"   )
      print(f'Launching a training on {num_processes} TPU cores.'   )
      xmp.spawn(A__	,    args=A__	,    nprocs=A__	,    start_method="fork"   )
   elif in_colab:
      # No need for a distributed launch otherwise as it's either CPU or one GPU.
      if torch.cuda.is_available():
         print("Launching training on one GPU."   )
      else:
         print("Launching training on one CPU."   )
      function(*A__   )
   else:
      if num_processes is None:
         raise ValueError(
             "You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call."   )

      if num_processes > 1:
         # Multi-GPU launch
         from torch.multiprocessing import start_processes
         from torch.multiprocessing.spawn import ProcessRaisedException

         if len(AcceleratorState._shared_state   ) > 0:
            raise ValueError(
                "To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized "
                "inside your training function. Restart your notebook and make sure no cells initializes an "
                "`Accelerator`."   )

         if torch.cuda.is_initialized():
            raise ValueError(
                "To launch a multi-GPU training from your notebook, you need to avoid running any instruction "
                "using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA "
                "function."   )

         # torch.distributed will expect a few environment variable to be here. We set the ones common to each
         # process here (the other ones will be set be the launcher).
         with patch_environment(
             world_size=A__	,    master_addr="127.0.01"	,    master_port=A__	,    mixed_precision=A__   ):
            lowerCamelCase__						=			PrepareForLaunch(A__	,    distributed_type="MULTI_GPU"   )
            print(f'Launching training on {num_processes} GPUs.'   )
            try:
               start_processes(A__	,    args=A__	,    nprocs=A__	,    start_method="fork"   )
            except ProcessRaisedException as e:
               if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
                  raise RuntimeError(
                      "CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. "
                      "This likely stems from an outside import causing issues once the `notebook_launcher()` is called. "
                      "Please review your imports and test them when running the `notebook_launcher()` to identify "
                      "which one is problematic."   ) from e

      else:
         # No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
         if is_mps_available():
            lowerCamelCase__						=			"1"
            print("Launching training on MPS."   )
         elif torch.cuda.is_available():
            print("Launching training on one GPU."   )
         else:
            print("Launching training on CPU."   )
         function(*A__   )



def 	UpperCAmelCase__   (				A__	,    A__=()	,    A__=2   )      ->      Any:




   """simple docstring"""
   from torch.multiprocessing import start_processes

   with tempfile.NamedTemporaryFile() as tmp_file:
      # torch.distributed will expect a few environment variable to be here. We set the ones common to each
      # process here (the other ones will be set be the launcher).
      with patch_environment(
          world_size=A__	,    master_addr="127.0.01"	,    master_port="29500"	,    accelerate_mixed_precision="no"	,    accelerate_debug_rdv_file=tmp_file.name	,    accelerate_use_cpu="yes"	,    ):
         lowerCamelCase__						=			PrepareForLaunch(A__	,    debug=A__   )
         start_processes(A__	,    args=A__	,    nprocs=A__	,    start_method="fork"   )







"""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 		_A       (			__a				,       __a				,       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 					_lowerCamelCase  (  self		)   ->					Tuple:
         return self._get_dummy_components()





      def 					_lowerCamelCase  (  self    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__=0		)   ->					Union[str, Any]:
         if str(SCREAMING_SNAKE_CASE__		).startswith("mps"		):
            lowerCamelCase__						=			torch.manual_seed(SCREAMING_SNAKE_CASE__		)
         else:
            lowerCamelCase__						=			torch.Generator(device=SCREAMING_SNAKE_CASE__		).manual_seed(SCREAMING_SNAKE_CASE__		)

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

         return inputs





      def 					_lowerCamelCase  (  self		)   ->					str:
         self._test_save_load_optional_components()





      @unittest.skipIf(torch_device != "cuda"    ,						reason="float16 requires CUDA"		)
      def 					_lowerCamelCase  (  self		)   ->					Optional[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 					_lowerCamelCase  (  self		)   ->					Union[str, Any]:
         self._test_attention_slicing_forward_pass(expected_max_diff=1e-2		)





      def 					_lowerCamelCase  (  self		)   ->					int:
         self._test_save_load_local()





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






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





      def 					_lowerCamelCase  (  self		)   ->					Union[str, Any]:
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()





      def 					_lowerCamelCase  (  self		)   ->					str:
         # if

         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=SCREAMING_SNAKE_CASE__    ,						tokenizer=SCREAMING_SNAKE_CASE__		)

         # 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(SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__		)

         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(SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__		)

         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(SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__		)





      def 					_lowerCamelCase  (  self    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__		)   ->					int:
         # pipeline 1

         _start_torch_memory_measurement()

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

         lowerCamelCase__						=			output.images[0]

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

         lowerCamelCase__						=			torch.cuda.max_memory_allocated()
         assert mem_bytes < 13 * 10**9

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

         # pipeline 2

         _start_torch_memory_measurement()

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

         lowerCamelCase__						=			floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(SCREAMING_SNAKE_CASE__		)

         lowerCamelCase__						=			pipe_a(
             prompt_embeds=SCREAMING_SNAKE_CASE__    ,						negative_prompt_embeds=SCREAMING_SNAKE_CASE__    ,						image=SCREAMING_SNAKE_CASE__    ,						generator=SCREAMING_SNAKE_CASE__    ,						num_inference_steps=2    ,						output_type="np"    ,						)

         lowerCamelCase__						=			output.images[0]

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

         lowerCamelCase__						=			torch.cuda.max_memory_allocated()
         assert mem_bytes < 4 * 10**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(SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__		)





      def 					_lowerCamelCase  (  self    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__		)   ->					Tuple:
         # pipeline 1

         _start_torch_memory_measurement()

         lowerCamelCase__						=			floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(SCREAMING_SNAKE_CASE__		)

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

         lowerCamelCase__						=			pipe_a(
             prompt_embeds=SCREAMING_SNAKE_CASE__    ,						negative_prompt_embeds=SCREAMING_SNAKE_CASE__    ,						image=SCREAMING_SNAKE_CASE__    ,						num_inference_steps=2    ,						generator=SCREAMING_SNAKE_CASE__    ,						output_type="np"    ,						)

         lowerCamelCase__						=			output.images[0]

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

         lowerCamelCase__						=			torch.cuda.max_memory_allocated()
         assert mem_bytes < 10 * 10**9

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

         # pipeline 2

         _start_torch_memory_measurement()

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

         lowerCamelCase__						=			floats_tensor((1, 3, 256, 256)    ,						rng=random.Random(0		)		).to(SCREAMING_SNAKE_CASE__		)
         lowerCamelCase__						=			floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(SCREAMING_SNAKE_CASE__		)

         lowerCamelCase__						=			pipe_a(
             prompt_embeds=SCREAMING_SNAKE_CASE__    ,						negative_prompt_embeds=SCREAMING_SNAKE_CASE__    ,						image=SCREAMING_SNAKE_CASE__    ,						original_image=SCREAMING_SNAKE_CASE__    ,						generator=SCREAMING_SNAKE_CASE__    ,						num_inference_steps=2    ,						output_type="np"    ,						)

         lowerCamelCase__						=			output.images[0]

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

         lowerCamelCase__						=			torch.cuda.max_memory_allocated()
         assert mem_bytes < 4 * 10**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(SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__		)





      def 					_lowerCamelCase  (  self    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__		)   ->					List[Any]:
         # pipeline 1

         _start_torch_memory_measurement()

         lowerCamelCase__						=			floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(SCREAMING_SNAKE_CASE__		)
         lowerCamelCase__						=			floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(1		)		).to(SCREAMING_SNAKE_CASE__		)

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

         lowerCamelCase__						=			output.images[0]

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

         lowerCamelCase__						=			torch.cuda.max_memory_allocated()
         assert mem_bytes < 10 * 10**9

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

         # pipeline 2

         _start_torch_memory_measurement()

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

         lowerCamelCase__						=			floats_tensor((1, 3, 64, 64)    ,						rng=random.Random(0		)		).to(SCREAMING_SNAKE_CASE__		)
         lowerCamelCase__						=			floats_tensor((1, 3, 256, 256)    ,						rng=random.Random(0		)		).to(SCREAMING_SNAKE_CASE__		)
         lowerCamelCase__						=			floats_tensor((1, 3, 256, 256)    ,						rng=random.Random(1		)		).to(SCREAMING_SNAKE_CASE__		)

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

         lowerCamelCase__						=			output.images[0]

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

         lowerCamelCase__						=			torch.cuda.max_memory_allocated()
         assert mem_bytes < 4 * 10**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(SCREAMING_SNAKE_CASE__    ,						SCREAMING_SNAKE_CASE__		)






def 	UpperCAmelCase__   (				)      ->      Union[str, Any]:




   """simple docstring"""
   torch.cuda.empty_cache()
   torch.cuda.reset_max_memory_allocated()
   torch.cuda.reset_peak_memory_stats()




