



"""simple docstring"""

import json
import os
from functools import lru_cache
from typing import Dict, List, Optional, Tuple, Union

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


lowercase__    :  Dict							    =	logging.get_logger(__name__)


lowercase__    :  Optional[int]							    =	{'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt'''}

# See all LED models at https://huggingface.co/models?filter=LED
lowercase__    :  List[str]							    =	{
    '''vocab_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json''',
    },
    '''merges_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt''',
    },
    '''tokenizer_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json''',
    },
}

lowercase__    :  List[str]							    =	{
    '''allenai/led-base-16384''': 1_63_84,
}




@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def __lowercase			(							):
    snake_case_ :     Dict				=	(
        list(range(ord('''!'''    )		,			ord('''~'''    ) + 1    )    ) + list(range(ord('''¡'''    )		,			ord('''¬'''    ) + 1    )    ) + list(range(ord('''®'''    )		,			ord('''ÿ'''    ) + 1    )    )
    )
    snake_case_ :     Dict				=	bs[:]
    snake_case_ :     Any				=	0
    for b in range(2**8    ):
        if b not in bs:
            bs.append(_a    )
            cs.append(2**8 + n    )
            n += 1
    snake_case_ :     Optional[Any]				=	[chr(_a    ) for n in cs]
    return dict(zip(_a		,			_a    )    )




def __lowercase			(							_a    ):
    snake_case_ :     Any				=	set()
    snake_case_ :     Any				=	word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)    )
        snake_case_ :     Optional[Any]				=	char
    return pairs






