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







 """simple docstring"""


 def __init__( self		:   List[str]					)					->      int:




    '''simple docstring'''






    a__  					=		[]


 def 							_lowerCAmelCase					( self		:   List[str]       , _snake_case		:   Dict       , _snake_case		:   Tuple       , _snake_case		:   Optional[Any]       , **_snake_case		:   Any					)					->      Optional[int]:




    '''simple docstring'''






    self.events.append('on_init_end'					)


 def 							_lowerCAmelCase					( self		:   List[str]       , _snake_case		:   List[Any]       , _snake_case		:   Any       , _snake_case		:   str       , **_snake_case		:   Union[str, Any]					)					->      Union[str, Any]:




    '''simple docstring'''






    self.events.append('on_train_begin'					)


 def 							_lowerCAmelCase					( self		:   int       , _snake_case		:   Optional[int]       , _snake_case		:   Union[str, Any]       , _snake_case		:   Any       , **_snake_case		:   Optional[int]					)					->      Tuple:




    '''simple docstring'''






    self.events.append('on_train_end'					)


 def 							_lowerCAmelCase					( self		:   List[Any]       , _snake_case		:   int       , _snake_case		:   Optional[int]       , _snake_case		:   Dict       , **_snake_case		:   List[Any]					)					->      str:




    '''simple docstring'''






    self.events.append('on_epoch_begin'					)


 def 							_lowerCAmelCase					( self		:   List[str]       , _snake_case		:   Tuple       , _snake_case		:   Optional[Any]       , _snake_case		:   List[Any]       , **_snake_case		:   List[str]					)					->      Optional[int]:




    '''simple docstring'''






    self.events.append('on_epoch_end'					)


 def 							_lowerCAmelCase					( self		:   List[str]       , _snake_case		:   List[Any]       , _snake_case		:   Union[str, Any]       , _snake_case		:   Optional[int]       , **_snake_case		:   Union[str, Any]					)					->      List[Any]:




    '''simple docstring'''






    self.events.append('on_step_begin'					)


 def 							_lowerCAmelCase					( self		:   str       , _snake_case		:   str       , _snake_case		:   Tuple       , _snake_case		:   Dict       , **_snake_case		:   Union[str, Any]					)					->      int:




    '''simple docstring'''






    self.events.append('on_step_end'					)


 def 							_lowerCAmelCase					( self		:   Any       , _snake_case		:   Union[str, Any]       , _snake_case		:   str       , _snake_case		:   Optional[int]       , **_snake_case		:   Optional[Any]					)					->      List[Any]:




    '''simple docstring'''






    self.events.append('on_evaluate'					)


 def 							_lowerCAmelCase					( self		:   Optional[int]       , _snake_case		:   Dict       , _snake_case		:   Dict       , _snake_case		:   Optional[Any]       , **_snake_case		:   Dict					)					->      Dict:




    '''simple docstring'''






    self.events.append('on_predict'					)


 def 							_lowerCAmelCase					( self		:   Tuple       , _snake_case		:   Tuple       , _snake_case		:   List[Any]       , _snake_case		:   Tuple       , **_snake_case		:   Any					)					->      Optional[int]:




    '''simple docstring'''






    self.events.append('on_save'					)


 def 							_lowerCAmelCase					( self		:   Union[str, Any]       , _snake_case		:   Optional[Any]       , _snake_case		:   List[Any]       , _snake_case		:   Tuple       , **_snake_case		:   Union[str, Any]					)					->      Tuple:




    '''simple docstring'''






    self.events.append('on_log'					)



 def 							_lowerCAmelCase					( self		:   Any       , _snake_case		:   List[Any]       , _snake_case		:   List[str]       , _snake_case		:   str       , **_snake_case		:   int					)					->      Union[str, Any]:




    '''simple docstring'''






    self.events.append('on_prediction_step'					)

