






'''simple docstring'''



import shutil
import tempfile
import unittest
from unittest.mock import patch

from transformers import (
    DefaultFlowCallback,
    IntervalStrategy,
    PrinterCallback,
    ProgressCallback,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    is_torch_available,
)
from transformers.testing_utils import require_torch


if is_torch_available():
      from transformers.trainer import DEFAULT_CALLBACKS

      from .test_trainer import RegressionDataset, RegressionModelConfig, RegressionPreTrainedModel
class UpperCAmelCase_       (     __A					):




      """simple docstring"""




      def __init__(			self      :		Optional[Any]		)	->	Optional[int]:


       '''simple docstring'''

       lowercase       : Union[str, Any]						=[]






      def 					A__   (			self      :		Optional[int]  ,   UpperCAmelCase      :		List[Any]  ,   UpperCAmelCase      :		str  ,   UpperCAmelCase      :		Any  ,   **UpperCAmelCase      :		Tuple		)	->	Optional[Any]:


       '''simple docstring'''

       self.events.append('''on_init_end'''		)






      def 					A__   (			self      :		Union[str, Any]  ,   UpperCAmelCase      :		List[Any]  ,   UpperCAmelCase      :		Any  ,   UpperCAmelCase      :		Any  ,   **UpperCAmelCase      :		Dict		)	->	int:


       '''simple docstring'''

       self.events.append('''on_train_begin'''		)






      def 					A__   (			self      :		Dict  ,   UpperCAmelCase      :		Optional[Any]  ,   UpperCAmelCase      :		Tuple  ,   UpperCAmelCase      :		Union[str, Any]  ,   **UpperCAmelCase      :		Tuple		)	->	Optional[Any]:


       '''simple docstring'''

       self.events.append('''on_train_end'''		)






      def 					A__   (			self      :		Any  ,   UpperCAmelCase      :		str  ,   UpperCAmelCase      :		Tuple  ,   UpperCAmelCase      :		Dict  ,   **UpperCAmelCase      :		Any		)	->	Dict:


       '''simple docstring'''

       self.events.append('''on_epoch_begin'''		)






      def 					A__   (			self      :		Optional[int]  ,   UpperCAmelCase      :		List[Any]  ,   UpperCAmelCase      :		Union[str, Any]  ,   UpperCAmelCase      :		List[str]  ,   **UpperCAmelCase      :		List[Any]		)	->	List[str]:


       '''simple docstring'''

       self.events.append('''on_epoch_end'''		)






      def 					A__   (			self      :		str  ,   UpperCAmelCase      :		Union[str, Any]  ,   UpperCAmelCase      :		List[str]  ,   UpperCAmelCase      :		Optional[Any]  ,   **UpperCAmelCase      :		Optional[int]		)	->	int:


       '''simple docstring'''

       self.events.append('''on_step_begin'''		)






      def 					A__   (			self      :		List[Any]  ,   UpperCAmelCase      :		Union[str, Any]  ,   UpperCAmelCase      :		List[Any]  ,   UpperCAmelCase      :		Tuple  ,   **UpperCAmelCase      :		List[Any]		)	->	Any:


       '''simple docstring'''

       self.events.append('''on_step_end'''		)






      def 					A__   (			self      :		Optional[int]  ,   UpperCAmelCase      :		Tuple  ,   UpperCAmelCase      :		Optional[Any]  ,   UpperCAmelCase      :		Union[str, Any]  ,   **UpperCAmelCase      :		Tuple		)	->	Dict:


       '''simple docstring'''

       self.events.append('''on_evaluate'''		)






      def 					A__   (			self      :		int  ,   UpperCAmelCase      :		Any  ,   UpperCAmelCase      :		Union[str, Any]  ,   UpperCAmelCase      :		List[str]  ,   **UpperCAmelCase      :		str		)	->	List[str]:


       '''simple docstring'''

       self.events.append('''on_predict'''		)






      def 					A__   (			self      :		Optional[Any]  ,   UpperCAmelCase      :		Tuple  ,   UpperCAmelCase      :		Optional[int]  ,   UpperCAmelCase      :		int  ,   **UpperCAmelCase      :		Dict		)	->	Union[str, Any]:


       '''simple docstring'''

       self.events.append('''on_save'''		)






      def 					A__   (			self      :		int  ,   UpperCAmelCase      :		List[str]  ,   UpperCAmelCase      :		int  ,   UpperCAmelCase      :		Any  ,   **UpperCAmelCase      :		int		)	->	str:


       '''simple docstring'''

       self.events.append('''on_log'''		)






      def 					A__   (			self      :		str  ,   UpperCAmelCase      :		List[str]  ,   UpperCAmelCase      :		List[str]  ,   UpperCAmelCase      :		str  ,   **UpperCAmelCase      :		Any		)	->	Tuple:


       '''simple docstring'''

       self.events.append('''on_prediction_step'''		)





