



'''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_(				SCREAMING_SNAKE_CASE_     ):
  '''simple docstring'''




  def __init__(    self      )    ->		Union[str, Any]:
      lowerCAmelCase__ :  Union[str, Any]							      =					[]




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Optional[int]:
      self.events.append("""on_init_end"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		int:
      self.events.append("""on_train_begin"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Optional[Any]:
      self.events.append("""on_train_end"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Dict:
      self.events.append("""on_epoch_begin"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Dict:
      self.events.append("""on_epoch_end"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Optional[int]:
      self.events.append("""on_step_begin"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Union[str, Any]:
      self.events.append("""on_step_end"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Optional[int]:
      self.events.append("""on_evaluate"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		int:
      self.events.append("""on_predict"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Any:
      self.events.append("""on_save"""      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Any:
      self.events.append("""on_log"""      )






  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase      )    ->		Optional[int]:
      self.events.append("""on_prediction_step"""      )







@require_torch
class        lowerCAmelCase_(				unittest.TestCase     ):
  '''simple docstring'''




  def 							UpperCAmelCase_	(    self      )    ->		Dict:
      lowerCAmelCase__ :  int							      =					tempfile.mkdtemp()




  def 							UpperCAmelCase_	(    self      )    ->		Tuple:
      shutil.rmtree(self.output_dir      )




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase=0  ,__UpperCAmelCase=0  ,__UpperCAmelCase=64  ,__UpperCAmelCase=64  ,__UpperCAmelCase=None  ,__UpperCAmelCase=False  ,**__UpperCAmelCase      )    ->		Optional[int]:
      # 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=__UpperCAmelCase      )
      lowerCAmelCase__ :  Tuple							      =					RegressionDataset(length=__UpperCAmelCase      )
      lowerCAmelCase__ :  Union[str, Any]							      =					RegressionModelConfig(a=__UpperCAmelCase  ,b=__UpperCAmelCase      )
      lowerCAmelCase__ :  Union[str, Any]							      =					RegressionPreTrainedModel(__UpperCAmelCase      )

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




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase  ,__UpperCAmelCase      )    ->		Dict:
      self.assertEqual(len(__UpperCAmelCase      )  ,len(__UpperCAmelCase      )      )

      # Order doesn't matter
      lowerCAmelCase__ :  Optional[int]							      =					sorted(__UpperCAmelCase  ,key=lambda __UpperCAmelCase      : cb.__name__ if isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ) else cb.__class__.__name__      )
      lowerCAmelCase__ :  Optional[int]							      =					sorted(__UpperCAmelCase  ,key=lambda __UpperCAmelCase      : cb.__name__ if isinstance(__UpperCAmelCase  ,__UpperCAmelCase      ) else cb.__class__.__name__      )

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




  def 							UpperCAmelCase_	(    self  ,__UpperCAmelCase      )    ->		Union[str, Any]:
      lowerCAmelCase__ :  Any							      =					["""on_init_end""", """on_train_begin"""]
      lowerCAmelCase__ :  Union[str, Any]							      =					0
      lowerCAmelCase__ :  Tuple							      =					len(trainer.get_eval_dataloader()      )
      lowerCAmelCase__ :  Optional[int]							      =					["""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(__UpperCAmelCase      ):
              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 							UpperCAmelCase_	(    self      )    ->		Dict:
      lowerCAmelCase__ :  Any							      =					self.get_trainer()
      lowerCAmelCase__ :  Tuple							      =					DEFAULT_CALLBACKS.copy() + [ProgressCallback]
      self.check_callbacks_equality(trainer.callback_handler.callbacks  ,__UpperCAmelCase      )

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

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




  def 							UpperCAmelCase_	(    self      )    ->		int:
      lowerCAmelCase__ :  Optional[int]							      =					DEFAULT_CALLBACKS.copy() + [ProgressCallback]
      lowerCAmelCase__ :  List[Any]							      =					self.get_trainer()

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

      lowerCAmelCase__ :  Optional[int]							      =					self.get_trainer()
      lowerCAmelCase__ :  Optional[int]							      =					trainer.pop_callback(__UpperCAmelCase      )
      self.assertEqual(cb.__class__  ,__UpperCAmelCase      )
      self.check_callbacks_equality(trainer.callback_handler.callbacks  ,__UpperCAmelCase      )

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

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

      lowerCAmelCase__ :  List[Any]							      =					self.get_trainer()
      lowerCAmelCase__ :  Dict							      =					trainer.callback_handler.callbacks[0]
      lowerCAmelCase__ :  List[Any]							      =					trainer.pop_callback(__UpperCAmelCase      )
      self.assertEqual(__UpperCAmelCase  ,__UpperCAmelCase      )
      self.check_callbacks_equality(trainer.callback_handler.callbacks  ,__UpperCAmelCase      )

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






  def 							UpperCAmelCase_	(    self      )    ->		Optional[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=__UpperCAmelCase      )

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

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

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

      lowerCAmelCase__ :  Tuple							      =					self.get_trainer(callbacks=[MyTestTrainerCallback]  ,eval_steps=5  ,evaluation_strategy="""steps"""      )
      trainer.train()
      lowerCAmelCase__ :  Tuple							      =					trainer.callback_handler.callbacks[-2].events
      self.assertEqual(__UpperCAmelCase  ,self.get_expected_events(__UpperCAmelCase      )      )

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

      # A bit of everything
      lowerCAmelCase__ :  List[str]							      =					self.get_trainer(
          callbacks=[MyTestTrainerCallback]  ,logging_steps=3  ,save_steps=10  ,eval_steps=5  ,evaluation_strategy="""steps"""  ,)
      trainer.train()
      lowerCAmelCase__ :  Optional[Any]							      =					trainer.callback_handler.callbacks[-2].events
      self.assertEqual(__UpperCAmelCase  ,self.get_expected_events(__UpperCAmelCase      )      )

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










from functools import lru_cache






@lru_cache
def 				lowerCamelCase__      (    __lowerCamelCase			:     int		):
  if num < 0:
    raise ValueError("""Number should not be negative."""		)

  return 1 if num in (0, 1) else num * factorial(num - 1		)


if __name__ == "__main__":
       import doctest

       doctest.testmod()


