


"""simple docstring"""







import argparse


lowerCamelCase_   :   int									=  """docs/source/_static/js/custom.js"""
def _A						(						lowercase		):







    """simple docstring"""
    with open(lowercase       ,    encoding='''utf-8'''       ,    newline='''\n'''		) as f:
        a 		=f.readlines()
    a 		=0

    # First let's put the right version
    while not lines[index].startswith('''const stableVersion ='''		):
        index += 1
    a 		=f'''const stableVersion = "v{version}"\n'''

    # Then update the dictionary
    while not lines[index].startswith('''const versionMapping = {'''		):
        index += 1

    # We go until the end
    while not lines[index].startswith('''}'''		):
        index += 1
    # We add the new version at the end
    lines[index - 1] += f'''    "v{version}": "v{version}",\n'''

    with open(lowercase       ,    '''w'''       ,    encoding='''utf-8'''       ,    newline='''\n'''		) as f:
        f.writelines(lowercase		)


if __name__ == "__main__":
       lowerCamelCase_   :   List[str]									=  argparse.ArgumentParser()
       parser.add_argument("""--version""", help="""Release version.""")
       lowerCamelCase_   :   Optional[Any]									=  parser.parse_args()
       update_custom_js(args.version)
'''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_	(snake_case__    ):
      '''simple docstring'''

      def __init__(       self						:				List[Any]       ):
             _UpperCAmelCase     :  Union[str, Any]							      =    []




      def      _A	(       self						:				Any						,					A						:				Union[str, Any]						,					A						:				Optional[int]						,					A						:				List[str]						,					**A						:				Tuple       ):
             self.events.append("on_init_end"       )




      def      _A	(       self						:				Any						,					A						:				str						,					A						:				List[Any]						,					A						:				List[Any]						,					**A						:				Tuple       ):
             self.events.append("on_train_begin"       )




      def      _A	(       self						:				Tuple						,					A						:				List[str]						,					A						:				Tuple						,					A						:				int						,					**A						:				List[str]       ):
             self.events.append("on_train_end"       )




      def      _A	(       self						:				Optional[Any]						,					A						:				Dict						,					A						:				Any						,					A						:				Optional[Any]						,					**A						:				List[Any]       ):
             self.events.append("on_epoch_begin"       )




      def      _A	(       self						:				Optional[Any]						,					A						:				List[Any]						,					A						:				List[str]						,					A						:				Optional[int]						,					**A						:				Optional[int]       ):
             self.events.append("on_epoch_end"       )




      def      _A	(       self						:				List[str]						,					A						:				Optional[int]						,					A						:				List[Any]						,					A						:				Union[str, Any]						,					**A						:				Any       ):
             self.events.append("on_step_begin"       )




      def      _A	(       self						:				Tuple						,					A						:				Union[str, Any]						,					A						:				int						,					A						:				Optional[int]						,					**A						:				int       ):
             self.events.append("on_step_end"       )




      def      _A	(       self						:				Optional[int]						,					A						:				Optional[Any]						,					A						:				Union[str, Any]						,					A						:				str						,					**A						:				Union[str, Any]       ):
             self.events.append("on_evaluate"       )




      def      _A	(       self						:				Optional[Any]						,					A						:				Optional[int]						,					A						:				Dict						,					A						:				List[Any]						,					**A						:				Dict       ):
             self.events.append("on_predict"       )




      def      _A	(       self						:				Dict						,					A						:				Dict						,					A						:				List[Any]						,					A						:				Dict						,					**A						:				str       ):
             self.events.append("on_save"       )




      def      _A	(       self						:				Tuple						,					A						:				Optional[Any]						,					A						:				Union[str, Any]						,					A						:				Optional[int]						,					**A						:				Dict       ):
             self.events.append("on_log"       )





      def      _A	(       self						:				Optional[int]						,					A						:				Optional[Any]						,					A						:				Tuple						,					A						:				Tuple						,					**A						:				List[str]       ):
             self.events.append("on_prediction_step"       )

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

      def      _A	(       self						:				Optional[int]       ):
             _UpperCAmelCase     :  Optional[Any]							      =    tempfile.mkdtemp()




      def      _A	(       self						:				List[Any]       ):
             shutil.rmtree(self.output_dir       )




      def      _A	(       self						:				Union[str, Any]						,					A						:				Optional[int]=0						,					A						:				Optional[Any]=0						,					A						:				Optional[Any]=64						,					A						:				Dict=64						,					A						:				Any=None						,					A						:				Tuple=False						,					**A						:				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.
             _UpperCAmelCase     :  str							      =    RegressionDataset(length=A       )
             _UpperCAmelCase     :  Union[str, Any]							      =    RegressionDataset(length=A       )
             _UpperCAmelCase     :  Any							      =    RegressionModelConfig(a=A						,					b=A       )
             _UpperCAmelCase     :  List[Any]							      =    RegressionPreTrainedModel(A       )

             _UpperCAmelCase     :  Dict							      =    TrainingArguments(self.output_dir						,					disable_tqdm=A						,					report_to=[]						,					**A       )
             return Trainer(
                 A						,					A						,					train_dataset=A						,					eval_dataset=A						,					callbacks=A						,					)




      def      _A	(       self						:				str						,					A						:				List[str]						,					A						:				List[str]       ):
             self.assertEqual(len(A       )						,					len(A       )       )

             # Order doesn't matter
             _UpperCAmelCase     :  Tuple							      =    sorted(A						,					key=lambda A       : cb.__name__ if isinstance(A						,					A       ) else cb.__class__.__name__       )
             _UpperCAmelCase     :  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      _A	(       self						:				int						,					A						:				List[str]       ):
             _UpperCAmelCase     :  List[str]							      =    ["on_init_end", "on_train_begin"]
             _UpperCAmelCase     :  str							      =    0
             _UpperCAmelCase     :  Optional[Any]							      =    len(trainer.get_eval_dataloader()       )
             _UpperCAmelCase     :  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(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      _A	(       self						:				str       ):
             _UpperCAmelCase     :  Any							      =    self.get_trainer()
             _UpperCAmelCase     :  int							      =    DEFAULT_CALLBACKS.copy() + [ProgressCallback]
             self.check_callbacks_equality(trainer.callback_handler.callbacks						,					A       )

             # Callbacks passed at init are added to the default callbacks
             _UpperCAmelCase     :  Optional[int]							      =    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
             _UpperCAmelCase     :  List[Any]							      =    self.get_trainer(disable_tqdm=A       )
             _UpperCAmelCase     :  Tuple							      =    DEFAULT_CALLBACKS.copy() + [PrinterCallback]
             self.check_callbacks_equality(trainer.callback_handler.callbacks						,					A       )




      def      _A	(       self						:				Optional[Any]       ):
             _UpperCAmelCase     :  Dict							      =    DEFAULT_CALLBACKS.copy() + [ProgressCallback]
             _UpperCAmelCase     :  Dict							      =    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       )

             _UpperCAmelCase     :  Optional[Any]							      =    self.get_trainer()
             _UpperCAmelCase     :  Any							      =    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
             _UpperCAmelCase     :  Union[str, Any]							      =    self.get_trainer()
             _UpperCAmelCase     :  List[Any]							      =    trainer.callback_handler.callbacks[0]
             trainer.remove_callback(A       )
             expected_callbacks.remove(A       )
             self.check_callbacks_equality(trainer.callback_handler.callbacks						,					A       )

             _UpperCAmelCase     :  List[Any]							      =    self.get_trainer()
             _UpperCAmelCase     :  List[Any]							      =    trainer.callback_handler.callbacks[0]
             _UpperCAmelCase     :  Union[str, Any]							      =    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      _A	(       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=A       )

             _UpperCAmelCase     :  Optional[int]							      =    self.get_trainer(callbacks=[MyTestTrainerCallback]       )
             trainer.train()
             _UpperCAmelCase     :  Union[str, Any]							      =    trainer.callback_handler.callbacks[-2].events
             self.assertEqual(A						,					self.get_expected_events(A       )       )

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

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

             _UpperCAmelCase     :  int							      =    self.get_trainer(callbacks=[MyTestTrainerCallback]						,					eval_steps=5						,					evaluation_strategy="steps"       )
             trainer.train()
             _UpperCAmelCase     :  Optional[int]							      =    trainer.callback_handler.callbacks[-2].events
             self.assertEqual(A						,					self.get_expected_events(A       )       )

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

             # A bit of everything
             _UpperCAmelCase     :  int							      =    self.get_trainer(
                 callbacks=[MyTestTrainerCallback]						,					logging_steps=3						,					save_steps=10						,					eval_steps=5						,					evaluation_strategy="steps"						,					)
             trainer.train()
             _UpperCAmelCase     :  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:
                    _UpperCAmelCase     :  Optional[Any]							      =    self.get_trainer(
                        callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]						,					)
                    assert str(A       ) in warn_mock.call_args[0][0]