@require_torch
class UpperCAmelCase_       (     unittest.TestCase					):




      """simple docstring"""




      def 					A__   (			self      :		Optional[Any]		)	->	int:


       '''simple docstring'''

       lowercase       : Optional[int]						=tempfile.mkdtemp()






      def 					A__   (			self      :		Optional[int]		)	->	Union[str, Any]:


       '''simple docstring'''

       shutil.rmtree(self.output_dir		)






      def 					A__   (			self      :		str  ,   UpperCAmelCase      :		List[Any]=0  ,   UpperCAmelCase      :		Tuple=0  ,   UpperCAmelCase      :		Dict=64  ,   UpperCAmelCase      :		List[Any]=64  ,   UpperCAmelCase      :		Optional[int]=None  ,   UpperCAmelCase      :		Optional[Any]=False  ,   **UpperCAmelCase      :		Optional[int]		)	->	Union[str, Any]:


       '''simple docstring'''

       lowercase       : Union[str, Any]						=RegressionDataset(length=UpperCAmelCase		)
       lowercase       : Any						=RegressionDataset(length=UpperCAmelCase		)
       lowercase       : Optional[int]						=RegressionModelConfig(a=UpperCAmelCase  ,   b=UpperCAmelCase		)
       lowercase       : Any						=RegressionPreTrainedModel(UpperCAmelCase		)

       lowercase       : Dict						=TrainingArguments(self.output_dir  ,   disable_tqdm=UpperCAmelCase  ,   report_to=[]  ,   **UpperCAmelCase		)
       return Trainer(
           UpperCAmelCase  ,   UpperCAmelCase  ,   train_dataset=UpperCAmelCase  ,   eval_dataset=UpperCAmelCase  ,   callbacks=UpperCAmelCase  ,   )






      def 					A__   (			self      :		Tuple  ,   UpperCAmelCase      :		Union[str, Any]  ,   UpperCAmelCase      :		List[str]		)	->	Tuple:


       '''simple docstring'''

       self.assertEqual(len(UpperCAmelCase		)  ,   len(UpperCAmelCase		)		)

       # Order doesn't matter
       lowercase       : Tuple						=sorted(UpperCAmelCase  ,   key=lambda UpperCAmelCase		: cb.__name__ if isinstance(UpperCAmelCase  ,   UpperCAmelCase		) else cb.__class__.__name__		)
       lowercase       : Any						=sorted(UpperCAmelCase  ,   key=lambda UpperCAmelCase		: cb.__name__ if isinstance(UpperCAmelCase  ,   UpperCAmelCase		) else cb.__class__.__name__		)

       for cba, cba in zip(UpperCAmelCase  ,   UpperCAmelCase		):
        if isinstance(UpperCAmelCase  ,   UpperCAmelCase		) and isinstance(UpperCAmelCase  ,   UpperCAmelCase		):
         self.assertEqual(UpperCAmelCase  ,   UpperCAmelCase		)
        elif isinstance(UpperCAmelCase  ,   UpperCAmelCase		) and not isinstance(UpperCAmelCase  ,   UpperCAmelCase		):
         self.assertEqual(UpperCAmelCase  ,   cba.__class__		)
        elif not isinstance(UpperCAmelCase  ,   UpperCAmelCase		) and isinstance(UpperCAmelCase  ,   UpperCAmelCase		):
         self.assertEqual(cba.__class__  ,   UpperCAmelCase		)
        else:
         self.assertEqual(UpperCAmelCase  ,   UpperCAmelCase		)






      def 					A__   (			self      :		Dict  ,   UpperCAmelCase      :		Optional[Any]		)	->	Any:


       '''simple docstring'''

       lowercase       : Tuple						=['''on_init_end''', '''on_train_begin''']
       lowercase       : List[str]						=0
       lowercase       : Union[str, Any]						=len(trainer.get_eval_dataloader()		)
       lowercase       : Optional[Any]						=['''on_prediction_step'''] * len(trainer.get_eval_dataloader()		) + ['''on_log''', '''on_evaluate''']
       for _ in range(trainer.state.num_train_epochs		):
        expected_events.append('''on_epoch_begin'''		)
        for _ in range(UpperCAmelCase		):
         step += 1
         expected_events += ["on_step_begin", "on_step_end"]
         if step % trainer.args.logging_steps == 0:
          expected_events.append('''on_log'''		)
         if trainer.args.evaluation_strategy == IntervalStrategy.STEPS and step % trainer.args.eval_steps == 0:
          expected_events += evaluation_events.copy()
         if step % trainer.args.save_steps == 0:
          expected_events.append('''on_save'''		)
        expected_events.append('''on_epoch_end'''		)
        if trainer.args.evaluation_strategy == IntervalStrategy.EPOCH:
         expected_events += evaluation_events.copy()
       expected_events += ["on_log", "on_train_end"]
       return expected_events






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


       '''simple docstring'''

       lowercase       : List[str]						=self.get_trainer()
       lowercase       : Dict						=DEFAULT_CALLBACKS.copy() + [ProgressCallback]
       self.check_callbacks_equality(trainer.callback_handler.callbacks  ,   UpperCAmelCase		)

       # Callbacks passed at init are added to the default callbacks
       lowercase       : List[Any]						=self.get_trainer(callbacks=[MyTestTrainerCallback]		)
       expected_callbacks.append(UpperCAmelCase		)
       self.check_callbacks_equality(trainer.callback_handler.callbacks  ,   UpperCAmelCase		)

       # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
       lowercase       : Tuple						=self.get_trainer(disable_tqdm=UpperCAmelCase		)
       lowercase       : str						=DEFAULT_CALLBACKS.copy() + [PrinterCallback]
       self.check_callbacks_equality(trainer.callback_handler.callbacks  ,   UpperCAmelCase		)






      def 					A__   (			self      :		int		)	->	Optional[Any]:


       '''simple docstring'''

       lowercase       : Any						=DEFAULT_CALLBACKS.copy() + [ProgressCallback]
       lowercase       : Optional[int]						=self.get_trainer()

       # We can add, pop, or remove by class name
       trainer.remove_callback(UpperCAmelCase		)
       expected_callbacks.remove(UpperCAmelCase		)
       self.check_callbacks_equality(trainer.callback_handler.callbacks  ,   UpperCAmelCase		)

       lowercase       : Union[str, Any]						=self.get_trainer()
       lowercase       : Union[str, Any]						=trainer.pop_callback(UpperCAmelCase		)
       self.assertEqual(cb.__class__  ,   UpperCAmelCase		)
       self.check_callbacks_equality(trainer.callback_handler.callbacks  ,   UpperCAmelCase		)

       trainer.add_callback(UpperCAmelCase		)
       expected_callbacks.insert(0  ,   UpperCAmelCase		)
       self.check_callbacks_equality(trainer.callback_handler.callbacks  ,   UpperCAmelCase		)

       # We can also add, pop, or remove by instance
       lowercase       : Optional[int]						=self.get_trainer()
       lowercase       : str						=trainer.callback_handler.callbacks[0]
       trainer.remove_callback(UpperCAmelCase		)
       expected_callbacks.remove(UpperCAmelCase		)
       self.check_callbacks_equality(trainer.callback_handler.callbacks  ,   UpperCAmelCase		)

       lowercase       : Union[str, Any]						=self.get_trainer()
       lowercase       : List[Any]						=trainer.callback_handler.callbacks[0]
       lowercase       : Optional[int]						=trainer.pop_callback(UpperCAmelCase		)
       self.assertEqual(UpperCAmelCase  ,   UpperCAmelCase		)
       self.check_callbacks_equality(trainer.callback_handler.callbacks  ,   UpperCAmelCase		)

       trainer.add_callback(UpperCAmelCase		)
       expected_callbacks.insert(0  ,   UpperCAmelCase		)
       self.check_callbacks_equality(trainer.callback_handler.callbacks  ,   UpperCAmelCase		)






      def 					A__   (			self      :		str		)	->	Optional[int]:


       '''simple docstring'''

       import warnings

       # XXX: for now ignore scatter_gather warnings in this test since it's not relevant to what's being tested
       warnings.simplefilter(action='''ignore'''  ,   category=UpperCAmelCase		)

       lowercase       : List[Any]						=self.get_trainer(callbacks=[MyTestTrainerCallback]		)
       trainer.train()
       lowercase       : List[Any]						=trainer.callback_handler.callbacks[-2].events
       self.assertEqual(UpperCAmelCase  ,   self.get_expected_events(UpperCAmelCase		)		)

       # Independent log/save/eval
       lowercase       : int						=self.get_trainer(callbacks=[MyTestTrainerCallback]  ,   logging_steps=5		)
       trainer.train()
       lowercase       : List[str]						=trainer.callback_handler.callbacks[-2].events
       self.assertEqual(UpperCAmelCase  ,   self.get_expected_events(UpperCAmelCase		)		)

       lowercase       : Dict						=self.get_trainer(callbacks=[MyTestTrainerCallback]  ,   save_steps=5		)
       trainer.train()
       lowercase       : Union[str, Any]						=trainer.callback_handler.callbacks[-2].events
       self.assertEqual(UpperCAmelCase  ,   self.get_expected_events(UpperCAmelCase		)		)

       lowercase       : int						=self.get_trainer(callbacks=[MyTestTrainerCallback]  ,   eval_steps=5  ,   evaluation_strategy='''steps'''		)
       trainer.train()
       lowercase       : List[str]						=trainer.callback_handler.callbacks[-2].events
       self.assertEqual(UpperCAmelCase  ,   self.get_expected_events(UpperCAmelCase		)		)

       lowercase       : Optional[Any]						=self.get_trainer(callbacks=[MyTestTrainerCallback]  ,   evaluation_strategy='''epoch'''		)
       trainer.train()
       lowercase       : Optional[int]						=trainer.callback_handler.callbacks[-2].events
       self.assertEqual(UpperCAmelCase  ,   self.get_expected_events(UpperCAmelCase		)		)

       # A bit of everything
       lowercase       : Dict						=self.get_trainer(
           callbacks=[MyTestTrainerCallback]  ,   logging_steps=3  ,   save_steps=10  ,   eval_steps=5  ,   evaluation_strategy='''steps'''  ,   )
       trainer.train()
       lowercase       : Optional[Any]						=trainer.callback_handler.callbacks[-2].events
       self.assertEqual(UpperCAmelCase  ,   self.get_expected_events(UpperCAmelCase		)		)

       # warning should be emitted for duplicated callbacks
       with patch('''transformers.trainer_callback.logger.warning'''		) as warn_mock:
        lowercase       : Optional[Any]						=self.get_trainer(
            callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]  ,   )
        assert str(UpperCAmelCase		) in warn_mock.call_args[0][0]








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_      :      Optional[int]    =     False
@skip_mps
class    __UpperCamelCase			(       _lowercase	,	_lowercase	,	_lowercase	,	unittest.TestCase				):

 """simple docstring"""



 _lowercase   :    List[Any]      						=					StableDiffusionAttendAndExcitePipeline
 _lowercase   :    str      						=					False
 _lowercase   :    List[str]      						=					TEXT_TO_IMAGE_PARAMS
 _lowercase   :    Optional[int]      						=					TEXT_TO_IMAGE_BATCH_PARAMS.union({'''token_indices'''}				)
 _lowercase   :    Union[str, Any]      						=					TEXT_TO_IMAGE_IMAGE_PARAMS
 _lowercase   :    List[Any]      						=					TEXT_TO_IMAGE_IMAGE_PARAMS

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Dict:
  super().setUpClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Tuple:
  super().tearDownClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  torch.manual_seed(0      )
  a__												= UNetaDConditionModel(
      block_out_channels=(3_2, 6_4)		,				layers_per_block=1		,				sample_size=3_2		,				in_channels=4		,				out_channels=4		,				down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,				up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		,				cross_attention_dim=3_2		,				attention_head_dim=(2, 4)		,				use_linear_projection=SCREAMING_SNAKE_CASE		,				)
  a__												= DDIMScheduler(
      beta_start=0.0_00_85		,				beta_end=0.0_12		,				beta_schedule='''scaled_linear'''		,				clip_sample=SCREAMING_SNAKE_CASE		,				set_alpha_to_one=SCREAMING_SNAKE_CASE		,				)
  torch.manual_seed(0      )
  a__												= AutoencoderKL(
      block_out_channels=[3_2, 6_4]		,				in_channels=3		,				out_channels=3		,				down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']		,				up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']		,				latent_channels=4		,				sample_size=1_2_8		,				)
  torch.manual_seed(0      )
  a__												= CLIPTextConfig(
      bos_token_id=0		,				eos_token_id=2		,				hidden_size=3_2		,				intermediate_size=3_7		,				layer_norm_eps=1e-0_5		,				num_attention_heads=4		,				num_hidden_layers=5		,				pad_token_id=1		,				vocab_size=1_0_0_0		,				hidden_act='''gelu'''		,				projection_dim=5_1_2		,				)
  a__												= CLIPTextModel(SCREAMING_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 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=0      ) ->			Dict:
  if str(SCREAMING_SNAKE_CASE      ).startswith('''mps'''      ):
   a__												= torch.manual_seed(SCREAMING_SNAKE_CASE      )
  else:
   a__												= torch.Generator(device=SCREAMING_SNAKE_CASE      ).manual_seed(SCREAMING_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 						_UpperCAmelCase       (	self      ) ->			str:
  a__												= '''cpu'''

  a__												= self.get_dummy_components()
  a__												= self.pipeline_class(**SCREAMING_SNAKE_CASE      )
  pipe.to(SCREAMING_SNAKE_CASE      )
  pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE      )

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

  self.assertEqual(image.shape		,				(1, 6_4, 6_4, 3)      )
  a__												= np.array(
      [0.63_90_53_64, 0.62_89_73_07, 0.48_59_90_17, 0.5_13_36_24, 0.5_55_00_48, 0.45_76_95_16, 0.50_32_69_73, 0.5_02_31_39, 0.45_38_44_96]      )
  a__												= np.abs(image_slice.flatten() - expected_slice      ).max()
  self.assertLessEqual(SCREAMING_SNAKE_CASE		,				1e-3      )

 def 						_UpperCAmelCase       (	self      ) ->			str:
  super().test_cpu_offload_forward_pass(expected_max_diff=5e-4      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
  self._test_inference_batch_consistent(batch_sizes=[1, 2]      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  self._test_inference_batch_single_identical(batch_size=2		,				expected_max_diff=7e-4      )

 def 						_UpperCAmelCase       (	self      ) ->			int:
  super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3      )

 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4      )

 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  super().test_save_load_local(expected_max_difference=5e-4      )



 def 						_UpperCAmelCase       (	self      ) ->			str:
  super().test_save_load_optional_components(expected_max_difference=4e-4      )





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

 """simple docstring"""



 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Optional[int]:
  super().setUpClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			str:
  super().tearDownClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self      ) ->			List[str]:
  super().tearDown()
  gc.collect()
  torch.cuda.empty_cache()



 def 						_UpperCAmelCase       (	self      ) ->			str:
  a__												= torch.manual_seed(5_1      )

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

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

  a__												= pipe(
      prompt=SCREAMING_SNAKE_CASE		,				token_indices=SCREAMING_SNAKE_CASE		,				guidance_scale=7.5		,				generator=SCREAMING_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





