

"""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 							SCREAMING_SNAKE_CASE   (						a_			):






					"""simple docstring"""



					def __init__(			self     :							List[str]						):
						lowerCAmelCase__      :		str    							=      []







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







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







					def 			__lowerCAmelCase					(			self     :							List[str]  ,lowercase_     :							str  ,lowercase_     :							Dict  ,lowercase_     :							Optional[int]  ,**lowercase_     :							Optional[int]						):
						self.events.append('''on_train_end'''						)







					def 			__lowerCAmelCase					(			self     :							int  ,lowercase_     :							Dict  ,lowercase_     :							Dict  ,lowercase_     :							int  ,**lowercase_     :							Optional[int]						):
						self.events.append('''on_epoch_begin'''						)







					def 			__lowerCAmelCase					(			self     :							Optional[int]  ,lowercase_     :							Optional[Any]  ,lowercase_     :							Optional[int]  ,lowercase_     :							Dict  ,**lowercase_     :							Optional[int]						):
						self.events.append('''on_epoch_end'''						)







					def 			__lowerCAmelCase					(			self     :							Any  ,lowercase_     :							Tuple  ,lowercase_     :							Union[str, Any]  ,lowercase_     :							List[Any]  ,**lowercase_     :							List[Any]						):
						self.events.append('''on_step_begin'''						)







					def 			__lowerCAmelCase					(			self     :							Dict  ,lowercase_     :							str  ,lowercase_     :							Optional[int]  ,lowercase_     :							List[str]  ,**lowercase_     :							str						):
						self.events.append('''on_step_end'''						)







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







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







					def 			__lowerCAmelCase					(			self     :							Tuple  ,lowercase_     :							Any  ,lowercase_     :							str  ,lowercase_     :							Optional[Any]  ,**lowercase_     :							Any						):
						self.events.append('''on_save'''						)







					def 			__lowerCAmelCase					(			self     :							Optional[Any]  ,lowercase_     :							List[Any]  ,lowercase_     :							Optional[Any]  ,lowercase_     :							Optional[int]  ,**lowercase_     :							Any						):
						self.events.append('''on_log'''						)







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







@require_torch
class 							SCREAMING_SNAKE_CASE   (						unittest.TestCase			):






					"""simple docstring"""



					def 			__lowerCAmelCase					(			self     :							List[Any]						):
						lowerCAmelCase__      :		str    							=      tempfile.mkdtemp()







					def 			__lowerCAmelCase					(			self     :							List[str]						):
						shutil.rmtree(self.output_dir						)







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

						lowerCAmelCase__      :		str    							=      TrainingArguments(self.output_dir  ,disable_tqdm=lowercase_  ,report_to=[]  ,**lowercase_						)
						return Trainer(
						    lowercase_  ,lowercase_  ,train_dataset=lowercase_  ,eval_dataset=lowercase_  ,callbacks=lowercase_  ,)







					def 			__lowerCAmelCase					(			self     :							str  ,lowercase_     :							Optional[Any]  ,lowercase_     :							Optional[int]						):
						self.assertEqual(len(lowercase_						)  ,len(lowercase_						)						)

						# Order doesn't matter
						lowerCAmelCase__      :		int    							=      sorted(lowercase_  ,key=lambda lowercase_						: cb.__name__ if isinstance(lowercase_  ,lowercase_						) else cb.__class__.__name__						)
						lowerCAmelCase__      :		Optional[Any]    							=      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 			__lowerCAmelCase					(			self     :							str  ,lowercase_     :							Optional[Any]						):
						lowerCAmelCase__      :		str    							=      ['''on_init_end''', '''on_train_begin''']
						lowerCAmelCase__      :		Optional[Any]    							=      0
						lowerCAmelCase__      :		Tuple    							=      len(trainer.get_eval_dataloader()						)
						lowerCAmelCase__      :		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(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 			__lowerCAmelCase					(			self     :							Optional[int]						):
						lowerCAmelCase__      :		str    							=      self.get_trainer()
						lowerCAmelCase__      :		List[Any]    							=      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						self.check_callbacks_equality(trainer.callback_handler.callbacks  ,lowercase_						)

						# Callbacks passed at init are added to the default callbacks
						lowerCAmelCase__      :		Dict    							=      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
						lowerCAmelCase__      :		int    							=      self.get_trainer(disable_tqdm=lowercase_						)
						lowerCAmelCase__      :		Dict    							=      DEFAULT_CALLBACKS.copy() + [PrinterCallback]
						self.check_callbacks_equality(trainer.callback_handler.callbacks  ,lowercase_						)







					def 			__lowerCAmelCase					(			self     :							Tuple						):
						lowerCAmelCase__      :		Tuple    							=      DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						lowerCAmelCase__      :		List[str]    							=      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_						)

						lowerCAmelCase__      :		Optional[Any]    							=      self.get_trainer()
						lowerCAmelCase__      :		Tuple    							=      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
						lowerCAmelCase__      :		Optional[Any]    							=      self.get_trainer()
						lowerCAmelCase__      :		Union[str, Any]    							=      trainer.callback_handler.callbacks[0]
						trainer.remove_callback(lowercase_						)
						expected_callbacks.remove(lowercase_						)
						self.check_callbacks_equality(trainer.callback_handler.callbacks  ,lowercase_						)

						lowerCAmelCase__      :		Any    							=      self.get_trainer()
						lowerCAmelCase__      :		List[Any]    							=      trainer.callback_handler.callbacks[0]
						lowerCAmelCase__      :		Dict    							=      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 			__lowerCAmelCase					(			self     :							List[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=lowercase_						)

						lowerCAmelCase__      :		Union[str, Any]    							=      self.get_trainer(callbacks=[MyTestTrainerCallback]						)
						trainer.train()
						lowerCAmelCase__      :		Union[str, Any]    							=      trainer.callback_handler.callbacks[-2].events
						self.assertEqual(lowercase_  ,self.get_expected_events(lowercase_						)						)

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

						lowerCAmelCase__      :		Tuple    							=      self.get_trainer(callbacks=[MyTestTrainerCallback]  ,save_steps=5						)
						trainer.train()
						lowerCAmelCase__      :		Dict    							=      trainer.callback_handler.callbacks[-2].events
						self.assertEqual(lowercase_  ,self.get_expected_events(lowercase_						)						)

						lowerCAmelCase__      :		Dict    							=      self.get_trainer(callbacks=[MyTestTrainerCallback]  ,eval_steps=5  ,evaluation_strategy='''steps'''						)
						trainer.train()
						lowerCAmelCase__      :		Union[str, Any]    							=      trainer.callback_handler.callbacks[-2].events
						self.assertEqual(lowercase_  ,self.get_expected_events(lowercase_						)						)

						lowerCAmelCase__      :		Any    							=      self.get_trainer(callbacks=[MyTestTrainerCallback]  ,evaluation_strategy='''epoch'''						)
						trainer.train()
						lowerCAmelCase__      :		List[Any]    							=      trainer.callback_handler.callbacks[-2].events
						self.assertEqual(lowercase_  ,self.get_expected_events(lowercase_						)						)

						# A bit of everything
						lowerCAmelCase__      :		Optional[Any]    							=      self.get_trainer(
						    callbacks=[MyTestTrainerCallback]  ,logging_steps=3  ,save_steps=1_0  ,eval_steps=5  ,evaluation_strategy='''steps'''  ,)
						trainer.train()
						lowerCAmelCase__      :		Optional[Any]    							=      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:
							lowerCAmelCase__      :		Union[str, Any]    							=      self.get_trainer(
							    callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]  ,)
							assert str(lowercase_						) in warn_mock.call_args[0][0]







"""simple docstring"""





import warnings

from ...utils import logging
from .image_processing_imagegpt import ImageGPTImageProcessor


__UpperCamelCase    : List[str]				 =						logging.get_logger(__name__)







class 							SCREAMING_SNAKE_CASE   (						a_			):






					"""simple docstring"""



					def __init__(			self     :							Any  ,*lowercase_     :							Dict  ,**lowercase_     :							List[str]						):
						warnings.warn(
						    '''The class ImageGPTFeatureExtractor is deprecated and will be removed in version 5 of Transformers.'''
						    ''' Please use ImageGPTImageProcessor instead.'''  ,lowercase_  ,)
						super().__init__(*lowercase_  ,**lowercase_						)




