



'''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				)   ->		Any:
      lowerCAmelCase__			:  Any											= []




  def 							UpperCAmelCase_     (   self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase				)   ->		str:
      self.events.append("""on_init_end"""				)




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




  def 							UpperCAmelCase_     (   self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase				)   ->		str:
      self.events.append("""on_train_end"""				)




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




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




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




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




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




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




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




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






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







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




  def 							UpperCAmelCase_     (   self				)   ->		List[str]:
      lowerCAmelCase__			:  Optional[Any]											= tempfile.mkdtemp()




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




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

      lowerCAmelCase__			:  Dict											= 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				)   ->		Optional[int]:
      self.assertEqual(len(__UpperCAmelCase				)  ,len(__UpperCAmelCase				)				)

      # Order doesn't matter
      lowerCAmelCase__			:  Optional[Any]											= 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				)   ->		Optional[Any]:
      lowerCAmelCase__			:  List[str]											= ["""on_init_end""", """on_train_begin"""]
      lowerCAmelCase__			:  Optional[Any]											= 0
      lowerCAmelCase__			:  str											= len(trainer.get_eval_dataloader()				)
      lowerCAmelCase__			:  Union[str, 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(__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				)   ->		Tuple:
      lowerCAmelCase__			:  Tuple											= 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__			:  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__			:  List[Any]											= self.get_trainer(disable_tqdm=__UpperCAmelCase				)
      lowerCAmelCase__			:  Optional[int]											= DEFAULT_CALLBACKS.copy() + [PrinterCallback]
      self.check_callbacks_equality(trainer.callback_handler.callbacks  ,__UpperCAmelCase				)




  def 							UpperCAmelCase_     (   self				)   ->		Dict:
      lowerCAmelCase__			:  Union[str, Any]											= DEFAULT_CALLBACKS.copy() + [ProgressCallback]
      lowerCAmelCase__			:  Optional[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__			:  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__			:  Dict											= self.get_trainer()
      lowerCAmelCase__			:  Optional[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__			:  str											= trainer.callback_handler.callbacks[0]
      lowerCAmelCase__			:  int											= 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				)   ->		str:
      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__			:  Tuple											= self.get_trainer(callbacks=[MyTestTrainerCallback]				)
      trainer.train()
      lowerCAmelCase__			:  Optional[int]											= trainer.callback_handler.callbacks[-2].events
      self.assertEqual(__UpperCAmelCase  ,self.get_expected_events(__UpperCAmelCase				)				)

      # Independent log/save/eval
      lowerCAmelCase__			:  Optional[int]											= 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__			:  str											= self.get_trainer(callbacks=[MyTestTrainerCallback]  ,save_steps=5				)
      trainer.train()
      lowerCAmelCase__			:  Any											= trainer.callback_handler.callbacks[-2].events
      self.assertEqual(__UpperCAmelCase  ,self.get_expected_events(__UpperCAmelCase				)				)

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

      lowerCAmelCase__			:  List[str]											= self.get_trainer(callbacks=[MyTestTrainerCallback]  ,evaluation_strategy="""epoch"""				)
      trainer.train()
      lowerCAmelCase__			:  Union[str, 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__			:  List[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__			:  Tuple											= self.get_trainer(
              callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]  ,)
          assert str(__UpperCAmelCase				) in warn_mock.call_args[0][0]







'''simple docstring'''





class 	lowerCAmelCase_:
  '''simple docstring'''




  def __init__(   self  ,__UpperCAmelCase				)   ->		Tuple:
      lowerCAmelCase__			:  Union[str, Any]											= n
      lowerCAmelCase__			:  int											= [None] * self.n
      lowerCAmelCase__			:  Union[str, Any]											= 0  # index of the first element
      lowerCAmelCase__			:  Union[str, Any]											= 0
      lowerCAmelCase__			:  Union[str, Any]											= 0




  def __len__(   self				)   ->		int:
      return self.size




  def 							UpperCAmelCase_     (   self				)   ->		bool:
      return self.size == 0




  def 							UpperCAmelCase_     (   self				)   ->		Optional[Any]:
      return False if self.is_empty() else self.array[self.front]




  def 							UpperCAmelCase_     (   self  ,__UpperCAmelCase				)   ->		Union[str, Any]:
      if self.size >= self.n:
          raise Exception("""QUEUE IS FULL"""				)

      lowerCAmelCase__			:  str											= data
      lowerCAmelCase__			:  List[str]											= (self.rear + 1) % self.n
      self.size += 1
      return self






  def 							UpperCAmelCase_     (   self				)   ->		int:
      if self.size == 0:
          raise Exception("""UNDERFLOW"""				)

      lowerCAmelCase__			:  int											= self.array[self.front]
      lowerCAmelCase__			:  Optional[Any]											= None
      lowerCAmelCase__			:  int											= (self.front + 1) % self.n
      self.size -= 1
      return temp


