




'''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 						a_				(			UpperCAmelCase__						):




	def __init__(       self     :  str      ):
							__snake_case						 =   []




	def   lowercase__   (       self     :  List[Any]       ,		__lowerCAmelCase     :  Dict       ,		__lowerCAmelCase     :  List[str]       ,		__lowerCAmelCase     :  List[str]       ,		**__lowerCAmelCase     :  Optional[Any]      ):
							self.events.append('on_init_end'      )




	def   lowercase__   (       self     :  Optional[int]       ,		__lowerCAmelCase     :  Any       ,		__lowerCAmelCase     :  Any       ,		__lowerCAmelCase     :  Tuple       ,		**__lowerCAmelCase     :  str      ):
							self.events.append('on_train_begin'      )




	def   lowercase__   (       self     :  List[str]       ,		__lowerCAmelCase     :  Union[str, Any]       ,		__lowerCAmelCase     :  List[str]       ,		__lowerCAmelCase     :  int       ,		**__lowerCAmelCase     :  List[str]      ):
							self.events.append('on_train_end'      )




	def   lowercase__   (       self     :  Tuple       ,		__lowerCAmelCase     :  str       ,		__lowerCAmelCase     :  Tuple       ,		__lowerCAmelCase     :  Dict       ,		**__lowerCAmelCase     :  str      ):
							self.events.append('on_epoch_begin'      )




	def   lowercase__   (       self     :  List[Any]       ,		__lowerCAmelCase     :  int       ,		__lowerCAmelCase     :  Tuple       ,		__lowerCAmelCase     :  Optional[Any]       ,		**__lowerCAmelCase     :  List[str]      ):
							self.events.append('on_epoch_end'      )




	def   lowercase__   (       self     :  List[Any]       ,		__lowerCAmelCase     :  Optional[Any]       ,		__lowerCAmelCase     :  List[str]       ,		__lowerCAmelCase     :  Optional[int]       ,		**__lowerCAmelCase     :  int      ):
							self.events.append('on_step_begin'      )




	def   lowercase__   (       self     :  Optional[int]       ,		__lowerCAmelCase     :  Tuple       ,		__lowerCAmelCase     :  List[Any]       ,		__lowerCAmelCase     :  Optional[int]       ,		**__lowerCAmelCase     :  Union[str, Any]      ):
							self.events.append('on_step_end'      )




	def   lowercase__   (       self     :  Union[str, Any]       ,		__lowerCAmelCase     :  int       ,		__lowerCAmelCase     :  Optional[int]       ,		__lowerCAmelCase     :  Optional[Any]       ,		**__lowerCAmelCase     :  Tuple      ):
							self.events.append('on_evaluate'      )




	def   lowercase__   (       self     :  Optional[Any]       ,		__lowerCAmelCase     :  str       ,		__lowerCAmelCase     :  List[str]       ,		__lowerCAmelCase     :  Union[str, Any]       ,		**__lowerCAmelCase     :  Any      ):
							self.events.append('on_predict'      )




	def   lowercase__   (       self     :  str       ,		__lowerCAmelCase     :  Dict       ,		__lowerCAmelCase     :  Optional[int]       ,		__lowerCAmelCase     :  int       ,		**__lowerCAmelCase     :  str      ):
							self.events.append('on_save'      )




	def   lowercase__   (       self     :  List[Any]       ,		__lowerCAmelCase     :  Tuple       ,		__lowerCAmelCase     :  List[str]       ,		__lowerCAmelCase     :  List[Any]       ,		**__lowerCAmelCase     :  Dict      ):
							self.events.append('on_log'      )




	def   lowercase__   (       self     :  int       ,		__lowerCAmelCase     :  str       ,		__lowerCAmelCase     :  Union[str, Any]       ,		__lowerCAmelCase     :  Tuple       ,		**__lowerCAmelCase     :  Dict      ):
							self.events.append('on_prediction_step'      )





@require_torch
class 						a_				(			unittest.TestCase						):




	def   lowercase__   (       self     :  Optional[Any]      ):
							__snake_case						 =   tempfile.mkdtemp()




	def   lowercase__   (       self     :  Union[str, Any]      ):
							shutil.rmtree(self.output_dir      )




	def   lowercase__   (       self     :  Dict       ,		__lowerCAmelCase     :  Tuple=0       ,		__lowerCAmelCase     :  Union[str, Any]=0       ,		__lowerCAmelCase     :  Dict=6_4       ,		__lowerCAmelCase     :  Any=6_4       ,		__lowerCAmelCase     :  Tuple=None       ,		__lowerCAmelCase     :  List[str]=False       ,		**__lowerCAmelCase     :  List[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.
							__snake_case						 =   RegressionDataset(length=__lowerCAmelCase      )
							__snake_case						 =   RegressionDataset(length=__lowerCAmelCase      )
							__snake_case						 =   RegressionModelConfig(a=__lowerCAmelCase       ,		b=__lowerCAmelCase      )
							__snake_case						 =   RegressionPreTrainedModel(__lowerCAmelCase      )

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




	def   lowercase__   (       self     :  str       ,		__lowerCAmelCase     :  str       ,		__lowerCAmelCase     :  int      ):
							self.assertEqual(len(__lowerCAmelCase      )       ,		len(__lowerCAmelCase      )      )

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

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




	def   lowercase__   (       self     :  Tuple       ,		__lowerCAmelCase     :  List[Any]      ):
							__snake_case						 =   ['on_init_end', 'on_train_begin']
							__snake_case						 =   0
							__snake_case						 =   len(trainer.get_eval_dataloader()      )
							__snake_case						 =   ['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(__lowerCAmelCase      ):
																			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   lowercase__   (       self     :  Optional[Any]      ):
							__snake_case						 =   self.get_trainer()
							__snake_case						 =   DEFAULT_CALLBACKS.copy() + [ProgressCallback]
							self.check_callbacks_equality(trainer.callback_handler.callbacks       ,		__lowerCAmelCase      )

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

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




	def   lowercase__   (       self     :  Dict      ):
							__snake_case						 =   DEFAULT_CALLBACKS.copy() + [ProgressCallback]
							__snake_case						 =   self.get_trainer()

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

							__snake_case						 =   self.get_trainer()
							__snake_case						 =   trainer.pop_callback(__lowerCAmelCase      )
							self.assertEqual(cb.__class__       ,		__lowerCAmelCase      )
							self.check_callbacks_equality(trainer.callback_handler.callbacks       ,		__lowerCAmelCase      )

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

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

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

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




	def   lowercase__   (       self     :  Optional[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=__lowerCAmelCase      )

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

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

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

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

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

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

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







'''simple docstring'''


def 						lowerCamelCase__   (				a							):
						if number < 0:
												raise ValueError('number must not be negative'							)
						return number & (number - 1) == 0


if __name__ == "__main__":
	import doctest

	doctest.testmod()