class        _UpperCAmelCase       (			lowerCAmelCase__):
 _lowerCAmelCase   :			List[str]							=		VOCAB_FILES_NAMES
 _lowerCAmelCase   :			Any							=		PRETRAINED_VOCAB_FILES_MAP
 _lowerCAmelCase   :			Tuple							=		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _lowerCAmelCase   :			Optional[int]							=		["""input_ids""", """attention_mask"""]

 def __init__(     self		:       Optional[Any]  ,     lowercase_		:       str  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       int="replace"  ,     lowercase_		:       Any="<s>"  ,     lowercase_		:       List[str]="</s>"  ,     lowercase_		:       int="</s>"  ,     lowercase_		:       List[str]="<s>"  ,     lowercase_		:       Tuple="<unk>"  ,     lowercase_		:       Optional[int]="<pad>"  ,     lowercase_		:       Union[str, Any]="<mask>"  ,     lowercase_		:       Dict=False  ,     **lowercase_		:       Optional[Any]  ,     ):
     snake_case_ :     Tuple				=	AddedToken(lowercase_  ,     lstrip=lowercase_  ,     rstrip=lowercase_				) if isinstance(lowercase_  ,     lowercase_				) else bos_token
     snake_case_ :     Optional[int]				=	AddedToken(lowercase_  ,     lstrip=lowercase_  ,     rstrip=lowercase_				) if isinstance(lowercase_  ,     lowercase_				) else eos_token
     snake_case_ :     str				=	AddedToken(lowercase_  ,     lstrip=lowercase_  ,     rstrip=lowercase_				) if isinstance(lowercase_  ,     lowercase_				) else sep_token
     snake_case_ :     int				=	AddedToken(lowercase_  ,     lstrip=lowercase_  ,     rstrip=lowercase_				) if isinstance(lowercase_  ,     lowercase_				) else cls_token
     snake_case_ :     int				=	AddedToken(lowercase_  ,     lstrip=lowercase_  ,     rstrip=lowercase_				) if isinstance(lowercase_  ,     lowercase_				) else unk_token
     snake_case_ :     Dict				=	AddedToken(lowercase_  ,     lstrip=lowercase_  ,     rstrip=lowercase_				) if isinstance(lowercase_  ,     lowercase_				) else pad_token

     # Mask token behave like a normal word, i.e. include the space before it
     snake_case_ :     Optional[int]				=	AddedToken(lowercase_  ,     lstrip=lowercase_  ,     rstrip=lowercase_				) if isinstance(lowercase_  ,     lowercase_				) else mask_token

     super().__init__(
         errors=lowercase_  ,     bos_token=lowercase_  ,     eos_token=lowercase_  ,     unk_token=lowercase_  ,     sep_token=lowercase_  ,     cls_token=lowercase_  ,     pad_token=lowercase_  ,     mask_token=lowercase_  ,     add_prefix_space=lowercase_  ,     **lowercase_  ,     )

     with open(lowercase_  ,     encoding='''utf-8'''				) as vocab_handle:
         snake_case_ :     Tuple				=	json.load(lowercase_				)
     snake_case_ :     Dict				=	{v: k for k, v in self.encoder.items()}
     snake_case_ :     Any				=	errors  # how to handle errors in decoding
     snake_case_ :     Union[str, Any]				=	bytes_to_unicode()
     snake_case_ :     Tuple				=	{v: k for k, v in self.byte_encoder.items()}
     with open(lowercase_  ,     encoding='''utf-8'''				) as merges_handle:
         snake_case_ :     List[Any]				=	merges_handle.read().split('''\n'''				)[1:-1]
     snake_case_ :     List[Any]				=	[tuple(merge.split()				) for merge in bpe_merges]
     snake_case_ :     List[str]				=	dict(zip(lowercase_  ,     range(len(lowercase_				)				)				)				)
     snake_case_ :     int				=	{}
     snake_case_ :     str				=	add_prefix_space

     # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
     snake_case_ :     int				=	re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''				)

 @property
 # Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
 def 						_snake_case					(     self		:       Union[str, Any]				):
     return len(self.encoder				)

 def 						_snake_case					(     self		:       Optional[Any]				):
     return dict(self.encoder  ,     **self.added_tokens_encoder				)

 def 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       Tuple				):
     if token in self.cache:
         return self.cache[token]
     snake_case_ :     Optional[Any]				=	tuple(lowercase_				)
     snake_case_ :     Dict				=	get_pairs(lowercase_				)

     if not pairs:
         return token

     while True:
         snake_case_ :     Optional[int]				=	min(lowercase_  ,     key=lambda lowercase_				: self.bpe_ranks.get(lowercase_  ,     float('''inf'''				)				)				)
         if bigram not in self.bpe_ranks:
             break
         snake_case_,	snake_case_ :     int				=	bigram
         snake_case_ :     Any				=	[]
         snake_case_ :     List[str]				=	0
         while i < len(lowercase_				):
             try:
                 snake_case_ :     Union[str, Any]				=	word.index(lowercase_  ,     lowercase_				)
             except ValueError:
                 new_word.extend(word[i:]				)
                 break
             else:
                 new_word.extend(word[i:j]				)
                 snake_case_ :     Union[str, Any]				=	j

             if word[i] == first and i < len(lowercase_				) - 1 and word[i + 1] == second:
                 new_word.append(first + second				)
                 i += 2
             else:
                 new_word.append(word[i]				)
                 i += 1
         snake_case_ :     Tuple				=	tuple(lowercase_				)
         snake_case_ :     Any				=	new_word
         if len(lowercase_				) == 1:
             break
         else:
             snake_case_ :     Optional[int]				=	get_pairs(lowercase_				)
     snake_case_ :     Dict				=	''' '''.join(lowercase_				)
     snake_case_ :     Optional[Any]				=	word
     return word

 def 						_snake_case					(     self		:       Any  ,     lowercase_		:       Union[str, Any]				):
     snake_case_ :     Dict				=	[]
     for token in re.findall(self.pat  ,     lowercase_				):
         snake_case_ :     str				=	''''''.join(
             self.byte_encoder[b] for b in token.encode('''utf-8'''				)				)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
         bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowercase_				).split(''' '''				)				)
     return bpe_tokens

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       List[str]				):
     return self.encoder.get(lowercase_  ,     self.encoder.get(self.unk_token				)				)

 def 						_snake_case					(     self		:       Union[str, Any]  ,     lowercase_		:       Dict				):
     return self.decoder.get(lowercase_				)

 def 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       Union[str, Any]				):
     snake_case_ :     Optional[Any]				=	''''''.join(lowercase_				)
     snake_case_ :     Any				=	bytearray([self.byte_decoder[c] for c in text]				).decode('''utf-8'''  ,     errors=self.errors				)
     return text

 def 						_snake_case					(     self		:       Optional[int]  ,     lowercase_		:       str  ,     lowercase_		:       Optional[str] = None				):
     if not os.path.isdir(lowercase_				):
         logger.error(f"Vocabulary path ({save_directory}) should be a directory"				)
         return
     snake_case_ :     str				=	os.path.join(
         lowercase_  ,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']				)
     snake_case_ :     int				=	os.path.join(
         lowercase_  ,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']				)

     with open(lowercase_  ,     '''w'''  ,     encoding='''utf-8'''				) as f:
         f.write(json.dumps(self.encoder  ,     indent=2  ,     sort_keys=lowercase_  ,     ensure_ascii=lowercase_				) + '''\n'''				)

     snake_case_ :     int				=	0
     with open(lowercase_  ,     '''w'''  ,     encoding='''utf-8'''				) as writer:
         writer.write('''#version: 0.2\n'''				)
         for bpe_tokens, token_index in sorted(self.bpe_ranks.items()  ,     key=lambda lowercase_				: kv[1]				):
             if index != token_index:
                 logger.warning(
                     f"Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."
                     ''' Please check that the tokenizer is not corrupted!'''				)
                 snake_case_ :     Union[str, Any]				=	token_index
             writer.write(''' '''.join(lowercase_				) + '''\n'''				)
             index += 1

     return vocab_file, merge_file

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       List[int]  ,     lowercase_		:       Optional[List[int]] = None				):
     if token_ids_a is None:
         return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
     snake_case_ :     Union[str, Any]				=	[self.cls_token_id]
     snake_case_ :     Union[str, Any]				=	[self.sep_token_id]
     return cls + token_ids_a + sep + sep + token_ids_a + sep

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

     if token_ids_a is None:
         return [1] + ([0] * len(lowercase_				)) + [1]
     return [1] + ([0] * len(lowercase_				)) + [1, 1] + ([0] * len(lowercase_				)) + [1]

 def 						_snake_case					(     self		:       int  ,     lowercase_		:       List[int]  ,     lowercase_		:       Optional[List[int]] = None				):
     snake_case_ :     Optional[Any]				=	[self.sep_token_id]
     snake_case_ :     List[Any]				=	[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]

 def 						_snake_case					(     self		:       Union[str, Any]  ,     lowercase_		:       str  ,     lowercase_		:       str=False  ,     **lowercase_		:       Dict				):
     snake_case_ :     Tuple				=	kwargs.pop('''add_prefix_space'''  ,     self.add_prefix_space				)
     if (is_split_into_words or add_prefix_space) and (len(lowercase_				) > 0 and not text[0].isspace()):
         snake_case_ :     List[str]				=	''' ''' + text
     return (text, kwargs)






 def 						_snake_case					(     self		:       Any  ,     lowercase_		:       Union[Dict[str, EncodedInput], BatchEncoding]  ,     lowercase_		:       Optional[int] = None  ,     lowercase_		:       PaddingStrategy = PaddingStrategy.DO_NOT_PAD  ,     lowercase_		:       Optional[int] = None  ,     lowercase_		:       Optional[bool] = None  ,     ):
     snake_case_ :     str				=	super()._pad(
         encoded_inputs=lowercase_  ,     max_length=lowercase_  ,     padding_strategy=lowercase_  ,     pad_to_multiple_of=lowercase_  ,     return_attention_mask=lowercase_  ,     )

     # Load from model defaults
     if return_attention_mask is None:
         snake_case_ :     int				=	'''attention_mask''' in self.model_input_names

     if return_attention_mask and "global_attention_mask" in encoded_inputs:
         snake_case_ :     Optional[Any]				=	encoded_inputs[self.model_input_names[0]]
         # `global_attention_mask` need to have the same length as other (sequential) inputs.
         snake_case_ :     Any				=	len(encoded_inputs['''global_attention_mask''']				) != len(lowercase_				)

         if needs_to_be_padded:
             snake_case_ :     int				=	len(lowercase_				) - len(encoded_inputs['''global_attention_mask''']				)

             if self.padding_side == "right":
                 # Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
                 snake_case_ :     List[Any]				=	(
                     encoded_inputs['''global_attention_mask'''] + [-1] * difference
                 )
             elif self.padding_side == "left":
                 snake_case_ :     int				=	[-1] * difference + encoded_inputs[
                     '''global_attention_mask'''
                 ]
             else:
                 raise ValueError('''Invalid padding strategy:''' + str(self.padding_side				)				)

     return encoded_inputs










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

 def __init__(     self		:       Optional[int]				):
     snake_case_ :     str				=	[]

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       Any  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       List[str]  ,     **lowercase_		:       Tuple				):
     self.events.append('''on_init_end'''				)

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       str  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]  ,     **lowercase_		:       List[str]				):
     self.events.append('''on_train_begin'''				)

 def 						_snake_case					(     self		:       Any  ,     lowercase_		:       List[str]  ,     lowercase_		:       Tuple  ,     lowercase_		:       List[Any]  ,     **lowercase_		:       Optional[int]				):
     self.events.append('''on_train_end'''				)

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       int  ,     lowercase_		:       Optional[Any]  ,     **lowercase_		:       List[Any]				):
     self.events.append('''on_epoch_begin'''				)

 def 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       List[str]  ,     lowercase_		:       Dict  ,     lowercase_		:       Union[str, Any]  ,     **lowercase_		:       Optional[Any]				):
     self.events.append('''on_epoch_end'''				)

 def 						_snake_case					(     self		:       List[str]  ,     lowercase_		:       Optional[Any]  ,     lowercase_		:       Optional[Any]  ,     lowercase_		:       int  ,     **lowercase_		:       Optional[Any]				):
     self.events.append('''on_step_begin'''				)

 def 						_snake_case					(     self		:       int  ,     lowercase_		:       int  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       List[Any]  ,     **lowercase_		:       List[str]				):
     self.events.append('''on_step_end'''				)

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       int  ,     lowercase_		:       Dict  ,     lowercase_		:       List[str]  ,     **lowercase_		:       List[str]				):
     self.events.append('''on_evaluate'''				)

 def 						_snake_case					(     self		:       Dict  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       Any  ,     lowercase_		:       List[Any]  ,     **lowercase_		:       str				):
     self.events.append('''on_predict'''				)

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       List[Any]  ,     lowercase_		:       int  ,     **lowercase_		:       Union[str, Any]				):
     self.events.append('''on_save'''				)

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       Tuple  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]  ,     **lowercase_		:       Any				):
     self.events.append('''on_log'''				)






 def 						_snake_case					(     self		:       Dict  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]  ,     lowercase_		:       Union[str, Any]  ,     **lowercase_		:       Optional[int]				):
     self.events.append('''on_prediction_step'''				)





