


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__   (       snake_case__      ):







    """simple docstring"""

    def __init__(			self       ):
         snake_case											=							[]
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_init_end'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_train_begin'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_train_end'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_epoch_begin'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_epoch_end'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_step_begin'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_step_end'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_evaluate'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_predict'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_save'''       )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_log'''       )




    def 						a_     (			self     ,   __snake_case     ,   __snake_case     ,   __snake_case     ,   **__snake_case       ):
         self.events.append('''on_prediction_step'''       )






@require_torch
class     A__   (       unittest.TestCase      ):







    """simple docstring"""

    def 						a_     (			self       ):
         snake_case											=							tempfile.mkdtemp()
    def 						a_     (			self       ):
         shutil.rmtree(self.output_dir       )
    def 						a_     (			self     ,   __snake_case=0     ,   __snake_case=0     ,   __snake_case=6_4     ,   __snake_case=6_4     ,   __snake_case=None     ,   __snake_case=False     ,   **__snake_case       ):
         # 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.
         snake_case											=							RegressionDataset(length=__snake_case       )
         snake_case											=							RegressionDataset(length=__snake_case       )
         snake_case											=							RegressionModelConfig(a=__snake_case     ,   b=__snake_case       )
         snake_case											=							RegressionPreTrainedModel(__snake_case       )

         snake_case											=							TrainingArguments(self.output_dir     ,   disable_tqdm=__snake_case     ,   report_to=[]     ,   **__snake_case       )
         return Trainer(
             __snake_case     ,   __snake_case     ,   train_dataset=__snake_case     ,   eval_dataset=__snake_case     ,   callbacks=__snake_case     ,   )
    def 						a_     (			self     ,   __snake_case     ,   __snake_case       ):
         self.assertEqual(len(__snake_case       )     ,   len(__snake_case       )       )

         # Order doesn't matter
         snake_case											=							sorted(__snake_case     ,   key=lambda __snake_case       : cb.__name__ if isinstance(__snake_case     ,   __snake_case       ) else cb.__class__.__name__       )
         snake_case											=							sorted(__snake_case     ,   key=lambda __snake_case       : cb.__name__ if isinstance(__snake_case     ,   __snake_case       ) else cb.__class__.__name__       )

         for cba, cba in zip(__snake_case     ,   __snake_case       ):
              if isinstance(__snake_case     ,   __snake_case       ) and isinstance(__snake_case     ,   __snake_case       ):
                   self.assertEqual(__snake_case     ,   __snake_case       )
              elif isinstance(__snake_case     ,   __snake_case       ) and not isinstance(__snake_case     ,   __snake_case       ):
                   self.assertEqual(__snake_case     ,   cba.__class__       )
              elif not isinstance(__snake_case     ,   __snake_case       ) and isinstance(__snake_case     ,   __snake_case       ):
                   self.assertEqual(cba.__class__     ,   __snake_case       )
              else:
                   self.assertEqual(__snake_case     ,   __snake_case       )
    def 						a_     (			self     ,   __snake_case       ):
         snake_case											=							['''on_init_end''', '''on_train_begin''']
         snake_case											=							0
         snake_case											=							len(trainer.get_eval_dataloader()       )
         snake_case											=							['''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(__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 						a_     (			self       ):
         snake_case											=							self.get_trainer()
         snake_case											=							DEFAULT_CALLBACKS.copy() + [ProgressCallback]
         self.check_callbacks_equality(trainer.callback_handler.callbacks     ,   __snake_case       )

         # Callbacks passed at init are added to the default callbacks
         snake_case											=							self.get_trainer(callbacks=[MyTestTrainerCallback]       )
         expected_callbacks.append(__snake_case       )
         self.check_callbacks_equality(trainer.callback_handler.callbacks     ,   __snake_case       )

         # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
         snake_case											=							self.get_trainer(disable_tqdm=__snake_case       )
         snake_case											=							DEFAULT_CALLBACKS.copy() + [PrinterCallback]
         self.check_callbacks_equality(trainer.callback_handler.callbacks     ,   __snake_case       )
    def 						a_     (			self       ):
         snake_case											=							DEFAULT_CALLBACKS.copy() + [ProgressCallback]
         snake_case											=							self.get_trainer()

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

         snake_case											=							self.get_trainer()
         snake_case											=							trainer.pop_callback(__snake_case       )
         self.assertEqual(cb.__class__     ,   __snake_case       )
         self.check_callbacks_equality(trainer.callback_handler.callbacks     ,   __snake_case       )

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

         # We can also add, pop, or remove by instance
         snake_case											=							self.get_trainer()
         snake_case											=							trainer.callback_handler.callbacks[0]
         trainer.remove_callback(__snake_case       )
         expected_callbacks.remove(__snake_case       )
         self.check_callbacks_equality(trainer.callback_handler.callbacks     ,   __snake_case       )

         snake_case											=							self.get_trainer()
         snake_case											=							trainer.callback_handler.callbacks[0]
         snake_case											=							trainer.pop_callback(__snake_case       )
         self.assertEqual(__snake_case     ,   __snake_case       )
         self.check_callbacks_equality(trainer.callback_handler.callbacks     ,   __snake_case       )

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




    def 						a_     (			self       ):
         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=__snake_case       )

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

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

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

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

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

         # A bit of everything
         snake_case											=							self.get_trainer(
             callbacks=[MyTestTrainerCallback]     ,   logging_steps=3     ,   save_steps=1_0     ,   eval_steps=5     ,   evaluation_strategy='''steps'''     ,   )
         trainer.train()
         snake_case											=							trainer.callback_handler.callbacks[-2].events
         self.assertEqual(__snake_case     ,   self.get_expected_events(__snake_case       )       )

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








from ..utils import DummyObject, requires_backends

class     A__   (       metaclass=snake_case__      ):







    """simple docstring"""

    __magic_name__         =						['flax', 'transformers']
    def __init__(			self     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(self     ,   ['''flax''', '''transformers''']       )
    @classmethod
    def 						a_     (			cls     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(cls     ,   ['''flax''', '''transformers''']       )




    @classmethod
    def 						a_     (			cls     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(cls     ,   ['''flax''', '''transformers''']       )






class     A__   (       metaclass=snake_case__      ):







    """simple docstring"""

    __magic_name__         =						['flax', 'transformers']
    def __init__(			self     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(self     ,   ['''flax''', '''transformers''']       )
    @classmethod
    def 						a_     (			cls     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(cls     ,   ['''flax''', '''transformers''']       )




    @classmethod
    def 						a_     (			cls     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(cls     ,   ['''flax''', '''transformers''']       )






class     A__   (       metaclass=snake_case__      ):







    """simple docstring"""

    __magic_name__         =						['flax', 'transformers']
    def __init__(			self     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(self     ,   ['''flax''', '''transformers''']       )
    @classmethod
    def 						a_     (			cls     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(cls     ,   ['''flax''', '''transformers''']       )




    @classmethod
    def 						a_     (			cls     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(cls     ,   ['''flax''', '''transformers''']       )






class     A__   (       metaclass=snake_case__      ):







    """simple docstring"""

    __magic_name__         =						['flax', 'transformers']
    def __init__(			self     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(self     ,   ['''flax''', '''transformers''']       )
    @classmethod
    def 						a_     (			cls     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(cls     ,   ['''flax''', '''transformers''']       )




    @classmethod
    def 						a_     (			cls     ,   *__snake_case     ,   **__snake_case       ):
         requires_backends(cls     ,   ['''flax''', '''transformers''']       )




