
import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPanoramaPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

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


enable_full_determinism()

@skip_mps
class    lowerCAmelCase__     (	__lowercase  ,	__lowercase  ,	unittest.TestCase			):
		a__       :						Any        =		StableDiffusionPanoramaPipeline
		a__       :						Any        =		TEXT_TO_IMAGE_PARAMS
		a__       :						List[str]        =		TEXT_TO_IMAGE_BATCH_PARAMS
		a__       :						Union[str, Any]        =		TEXT_TO_IMAGE_IMAGE_PARAMS
		a__       :						Dict        =		TEXT_TO_IMAGE_IMAGE_PARAMS






		def  __A       (		self		:							List[Any]  )     ->							str:
						torch.manual_seed(0  )
						__lowerCamelCase    					=	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							,			)
						__lowerCamelCase    					=	DDIMScheduler()
						torch.manual_seed(0  )
						__lowerCamelCase    					=	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							,			)
						torch.manual_seed(0  )
						__lowerCamelCase    					=	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=10_00							,			)
						__lowerCamelCase    					=	CLIPTextModel(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''  )

						__lowerCamelCase    					=	{
						    '''unet''': unet,
						    '''scheduler''': scheduler,
						    '''vae''': vae,
						    '''text_encoder''': text_encoder,
						    '''tokenizer''': tokenizer,
						    '''safety_checker''': None,
						    '''feature_extractor''': None,
						}
						return components






		def  __A       (		self		:							Optional[int]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							int=0  )     ->							int:
						__lowerCamelCase    					=	torch.manual_seed(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	{
						    '''prompt''': '''a photo of the dolomites''',
						    '''generator''': generator,
						    # Setting height and width to None to prevent OOMs on CPU.
						    '''height''': None,
						    '''width''': None,
						    '''num_inference_steps''': 1,
						    '''guidance_scale''': 6.0,
						    '''output_type''': '''numpy''',
						}
						return inputs






		def  __A       (		self		:							List[Any]  )     ->							List[Any]:
						__lowerCamelCase    					=	'''cpu'''  # ensure determinism for the device-dependent torch.Generator
						__lowerCamelCase    					=	self.get_dummy_components()
						__lowerCamelCase    					=	StableDiffusionPanoramaPipeline(**SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	sd_pipe.to(SCREAMING_SNAKE_CASE__  )
						sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	self.get_dummy_inputs(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	sd_pipe(**SCREAMING_SNAKE_CASE__  ).images
						__lowerCamelCase    					=	image[0, -3:, -3:, -1]
						assert image.shape == (1, 64, 64, 3)

						__lowerCamelCase    					=	np.array([0.6186, 0.5374, 0.4915, 0.4135, 0.4114, 0.4563, 0.5128, 0.4977, 0.4757]  )

						assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2






		def  __A       (		self		:							Optional[int]  )     ->							str:
						super().test_inference_batch_consistent(batch_sizes=[1, 2]  )






		def  __A       (		self		:							Tuple  )     ->							Union[str, Any]:
						super().test_inference_batch_single_identical(batch_size=2							,			expected_max_diff=3.25e-3  )






		def  __A       (		self		:							Optional[Any]  )     ->							List[str]:
						__lowerCamelCase    					=	'''cpu'''  # ensure determinism for the device-dependent torch.Generator
						__lowerCamelCase    					=	self.get_dummy_components()
						__lowerCamelCase    					=	StableDiffusionPanoramaPipeline(**SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	sd_pipe.to(SCREAMING_SNAKE_CASE__  )
						sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	self.get_dummy_inputs(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	'''french fries'''
						__lowerCamelCase    					=	sd_pipe(**SCREAMING_SNAKE_CASE__							,			negative_prompt=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	output.images
						__lowerCamelCase    					=	image[0, -3:, -3:, -1]

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

						__lowerCamelCase    					=	np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]  )

						assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2






		def  __A       (		self		:							Any  )     ->							Union[str, Any]:
						__lowerCamelCase    					=	'''cpu'''  # ensure determinism for the device-dependent torch.Generator
						__lowerCamelCase    					=	self.get_dummy_components()
						__lowerCamelCase    					=	StableDiffusionPanoramaPipeline(**SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	sd_pipe.to(SCREAMING_SNAKE_CASE__  )
						sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	self.get_dummy_inputs(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	sd_pipe(**SCREAMING_SNAKE_CASE__							,			view_batch_size=2  )
						__lowerCamelCase    					=	output.images
						__lowerCamelCase    					=	image[0, -3:, -3:, -1]

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

						__lowerCamelCase    					=	np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]  )

						assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2






		def  __A       (		self		:							Tuple  )     ->							List[Any]:
						__lowerCamelCase    					=	'''cpu'''  # ensure determinism for the device-dependent torch.Generator
						__lowerCamelCase    					=	self.get_dummy_components()
						__lowerCamelCase    					=	EulerAncestralDiscreteScheduler(
						    beta_start=0.00085							,			beta_end=0.012							,			beta_schedule='''scaled_linear'''  )
						__lowerCamelCase    					=	StableDiffusionPanoramaPipeline(**SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	sd_pipe.to(SCREAMING_SNAKE_CASE__  )
						sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	self.get_dummy_inputs(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	sd_pipe(**SCREAMING_SNAKE_CASE__  ).images
						__lowerCamelCase    					=	image[0, -3:, -3:, -1]

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

						__lowerCamelCase    					=	np.array([0.4024, 0.6510, 0.4901, 0.5378, 0.5813, 0.5622, 0.4795, 0.4467, 0.4952]  )

						assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2







		def  __A       (		self		:							List[Any]  )     ->							Any:
						__lowerCamelCase    					=	'''cpu'''  # ensure determinism for the device-dependent torch.Generator
						__lowerCamelCase    					=	self.get_dummy_components()
						__lowerCamelCase    					=	PNDMScheduler(
						    beta_start=0.00085							,			beta_end=0.012							,			beta_schedule='''scaled_linear'''							,			skip_prk_steps=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	StableDiffusionPanoramaPipeline(**SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	sd_pipe.to(SCREAMING_SNAKE_CASE__  )
						sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	self.get_dummy_inputs(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	sd_pipe(**SCREAMING_SNAKE_CASE__  ).images
						__lowerCamelCase    					=	image[0, -3:, -3:, -1]

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

						__lowerCamelCase    					=	np.array([0.6391, 0.6291, 0.4861, 0.5134, 0.5552, 0.4578, 0.5032, 0.5023, 0.4539]  )

						assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2





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






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






		def  __A       (		self		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							str=0  )     ->							List[str]:
						__lowerCamelCase    					=	torch.manual_seed(SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	{
						    '''prompt''': '''a photo of the dolomites''',
						    '''generator''': generator,
						    '''num_inference_steps''': 3,
						    '''guidance_scale''': 7.5,
						    '''output_type''': '''numpy''',
						}
						return inputs






		def  __A       (		self		:							Union[str, Any]  )     ->							Dict:
						__lowerCamelCase    					=	'''stabilityai/stable-diffusion-2-base'''
						__lowerCamelCase    					=	DDIMScheduler.from_pretrained(SCREAMING_SNAKE_CASE__							,			subfolder='''scheduler'''  )
						__lowerCamelCase    					=	StableDiffusionPanoramaPipeline.from_pretrained(SCREAMING_SNAKE_CASE__							,			scheduler=SCREAMING_SNAKE_CASE__							,			safety_checker=SCREAMING_SNAKE_CASE__  )
						pipe.to(SCREAMING_SNAKE_CASE__  )
						pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )
						pipe.enable_attention_slicing()

						__lowerCamelCase    					=	self.get_inputs()
						__lowerCamelCase    					=	pipe(**SCREAMING_SNAKE_CASE__  ).images
						__lowerCamelCase    					=	image[0, -3:, -3:, -1].flatten()

						assert image.shape == (1, 5_12, 20_48, 3)

						__lowerCamelCase    					=	np.array(
						    [
						        0.36968392,
						        0.27025372,
						        0.32446766,
						        0.28379387,
						        0.36363274,
						        0.30733347,
						        0.27100027,
						        0.27054125,
						        0.25536096,
						    ]  )

						assert np.abs(expected_slice - image_slice  ).max() < 1e-2






		def  __A       (		self		:							List[str]  )     ->							Tuple:
						__lowerCamelCase    					=	StableDiffusionPanoramaPipeline.from_pretrained(
						    '''stabilityai/stable-diffusion-2-base'''							,			safety_checker=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	LMSDiscreteScheduler.from_config(pipe.scheduler.config  )
						pipe.to(SCREAMING_SNAKE_CASE__  )
						pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )
						pipe.enable_attention_slicing()

						__lowerCamelCase    					=	self.get_inputs()
						__lowerCamelCase    					=	pipe(**SCREAMING_SNAKE_CASE__  ).images
						__lowerCamelCase    					=	image[0, -3:, -3:, -1].flatten()

						assert image.shape == (1, 5_12, 20_48, 3)

						__lowerCamelCase    					=	np.array(
						    [
						        [
						            0.0,
						            0.0,
						            0.0,
						            0.0,
						            0.0,
						            0.0,
						            0.0,
						            0.0,
						            0.0,
						        ]
						    ]  )

						assert np.abs(expected_slice - image_slice  ).max() < 1e-3






		def  __A       (		self		:							Tuple  )     ->							int:
						__lowerCamelCase    					=	0

						def callback_fn(SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							torch.FloatTensor  ) -> None:
										__lowerCamelCase    					=	True
										nonlocal number_of_steps
										number_of_steps += 1
										if step == 1:
														__lowerCamelCase    					=	latents.detach().cpu().numpy()
														assert latents.shape == (1, 4, 64, 2_56)
														__lowerCamelCase    					=	latents[0, -3:, -3:, -1]

														__lowerCamelCase    					=	np.array(
														    [
														        0.18681869,
														        0.33907816,
														        0.5361276,
														        0.14432865,
														        -0.02856611,
														        -0.73941123,
														        0.23397987,
														        0.47322682,
														        -0.37823164,
														    ]  )
														assert np.abs(latents_slice.flatten() - expected_slice  ).max() < 5e-2
										elif step == 2:
														__lowerCamelCase    					=	latents.detach().cpu().numpy()
														assert latents.shape == (1, 4, 64, 2_56)
														__lowerCamelCase    					=	latents[0, -3:, -3:, -1]

														__lowerCamelCase    					=	np.array(
														    [
														        0.18539645,
														        0.33987248,
														        0.5378559,
														        0.14437142,
														        -0.02455261,
														        -0.7338317,
														        0.23990755,
														        0.47356272,
														        -0.3786505,
														    ]  )

														assert np.abs(latents_slice.flatten() - expected_slice  ).max() < 5e-2

						__lowerCamelCase    					=	False

						__lowerCamelCase    					=	'''stabilityai/stable-diffusion-2-base'''
						__lowerCamelCase    					=	DDIMScheduler.from_pretrained(SCREAMING_SNAKE_CASE__							,			subfolder='''scheduler'''  )
						__lowerCamelCase    					=	StableDiffusionPanoramaPipeline.from_pretrained(SCREAMING_SNAKE_CASE__							,			scheduler=SCREAMING_SNAKE_CASE__							,			safety_checker=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	pipe.to(SCREAMING_SNAKE_CASE__  )
						pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )
						pipe.enable_attention_slicing()

						__lowerCamelCase    					=	self.get_inputs()
						pipe(**SCREAMING_SNAKE_CASE__							,			callback=SCREAMING_SNAKE_CASE__							,			callback_steps=1  )
						assert callback_fn.has_been_called
						assert number_of_steps == 3







		def  __A       (		self		:							Union[str, Any]  )     ->							List[Any]:
						torch.cuda.empty_cache()
						torch.cuda.reset_max_memory_allocated()
						torch.cuda.reset_peak_memory_stats()

						__lowerCamelCase    					=	'''stabilityai/stable-diffusion-2-base'''
						__lowerCamelCase    					=	DDIMScheduler.from_pretrained(SCREAMING_SNAKE_CASE__							,			subfolder='''scheduler'''  )
						__lowerCamelCase    					=	StableDiffusionPanoramaPipeline.from_pretrained(SCREAMING_SNAKE_CASE__							,			scheduler=SCREAMING_SNAKE_CASE__							,			safety_checker=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	pipe.to(SCREAMING_SNAKE_CASE__  )
						pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__  )
						pipe.enable_attention_slicing(1  )
						pipe.enable_sequential_cpu_offload()

						__lowerCamelCase    					=	self.get_inputs()
						__lowerCamelCase    					=	pipe(**SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	torch.cuda.max_memory_allocated()
						# make sure that less than 5.2 GB is allocated
						assert mem_bytes < 5.5 * 10**9





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






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






		def  __A       (		self		:							Any							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							str							,			**SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							Tuple:
						self.events.append('''on_init_end'''  )






		def  __A       (		self		:							Tuple							,			SCREAMING_SNAKE_CASE__		:							Optional[int]							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]							,			SCREAMING_SNAKE_CASE__		:							Optional[int]							,			**SCREAMING_SNAKE_CASE__		:							Optional[Any]  )     ->							Any:
						self.events.append('''on_train_begin'''  )






		def  __A       (		self		:							str							,			SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							Any							,			SCREAMING_SNAKE_CASE__		:							Dict							,			**SCREAMING_SNAKE_CASE__		:							int  )     ->							Any:
						self.events.append('''on_train_end'''  )






		def  __A       (		self		:							Union[str, Any]							,			SCREAMING_SNAKE_CASE__		:							Dict							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			**SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							int:
						self.events.append('''on_epoch_begin'''  )






		def  __A       (		self		:							Optional[Any]							,			SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							Any							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]							,			**SCREAMING_SNAKE_CASE__		:							Tuple  )     ->							Optional[Any]:
						self.events.append('''on_epoch_end'''  )






		def  __A       (		self		:							str							,			SCREAMING_SNAKE_CASE__		:							Tuple							,			SCREAMING_SNAKE_CASE__		:							Optional[Any]							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]							,			**SCREAMING_SNAKE_CASE__		:							str  )     ->							List[Any]:
						self.events.append('''on_step_begin'''  )






		def  __A       (		self		:							Any							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							Any							,			**SCREAMING_SNAKE_CASE__		:							Optional[int]  )     ->							Tuple:
						self.events.append('''on_step_end'''  )






		def  __A       (		self		:							int							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							int							,			**SCREAMING_SNAKE_CASE__		:							List[Any]  )     ->							Union[str, Any]:
						self.events.append('''on_evaluate'''  )






		def  __A       (		self		:							Union[str, Any]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							str							,			**SCREAMING_SNAKE_CASE__		:							str  )     ->							str:
						self.events.append('''on_predict'''  )






		def  __A       (		self		:							Optional[int]							,			SCREAMING_SNAKE_CASE__		:							int							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							Optional[int]							,			**SCREAMING_SNAKE_CASE__		:							Optional[Any]  )     ->							Tuple:
						self.events.append('''on_save'''  )






		def  __A       (		self		:							Optional[int]							,			SCREAMING_SNAKE_CASE__		:							Dict							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]							,			SCREAMING_SNAKE_CASE__		:							Union[str, Any]							,			**SCREAMING_SNAKE_CASE__		:							List[str]  )     ->							List[str]:
						self.events.append('''on_log'''  )







		def  __A       (		self		:							int							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							List[Any]							,			**SCREAMING_SNAKE_CASE__		:							Optional[int]  )     ->							str:
						self.events.append('''on_prediction_step'''  )





@require_torch
class    lowerCAmelCase__     (	unittest.TestCase			):






		def  __A       (		self		:							Union[str, Any]  )     ->							Optional[Any]:
						__lowerCamelCase    					=	tempfile.mkdtemp()






		def  __A       (		self		:							int  )     ->							List[str]:
						shutil.rmtree(self.output_dir  )






		def  __A       (		self		:							Tuple							,			SCREAMING_SNAKE_CASE__		:							List[Any]=0							,			SCREAMING_SNAKE_CASE__		:							Any=0							,			SCREAMING_SNAKE_CASE__		:							List[str]=64							,			SCREAMING_SNAKE_CASE__		:							Optional[int]=64							,			SCREAMING_SNAKE_CASE__		:							Optional[int]=None							,			SCREAMING_SNAKE_CASE__		:							Optional[int]=False							,			**SCREAMING_SNAKE_CASE__		:							Optional[Any]  )     ->							Union[str, Any]:
						# disable_tqdm in TrainingArguments has a flaky default since it depends on the level of logging. We make sure
						# its set to False since the tests later on depend on its value.
						__lowerCamelCase    					=	RegressionDataset(length=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	RegressionDataset(length=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	RegressionModelConfig(a=SCREAMING_SNAKE_CASE__							,			b=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	RegressionPreTrainedModel(SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	TrainingArguments(self.output_dir							,			disable_tqdm=SCREAMING_SNAKE_CASE__							,			report_to=[]							,			**SCREAMING_SNAKE_CASE__  )
						return Trainer(
						    SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__							,			train_dataset=SCREAMING_SNAKE_CASE__							,			eval_dataset=SCREAMING_SNAKE_CASE__							,			callbacks=SCREAMING_SNAKE_CASE__							,			)






		def  __A       (		self		:							List[Any]							,			SCREAMING_SNAKE_CASE__		:							List[str]							,			SCREAMING_SNAKE_CASE__		:							str  )     ->							Optional[int]:
						self.assertEqual(len(SCREAMING_SNAKE_CASE__  )							,			len(SCREAMING_SNAKE_CASE__  )  )

						# Order doesn't matter
						__lowerCamelCase    					=	sorted(SCREAMING_SNAKE_CASE__							,			key=lambda SCREAMING_SNAKE_CASE__  : cb.__name__ if isinstance(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  ) else cb.__class__.__name__  )
						__lowerCamelCase    					=	sorted(SCREAMING_SNAKE_CASE__							,			key=lambda SCREAMING_SNAKE_CASE__  : cb.__name__ if isinstance(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  ) else cb.__class__.__name__  )

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






		def  __A       (		self		:							Union[str, Any]							,			SCREAMING_SNAKE_CASE__		:							Optional[int]  )     ->							Tuple:
						__lowerCamelCase    					=	['''on_init_end''', '''on_train_begin''']
						__lowerCamelCase    					=	0
						__lowerCamelCase    					=	len(trainer.get_eval_dataloader()  )
						__lowerCamelCase    					=	['''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(SCREAMING_SNAKE_CASE__  ):
														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		:							Union[str, Any]  )     ->							int:
						__lowerCamelCase    					=	self.get_trainer()
						__lowerCamelCase    					=	DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )

						# Callbacks passed at init are added to the default callbacks
						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]  )
						expected_callbacks.append(SCREAMING_SNAKE_CASE__  )
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )

						# TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
						__lowerCamelCase    					=	self.get_trainer(disable_tqdm=SCREAMING_SNAKE_CASE__  )
						__lowerCamelCase    					=	DEFAULT_CALLBACKS.copy() + [PrinterCallback]
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )






		def  __A       (		self		:							List[Any]  )     ->							str:
						__lowerCamelCase    					=	DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						__lowerCamelCase    					=	self.get_trainer()

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

						__lowerCamelCase    					=	self.get_trainer()
						__lowerCamelCase    					=	trainer.pop_callback(SCREAMING_SNAKE_CASE__  )
						self.assertEqual(cb.__class__							,			SCREAMING_SNAKE_CASE__  )
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )

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

						# We can also add, pop, or remove by instance
						__lowerCamelCase    					=	self.get_trainer()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[0]
						trainer.remove_callback(SCREAMING_SNAKE_CASE__  )
						expected_callbacks.remove(SCREAMING_SNAKE_CASE__  )
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )

						__lowerCamelCase    					=	self.get_trainer()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[0]
						__lowerCamelCase    					=	trainer.pop_callback(SCREAMING_SNAKE_CASE__  )
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			SCREAMING_SNAKE_CASE__  )
						self.check_callbacks_equality(trainer.callback_handler.callbacks							,			SCREAMING_SNAKE_CASE__  )

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







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

						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

						# Independent log/save/eval
						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]							,			logging_steps=5  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]							,			save_steps=5  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]							,			eval_steps=5							,			evaluation_strategy='''steps'''  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

						__lowerCamelCase    					=	self.get_trainer(callbacks=[MyTestTrainerCallback]							,			evaluation_strategy='''epoch'''  )
						trainer.train()
						__lowerCamelCase    					=	trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE__							,			self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

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

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



