


'''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 _lowerCamelCase	(						lowercase__							):



		'''simple docstring'''







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







		def        __lowerCAmelCase		( self		:						Tuple  ,					_A		:						str  ,					_A		:						int  ,					_A		:						Any  ,					**_A		:						Optional[int]      )  ->			Optional[Any]:
			self.events.append('on_init_end'      )







		def        __lowerCAmelCase		( self		:						Any  ,					_A		:						Tuple  ,					_A		:						List[str]  ,					_A		:						List[Any]  ,					**_A		:						List[str]      )  ->			Any:
			self.events.append('on_train_begin'      )







		def        __lowerCAmelCase		( self		:						List[str]  ,					_A		:						Optional[Any]  ,					_A		:						List[str]  ,					_A		:						str  ,					**_A		:						Union[str, Any]      )  ->			List[Any]:
			self.events.append('on_train_end'      )







		def        __lowerCAmelCase		( self		:						Tuple  ,					_A		:						Tuple  ,					_A		:						Union[str, Any]  ,					_A		:						Tuple  ,					**_A		:						List[Any]      )  ->			Dict:
			self.events.append('on_epoch_begin'      )







		def        __lowerCAmelCase		( self		:						Union[str, Any]  ,					_A		:						List[str]  ,					_A		:						int  ,					_A		:						List[Any]  ,					**_A		:						str      )  ->			int:
			self.events.append('on_epoch_end'      )







		def        __lowerCAmelCase		( self		:						Union[str, Any]  ,					_A		:						Tuple  ,					_A		:						Tuple  ,					_A		:						int  ,					**_A		:						Optional[Any]      )  ->			Any:
			self.events.append('on_step_begin'      )







		def        __lowerCAmelCase		( self		:						Optional[Any]  ,					_A		:						int  ,					_A		:						Dict  ,					_A		:						Union[str, Any]  ,					**_A		:						Union[str, Any]      )  ->			List[str]:
			self.events.append('on_step_end'      )







		def        __lowerCAmelCase		( self		:						Dict  ,					_A		:						Any  ,					_A		:						Optional[Any]  ,					_A		:						Optional[Any]  ,					**_A		:						List[str]      )  ->			Tuple:
			self.events.append('on_evaluate'      )







		def        __lowerCAmelCase		( self		:						Optional[Any]  ,					_A		:						Tuple  ,					_A		:						Union[str, Any]  ,					_A		:						Optional[int]  ,					**_A		:						Union[str, Any]      )  ->			Tuple:
			self.events.append('on_predict'      )







		def        __lowerCAmelCase		( self		:						Dict  ,					_A		:						List[str]  ,					_A		:						Any  ,					_A		:						Tuple  ,					**_A		:						List[str]      )  ->			List[str]:
			self.events.append('on_save'      )







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







		def        __lowerCAmelCase		( self		:						int  ,					_A		:						Dict  ,					_A		:						str  ,					_A		:						List[Any]  ,					**_A		:						int      )  ->			str:
			self.events.append('on_prediction_step'      )







@require_torch
class _lowerCamelCase	(						unittest.TestCase							):



		'''simple docstring'''







		def        __lowerCAmelCase		( self		:						int      )  ->			Any:
			__magic_name__	:   Union[str, Any]					     =					tempfile.mkdtemp()







		def        __lowerCAmelCase		( self		:						str      )  ->			Any:
			shutil.rmtree(self.output_dir      )







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

			__magic_name__	:   Optional[Any]					     =					TrainingArguments(self.output_dir  ,					disable_tqdm=_A  ,					report_to=[]  ,					**_A      )
			return Trainer(
			    _A  ,					_A  ,					train_dataset=_A  ,					eval_dataset=_A  ,					callbacks=_A  ,					)







		def        __lowerCAmelCase		( self		:						Union[str, Any]  ,					_A		:						Any  ,					_A		:						List[Any]      )  ->			Optional[int]:
			self.assertEqual(len(_A      )  ,					len(_A      )      )

			# Order doesn't matter
			__magic_name__	:   Dict					     =					sorted(_A  ,					key=lambda _A      : cb.__name__ if isinstance(_A  ,					_A      ) else cb.__class__.__name__      )
			__magic_name__	:   Union[str, Any]					     =					sorted(_A  ,					key=lambda _A      : cb.__name__ if isinstance(_A  ,					_A      ) else cb.__class__.__name__      )

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







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

			# Callbacks passed at init are added to the default callbacks
			__magic_name__	:   Optional[int]					     =					self.get_trainer(callbacks=[MyTestTrainerCallback]      )
			expected_callbacks.append(_A      )
			self.check_callbacks_equality(trainer.callback_handler.callbacks  ,					_A      )

			# TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
			__magic_name__	:   Optional[int]					     =					self.get_trainer(disable_tqdm=_A      )
			__magic_name__	:   Optional[int]					     =					DEFAULT_CALLBACKS.copy() + [PrinterCallback]
			self.check_callbacks_equality(trainer.callback_handler.callbacks  ,					_A      )







		def        __lowerCAmelCase		( self		:						List[Any]      )  ->			str:
			__magic_name__	:   Optional[Any]					     =					DEFAULT_CALLBACKS.copy() + [ProgressCallback]
			__magic_name__	:   List[str]					     =					self.get_trainer()

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

			__magic_name__	:   List[Any]					     =					self.get_trainer()
			__magic_name__	:   Any					     =					trainer.pop_callback(_A      )
			self.assertEqual(cb.__class__  ,					_A      )
			self.check_callbacks_equality(trainer.callback_handler.callbacks  ,					_A      )

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

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

			__magic_name__	:   Dict					     =					self.get_trainer()
			__magic_name__	:   Dict					     =					trainer.callback_handler.callbacks[0]
			__magic_name__	:   Dict					     =					trainer.pop_callback(_A      )
			self.assertEqual(_A  ,					_A      )
			self.check_callbacks_equality(trainer.callback_handler.callbacks  ,					_A      )

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







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

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

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

			__magic_name__	:   List[Any]					     =					self.get_trainer(callbacks=[MyTestTrainerCallback]  ,					save_steps=5      )
			trainer.train()
			__magic_name__	:   List[str]					     =					trainer.callback_handler.callbacks[-2].events
			self.assertEqual(_A  ,					self.get_expected_events(_A      )      )

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

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

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

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



'''simple docstring'''

from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


lowerCAmelCase				:str        =      {
    '''configuration_vision_encoder_decoder''': ['''VisionEncoderDecoderConfig''', '''VisionEncoderDecoderOnnxConfig''']
}

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							lowerCAmelCase				:List[str]        =      ['''VisionEncoderDecoderModel''']

try:
							if not is_tf_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							lowerCAmelCase				:List[str]        =      ['''TFVisionEncoderDecoderModel''']

try:
							if not is_flax_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							lowerCAmelCase				:Optional[Any]        =      ['''FlaxVisionEncoderDecoderModel''']

if TYPE_CHECKING:
							from .configuration_vision_encoder_decoder import VisionEncoderDecoderConfig, VisionEncoderDecoderOnnxConfig

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_vision_encoder_decoder import VisionEncoderDecoderModel

							try:
														if not is_tf_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_tf_vision_encoder_decoder import TFVisionEncoderDecoderModel

							try:
														if not is_flax_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .modeling_flax_vision_encoder_decoder import FlaxVisionEncoderDecoderModel

else:
							import sys

							lowerCAmelCase				:Tuple        =      _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)