




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







						'''simple docstring'''





						def __init__(						self							:     Dict						):






							"""simple docstring"""






							snake_case_		    =     []






						def        snake_case__						(						self							:     List[str]       ,	__lowercase							:     str       ,	__lowercase							:     Dict       ,	__lowercase							:     List[str]       ,	**__lowercase							:     int						):






							"""simple docstring"""






							self.events.append("on_init_end"						)






						def        snake_case__						(						self							:     Tuple       ,	__lowercase							:     Optional[int]       ,	__lowercase							:     List[Any]       ,	__lowercase							:     Dict       ,	**__lowercase							:     Optional[Any]						):






							"""simple docstring"""






							self.events.append("on_train_begin"						)






						def        snake_case__						(						self							:     Dict       ,	__lowercase							:     str       ,	__lowercase							:     List[str]       ,	__lowercase							:     List[Any]       ,	**__lowercase							:     str						):






							"""simple docstring"""






							self.events.append("on_train_end"						)






						def        snake_case__						(						self							:     Any       ,	__lowercase							:     Dict       ,	__lowercase							:     Optional[Any]       ,	__lowercase							:     str       ,	**__lowercase							:     List[str]						):






							"""simple docstring"""






							self.events.append("on_epoch_begin"						)






						def        snake_case__						(						self							:     Dict       ,	__lowercase							:     Optional[Any]       ,	__lowercase							:     Any       ,	__lowercase							:     Dict       ,	**__lowercase							:     int						):






							"""simple docstring"""






							self.events.append("on_epoch_end"						)






						def        snake_case__						(						self							:     int       ,	__lowercase							:     Tuple       ,	__lowercase							:     List[Any]       ,	__lowercase							:     Tuple       ,	**__lowercase							:     Optional[Any]						):






							"""simple docstring"""






							self.events.append("on_step_begin"						)






						def        snake_case__						(						self							:     Dict       ,	__lowercase							:     Union[str, Any]       ,	__lowercase							:     Any       ,	__lowercase							:     Union[str, Any]       ,	**__lowercase							:     int						):






							"""simple docstring"""






							self.events.append("on_step_end"						)






						def        snake_case__						(						self							:     List[str]       ,	__lowercase							:     Any       ,	__lowercase							:     Optional[Any]       ,	__lowercase							:     Any       ,	**__lowercase							:     str						):






							"""simple docstring"""






							self.events.append("on_evaluate"						)






						def        snake_case__						(						self							:     int       ,	__lowercase							:     Dict       ,	__lowercase							:     Tuple       ,	__lowercase							:     Tuple       ,	**__lowercase							:     Union[str, Any]						):






							"""simple docstring"""






							self.events.append("on_predict"						)






						def        snake_case__						(						self							:     Optional[Any]       ,	__lowercase							:     str       ,	__lowercase							:     Tuple       ,	__lowercase							:     Any       ,	**__lowercase							:     Dict						):






							"""simple docstring"""






							self.events.append("on_save"						)






						def        snake_case__						(						self							:     Tuple       ,	__lowercase							:     Union[str, Any]       ,	__lowercase							:     Any       ,	__lowercase							:     Union[str, Any]       ,	**__lowercase							:     Dict						):






							"""simple docstring"""






							self.events.append("on_log"						)






						def        snake_case__						(						self							:     Optional[int]       ,	__lowercase							:     Tuple       ,	__lowercase							:     List[str]       ,	__lowercase							:     Optional[Any]       ,	**__lowercase							:     List[Any]						):






							"""simple docstring"""






							self.events.append("on_prediction_step"						)







