


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__						(       lowercase__    ):
      def __init__( self       : int  )	->    Any:
             a_      :	Optional[Any]							=						[]
      def  SCREAMING_SNAKE_CASE							( self       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : Optional[int]  ,							SCREAMING_SNAKE_CASE__       : List[str]  ,							SCREAMING_SNAKE_CASE__       : List[str]  ,							**SCREAMING_SNAKE_CASE__       : List[str]  )	->    Union[str, Any]:
             self.events.append('on_init_end'  )
      def  SCREAMING_SNAKE_CASE							( self       : Optional[Any]  ,							SCREAMING_SNAKE_CASE__       : List[str]  ,							SCREAMING_SNAKE_CASE__       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : List[str]  ,							**SCREAMING_SNAKE_CASE__       : int  )	->    int:
             self.events.append('on_train_begin'  )
      def  SCREAMING_SNAKE_CASE							( self       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : Any  ,							SCREAMING_SNAKE_CASE__       : Dict  ,							SCREAMING_SNAKE_CASE__       : List[str]  ,							**SCREAMING_SNAKE_CASE__       : List[str]  )	->    List[str]:
             self.events.append('on_train_end'  )
      def  SCREAMING_SNAKE_CASE							( self       : Any  ,							SCREAMING_SNAKE_CASE__       : int  ,							SCREAMING_SNAKE_CASE__       : Dict  ,							SCREAMING_SNAKE_CASE__       : Union[str, Any]  ,							**SCREAMING_SNAKE_CASE__       : str  )	->    Optional[Any]:
             self.events.append('on_epoch_begin'  )
      def  SCREAMING_SNAKE_CASE							( self       : int  ,							SCREAMING_SNAKE_CASE__       : str  ,							SCREAMING_SNAKE_CASE__       : Dict  ,							SCREAMING_SNAKE_CASE__       : Tuple  ,							**SCREAMING_SNAKE_CASE__       : Optional[Any]  )	->    Union[str, Any]:
             self.events.append('on_epoch_end'  )
      def  SCREAMING_SNAKE_CASE							( self       : str  ,							SCREAMING_SNAKE_CASE__       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : int  ,							SCREAMING_SNAKE_CASE__       : Optional[int]  ,							**SCREAMING_SNAKE_CASE__       : Optional[Any]  )	->    int:
             self.events.append('on_step_begin'  )
      def  SCREAMING_SNAKE_CASE							( self       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : List[Any]  ,							SCREAMING_SNAKE_CASE__       : Optional[int]  ,							SCREAMING_SNAKE_CASE__       : Optional[int]  ,							**SCREAMING_SNAKE_CASE__       : Dict  )	->    Optional[Any]:
             self.events.append('on_step_end'  )
      def  SCREAMING_SNAKE_CASE							( self       : Optional[Any]  ,							SCREAMING_SNAKE_CASE__       : str  ,							SCREAMING_SNAKE_CASE__       : Any  ,							SCREAMING_SNAKE_CASE__       : Tuple  ,							**SCREAMING_SNAKE_CASE__       : List[str]  )	->    str:
             self.events.append('on_evaluate'  )
      def  SCREAMING_SNAKE_CASE							( self       : Optional[int]  ,							SCREAMING_SNAKE_CASE__       : Tuple  ,							SCREAMING_SNAKE_CASE__       : Tuple  ,							SCREAMING_SNAKE_CASE__       : Optional[int]  ,							**SCREAMING_SNAKE_CASE__       : int  )	->    Union[str, Any]:
             self.events.append('on_predict'  )
      def  SCREAMING_SNAKE_CASE							( self       : str  ,							SCREAMING_SNAKE_CASE__       : Optional[int]  ,							SCREAMING_SNAKE_CASE__       : Dict  ,							SCREAMING_SNAKE_CASE__       : str  ,							**SCREAMING_SNAKE_CASE__       : Optional[Any]  )	->    Tuple:
             self.events.append('on_save'  )
      def  SCREAMING_SNAKE_CASE							( self       : Any  ,							SCREAMING_SNAKE_CASE__       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : Dict  ,							SCREAMING_SNAKE_CASE__       : Any  ,							**SCREAMING_SNAKE_CASE__       : List[Any]  )	->    int:
             self.events.append('on_log'  )

      def  SCREAMING_SNAKE_CASE							( self       : int  ,							SCREAMING_SNAKE_CASE__       : Any  ,							SCREAMING_SNAKE_CASE__       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : Tuple  ,							**SCREAMING_SNAKE_CASE__       : Union[str, Any]  )	->    List[Any]:
             self.events.append('on_prediction_step'  )
