




'''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 		lowercase_   (			A       ):


			"""simple docstring"""

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

						"""simple docstring"""
						_SCREAMING_SNAKE_CASE	=				[]



			def 							lowerCAmelCase_      (   self :     Tuple  ,				__lowerCamelCase :     List[str]  ,				__lowerCamelCase :     List[Any]  ,				__lowerCamelCase :     int  ,				**__lowerCamelCase :     Union[str, Any]   ):

						"""simple docstring"""
						self.events.append("on_init_end"   )



			def 							lowerCAmelCase_      (   self :     List[str]  ,				__lowerCamelCase :     str  ,				__lowerCamelCase :     List[Any]  ,				__lowerCamelCase :     int  ,				**__lowerCamelCase :     Dict   ):

						"""simple docstring"""
						self.events.append("on_train_begin"   )



			def 							lowerCAmelCase_      (   self :     List[Any]  ,				__lowerCamelCase :     List[Any]  ,				__lowerCamelCase :     str  ,				__lowerCamelCase :     Optional[Any]  ,				**__lowerCamelCase :     Union[str, Any]   ):

						"""simple docstring"""
						self.events.append("on_train_end"   )



			def 							lowerCAmelCase_      (   self :     int  ,				__lowerCamelCase :     Union[str, Any]  ,				__lowerCamelCase :     Optional[Any]  ,				__lowerCamelCase :     Optional[Any]  ,				**__lowerCamelCase :     Union[str, Any]   ):

						"""simple docstring"""
						self.events.append("on_epoch_begin"   )



			def 							lowerCAmelCase_      (   self :     Union[str, Any]  ,				__lowerCamelCase :     Tuple  ,				__lowerCamelCase :     Optional[int]  ,				__lowerCamelCase :     Any  ,				**__lowerCamelCase :     int   ):

						"""simple docstring"""
						self.events.append("on_epoch_end"   )



			def 							lowerCAmelCase_      (   self :     List[Any]  ,				__lowerCamelCase :     Dict  ,				__lowerCamelCase :     Dict  ,				__lowerCamelCase :     Any  ,				**__lowerCamelCase :     Union[str, Any]   ):

						"""simple docstring"""
						self.events.append("on_step_begin"   )



			def 							lowerCAmelCase_      (   self :     List[str]  ,				__lowerCamelCase :     List[str]  ,				__lowerCamelCase :     Tuple  ,				__lowerCamelCase :     Optional[int]  ,				**__lowerCamelCase :     str   ):

						"""simple docstring"""
						self.events.append("on_step_end"   )



			def 							lowerCAmelCase_      (   self :     Dict  ,				__lowerCamelCase :     Optional[int]  ,				__lowerCamelCase :     Any  ,				__lowerCamelCase :     Dict  ,				**__lowerCamelCase :     Tuple   ):

						"""simple docstring"""
						self.events.append("on_evaluate"   )



			def 							lowerCAmelCase_      (   self :     Any  ,				__lowerCamelCase :     List[Any]  ,				__lowerCamelCase :     Optional[Any]  ,				__lowerCamelCase :     Any  ,				**__lowerCamelCase :     int   ):

						"""simple docstring"""
						self.events.append("on_predict"   )



			def 							lowerCAmelCase_      (   self :     Dict  ,				__lowerCamelCase :     Optional[Any]  ,				__lowerCamelCase :     List[Any]  ,				__lowerCamelCase :     Optional[Any]  ,				**__lowerCamelCase :     Any   ):

						"""simple docstring"""
						self.events.append("on_save"   )



			def 							lowerCAmelCase_      (   self :     int  ,				__lowerCamelCase :     int  ,				__lowerCamelCase :     Any  ,				__lowerCamelCase :     List[str]  ,				**__lowerCamelCase :     List[str]   ):

						"""simple docstring"""
						self.events.append("on_log"   )



			def 							lowerCAmelCase_      (   self :     Dict  ,				__lowerCamelCase :     Union[str, Any]  ,				__lowerCamelCase :     List[str]  ,				__lowerCamelCase :     List[Any]  ,				**__lowerCamelCase :     List[str]   ):

						"""simple docstring"""
						self.events.append("on_prediction_step"   )







@require_torch
class 		lowercase_   (			unittest.TestCase       ):


			"""simple docstring"""

			def 							lowerCAmelCase_      (   self :     Optional[int]   ):

						"""simple docstring"""
						_SCREAMING_SNAKE_CASE	=				tempfile.mkdtemp()



			def 							lowerCAmelCase_      (   self :     Union[str, Any]   ):

						"""simple docstring"""
						shutil.rmtree(self.output_dir   )



			def 							lowerCAmelCase_      (   self :     List[Any]  ,				__lowerCamelCase :     Tuple=0  ,				__lowerCamelCase :     Tuple=0  ,				__lowerCamelCase :     Any=6_4  ,				__lowerCamelCase :     Optional[Any]=6_4  ,				__lowerCamelCase :     Optional[Any]=None  ,				__lowerCamelCase :     Union[str, Any]=False  ,				**__lowerCamelCase :     Optional[Any]   ):

						"""simple docstring"""
						_SCREAMING_SNAKE_CASE	=				RegressionDataset(length=__lowerCamelCase   )
						_SCREAMING_SNAKE_CASE	=				RegressionDataset(length=__lowerCamelCase   )
						_SCREAMING_SNAKE_CASE	=				RegressionModelConfig(a=__lowerCamelCase  ,				b=__lowerCamelCase   )
						_SCREAMING_SNAKE_CASE	=				RegressionPreTrainedModel(__lowerCamelCase   )

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



			def 							lowerCAmelCase_      (   self :     Optional[int]  ,				__lowerCamelCase :     Tuple  ,				__lowerCamelCase :     Dict   ):

						"""simple docstring"""
						self.assertEqual(len(__lowerCamelCase   )  ,				len(__lowerCamelCase   )   )

						# Order doesn't matter
						_SCREAMING_SNAKE_CASE	=				sorted(__lowerCamelCase  ,				key=lambda __lowerCamelCase   : cb.__name__ if isinstance(__lowerCamelCase  ,				__lowerCamelCase   ) else cb.__class__.__name__   )
						_SCREAMING_SNAKE_CASE	=				sorted(__lowerCamelCase  ,				key=lambda __lowerCamelCase   : 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 							lowerCAmelCase_      (   self :     str  ,				__lowerCamelCase :     Dict   ):

						"""simple docstring"""
						_SCREAMING_SNAKE_CASE	=				["on_init_end", "on_train_begin"]
						_SCREAMING_SNAKE_CASE	=				0
						_SCREAMING_SNAKE_CASE	=				len(trainer.get_eval_dataloader()   )
						_SCREAMING_SNAKE_CASE	=				["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 							lowerCAmelCase_      (   self :     List[str]   ):

						"""simple docstring"""
						_SCREAMING_SNAKE_CASE	=				self.get_trainer()
						_SCREAMING_SNAKE_CASE	=				DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						self.check_callbacks_equality(trainer.callback_handler.callbacks  ,				__lowerCamelCase   )

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



			def 							lowerCAmelCase_      (   self :     List[str]   ):

						"""simple docstring"""
						_SCREAMING_SNAKE_CASE	=				DEFAULT_CALLBACKS.copy() + [ProgressCallback]
						_SCREAMING_SNAKE_CASE	=				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   )

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

						_SCREAMING_SNAKE_CASE	=				self.get_trainer()
						_SCREAMING_SNAKE_CASE	=				trainer.callback_handler.callbacks[0]
						_SCREAMING_SNAKE_CASE	=				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 							lowerCAmelCase_      (   self :     List[str]   ):

						"""simple docstring"""
						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   )

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

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

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

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

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

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







'''simple docstring'''



import argparse
import os

import transformers

from .convert_slow_tokenizer import SLOW_TO_FAST_CONVERTERS
from .utils import logging


logging.set_verbosity_info()

lowerCamelCase_			=						logging.get_logger(__name__)


lowerCamelCase_			=						{name: getattr(transformers, name + 'Fast') for name in SLOW_TO_FAST_CONVERTERS}
def 			SCREAMING_SNAKE_CASE_			(							__A    :			Optional[int]	,					__A    :			Tuple	,					__A    :			List[str]	,					__A    :			List[str]  )			->			List[Any]:
   if tokenizer_name is not None and tokenizer_name not in TOKENIZER_CLASSES:
      raise ValueError(f"""Unrecognized tokenizer name, should be one of {list(TOKENIZER_CLASSES.keys()  )}."""  )

   if tokenizer_name is None:
      _SCREAMING_SNAKE_CASE	=				TOKENIZER_CLASSES
   else:
      _SCREAMING_SNAKE_CASE	=				{tokenizer_name: getattr(__A	,					tokenizer_name + "Fast"  )}

   logger.info(f"""Loading tokenizer classes: {tokenizer_names}"""  )

   for tokenizer_name in tokenizer_names:
      _SCREAMING_SNAKE_CASE	=				TOKENIZER_CLASSES[tokenizer_name]

      _SCREAMING_SNAKE_CASE	=				True
      if checkpoint_name is None:
         _SCREAMING_SNAKE_CASE	=				list(tokenizer_class.max_model_input_sizes.keys()  )
      else:
         _SCREAMING_SNAKE_CASE	=				[checkpoint_name]

      logger.info(f"""For tokenizer {tokenizer_class.__class__.__name__} loading checkpoints: {checkpoint_names}"""  )

      for checkpoint in checkpoint_names:
         logger.info(f"""Loading {tokenizer_class.__class__.__name__} {checkpoint}"""  )

         # Load tokenizer
         _SCREAMING_SNAKE_CASE	=				tokenizer_class.from_pretrained(__A	,					force_download=__A  )

         # Save fast tokenizer
         logger.info(f"""Save fast tokenizer to {dump_path} with prefix {checkpoint} add_prefix {add_prefix}"""  )

         # For organization names we create sub-directories
         if "/" in checkpoint:
            _SCREAMING_SNAKE_CASE,	_SCREAMING_SNAKE_CASE	=				checkpoint.split("/"  )
            _SCREAMING_SNAKE_CASE	=				os.path.join(__A	,					__A  )
         elif add_prefix:
            _SCREAMING_SNAKE_CASE	=				checkpoint
            _SCREAMING_SNAKE_CASE	=				dump_path
         else:
            _SCREAMING_SNAKE_CASE	=				None
            _SCREAMING_SNAKE_CASE	=				dump_path

         logger.info(f"""=> {dump_path_full} with prefix {checkpoint_prefix_name}, add_prefix {add_prefix}"""  )

         if checkpoint in list(tokenizer.pretrained_vocab_files_map.values()  )[0]:
            _SCREAMING_SNAKE_CASE	=				list(tokenizer.pretrained_vocab_files_map.values()  )[0][checkpoint]
            _SCREAMING_SNAKE_CASE	=				file_path.split(__A  )[-1][0]
            if next_char == "/":
               _SCREAMING_SNAKE_CASE	=				os.path.join(__A	,					__A  )
               _SCREAMING_SNAKE_CASE	=				None

            logger.info(f"""=> {dump_path_full} with prefix {checkpoint_prefix_name}, add_prefix {add_prefix}"""  )

         _SCREAMING_SNAKE_CASE	=				tokenizer.save_pretrained(
             __A	,					legacy_format=__A	,					filename_prefix=__A  )
         logger.info(f"""=> File names {file_names}"""  )

         for file_name in file_names:
            if not file_name.endswith("tokenizer.json"  ):
               os.remove(__A  )
               logger.info(f"""=> removing {file_name}"""  )


if __name__ == "__main__":
 lowerCamelCase_			=						argparse.ArgumentParser()
 # Required parameters
 parser.add_argument(
     '--dump_path', default=None, type=str, required=True, help='Path to output generated fast tokenizer files.'
 )
 parser.add_argument(
     '--tokenizer_name',
     default=None,
     type=str,
     help=(
         f'''Optional tokenizer type selected in the list of {list(TOKENIZER_CLASSES.keys())}. If not given, will '''
         'download and convert all the checkpoints from AWS.'
     ),
 )
 parser.add_argument(
     '--checkpoint_name',
     default=None,
     type=str,
     help='Optional checkpoint name. If not given, will download and convert the canonical checkpoints from AWS.',
 )
 parser.add_argument(
     '--force_download',
     action='store_true',
     help='Re-download checkpoints.',
 )
 lowerCamelCase_			=						parser.parse_args()

 convert_slow_checkpoint_to_fast(args.tokenizer_name, args.checkpoint_name, args.dump_path, args.force_download)