@require_torch
class        _UpperCAmelCase       (			unittest.TestCase):

 def 						_snake_case					(     self		:       List[str]				):
     snake_case_ :     Tuple				=	tempfile.mkdtemp()

 def 						_snake_case					(     self		:       Tuple				):
     shutil.rmtree(self.output_dir				)

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

     snake_case_ :     Any				=	TrainingArguments(self.output_dir  ,     disable_tqdm=lowercase_  ,     report_to=[]  ,     **lowercase_				)
     return Trainer(
         lowercase_  ,     lowercase_  ,     train_dataset=lowercase_  ,     eval_dataset=lowercase_  ,     callbacks=lowercase_  ,     )

 def 						_snake_case					(     self		:       Optional[int]  ,     lowercase_		:       Any  ,     lowercase_		:       List[Any]				):
     self.assertEqual(len(lowercase_				)  ,     len(lowercase_				)				)

     # Order doesn't matter
     snake_case_ :     Any				=	sorted(lowercase_  ,     key=lambda lowercase_				: cb.__name__ if isinstance(lowercase_  ,     lowercase_				) else cb.__class__.__name__				)
     snake_case_ :     List[str]				=	sorted(lowercase_  ,     key=lambda lowercase_				: cb.__name__ if isinstance(lowercase_  ,     lowercase_				) else cb.__class__.__name__				)

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

 def 						_snake_case					(     self		:       Optional[Any]  ,     lowercase_		:       Tuple				):
     snake_case_ :     Tuple				=	['''on_init_end''', '''on_train_begin''']
     snake_case_ :     List[Any]				=	0
     snake_case_ :     Union[str, Any]				=	len(trainer.get_eval_dataloader()				)
     snake_case_ :     List[Any]				=	['''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(lowercase_				):
             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 						_snake_case					(     self		:       List[str]				):
     snake_case_ :     Union[str, Any]				=	self.get_trainer()
     snake_case_ :     Dict				=	DEFAULT_CALLBACKS.copy() + [ProgressCallback]
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     # Callbacks passed at init are added to the default callbacks
     snake_case_ :     Optional[Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]				)
     expected_callbacks.append(lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

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

 def 						_snake_case					(     self		:       int				):
     snake_case_ :     int				=	DEFAULT_CALLBACKS.copy() + [ProgressCallback]
     snake_case_ :     List[Any]				=	self.get_trainer()

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

     snake_case_ :     Dict				=	self.get_trainer()
     snake_case_ :     Optional[int]				=	trainer.pop_callback(lowercase_				)
     self.assertEqual(cb.__class__  ,     lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

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

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

     snake_case_ :     List[Any]				=	self.get_trainer()
     snake_case_ :     Optional[int]				=	trainer.callback_handler.callbacks[0]
     snake_case_ :     Optional[Any]				=	trainer.pop_callback(lowercase_				)
     self.assertEqual(lowercase_  ,     lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

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






 def 						_snake_case					(     self		:       List[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=lowercase_				)

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

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

     snake_case_ :     List[Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     save_steps=5				)
     trainer.train()
     snake_case_ :     int				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     snake_case_ :     List[Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     eval_steps=5  ,     evaluation_strategy='''steps'''				)
     trainer.train()
     snake_case_ :     Union[str, Any]				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

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

     # A bit of everything
     snake_case_ :     str				=	self.get_trainer(
         callbacks=[MyTestTrainerCallback]  ,     logging_steps=3  ,     save_steps=10  ,     eval_steps=5  ,     evaluation_strategy='''steps'''  ,     )
     trainer.train()
     snake_case_ :     str				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     # warning should be emitted for duplicated callbacks
     with patch('''transformers.trainer_callback.logger.warning'''				) as warn_mock:
         snake_case_ :     Dict				=	self.get_trainer(
             callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]  ,     )
         assert str(lowercase_				) in warn_mock.call_args[0][0]