@require_torch
class 				SCREAMING_SNAKE_CASE__						(       unittest.TestCase    ):
      def  SCREAMING_SNAKE_CASE							( self       : Tuple  )	->    Optional[int]:
             a_      :	List[str]							=						tempfile.mkdtemp()
      def  SCREAMING_SNAKE_CASE							( self       : List[str]  )	->    int:
             shutil.rmtree(self.output_dir  )
      def  SCREAMING_SNAKE_CASE							( self       : Optional[Any]  ,							SCREAMING_SNAKE_CASE__       : int=0  ,							SCREAMING_SNAKE_CASE__       : Tuple=0  ,							SCREAMING_SNAKE_CASE__       : Dict=6_4  ,							SCREAMING_SNAKE_CASE__       : List[Any]=6_4  ,							SCREAMING_SNAKE_CASE__       : Dict=None  ,							SCREAMING_SNAKE_CASE__       : Dict=False  ,							**SCREAMING_SNAKE_CASE__       : Any  )	->    Dict:
             # 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.
             a_      :	Optional[Any]							=						RegressionDataset(length=SCREAMING_SNAKE_CASE__  )
             a_      :	Tuple							=						RegressionDataset(length=SCREAMING_SNAKE_CASE__  )
             a_      :	List[str]							=						RegressionModelConfig(a=SCREAMING_SNAKE_CASE__  ,							b=SCREAMING_SNAKE_CASE__  )
             a_      :	Tuple							=						RegressionPreTrainedModel(SCREAMING_SNAKE_CASE__  )

             a_      :	Any							=						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  SCREAMING_SNAKE_CASE							( self       : Union[str, Any]  ,							SCREAMING_SNAKE_CASE__       : Tuple  ,							SCREAMING_SNAKE_CASE__       : Any  )	->    Optional[int]:
             self.assertEqual(len(SCREAMING_SNAKE_CASE__  )  ,							len(SCREAMING_SNAKE_CASE__  )  )

             # Order doesn't matter
             a_      :	Tuple							=						sorted(SCREAMING_SNAKE_CASE__  ,							key=lambda SCREAMING_SNAKE_CASE__  : cb.__name__ if isinstance(SCREAMING_SNAKE_CASE__  ,							SCREAMING_SNAKE_CASE__  ) else cb.__class__.__name__  )
             a_      :	str							=						sorted(SCREAMING_SNAKE_CASE__  ,							key=lambda SCREAMING_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  SCREAMING_SNAKE_CASE							( self       : Any  ,							SCREAMING_SNAKE_CASE__       : Any  )	->    Tuple:
             a_      :	Union[str, Any]							=						['on_init_end', 'on_train_begin']
             a_      :	Tuple							=						0
             a_      :	Tuple							=						len(trainer.get_eval_dataloader()  )
             a_      :	List[str]							=						['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  SCREAMING_SNAKE_CASE							( self       : List[Any]  )	->    Tuple:
             a_      :	int							=						self.get_trainer()
             a_      :	str							=						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_      :	List[str]							=						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_      :	int							=						self.get_trainer(disable_tqdm=SCREAMING_SNAKE_CASE__  )
             a_      :	Any							=						DEFAULT_CALLBACKS.copy() + [PrinterCallback]
             self.check_callbacks_equality(trainer.callback_handler.callbacks  ,							SCREAMING_SNAKE_CASE__  )
      def  SCREAMING_SNAKE_CASE							( self       : Tuple  )	->    str:
             a_      :	Tuple							=						DEFAULT_CALLBACKS.copy() + [ProgressCallback]
             a_      :	int							=						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_      :	List[Any]							=						self.get_trainer()
             a_      :	Tuple							=						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_      :	Tuple							=						self.get_trainer()
             a_      :	Union[str, Any]							=						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_      :	Tuple							=						self.get_trainer()
             a_      :	List[str]							=						trainer.callback_handler.callbacks[0]
             a_      :	Any							=						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  SCREAMING_SNAKE_CASE							( self       : int  )	->    Optional[int]:
             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_      :	List[Any]							=						self.get_trainer(callbacks=[MyTestTrainerCallback]  )
             trainer.train()
             a_      :	List[str]							=						trainer.callback_handler.callbacks[-2].events
             self.assertEqual(SCREAMING_SNAKE_CASE__  ,							self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

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

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

             a_      :	Optional[int]							=						self.get_trainer(callbacks=[MyTestTrainerCallback]  ,							eval_steps=5  ,							evaluation_strategy='steps'  )
             trainer.train()
             a_      :	Dict							=						trainer.callback_handler.callbacks[-2].events
             self.assertEqual(SCREAMING_SNAKE_CASE__  ,							self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

             a_      :	Any							=						self.get_trainer(callbacks=[MyTestTrainerCallback]  ,							evaluation_strategy='epoch'  )
             trainer.train()
             a_      :	Dict							=						trainer.callback_handler.callbacks[-2].events
             self.assertEqual(SCREAMING_SNAKE_CASE__  ,							self.get_expected_events(SCREAMING_SNAKE_CASE__  )  )

             # A bit of everything
             a_      :	Optional[Any]							=						self.get_trainer(
                 callbacks=[MyTestTrainerCallback]  ,							logging_steps=3  ,							save_steps=1_0  ,							eval_steps=5  ,							evaluation_strategy='steps'  ,							)
             trainer.train()
             a_      :	Dict							=						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_      :	Optional[Any]							=						self.get_trainer(
                        callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]  ,							)
                    assert str(SCREAMING_SNAKE_CASE__  ) in warn_mock.call_args[0][0]






from .dependency_versions_table import deps
from .utils.versions import require_version, require_version_core


# define which module versions we always want to check at run time
# (usually the ones defined in `install_requires` in setup.py)
#
# order specific notes:
# - tqdm must be checked before tokenizers

UpperCAmelCase_   :				Dict						      = [
    'python',
    'tqdm',
    'regex',
    'requests',
    'packaging',
    'filelock',
    'numpy',
    'tokenizers',
    'huggingface-hub',
    'safetensors',
    'accelerate',
    'pyyaml',
]

for pkg in pkgs_to_check_at_runtime:
   if pkg in deps:
      if pkg == "tokenizers":
         # must be loaded here, or else tqdm check may fail
         from .utils import is_tokenizers_available

         if not is_tokenizers_available():
            continue  # not required, check version only if installed
      elif pkg == "accelerate":
         # must be loaded here, or else tqdm check may fail
         from .utils import is_accelerate_available

         # Maybe switch to is_torch_available in the future here so that Accelerate is hard dep of
         # Transformers with PyTorch
         if not is_accelerate_available():
            continue  # not required, check version only if installed

      require_version_core(deps[pkg])
   else:
      raise ValueError(F'can\'t find {pkg} in {deps.keys()}, check dependency_versions_table.py')







def  SCREAMING_SNAKE_CASE_		(    __A				:    Optional[Any]						,					__A				:    int=None       )			->	int:

       """simple docstring"""

       require_version(deps[pkg]						,					__A       )


