



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

			def __init__(self					)       ->   Optional[Any]:
						UpperCAmelCase_:		Dict							      =     []

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   Dict:
						self.events.append("""on_init_end"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   Dict:
						self.events.append("""on_train_begin"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   Dict:
						self.events.append("""on_train_end"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   int:
						self.events.append("""on_epoch_begin"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   Dict:
						self.events.append("""on_epoch_end"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   Any:
						self.events.append("""on_step_begin"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   List[str]:
						self.events.append("""on_step_end"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   int:
						self.events.append("""on_evaluate"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   Union[str, Any]:
						self.events.append("""on_predict"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   str:
						self.events.append("""on_save"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   Tuple:
						self.events.append("""on_log"""					)

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_,   **SCREAMING_SNAKE_CASE_					)       ->   str:
						self.events.append("""on_prediction_step"""					)







@require_torch
class 				_a				(   unittest.TestCase					):

			def      __snake_case       (self					)       ->   Optional[int]:
						UpperCAmelCase_:		Optional[Any]							      =     tempfile.mkdtemp()

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

			def      __snake_case       (self,   SCREAMING_SNAKE_CASE_=0,   SCREAMING_SNAKE_CASE_=0,   SCREAMING_SNAKE_CASE_=64,   SCREAMING_SNAKE_CASE_=64,   SCREAMING_SNAKE_CASE_=None,   SCREAMING_SNAKE_CASE_=False,   **SCREAMING_SNAKE_CASE_					)       ->   Optional[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.
						UpperCAmelCase_:		str							      =     RegressionDataset(length=SCREAMING_SNAKE_CASE_					)
						UpperCAmelCase_:		Dict							      =     RegressionDataset(length=SCREAMING_SNAKE_CASE_					)
						UpperCAmelCase_:		Tuple							      =     RegressionModelConfig(a=SCREAMING_SNAKE_CASE_,   b=SCREAMING_SNAKE_CASE_					)
						UpperCAmelCase_:		Tuple							      =     RegressionPreTrainedModel(SCREAMING_SNAKE_CASE_					)

						UpperCAmelCase_:		Optional[Any]							      =     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      __snake_case       (self,   SCREAMING_SNAKE_CASE_,   SCREAMING_SNAKE_CASE_					)       ->   Optional[int]:
						self.assertEqual(len(SCREAMING_SNAKE_CASE_					),   len(SCREAMING_SNAKE_CASE_					)					)

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

			def      __snake_case       (self					)       ->   Optional[int]:
						UpperCAmelCase_:		Optional[int]							      =     DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						UpperCAmelCase_:		int							      =     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_					)

						UpperCAmelCase_:		List[Any]							      =     self.get_trainer()
						UpperCAmelCase_:		int							      =     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
						UpperCAmelCase_:		Optional[int]							      =     self.get_trainer()
						UpperCAmelCase_:		Optional[int]							      =     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_					)

						UpperCAmelCase_:		Optional[int]							      =     self.get_trainer()
						UpperCAmelCase_:		int							      =     trainer.callback_handler.callbacks[0]
						UpperCAmelCase_:		Tuple							      =     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      __snake_case       (self					)       ->   Optional[int]:
						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_					)

						UpperCAmelCase_:		int							      =     self.get_trainer(callbacks=[MyTestTrainerCallback]					)
						trainer.train()
						UpperCAmelCase_:		List[str]							      =     trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE_,   self.get_expected_events(SCREAMING_SNAKE_CASE_					)					)

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

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

						UpperCAmelCase_:		List[str]							      =     self.get_trainer(callbacks=[MyTestTrainerCallback],   eval_steps=5,   evaluation_strategy="""steps"""					)
						trainer.train()
						UpperCAmelCase_:		Union[str, Any]							      =     trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE_,   self.get_expected_events(SCREAMING_SNAKE_CASE_					)					)

						UpperCAmelCase_:		Any							      =     self.get_trainer(callbacks=[MyTestTrainerCallback],   evaluation_strategy="""epoch"""					)
						trainer.train()
						UpperCAmelCase_:		Dict							      =     trainer.callback_handler.callbacks[-2].events
						self.assertEqual(SCREAMING_SNAKE_CASE_,   self.get_expected_events(SCREAMING_SNAKE_CASE_					)					)

						# A bit of everything
						UpperCAmelCase_:		Optional[int]							      =     self.get_trainer(
						    callbacks=[MyTestTrainerCallback],   logging_steps=3,   save_steps=10,   eval_steps=5,   evaluation_strategy="""steps""",   )
						trainer.train()
						UpperCAmelCase_:		str							      =     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:
									UpperCAmelCase_:		Any							      =     self.get_trainer(
									    callbacks=[MyTestTrainerCallback, MyTestTrainerCallback],   )
									assert str(SCREAMING_SNAKE_CASE_					) in warn_mock.call_args[0][0]











import logging
from dataclasses import dataclass, field
from typing import Optional

from seqaseq_trainer import arg_to_scheduler

from transformers import TrainingArguments


a      :  Optional[int]			     =			logging.getLogger(__name__)







@dataclass
class 				_a				(   _lowerCAmelCase					):
			A      		=	field(
			    default=0.0 ,		metadata={'''help''': '''The label smoothing epsilon to apply (if not zero).'''}					)
			A      		=	field(default=_lowerCAmelCase ,		metadata={'''help''': '''Whether to SortishSamler or not.'''}					)
			A      		=	field(
			    default=_lowerCAmelCase ,		metadata={'''help''': '''Whether to use generate to calculate generative metrics (ROUGE, BLEU).'''}					)
			A      		=	field(default=_lowerCAmelCase ,		metadata={'''help''': '''whether to use adafactor'''}					)
			A      		=	field(
			    default=_lowerCAmelCase ,		metadata={'''help''': '''Encoder layer dropout probability. Goes into model.config.'''}					)
			A      		=	field(
			    default=_lowerCAmelCase ,		metadata={'''help''': '''Decoder layer dropout probability. Goes into model.config.'''}					)
			A      		=	field(default=_lowerCAmelCase ,		metadata={'''help''': '''Dropout probability. Goes into model.config.'''}					)
			A      		=	field(
			    default=_lowerCAmelCase ,		metadata={'''help''': '''Attention dropout probability. Goes into model.config.'''}					)
			A      		=	field(
			    default='''linear''' ,		metadata={'''help''': F'''Which lr scheduler to use. Selected in {sorted(arg_to_scheduler.keys()					)}'''} ,		)






