


'''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 							__magic_name__	(       lowerCAmelCase    ,		lowerCAmelCase    ,		unittest.TestCase ):

     """simple docstring"""






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


     def lowerCAmelCase__  (   self		):


          '''simple docstring'''





          return self._get_dummy_components()


     def lowerCAmelCase__  (   self			,     lowerCamelCase			,     lowerCamelCase=0		):


          '''simple docstring'''





          if str(lowerCamelCase		).startswith("mps"		):
               __A	:       Dict      =							torch.manual_seed(lowerCamelCase		)
          else:
               __A	:       List[str]      =							torch.Generator(device=lowerCamelCase		).manual_seed(lowerCamelCase		)

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

          return inputs


     def lowerCAmelCase__  (   self		):


          '''simple docstring'''





          self._test_save_load_optional_components()


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


          '''simple docstring'''





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


     def lowerCAmelCase__  (   self		):


          '''simple docstring'''





          self._test_attention_slicing_forward_pass(expected_max_diff=1E-2		)


     def lowerCAmelCase__  (   self		):


          '''simple docstring'''





          self._test_save_load_local()


     def lowerCAmelCase__  (   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 lowerCAmelCase__  (   self		):


          '''simple docstring'''





          self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3		)







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

     """simple docstring"""






     def lowerCAmelCase__  (   self		):


          '''simple docstring'''





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


     def lowerCAmelCase__  (   self		):


          '''simple docstring'''





          __A	:       List[str]      =							IFPipeline.from_pretrained("DeepFloyd/IF-I-XL-v1.0"			,     variant="fp16"			,     torch_dtype=torch.floataa		)

          __A	:       Dict      =							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"		)

          __A  ,__A	:       Optional[Any]      =							pipe_a.encode_prompt("anime turtle"			,     device="cuda"		)

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

          __A	:       Dict      =							None
          __A	:       Union[str, Any]      =							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

          __A	:       List[Any]      =							IFImgaImgPipeline(**pipe_a.components		)
          __A	:       Optional[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

          __A	:       Optional[Any]      =							IFInpaintingPipeline(**pipe_a.components		)
          __A	:       str      =							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 lowerCAmelCase__  (   self			,     lowerCamelCase			,     lowerCamelCase			,     lowerCamelCase			,     lowerCamelCase		):


          '''simple docstring'''





          _start_torch_memory_measurement()

          __A	:       str      =							torch.Generator(device="cpu"		).manual_seed(0		)
          __A	:       str      =							pipe_a(
              prompt_embeds=lowerCamelCase			,     negative_prompt_embeds=lowerCamelCase			,     num_inference_steps=2			,     generator=lowerCamelCase			,     output_type="np"			,     )

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

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

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

          __A	:       Union[str, Any]      =							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()

          __A	:       int      =							torch.Generator(device="cpu"		).manual_seed(0		)

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

          __A	:       Any      =							pipe_a(
              prompt_embeds=lowerCamelCase			,     negative_prompt_embeds=lowerCamelCase			,     image=lowerCamelCase			,     generator=lowerCamelCase			,     num_inference_steps=2			,     output_type="np"			,     )

          __A	:       Tuple      =							output.images[0]

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

          __A	:       Union[str, Any]      =							torch.cuda.max_memory_allocated()
          assert mem_bytes < 4 * 10**9

          __A	:       Union[str, 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 lowerCAmelCase__  (   self			,     lowerCamelCase			,     lowerCamelCase			,     lowerCamelCase			,     lowerCamelCase		):


          '''simple docstring'''





          _start_torch_memory_measurement()

          __A	:       List[str]      =							floats_tensor((1, 3, 64, 64)			,     rng=random.Random(0		)		).to(lowerCamelCase		)

          __A	:       Union[str, Any]      =							torch.Generator(device="cpu"		).manual_seed(0		)

          __A	:       int      =							pipe_a(
              prompt_embeds=lowerCamelCase			,     negative_prompt_embeds=lowerCamelCase			,     image=lowerCamelCase			,     num_inference_steps=2			,     generator=lowerCamelCase			,     output_type="np"			,     )

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

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

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

          __A	:       int      =							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()

          __A	:       Optional[Any]      =							torch.Generator(device="cpu"		).manual_seed(0		)

          __A	:       Optional[Any]      =							floats_tensor((1, 3, 256, 256)			,     rng=random.Random(0		)		).to(lowerCamelCase		)
          __A	:       Union[str, Any]      =							floats_tensor((1, 3, 64, 64)			,     rng=random.Random(0		)		).to(lowerCamelCase		)

          __A	:       Optional[int]      =							pipe_a(
              prompt_embeds=lowerCamelCase			,     negative_prompt_embeds=lowerCamelCase			,     image=lowerCamelCase			,     original_image=lowerCamelCase			,     generator=lowerCamelCase			,     num_inference_steps=2			,     output_type="np"			,     )

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

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

          __A	:       Union[str, Any]      =							torch.cuda.max_memory_allocated()
          assert mem_bytes < 4 * 10**9

          __A	:       Optional[int]      =							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 lowerCAmelCase__  (   self			,     lowerCamelCase			,     lowerCamelCase			,     lowerCamelCase			,     lowerCamelCase		):


          '''simple docstring'''





          _start_torch_memory_measurement()

          __A	:       List[Any]      =							floats_tensor((1, 3, 64, 64)			,     rng=random.Random(0		)		).to(lowerCamelCase		)
          __A	:       str      =							floats_tensor((1, 3, 64, 64)			,     rng=random.Random(1		)		).to(lowerCamelCase		)

          __A	:       Union[str, Any]      =							torch.Generator(device="cpu"		).manual_seed(0		)
          __A	:       Dict      =							pipe_a(
              prompt_embeds=lowerCamelCase			,     negative_prompt_embeds=lowerCamelCase			,     image=lowerCamelCase			,     mask_image=lowerCamelCase			,     num_inference_steps=2			,     generator=lowerCamelCase			,     output_type="np"			,     )

          __A	:       Any      =							output.images[0]

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

          __A	:       Union[str, Any]      =							torch.cuda.max_memory_allocated()
          assert mem_bytes < 10 * 10**9

          __A	:       Optional[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()

          __A	:       Dict      =							torch.Generator(device="cpu"		).manual_seed(0		)

          __A	:       Optional[Any]      =							floats_tensor((1, 3, 64, 64)			,     rng=random.Random(0		)		).to(lowerCamelCase		)
          __A	:       int      =							floats_tensor((1, 3, 256, 256)			,     rng=random.Random(0		)		).to(lowerCamelCase		)
          __A	:       int      =							floats_tensor((1, 3, 256, 256)			,     rng=random.Random(1		)		).to(lowerCamelCase		)

          __A	:       List[str]      =							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"			,     )

          __A	:       str      =							output.images[0]

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

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

          __A	:       Optional[Any]      =							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 		_lowercase ():




     '''simple docstring'''





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





'''simple docstring'''


import math
import time

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
      import torch_xla.core.xla_model as xm
      import torch_xla.debug.metrics as met







class 							__magic_name__	(       lowerCAmelCase ):

     """simple docstring"""






     def __init__(   self			,     *lowerCamelCase			,     lowerCamelCase=None			,     lowerCamelCase=None			,     **lowerCamelCase		):


          '''simple docstring'''





          super().__init__(*lowerCamelCase			,     **lowerCamelCase		)
          __A	:       Tuple      =							eval_examples
          __A	:       List[str]      =							post_process_function


     def lowerCAmelCase__  (   self			,     lowerCamelCase=None			,     lowerCamelCase=None			,     lowerCamelCase=None			,     lowerCamelCase = "eval"		):


          '''simple docstring'''





          __A	:       Any      =							self.eval_dataset if eval_dataset is None else eval_dataset
          __A	:       Dict      =							self.get_eval_dataloader(lowerCamelCase		)
          __A	:       str      =							self.eval_examples if eval_examples is None else eval_examples

          # Temporarily disable metric computation, we will do it in the loop here.
          __A	:       Any      =							self.compute_metrics
          __A	:       Optional[int]      =							None
          __A	:       Tuple      =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
          __A	:       Dict      =							time.time()
          try:
               __A	:       List[str]      =							eval_loop(
                   lowerCamelCase			,     description="Evaluation"			,     prediction_loss_only=True if compute_metrics is None else None			,     ignore_keys=lowerCamelCase			,     metric_key_prefix=lowerCamelCase			,     )
          finally:
               __A	:       List[str]      =							compute_metrics
          __A	:       List[Any]      =							self.args.eval_batch_size * self.args.world_size
          if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
               start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
          output.metrics.update(
              speed_metrics(
                  lowerCamelCase			,     lowerCamelCase			,     num_samples=output.num_samples			,     num_steps=math.ceil(output.num_samples / total_batch_size		)			,     )		)
          if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
               # Only the main node write the results by default
               __A	:       List[Any]      =							self.post_process_function(lowerCamelCase			,     lowerCamelCase			,     output.predictions		)
               __A	:       List[str]      =							self.compute_metrics(lowerCamelCase		)

               # Prefix all keys with metric_key_prefix + '_'
               for key in list(metrics.keys()		):
                    if not key.startswith(f"{metric_key_prefix}_"		):
                         __A	:       Union[str, Any]      =							metrics.pop(lowerCamelCase		)
               metrics.update(output.metrics		)
          else:
               __A	:       Any      =							output.metrics

          if self.args.should_log:
               # Only the main node log the results by default
               self.log(lowerCamelCase		)

          if self.args.tpu_metrics_debug or self.args.debug:
               # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
               xm.master_print(met.metrics_report()		)

          __A	:       str      =							self.callback_handler.on_evaluate(self.args			,     self.state			,     self.control			,     lowerCamelCase		)
          return metrics


     def lowerCAmelCase__  (   self			,     lowerCamelCase			,     lowerCamelCase			,     lowerCamelCase=None			,     lowerCamelCase = "test"		):


          '''simple docstring'''





          __A	:       Tuple      =							self.get_test_dataloader(lowerCamelCase		)

          # Temporarily disable metric computation, we will do it in the loop here.
          __A	:       Union[str, Any]      =							self.compute_metrics
          __A	:       Union[str, Any]      =							None
          __A	:       Optional[Any]      =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
          __A	:       Tuple      =							time.time()
          try:
               __A	:       List[str]      =							eval_loop(
                   lowerCamelCase			,     description="Prediction"			,     prediction_loss_only=True if compute_metrics is None else None			,     ignore_keys=lowerCamelCase			,     metric_key_prefix=lowerCamelCase			,     )
          finally:
               __A	:       int      =							compute_metrics
          __A	:       List[Any]      =							self.args.eval_batch_size * self.args.world_size
          if f"{metric_key_prefix}_jit_compilation_time" in output.metrics:
               start_time += output.metrics[f"{metric_key_prefix}_jit_compilation_time"]
          output.metrics.update(
              speed_metrics(
                  lowerCamelCase			,     lowerCamelCase			,     num_samples=output.num_samples			,     num_steps=math.ceil(output.num_samples / total_batch_size		)			,     )		)

          if self.post_process_function is None or self.compute_metrics is None:
               return output

          __A	:       Dict      =							self.post_process_function(lowerCamelCase			,     lowerCamelCase			,     output.predictions			,     "predict"		)
          __A	:       Tuple      =							self.compute_metrics(lowerCamelCase		)

          # Prefix all keys with metric_key_prefix + '_'
          for key in list(metrics.keys()		):
               if not key.startswith(f"{metric_key_prefix}_"		):
                    __A	:       Any      =							metrics.pop(lowerCamelCase		)
          metrics.update(output.metrics		)
          return PredictionOutput(predictions=predictions.predictions			,     label_ids=predictions.label_ids			,     metrics=lowerCamelCase		)

