

"""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  (     lowercase       ):







 """simple docstring"""







 def __init__(      self							:      List[str]     ):
    __lowercase								=						[]


 def  _lowercase			(      self							:      Optional[Any], UpperCAmelCase__							:      Optional[Any], UpperCAmelCase__							:      Tuple, UpperCAmelCase__							:      List[str], **UpperCAmelCase__							:      List[str]     ):
    self.events.append("on_init_end"     )


 def  _lowercase			(      self							:      List[Any], UpperCAmelCase__							:      Dict, UpperCAmelCase__							:      List[Any], UpperCAmelCase__							:      Optional[int], **UpperCAmelCase__							:      int     ):
    self.events.append("on_train_begin"     )


 def  _lowercase			(      self							:      Optional[Any], UpperCAmelCase__							:      Any, UpperCAmelCase__							:      Optional[Any], UpperCAmelCase__							:      Dict, **UpperCAmelCase__							:      int     ):
    self.events.append("on_train_end"     )


 def  _lowercase			(      self							:      int, UpperCAmelCase__							:      Tuple, UpperCAmelCase__							:      Union[str, Any], UpperCAmelCase__							:      int, **UpperCAmelCase__							:      Any     ):
    self.events.append("on_epoch_begin"     )


 def  _lowercase			(      self							:      int, UpperCAmelCase__							:      Dict, UpperCAmelCase__							:      int, UpperCAmelCase__							:      int, **UpperCAmelCase__							:      Optional[Any]     ):
    self.events.append("on_epoch_end"     )


 def  _lowercase			(      self							:      Any, UpperCAmelCase__							:      Tuple, UpperCAmelCase__							:      Any, UpperCAmelCase__							:      Union[str, Any], **UpperCAmelCase__							:      Optional[int]     ):
    self.events.append("on_step_begin"     )


 def  _lowercase			(      self							:      List[str], UpperCAmelCase__							:      int, UpperCAmelCase__							:      List[str], UpperCAmelCase__							:      str, **UpperCAmelCase__							:      List[Any]     ):
    self.events.append("on_step_end"     )


 def  _lowercase			(      self							:      str, UpperCAmelCase__							:      List[Any], UpperCAmelCase__							:      List[str], UpperCAmelCase__							:      Optional[int], **UpperCAmelCase__							:      List[str]     ):
    self.events.append("on_evaluate"     )


 def  _lowercase			(      self							:      int, UpperCAmelCase__							:      Dict, UpperCAmelCase__							:      List[str], UpperCAmelCase__							:      Tuple, **UpperCAmelCase__							:      Optional[int]     ):
    self.events.append("on_predict"     )


 def  _lowercase			(      self							:      Optional[int], UpperCAmelCase__							:      Any, UpperCAmelCase__							:      List[Any], UpperCAmelCase__							:      List[str], **UpperCAmelCase__							:      List[Any]     ):
    self.events.append("on_save"     )


 def  _lowercase			(      self							:      Optional[Any], UpperCAmelCase__							:      List[Any], UpperCAmelCase__							:      Union[str, Any], UpperCAmelCase__							:      Dict, **UpperCAmelCase__							:      Union[str, Any]     ):
    self.events.append("on_log"     )



 def  _lowercase			(      self							:      Optional[int], UpperCAmelCase__							:      str, UpperCAmelCase__							:      Union[str, Any], UpperCAmelCase__							:      Union[str, Any], **UpperCAmelCase__							:      List[str]     ):
    self.events.append("on_prediction_step"     )







@require_torch
class  _lowerCAmelCase  (     unittest.TestCase       ):







 """simple docstring"""







 def  _lowercase			(      self							:      Optional[int]     ):
    __lowercase								=						tempfile.mkdtemp()


 def  _lowercase			(      self							:      int     ):
    shutil.rmtree(self.output_dir     )


 def  _lowercase			(      self							:      List[Any], UpperCAmelCase__							:      str=0, UpperCAmelCase__							:      Any=0, UpperCAmelCase__							:      str=6_4, UpperCAmelCase__							:      List[Any]=6_4, UpperCAmelCase__							:      Union[str, Any]=None, UpperCAmelCase__							:      Optional[int]=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.
    __lowercase								=						RegressionDataset(length=UpperCAmelCase__     )
    __lowercase								=						RegressionDataset(length=UpperCAmelCase__     )
    __lowercase								=						RegressionModelConfig(a=UpperCAmelCase__, b=UpperCAmelCase__     )
    __lowercase								=						RegressionPreTrainedModel(UpperCAmelCase__     )

    __lowercase								=						TrainingArguments(self.output_dir, disable_tqdm=UpperCAmelCase__, report_to=[], **UpperCAmelCase__     )
    return Trainer(
        UpperCAmelCase__, UpperCAmelCase__, train_dataset=UpperCAmelCase__, eval_dataset=UpperCAmelCase__, callbacks=UpperCAmelCase__, )


 def  _lowercase			(      self							:      str, UpperCAmelCase__							:      Any, UpperCAmelCase__							:      Union[str, Any]     ):
    self.assertEqual(len(UpperCAmelCase__     ), len(UpperCAmelCase__     )     )

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

    # Callbacks passed at init are added to the default callbacks
    __lowercase								=						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
    __lowercase								=						self.get_trainer(disable_tqdm=UpperCAmelCase__     )
    __lowercase								=						DEFAULT_CALLBACKS.copy() + [PrinterCallback]
    self.check_callbacks_equality(trainer.callback_handler.callbacks, UpperCAmelCase__     )


 def  _lowercase			(      self							:      Union[str, Any]     ):
    __lowercase								=						DEFAULT_CALLBACKS.copy() + [ProgressCallback]
    __lowercase								=						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__     )

    __lowercase								=						self.get_trainer()
    __lowercase								=						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
    __lowercase								=						self.get_trainer()
    __lowercase								=						trainer.callback_handler.callbacks[0]
    trainer.remove_callback(UpperCAmelCase__     )
    expected_callbacks.remove(UpperCAmelCase__     )
    self.check_callbacks_equality(trainer.callback_handler.callbacks, UpperCAmelCase__     )

    __lowercase								=						self.get_trainer()
    __lowercase								=						trainer.callback_handler.callbacks[0]
    __lowercase								=						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  _lowercase			(      self							:      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=UpperCAmelCase__     )

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

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

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

    __lowercase								=						self.get_trainer(callbacks=[MyTestTrainerCallback], eval_steps=5, evaluation_strategy="steps"     )
    trainer.train()
    __lowercase								=						trainer.callback_handler.callbacks[-2].events
    self.assertEqual(UpperCAmelCase__, self.get_expected_events(UpperCAmelCase__     )     )

    __lowercase								=						self.get_trainer(callbacks=[MyTestTrainerCallback], evaluation_strategy="epoch"     )
    trainer.train()
    __lowercase								=						trainer.callback_handler.callbacks[-2].events
    self.assertEqual(UpperCAmelCase__, self.get_expected_events(UpperCAmelCase__     )     )

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







"""simple docstring"""





import re

from filelock import FileLock


try:
   import nltk

   _a							=							True
except (ImportError, ModuleNotFoundError):
   _a							=							False

if NLTK_AVAILABLE:
   with FileLock('.lock') as lock:
      nltk.download('punkt', quiet=True)




def   _A  (  UpperCamelCase_					:	str)     ->						str:
   '''simple docstring'''


   re.sub("<n>",      "",      UpperCamelCase_)  # remove pegasus newline char
   assert NLTK_AVAILABLE, "nltk must be installed to separate newlines between sentences. (pip install nltk)"
   return "\n".join(nltk.sent_tokenize(UpperCamelCase_))




