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






     def __init__(		self  ):
       _lowerCAmelCase						:       Optional[int]  			=     []






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_init_end"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_train_begin"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_train_end"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_epoch_begin"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_epoch_end"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_step_begin"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_step_end"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_evaluate"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_predict"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_save"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_log"""  )






     def     __A  (		self				,     a__				,     a__				,     a__				,     **a__  ):
       self.events.append("""on_prediction_step"""  )







@require_torch
class 						__A							( unittest.TestCase		):






     def     __A  (		self  ):
       _lowerCAmelCase						:       Any  			=     tempfile.mkdtemp()






     def     __A  (		self  ):
       shutil.rmtree(self.output_dir  )






     def     __A  (		self				,     a__=0				,     a__=0				,     a__=64				,     a__=64				,     a__=None				,     a__=False				,     **a__  ):
       # 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.
       _lowerCAmelCase						:       Tuple  			=     RegressionDataset(length=lowerCamelCase__  )
       _lowerCAmelCase						:       Tuple  			=     RegressionDataset(length=lowerCamelCase__  )
       _lowerCAmelCase						:       str  			=     RegressionModelConfig(a=lowerCamelCase__				,     b=lowerCamelCase__  )
       _lowerCAmelCase						:       Tuple  			=     RegressionPreTrainedModel(lowerCamelCase__  )

       _lowerCAmelCase						:       str  			=     TrainingArguments(self.output_dir				,     disable_tqdm=lowerCamelCase__				,     report_to=[]				,     **lowerCamelCase__  )
       return Trainer(
           lowerCamelCase__				,     lowerCamelCase__				,     train_dataset=lowerCamelCase__				,     eval_dataset=lowerCamelCase__				,     callbacks=lowerCamelCase__				,     )






     def     __A  (		self				,     a__				,     a__  ):
       self.assertEqual(len(lowerCamelCase__  )				,     len(lowerCamelCase__  )  )

       # Order doesn't matter
       _lowerCAmelCase						:       Optional[int]  			=     sorted(lowerCamelCase__				,     key=lambda a__  : cb.__name__ if isinstance(lowerCamelCase__				,     lowerCamelCase__  ) else cb.__class__.__name__  )
       _lowerCAmelCase						:       Optional[int]  			=     sorted(lowerCamelCase__				,     key=lambda a__  : cb.__name__ if isinstance(lowerCamelCase__				,     lowerCamelCase__  ) else cb.__class__.__name__  )

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






     def     __A  (		self				,     a__  ):
       _lowerCAmelCase						:       List[Any]  			=     ["""on_init_end""", """on_train_begin"""]
       _lowerCAmelCase						:       Union[str, Any]  			=     0
       _lowerCAmelCase						:       Optional[int]  			=     len(trainer.get_eval_dataloader()  )
       _lowerCAmelCase						:       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(lowerCamelCase__  ):
           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  ):
       _lowerCAmelCase						:       Optional[Any]  			=     self.get_trainer()
       _lowerCAmelCase						:       Union[str, Any]  			=     DEFAULT_CALLBACKS.copy() + [ProgressCallback]
       self.check_callbacks_equality(trainer.callback_handler.callbacks				,     lowerCamelCase__  )

       # Callbacks passed at init are added to the default callbacks
       _lowerCAmelCase						:       Dict  			=     self.get_trainer(callbacks=[MyTestTrainerCallback]  )
       expected_callbacks.append(lowerCamelCase__  )
       self.check_callbacks_equality(trainer.callback_handler.callbacks				,     lowerCamelCase__  )

       # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
       _lowerCAmelCase						:       str  			=     self.get_trainer(disable_tqdm=lowerCamelCase__  )
       _lowerCAmelCase						:       Optional[Any]  			=     DEFAULT_CALLBACKS.copy() + [PrinterCallback]
       self.check_callbacks_equality(trainer.callback_handler.callbacks				,     lowerCamelCase__  )






     def     __A  (		self  ):
       _lowerCAmelCase						:       int  			=     DEFAULT_CALLBACKS.copy() + [ProgressCallback]
       _lowerCAmelCase						:       str  			=     self.get_trainer()

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

       _lowerCAmelCase						:       Tuple  			=     self.get_trainer()
       _lowerCAmelCase						:       Tuple  			=     trainer.pop_callback(lowerCamelCase__  )
       self.assertEqual(cb.__class__				,     lowerCamelCase__  )
       self.check_callbacks_equality(trainer.callback_handler.callbacks				,     lowerCamelCase__  )

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

       # We can also add, pop, or remove by instance
       _lowerCAmelCase						:       Tuple  			=     self.get_trainer()
       _lowerCAmelCase						:       Optional[int]  			=     trainer.callback_handler.callbacks[0]
       trainer.remove_callback(lowerCamelCase__  )
       expected_callbacks.remove(lowerCamelCase__  )
       self.check_callbacks_equality(trainer.callback_handler.callbacks				,     lowerCamelCase__  )

       _lowerCAmelCase						:       Optional[Any]  			=     self.get_trainer()
       _lowerCAmelCase						:       Any  			=     trainer.callback_handler.callbacks[0]
       _lowerCAmelCase						:       Union[str, Any]  			=     trainer.pop_callback(lowerCamelCase__  )
       self.assertEqual(lowerCamelCase__				,     lowerCamelCase__  )
       self.check_callbacks_equality(trainer.callback_handler.callbacks				,     lowerCamelCase__  )

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






     def     __A  (		self  ):
       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=lowerCamelCase__  )

       _lowerCAmelCase						:       Optional[int]  			=     self.get_trainer(callbacks=[MyTestTrainerCallback]  )
       trainer.train()
       _lowerCAmelCase						:       List[str]  			=     trainer.callback_handler.callbacks[-2].events
       self.assertEqual(lowerCamelCase__				,     self.get_expected_events(lowerCamelCase__  )  )

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

       _lowerCAmelCase						:       Union[str, Any]  			=     self.get_trainer(callbacks=[MyTestTrainerCallback]				,     save_steps=5  )
       trainer.train()
       _lowerCAmelCase						:       str  			=     trainer.callback_handler.callbacks[-2].events
       self.assertEqual(lowerCamelCase__				,     self.get_expected_events(lowerCamelCase__  )  )

       _lowerCAmelCase						:       Tuple  			=     self.get_trainer(callbacks=[MyTestTrainerCallback]				,     eval_steps=5				,     evaluation_strategy="""steps"""  )
       trainer.train()
       _lowerCAmelCase						:       List[Any]  			=     trainer.callback_handler.callbacks[-2].events
       self.assertEqual(lowerCamelCase__				,     self.get_expected_events(lowerCamelCase__  )  )

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

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

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





"""simple docstring"""






import warnings
from typing import Dict, List, Optional, Tuple

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase__  :  Optional[int]						     =	logging.get_logger(__name__)
class 				snake_case       ( __UpperCAmelCase				):




					"""simple docstring"""

					snake_case__				    = ["input_ids", "attention_mask"]







					def __init__(				self   :	Optional[int]    ,lowerCamelCase__   :	int="</s>"    ,lowerCamelCase__   :	str="<unk>"    ,lowerCamelCase__   :	Union[str, Any]="<pad>"    ,lowerCamelCase__   :	int=125    ,lowerCamelCase__   :	str=None    ,**lowerCamelCase__   :	Union[str, Any]    ,):
									# Add extra_ids to the special token list
									if extra_ids > 0 and additional_special_tokens is None:
													UpperCAmelCase__             =							[f'''<extra_id_{i}>''' for i in range(lowerCamelCase__			)]
									elif extra_ids > 0 and additional_special_tokens is not None:
													# Check that we have the right number of extra_id special tokens
													UpperCAmelCase__             =							len(set(filter(lambda lowerCamelCase__			: bool('extra_id' in str(lowerCamelCase__			)			)    ,lowerCamelCase__			)			)			)
													if extra_tokens != extra_ids:
																	raise ValueError(
																	    f'''Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are'''
																	    ' provided to ByT5Tokenizer. In this case the additional_special_tokens must include the'
																	    ' extra_ids tokens'			)

									UpperCAmelCase__             =							AddedToken(lowerCamelCase__    ,lstrip=lowerCamelCase__    ,rstrip=lowerCamelCase__			) if isinstance(lowerCamelCase__    ,lowerCamelCase__			) else pad_token
									UpperCAmelCase__             =							AddedToken(lowerCamelCase__    ,lstrip=lowerCamelCase__    ,rstrip=lowerCamelCase__			) if isinstance(lowerCamelCase__    ,lowerCamelCase__			) else eos_token
									UpperCAmelCase__             =							AddedToken(lowerCamelCase__    ,lstrip=lowerCamelCase__    ,rstrip=lowerCamelCase__			) if isinstance(lowerCamelCase__    ,lowerCamelCase__			) else unk_token

									super().__init__(
									    eos_token=lowerCamelCase__    ,unk_token=lowerCamelCase__    ,pad_token=lowerCamelCase__    ,extra_ids=lowerCamelCase__    ,additional_special_tokens=lowerCamelCase__    ,**lowerCamelCase__    ,)

									UpperCAmelCase__             =							extra_ids

									UpperCAmelCase__             =							2**8  # utf is 8 bits

									# define special tokens dict
									UpperCAmelCase__             =							{
									    self.pad_token: 0,
									    self.eos_token: 1,
									    self.unk_token: 2,
									}
									UpperCAmelCase__             =							len(self.special_tokens_encoder			)
									UpperCAmelCase__             =							len(lowerCamelCase__			)
									for i, token in enumerate(lowerCamelCase__			):
													UpperCAmelCase__             =							self.vocab_size + i - n
									UpperCAmelCase__             =							{v: k for k, v in self.special_tokens_encoder.items()}







					@property
					def 		__lowerCAmelCase						(				self   :	Union[str, Any]			):
									return self._utf_vocab_size + self._num_special_tokens + self._extra_ids







					def 		__lowerCAmelCase						(				self   :	Optional[Any]    ,lowerCamelCase__   :	List[int]    ,lowerCamelCase__   :	Optional[List[int]] = None    ,lowerCamelCase__   :	bool = False			):
									if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=lowerCamelCase__    ,token_ids_a=lowerCamelCase__    ,already_has_special_tokens=lowerCamelCase__			)

									# normal case: some special tokens
									if token_ids_a is None:
													return ([0] * len(lowerCamelCase__			)) + [1]
									return ([0] * len(lowerCamelCase__			)) + [1] + ([0] * len(lowerCamelCase__			)) + [1]







					def 		__lowerCAmelCase						(				self   :	str    ,lowerCamelCase__   :	List[int]			):
									if len(lowerCamelCase__			) > 0 and token_ids[-1] == self.eos_token_id:
													warnings.warn(
													    f'''This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated'''
													    ' eos tokens being added.'			)
													return token_ids
									else:
													return token_ids + [self.eos_token_id]







					def 		__lowerCAmelCase						(				self   :	str    ,lowerCamelCase__   :	List[int]    ,lowerCamelCase__   :	Optional[List[int]] = None			):
									UpperCAmelCase__             =							[self.eos_token_id]

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







					def 		__lowerCAmelCase						(				self   :	Dict    ,lowerCamelCase__   :	List[int]    ,lowerCamelCase__   :	Optional[List[int]] = None			):
									UpperCAmelCase__             =							self._add_eos_if_not_present(lowerCamelCase__			)
									if token_ids_a is None:
													return token_ids_a
									else:
													UpperCAmelCase__             =							self._add_eos_if_not_present(lowerCamelCase__			)
													return token_ids_a + token_ids_a







					def 		__lowerCAmelCase						(				self   :	Optional[Any]    ,lowerCamelCase__   :	str			):
									UpperCAmelCase__             =							[chr(lowerCamelCase__			) for i in text.encode('utf-8'			)]
									return tokens







					def 		__lowerCAmelCase						(				self   :	List[Any]    ,lowerCamelCase__   :	str			):
									if token in self.special_tokens_encoder:
													UpperCAmelCase__             =							self.special_tokens_encoder[token]
									elif token in self.added_tokens_encoder:
													UpperCAmelCase__             =							self.added_tokens_encoder[token]
									elif len(lowerCamelCase__			) != 1:
													UpperCAmelCase__             =							self.unk_token_id
									else:
													UpperCAmelCase__             =							ord(lowerCamelCase__			) + self._num_special_tokens
									return token_id







					def 		__lowerCAmelCase						(				self   :	Union[str, Any]    ,lowerCamelCase__   :	Any			):
									if index in self.special_tokens_decoder:
													UpperCAmelCase__             =							self.special_tokens_decoder[index]
									else:
													UpperCAmelCase__             =							chr(index - self._num_special_tokens			)
									return token







					def 		__lowerCAmelCase						(				self   :	Optional[Any]    ,lowerCamelCase__   :	Union[str, Any]			):
									UpperCAmelCase__             =							b''
									for token in tokens:
													if token in self.special_tokens_decoder:
																	UpperCAmelCase__             =							self.special_tokens_decoder[token].encode('utf-8'			)
													elif token in self.added_tokens_decoder:
																	UpperCAmelCase__             =							self.special_tokens_decoder[token].encode('utf-8'			)
													elif token in self.special_tokens_encoder:
																	UpperCAmelCase__             =							token.encode('utf-8'			)
													elif token in self.added_tokens_encoder:
																	UpperCAmelCase__             =							token.encode('utf-8'			)
													else:
																	UpperCAmelCase__             =							bytes([ord(lowerCamelCase__			)]			)
													bstring += tok_string
									UpperCAmelCase__             =							bstring.decode('utf-8'    ,errors='ignore'			)
									return string







					def 		__lowerCAmelCase						(				self   :	str    ,lowerCamelCase__   :	str    ,lowerCamelCase__   :	Optional[str] = None			):
									return ()



