
import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


A     :	List[Any]    = False







@skip_mps
class 			__A( a  ,   a  ,   a  ,   unittest.TestCase			):
				snake_case_						   =    StableDiffusionAttendAndExcitePipeline
				snake_case_						   =    False
				snake_case_						   =    TEXT_TO_IMAGE_PARAMS
				snake_case_						   =    TEXT_TO_IMAGE_BATCH_PARAMS.union({'''token_indices'''}			)
				snake_case_						   =    TEXT_TO_IMAGE_IMAGE_PARAMS
				snake_case_						   =    TEXT_TO_IMAGE_IMAGE_PARAMS




				@classmethod
				def 				SCREAMING_SNAKE_CASE_		(     cls				)    ->      List[str]:


								'''simple docstring'''




								super().setUpClass()
								torch.use_deterministic_algorithms(_snake_case				)




				@classmethod
				def 				SCREAMING_SNAKE_CASE_		(     cls				)    ->      int:


								'''simple docstring'''




								super().tearDownClass()
								torch.use_deterministic_algorithms(_snake_case				)




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


								'''simple docstring'''




								torch.manual_seed(0				)
								__a										=    UNetaDConditionModel(
								    block_out_channels=(32, 64)		,     layers_per_block=1		,     sample_size=32		,     in_channels=4		,     out_channels=4		,     down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,     up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		,     cross_attention_dim=32		,     attention_head_dim=(2, 4)		,     use_linear_projection=_snake_case		,     )
								__a										=    DDIMScheduler(
								    beta_start=0.0_0085		,     beta_end=0.012		,     beta_schedule='''scaled_linear'''		,     clip_sample=_snake_case		,     set_alpha_to_one=_snake_case		,     )
								torch.manual_seed(0				)
								__a										=    AutoencoderKL(
								    block_out_channels=[32, 64]		,     in_channels=3		,     out_channels=3		,     down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']		,     up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']		,     latent_channels=4		,     sample_size=128		,     )
								torch.manual_seed(0				)
								__a										=    CLIPTextConfig(
								    bos_token_id=0		,     eos_token_id=2		,     hidden_size=32		,     intermediate_size=37		,     layer_norm_eps=1E-05		,     num_attention_heads=4		,     num_hidden_layers=5		,     pad_token_id=1		,     vocab_size=1_000		,     hidden_act='''gelu'''		,     projection_dim=512		,     )
								__a										=    CLIPTextModel(_snake_case				)
								__a										=    CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''				)

								__a										=    {
								    '''unet''': unet,
								    '''scheduler''': scheduler,
								    '''vae''': vae,
								    '''text_encoder''': text_encoder,
								    '''tokenizer''': tokenizer,
								    '''safety_checker''': None,
								    '''feature_extractor''': None,
								}

								return components




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case=0				)    ->      Any:


								'''simple docstring'''




								if str(_snake_case				).startswith('''mps'''				):
												__a										=    torch.manual_seed(_snake_case				)
								else:
												__a										=    torch.Generator(device=_snake_case				).manual_seed(_snake_case				)
								__a										=    __a										=    {
								    '''prompt''': '''a cat and a frog''',
								    '''token_indices''': [2, 5],
								    '''generator''': generator,
								    '''num_inference_steps''': 1,
								    '''guidance_scale''': 6.0,
								    '''output_type''': '''numpy''',
								    '''max_iter_to_alter''': 2,
								    '''thresholds''': {0: 0.7},
								}
								return inputs




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Tuple:


								'''simple docstring'''




								__a										=    '''cpu'''

								__a										=    self.get_dummy_components()
								__a										=    self.pipeline_class(**_snake_case				)
								pipe.to(_snake_case				)
								pipe.set_progress_bar_config(disable=_snake_case				)

								__a										=    self.get_dummy_inputs(_snake_case				)
								__a										=    pipe(**_snake_case				).images
								__a										=    image[0, -3:, -3:, -1]

								self.assertEqual(image.shape		,     (1, 64, 64, 3)				)
								__a										=    np.array(
								    [0.6390_5364, 0.6289_7307, 0.4859_9017, 0.513_3624, 0.555_0048, 0.4576_9516, 0.5032_6973, 0.502_3139, 0.4538_4496]				)
								__a										=    np.abs(image_slice.flatten() - expected_slice				).max()
								self.assertLessEqual(_snake_case		,     1E-3				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[int]:


								'''simple docstring'''




								super().test_cpu_offload_forward_pass(expected_max_diff=5E-4				)




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


								'''simple docstring'''




								self._test_inference_batch_consistent(batch_sizes=[1, 2]				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      str:


								'''simple docstring'''




								self._test_inference_batch_single_identical(batch_size=2		,     expected_max_diff=7E-4				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[int]:


								'''simple docstring'''




								super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      str:


								'''simple docstring'''




								super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5E-4				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Dict:


								'''simple docstring'''




								super().test_save_load_local(expected_max_difference=5E-4				)




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


								'''simple docstring'''




								super().test_save_load_optional_components(expected_max_difference=4E-4				)







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




				@classmethod
				def 				SCREAMING_SNAKE_CASE_		(     cls				)    ->      Dict:


								'''simple docstring'''




								super().setUpClass()
								torch.use_deterministic_algorithms(_snake_case				)




				@classmethod
				def 				SCREAMING_SNAKE_CASE_		(     cls				)    ->      Union[str, Any]:


								'''simple docstring'''




								super().tearDownClass()
								torch.use_deterministic_algorithms(_snake_case				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      List[str]:


								'''simple docstring'''




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




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      List[Any]:


								'''simple docstring'''




								__a										=    torch.manual_seed(51				)

								__a										=    StableDiffusionAttendAndExcitePipeline.from_pretrained(
								    '''CompVis/stable-diffusion-v1-4'''		,     safety_checker=_snake_case		,     torch_dtype=torch.floataa				)
								pipe.to('''cuda'''				)

								__a										=    '''a painting of an elephant with glasses'''
								__a										=    [5, 7]

								__a										=    pipe(
								    prompt=_snake_case		,     token_indices=_snake_case		,     guidance_scale=7.5		,     generator=_snake_case		,     num_inference_steps=5		,     max_iter_to_alter=5		,     output_type='''numpy'''		,     ).images[0]

								__a										=    load_numpy(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy'''				)
								assert np.abs((expected_image - image).max()				) < 5E-1




import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, 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     _A       (	__UpperCAmelCase    ):



       def __init__(		self							:     List[Any]     ,		*__SCREAMING_SNAKE_CASE							:     Tuple     ,		__SCREAMING_SNAKE_CASE							:     Any=None     ,		__SCREAMING_SNAKE_CASE							:     Union[str, Any]=None     ,		**__SCREAMING_SNAKE_CASE							:     str):


           '''simple docstring'''

           super().__init__(*__SCREAMING_SNAKE_CASE     ,		**__SCREAMING_SNAKE_CASE)
           __a						       =				eval_examples
           __a						       =				post_process_function



       def 		_lowerCamelCase			(		self							:     Dict     ,		__SCREAMING_SNAKE_CASE							:     Optional[Dataset] = None     ,		__SCREAMING_SNAKE_CASE							:     List[Any]=None     ,		__SCREAMING_SNAKE_CASE							:     Optional[List[str]] = None     ,		__SCREAMING_SNAKE_CASE							:     str = "eval"     ,		**__SCREAMING_SNAKE_CASE							:     Any     ,		):


           '''simple docstring'''

           __a						       =				gen_kwargs.copy()
           __a						       =				(
               gen_kwargs['''max_length'''] if gen_kwargs.get('''max_length''') is not None else self.args.generation_max_length
           )
           __a						       =				(
               gen_kwargs['''num_beams'''] if gen_kwargs.get('''num_beams''') is not None else self.args.generation_num_beams
           )
           __a						       =				gen_kwargs

           __a						       =				self.eval_dataset if eval_dataset is None else eval_dataset
           __a						       =				self.get_eval_dataloader(__SCREAMING_SNAKE_CASE)
           __a						       =				self.eval_examples if eval_examples is None else eval_examples

           # Temporarily disable metric computation, we will do it in the loop here.
           __a						       =				self.compute_metrics
           __a						       =				None
           __a						       =				time.time()
           __a						       =				self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
           try:
               __a						       =				eval_loop(
                   __SCREAMING_SNAKE_CASE     ,		description='''Evaluation'''     ,		prediction_loss_only=True if compute_metrics is None else None     ,		ignore_keys=__SCREAMING_SNAKE_CASE     ,		metric_key_prefix=__SCREAMING_SNAKE_CASE     ,		)
           finally:
               __a						       =				compute_metrics
           __a						       =				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(
                   __SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		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						       =				self.post_process_function(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE)
               __a						       =				self.compute_metrics(__SCREAMING_SNAKE_CASE)

               # Prefix all keys with metric_key_prefix + '_'
               for key in list(metrics.keys()):
                   if not key.startswith(F'{metric_key_prefix}_'):
                       __a						       =				metrics.pop(__SCREAMING_SNAKE_CASE)

               metrics.update(output.metrics)
           else:
               __a						       =				output.metrics

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

           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						       =				self.callback_handler.on_evaluate(self.args     ,		self.state     ,		self.control     ,		__SCREAMING_SNAKE_CASE)
           return metrics





       def 		_lowerCamelCase			(		self							:     Tuple     ,		__SCREAMING_SNAKE_CASE							:     Any     ,		__SCREAMING_SNAKE_CASE							:     int     ,		__SCREAMING_SNAKE_CASE							:     Tuple=None     ,		__SCREAMING_SNAKE_CASE							:     str = "test"     ,		**__SCREAMING_SNAKE_CASE							:     Dict):


           '''simple docstring'''

           __a						       =				gen_kwargs.copy()

           __a						       =				self.get_test_dataloader(__SCREAMING_SNAKE_CASE)

           # Temporarily disable metric computation, we will do it in the loop here.
           __a						       =				self.compute_metrics
           __a						       =				None
           __a						       =				time.time()
           __a						       =				self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
           try:
               __a						       =				eval_loop(
                   __SCREAMING_SNAKE_CASE     ,		description='''Prediction'''     ,		prediction_loss_only=True if compute_metrics is None else None     ,		ignore_keys=__SCREAMING_SNAKE_CASE     ,		metric_key_prefix=__SCREAMING_SNAKE_CASE     ,		)
           finally:
               __a						       =				compute_metrics

           __a						       =				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(
                   __SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		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						       =				self.post_process_function(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		'''predict''')
           __a						       =				self.compute_metrics(__SCREAMING_SNAKE_CASE)

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