@require_torch
class SCREAMING_SNAKE_CASE  (     unittest.TestCase						):







 """simple docstring"""


 def 							_lowerCAmelCase					( self		:   Union[str, Any]					)					->      Optional[int]:




    '''simple docstring'''






    a__  					=		tempfile.mkdtemp()


 def 							_lowerCAmelCase					( self		:   Tuple					)					->      List[str]:




    '''simple docstring'''






    shutil.rmtree(self.output_dir					)


 def 							_lowerCAmelCase					( self		:   int       , _snake_case		:   int=0       , _snake_case		:   int=0       , _snake_case		:   Any=64       , _snake_case		:   Dict=64       , _snake_case		:   Any=None       , _snake_case		:   List[Any]=False       , **_snake_case		:   Union[str, Any]					)					->      Tuple:




    '''simple docstring'''






    a__  					=		RegressionDataset(length=SCREAMING_SNAKE_CASE_					)
    a__  					=		RegressionDataset(length=SCREAMING_SNAKE_CASE_					)
    a__  					=		RegressionModelConfig(a=SCREAMING_SNAKE_CASE_       , b=SCREAMING_SNAKE_CASE_					)
    a__  					=		RegressionPreTrainedModel(SCREAMING_SNAKE_CASE_					)

    a__  					=		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 							_lowerCAmelCase					( self		:   List[Any]       , _snake_case		:   Tuple       , _snake_case		:   Union[str, Any]					)					->      List[str]:




    '''simple docstring'''






    self.assertEqual(len(SCREAMING_SNAKE_CASE_					)       , len(SCREAMING_SNAKE_CASE_					)					)

    # Order doesn't matter
    a__  					=		sorted(SCREAMING_SNAKE_CASE_       , key=lambda _snake_case					: cb.__name__ if isinstance(SCREAMING_SNAKE_CASE_       , SCREAMING_SNAKE_CASE_					) else cb.__class__.__name__					)
    a__  					=		sorted(SCREAMING_SNAKE_CASE_       , key=lambda _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 							_lowerCAmelCase					( self		:   Optional[int]       , _snake_case		:   str					)					->      Any:




    '''simple docstring'''






    a__  					=		["""on_init_end""", """on_train_begin"""]
    a__  					=		0
    a__  					=		len(trainer.get_eval_dataloader()					)
    a__  					=		["""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 							_lowerCAmelCase					( self		:   List[str]					)					->      Tuple:




    '''simple docstring'''






    a__  					=		self.get_trainer()
    a__  					=		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
    a__  					=		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
    a__  					=		self.get_trainer(disable_tqdm=SCREAMING_SNAKE_CASE_					)
    a__  					=		DEFAULT_CALLBACKS.copy() + [PrinterCallback]
    self.check_callbacks_equality(trainer.callback_handler.callbacks       , SCREAMING_SNAKE_CASE_					)


 def 							_lowerCAmelCase					( self		:   Dict					)					->      Any:




    '''simple docstring'''






    a__  					=		DEFAULT_CALLBACKS.copy() + [ProgressCallback]
    a__  					=		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_					)

    a__  					=		self.get_trainer()
    a__  					=		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
    a__  					=		self.get_trainer()
    a__  					=		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_					)

    a__  					=		self.get_trainer()
    a__  					=		trainer.callback_handler.callbacks[0]
    a__  					=		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 							_lowerCAmelCase					( self		:   Optional[Any]					)					->      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=SCREAMING_SNAKE_CASE_					)

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

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

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

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

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

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




def 		UpperCamelCase						(   lowercase_ )		->						float:


  '''simple docstring'''






  if not nums:  # Makes sure that the list is not empty
    raise ValueError("""List is empty""" )

  lowercase__	: int     		=   sum(lowercase_ ) / len(lowercase_ )  # Calculate the average
  return sum(abs(x - average ) for x in nums ) / len(lowercase_ )


if __name__ == "__main__":
     import doctest

     doctest.testmod()






