'''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]



import json
from typing import List, Optional, Tuple

from tokenizers import pre_tokenizers, processors

from ...tokenization_utils_base import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_bart import BartTokenizer


__snake_case								=  logging.get_logger(__name__)


__snake_case								=  {"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt""", """tokenizer_file""": """tokenizer.json"""}

# See all BART models at https://huggingface.co/models?filter=bart
__snake_case								=  {
    """vocab_file""": {
        """facebook/bart-base""": """https://huggingface.co/facebook/bart-base/resolve/main/vocab.json""",
        """facebook/bart-large""": """https://huggingface.co/facebook/bart-large/resolve/main/vocab.json""",
        """facebook/bart-large-mnli""": """https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json""",
        """facebook/bart-large-cnn""": """https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json""",
        """facebook/bart-large-xsum""": """https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json""",
        """yjernite/bart_eli5""": """https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """facebook/bart-base""": """https://huggingface.co/facebook/bart-base/resolve/main/merges.txt""",
        """facebook/bart-large""": """https://huggingface.co/facebook/bart-large/resolve/main/merges.txt""",
        """facebook/bart-large-mnli""": """https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt""",
        """facebook/bart-large-cnn""": """https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt""",
        """facebook/bart-large-xsum""": """https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt""",
        """yjernite/bart_eli5""": """https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt""",
    },
    """tokenizer_file""": {
        """facebook/bart-base""": """https://huggingface.co/facebook/bart-base/resolve/main/tokenizer.json""",
        """facebook/bart-large""": """https://huggingface.co/facebook/bart-large/resolve/main/tokenizer.json""",
        """facebook/bart-large-mnli""": """https://huggingface.co/facebook/bart-large-mnli/resolve/main/tokenizer.json""",
        """facebook/bart-large-cnn""": """https://huggingface.co/facebook/bart-large-cnn/resolve/main/tokenizer.json""",
        """facebook/bart-large-xsum""": """https://huggingface.co/facebook/bart-large-xsum/resolve/main/tokenizer.json""",
        """yjernite/bart_eli5""": """https://huggingface.co/yjernite/bart_eli5/resolve/main/tokenizer.json""",
    },
}

__snake_case								=  {
    """facebook/bart-base""": 10_24,
    """facebook/bart-large""": 10_24,
    """facebook/bart-large-mnli""": 10_24,
    """facebook/bart-large-cnn""": 10_24,
    """facebook/bart-large-xsum""": 10_24,
    """yjernite/bart_eli5""": 10_24,
}




class 						lowercase__  (						_UpperCAmelCase  ):
	A__    :   Tuple				       =VOCAB_FILES_NAMES
	A__    :   Any				       =PRETRAINED_VOCAB_FILES_MAP
	A__    :   str				       =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	A__    :   Tuple				       =["""input_ids""", """attention_mask"""]
	A__    :   Optional[int]				       =BartTokenizer
	def __init__(			self							:   str       ,   UpperCAmelCase_							:   Optional[Any]=None       ,   UpperCAmelCase_							:   int=None       ,   UpperCAmelCase_							:   str=None       ,   UpperCAmelCase_							:   Optional[Any]="replace"       ,   UpperCAmelCase_							:   int="<s>"       ,   UpperCAmelCase_							:   List[Any]="</s>"       ,   UpperCAmelCase_							:   int="</s>"       ,   UpperCAmelCase_							:   Tuple="<s>"       ,   UpperCAmelCase_							:   Any="<unk>"       ,   UpperCAmelCase_							:   Any="<pad>"       ,   UpperCAmelCase_							:   Optional[Any]="<mask>"       ,   UpperCAmelCase_							:   Optional[int]=False       ,   UpperCAmelCase_							:   List[str]=True       ,   **UpperCAmelCase_							:   List[Any]       ,   ):
						super().__init__(
						    UpperCAmelCase_       ,   UpperCAmelCase_       ,   tokenizer_file=UpperCAmelCase_       ,   errors=UpperCAmelCase_       ,   bos_token=UpperCAmelCase_       ,   eos_token=UpperCAmelCase_       ,   sep_token=UpperCAmelCase_       ,   cls_token=UpperCAmelCase_       ,   unk_token=UpperCAmelCase_       ,   pad_token=UpperCAmelCase_       ,   mask_token=UpperCAmelCase_       ,   add_prefix_space=UpperCAmelCase_       ,   trim_offsets=UpperCAmelCase_       ,   **UpperCAmelCase_       ,   )

						SCREAMING_SNAKE_CASE__													= json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()							)
						if pre_tok_state.get('add_prefix_space'       ,   UpperCAmelCase_							) != add_prefix_space:
											SCREAMING_SNAKE_CASE__													= getattr(UpperCAmelCase_       ,   pre_tok_state.pop('type'							)							)
											SCREAMING_SNAKE_CASE__													= add_prefix_space
											SCREAMING_SNAKE_CASE__													= pre_tok_class(**UpperCAmelCase_							)

						SCREAMING_SNAKE_CASE__													= add_prefix_space

						# the pre_tokenizer is already updated in the GPT2TokenizerFast `__init__`
						SCREAMING_SNAKE_CASE__													= 'post_processor'
						SCREAMING_SNAKE_CASE__													= getattr(self.backend_tokenizer       ,   UpperCAmelCase_       ,   UpperCAmelCase_							)
						if tokenizer_component_instance:
											SCREAMING_SNAKE_CASE__													= json.loads(tokenizer_component_instance.__getstate__()							)

											# The lists 'sep' and 'cls' must be cased in tuples for the object `post_processor_class`
											if "sep" in state:
																SCREAMING_SNAKE_CASE__													= tuple(state['sep']							)
											if "cls" in state:
																SCREAMING_SNAKE_CASE__													= tuple(state['cls']							)

											SCREAMING_SNAKE_CASE__													= False

											if state.get('add_prefix_space'       ,   UpperCAmelCase_							) != add_prefix_space:
																SCREAMING_SNAKE_CASE__													= add_prefix_space
																SCREAMING_SNAKE_CASE__													= True

											if state.get('trim_offsets'       ,   UpperCAmelCase_							) != trim_offsets:
																SCREAMING_SNAKE_CASE__													= trim_offsets
																SCREAMING_SNAKE_CASE__													= True

											if changes_to_apply:
																SCREAMING_SNAKE_CASE__													= getattr(UpperCAmelCase_       ,   state.pop('type'							)							)
																SCREAMING_SNAKE_CASE__													= component_class(**UpperCAmelCase_							)
																setattr(self.backend_tokenizer       ,   UpperCAmelCase_       ,   UpperCAmelCase_							)
	@property
	def      A_				(			self							:   Tuple							):
						if self._mask_token is None:
											if self.verbose:
																logger.error('Using mask_token, but it is not set yet.'							)
											return None
						return str(self._mask_token							)
	@mask_token.setter
	def      A_				(			self							:   Any       ,   UpperCAmelCase_							:   List[Any]							):
						SCREAMING_SNAKE_CASE__													= AddedToken(UpperCAmelCase_       ,   lstrip=UpperCAmelCase_       ,   rstrip=UpperCAmelCase_							) if isinstance(UpperCAmelCase_       ,   UpperCAmelCase_							) else value
						SCREAMING_SNAKE_CASE__													= value
	def      A_				(			self							:   List[str]       ,   *UpperCAmelCase_							:   Tuple       ,   **UpperCAmelCase_							:   str							):
						SCREAMING_SNAKE_CASE__													= kwargs.get('is_split_into_words'       ,   UpperCAmelCase_							)

						if is_split_into_words and not self.add_prefix_space:
											raise ValueError(
											    F'You need to instantiate {self.__class__.__name__} with add_prefix_space=True '
											    'to use it with pretokenized inputs.'							)

						return super()._batch_encode_plus(*UpperCAmelCase_       ,   **UpperCAmelCase_							)
	def      A_				(			self							:   List[str]       ,   *UpperCAmelCase_							:   Dict       ,   **UpperCAmelCase_							:   int							):
						SCREAMING_SNAKE_CASE__													= kwargs.get('is_split_into_words'       ,   UpperCAmelCase_							)

						if is_split_into_words and not self.add_prefix_space:
											raise ValueError(
											    F'You need to instantiate {self.__class__.__name__} with add_prefix_space=True '
											    'to use it with pretokenized inputs.'							)

						return super()._encode_plus(*UpperCAmelCase_       ,   **UpperCAmelCase_							)
	def      A_				(			self							:   List[str]       ,   UpperCAmelCase_							:   str       ,   UpperCAmelCase_							:   Optional[str] = None							):
						SCREAMING_SNAKE_CASE__													= self._tokenizer.model.save(UpperCAmelCase_       ,   name=UpperCAmelCase_							)
						return tuple(UpperCAmelCase_							)
	def      A_				(			self							:   Optional[Any]       ,   UpperCAmelCase_							:   Optional[Any]       ,   UpperCAmelCase_							:   Union[str, Any]=None							):
						SCREAMING_SNAKE_CASE__													= [self.bos_token_id] + token_ids_a + [self.eos_token_id]
						if token_ids_a is None:
											return output

						return output + [self.eos_token_id] + token_ids_a + [self.eos_token_id]





	def      A_				(			self							:   str       ,   UpperCAmelCase_							:   List[int]       ,   UpperCAmelCase_							:   Optional[List[int]] = None							):
						SCREAMING_SNAKE_CASE__													= [self.sep_token_id]
						SCREAMING_SNAKE_CASE__													= [self.cls_token_id]

						if token_ids_a is None:
											return len(cls + token_ids_a + sep							) * [0]
						return len(cls + token_ids_a + sep + sep + token_ids_a + sep							) * [0]





