






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

       '''simple docstring'''



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






            '''simple docstring'''





            __snake_case			:Any          =						[]



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      Any:






            '''simple docstring'''





            self.events.append("""on_init_end"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      str:






            '''simple docstring'''





            self.events.append("""on_train_begin"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      Dict:






            '''simple docstring'''





            self.events.append("""on_train_end"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      Any:






            '''simple docstring'''





            self.events.append("""on_epoch_begin"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      Dict:






            '''simple docstring'''





            self.events.append("""on_epoch_end"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      Optional[Any]:






            '''simple docstring'''





            self.events.append("""on_step_begin"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      Tuple:






            '''simple docstring'''





            self.events.append("""on_step_end"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      Any:






            '''simple docstring'''





            self.events.append("""on_evaluate"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      List[str]:






            '''simple docstring'''





            self.events.append("""on_predict"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      int:






            '''simple docstring'''





            self.events.append("""on_save"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      Optional[int]:






            '''simple docstring'''





            self.events.append("""on_log"""    )



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						**a__    ) ->      Dict:






            '''simple docstring'''





            self.events.append("""on_prediction_step"""    )
@require_torch
class 			snake_case__   (							unittest.TestCase):

       '''simple docstring'''



       def 					__lowercase						(							self    ) ->      Union[str, Any]:






            '''simple docstring'''





            __snake_case			:int          =						tempfile.mkdtemp()



       def 					__lowercase						(							self    ) ->      Tuple:






            '''simple docstring'''





            shutil.rmtree(self.output_dir    )



       def 					__lowercase						(							self			,						a__=0			,						a__=0			,						a__=64			,						a__=64			,						a__=None			,						a__=False			,						**a__    ) ->      Optional[Any]:






            '''simple docstring'''





            __snake_case			:List[Any]          =						RegressionDataset(length=a__    )
            __snake_case			:int          =						RegressionDataset(length=a__    )
            __snake_case			:str          =						RegressionModelConfig(a=a__			,						b=a__    )
            __snake_case			:str          =						RegressionPreTrainedModel(a__    )

            __snake_case			:Any          =						TrainingArguments(self.output_dir			,						disable_tqdm=a__			,						report_to=[]			,						**a__    )
            return Trainer(
                a__			,						a__			,						train_dataset=a__			,						eval_dataset=a__			,						callbacks=a__			,						)



       def 					__lowercase						(							self			,						a__			,						a__    ) ->      Dict:






            '''simple docstring'''





            self.assertEqual(len(a__    )			,						len(a__    )    )

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

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



       def 					__lowercase						(							self			,						a__    ) ->      List[str]:






            '''simple docstring'''





            __snake_case			:Union[str, Any]          =						["""on_init_end""", """on_train_begin"""]
            __snake_case			:Optional[int]          =						0
            __snake_case			:Dict          =						len(trainer.get_eval_dataloader()    )
            __snake_case			: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(a__    ):
                      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 					__lowercase						(							self    ) ->      List[str]:






            '''simple docstring'''





            __snake_case			:List[Any]          =						self.get_trainer()
            __snake_case			:str          =						DEFAULT_CALLBACKS.copy() + [ProgressCallback]
            self.check_callbacks_equality(trainer.callback_handler.callbacks			,						a__    )

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

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



       def 					__lowercase						(							self    ) ->      Optional[int]:






            '''simple docstring'''





            __snake_case			:List[Any]          =						DEFAULT_CALLBACKS.copy() + [ProgressCallback]
            __snake_case			:List[str]          =						self.get_trainer()

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

            __snake_case			:str          =						self.get_trainer()
            __snake_case			:Tuple          =						trainer.pop_callback(a__    )
            self.assertEqual(cb.__class__			,						a__    )
            self.check_callbacks_equality(trainer.callback_handler.callbacks			,						a__    )

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

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

            __snake_case			:Any          =						self.get_trainer()
            __snake_case			:Optional[int]          =						trainer.callback_handler.callbacks[0]
            __snake_case			:str          =						trainer.pop_callback(a__    )
            self.assertEqual(a__			,						a__    )
            self.check_callbacks_equality(trainer.callback_handler.callbacks			,						a__    )

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



       def 					__lowercase						(							self    ) ->      Dict:






            '''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=a__    )

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

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

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

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

            __snake_case			:Tuple          =						self.get_trainer(callbacks=[MyTestTrainerCallback]			,						evaluation_strategy="""epoch"""    )
            trainer.train()
            __snake_case			:Union[str, Any]          =						trainer.callback_handler.callbacks[-2].events
            self.assertEqual(a__			,						self.get_expected_events(a__    )    )

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

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










from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


lowerCamelCase__							=     {
    """configuration_canine""": ["""CANINE_PRETRAINED_CONFIG_ARCHIVE_MAP""", """CanineConfig"""],
    """tokenization_canine""": ["""CanineTokenizer"""],
}

try:
     if not is_torch_available():
          raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
     pass
else:
     lowerCamelCase__							=     [
         """CANINE_PRETRAINED_MODEL_ARCHIVE_LIST""",
         """CanineForMultipleChoice""",
         """CanineForQuestionAnswering""",
         """CanineForSequenceClassification""",
         """CanineForTokenClassification""",
         """CanineLayer""",
         """CanineModel""",
         """CaninePreTrainedModel""",
         """load_tf_weights_in_canine""",
     ]


if TYPE_CHECKING:
     from .configuration_canine import CANINE_PRETRAINED_CONFIG_ARCHIVE_MAP, CanineConfig
     from .tokenization_canine import CanineTokenizer

     try:
          if not is_torch_available():
               raise OptionalDependencyNotAvailable()
     except OptionalDependencyNotAvailable:
          pass
     else:
          from .modeling_canine import (
              CANINE_PRETRAINED_MODEL_ARCHIVE_LIST,
              CanineForMultipleChoice,
              CanineForQuestionAnswering,
              CanineForSequenceClassification,
              CanineForTokenClassification,
              CanineLayer,
              CanineModel,
              CaninePreTrainedModel,
              load_tf_weights_in_canine,
          )


else:
     import sys

     lowerCamelCase__							=     _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)