@require_torch
class 		UpperCAmelCase   (				unittest.TestCase			):







						'''simple docstring'''





						def        snake_case__						(						self							:     Optional[int]						):






							"""simple docstring"""






							snake_case_		    =     tempfile.mkdtemp()






						def        snake_case__						(						self							:     Optional[Any]						):






							"""simple docstring"""






							shutil.rmtree(self.output_dir						)






						def        snake_case__						(						self							:     str       ,	__lowercase							:     Dict=0       ,	__lowercase							:     Any=0       ,	__lowercase							:     Tuple=64       ,	__lowercase							:     Tuple=64       ,	__lowercase							:     int=None       ,	__lowercase							:     List[str]=False       ,	**__lowercase							:     Optional[int]						):






							"""simple docstring"""






							snake_case_		    =     RegressionDataset(length=__lowercase						)
							snake_case_		    =     RegressionDataset(length=__lowercase						)
							snake_case_		    =     RegressionModelConfig(a=__lowercase       ,	b=__lowercase						)
							snake_case_		    =     RegressionPreTrainedModel(__lowercase						)

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






						def        snake_case__						(						self							:     Optional[int]       ,	__lowercase							:     List[str]       ,	__lowercase							:     str						):






							"""simple docstring"""






							self.assertEqual(len(__lowercase						)       ,	len(__lowercase						)						)

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

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






						def        snake_case__						(						self							:     str       ,	__lowercase							:     List[str]						):






							"""simple docstring"""






							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(__lowercase						):
									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        snake_case__						(						self							:     Optional[Any]						):






							"""simple docstring"""






							snake_case_		    =     self.get_trainer()
							snake_case_		    =     DEFAULT_CALLBACKS.copy() + [ProgressCallback]
							self.check_callbacks_equality(trainer.callback_handler.callbacks       ,	__lowercase						)

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

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






						def        snake_case__						(						self							:     str						):






							"""simple docstring"""






							snake_case_		    =     DEFAULT_CALLBACKS.copy() + [ProgressCallback]
							snake_case_		    =     self.get_trainer()

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

							snake_case_		    =     self.get_trainer()
							snake_case_		    =     trainer.pop_callback(__lowercase						)
							self.assertEqual(cb.__class__       ,	__lowercase						)
							self.check_callbacks_equality(trainer.callback_handler.callbacks       ,	__lowercase						)

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

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

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

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






						def        snake_case__						(						self							:     List[str]						):






							"""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=__lowercase						)

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

							# 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(__lowercase       ,	self.get_expected_events(__lowercase						)						)

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

							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(__lowercase       ,	self.get_expected_events(__lowercase						)						)

							snake_case_		    =     self.get_trainer(callbacks=[MyTestTrainerCallback]       ,	evaluation_strategy="epoch"						)
							trainer.train()
							snake_case_		    =     trainer.callback_handler.callbacks[-2].events
							self.assertEqual(__lowercase       ,	self.get_expected_events(__lowercase						)						)

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

							# 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(__lowercase						) in warn_mock.call_args[0][0]






from timeit import timeit

lowercase__				:     Union[str, Any] 		=						{
    "MALAYALAM": True,
    "String": False,
    "rotor": True,
    "level": True,
    "A": True,
    "BB": True,
    "ABC": False,
    "amanaplanacanalpanama": True,  # "a man a plan a canal panama"
}
# Ensure our test data is valid
assert all((key == key[::-1]) is value for key, value in test_data.items())


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	snake_case_		    =     0
	snake_case_		    =     len(_A							) - 1
	while start_i < end_i:
		if s[start_i] == s[end_i]:
			start_i += 1
			end_i -= 1
		else:
			return False
	return True


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	snake_case_		    =     len(_A							) // 2
	snake_case_		    =     len(_A							)

	# We need to traverse till half of the length of string
	# as we can get access of the i'th last element from
	# i'th index.
	# eg: [0,1,2,3,4,5] => 4th index can be accessed
	# with the help of 1st index (i==n-i-1)
	# where n is length of string
	return all(s[i] == s[n - i - 1] for i in range(_A							)							)


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	if len(_A							) <= 2:
		return True
	if s[0] == s[len(_A							) - 1]:
		return is_palindrome_recursive(s[1:-1]							)
	else:
		return False


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	return s == s[::-1]


def 						lowerCamelCase__    (		_A							):







	'''simple docstring'''





	snake_case_		    =     f"all({name}(key) is value for key, value in test_data.items())"
	snake_case_		    =     f"from __main__ import test_data, {name}"
	snake_case_		    =     500000
	snake_case_		    =     timeit(stmt=_A     ,	setup=_A     ,	number=_A							)
	print(f"{name:<35} finished {number:,} runs in {result:.5f} seconds"							)


if __name__ == "__main__":
		for key, value in test_data.items():
				assert is_palindrome(key) is is_palindrome_recursive(key)
				assert is_palindrome(key) is is_palindrome_slice(key)
				print(f'''{key:21} {value}''')
		print("a man a plan a canal panama")

		# finished 500,000 runs in 0.46793 seconds
		benchmark_function("is_palindrome_slice")
		# finished 500,000 runs in 0.85234 seconds
		benchmark_function("is_palindrome")
		# finished 500,000 runs in 1.32028 seconds
		benchmark_function("is_palindrome_recursive")
		# finished 500,000 runs in 2.08679 seconds
		benchmark_function("is_palindrome_traversal")
