





import unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from transformers import AutoConfig, TFGPTaLMHeadModel, is_keras_nlp_available, is_tf_available
from transformers.models.gpta.tokenization_gpta import GPTaTokenizer
from transformers.testing_utils import require_keras_nlp, require_tf, slow


if is_tf_available():
      import tensorflow as tf

if is_keras_nlp_available():
      from transformers.models.gpta import TFGPTaTokenizer


lowerCAmelCase_						   =    ["""gpt2"""]
lowerCAmelCase_						   =    """gpt2"""

if is_tf_available():



      class        lowerCAmelCase_    (			tf.Module							):


        '''simple docstring'''







        def __init__(      self      :					List[str]      ,				UpperCamelCase      :					Dict							):


          '''simple docstring'''





          super().__init__()
          _snake_case   :				Optional[int]														=	tokenizer
          _snake_case   :				Union[str, Any]														=	AutoConfig.from_pretrained(UpperCamelCase							)
          _snake_case   :				int														=	TFGPTaLMHeadModel.from_config(UpperCamelCase							)







        @tf.function(input_signature=(tf.TensorSpec((None,)      ,				tf.string      ,				name='text'							),)							)
        def UpperCamelCase_ (      self      :					Optional[int]      ,				UpperCamelCase      :					Optional[int]							):


          '''simple docstring'''





          _snake_case   :				Dict														=	self.tokenizer(UpperCamelCase							)
          _snake_case   :				Union[str, Any]														=	tokenized['input_ids'].to_tensor()

          _snake_case   :				Any														=	tf.cast(input_ids_dense > 0      ,				tf.intaa							)
          # input_mask = tf.reshape(input_mask, [-1, MAX_SEQ_LEN])

          _snake_case   :				Tuple														=	self.model(input_ids=UpperCamelCase      ,				attention_mask=UpperCamelCase							)['logits']

          return outputs



@require_tf
@require_keras_nlp
class        lowerCAmelCase_    (			unittest.TestCase							):


  '''simple docstring'''







  def UpperCamelCase_ (      self      :					Any							):


    '''simple docstring'''





    super().setUp()

    _snake_case   :				Optional[int]														=	[GPTaTokenizer.from_pretrained(UpperCamelCase							) for checkpoint in (TOKENIZER_CHECKPOINTS)]
    _snake_case   :				Tuple														=	[TFGPTaTokenizer.from_pretrained(UpperCamelCase							) for checkpoint in TOKENIZER_CHECKPOINTS]
    assert len(self.tokenizers							) == len(self.tf_tokenizers							)

    _snake_case   :				Any														=	[
        'This is a straightforward English test sentence.',
        'This one has some weird characters\rto\nsee\r\nif  those\u00E9break things.',
        'Now we\'re going to add some Chinese: 一 二 三 一二三',
        'And some much more rare Chinese: 齉 堃 齉堃',
        'Je vais aussi écrire en français pour tester les accents',
        'Classical Irish also has some unusual characters, so in they go: Gaelaċ, ꝼ',
    ]
    _snake_case   :				Tuple														=	list(zip(self.test_sentences      ,				self.test_sentences[::-1]							)							)







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


    '''simple docstring'''





    for tokenizer, tf_tokenizer in zip(self.tokenizers      ,				self.tf_tokenizers							):
      for test_inputs in self.test_sentences:
        _snake_case   :				Optional[int]														=	tokenizer([test_inputs]      ,				return_tensors='tf'							)
        _snake_case   :				Tuple														=	tf_tokenizer([test_inputs]							)

        for key in python_outputs.keys():
          # convert them to numpy to avoid messing with ragged tensors
          _snake_case   :				Dict														=	python_outputs[key].numpy()
          _snake_case   :				Optional[Any]														=	tf_outputs[key].numpy()

          self.assertTrue(tf.reduce_all(python_outputs_values.shape == tf_outputs_values.shape							)							)
          self.assertTrue(tf.reduce_all(tf.cast(UpperCamelCase      ,				tf.intaa							) == tf_outputs_values							)							)







  @slow
  def UpperCamelCase_ (      self      :					Optional[Any]							):


    '''simple docstring'''





    for tf_tokenizer in self.tf_tokenizers:
      _snake_case   :				str														=	tf.function(UpperCamelCase							)
      for test_inputs in self.test_sentences:
        _snake_case   :				int														=	tf.constant(UpperCamelCase							)
        _snake_case   :				Tuple														=	compiled_tokenizer(UpperCamelCase							)
        _snake_case   :				int														=	tf_tokenizer(UpperCamelCase							)

        for key in eager_outputs.keys():
          self.assertTrue(tf.reduce_all(eager_outputs[key] == compiled_outputs[key]							)							)







  @slow
  def UpperCamelCase_ (      self      :					Tuple							):


    '''simple docstring'''





    for tf_tokenizer in self.tf_tokenizers:
      _snake_case   :				Union[str, Any]														=	ModelToSave(tokenizer=UpperCamelCase							)
      _snake_case   :				Optional[Any]														=	tf.convert_to_tensor([self.test_sentences[0]]							)
      _snake_case   :				Tuple														=	model.serving(UpperCamelCase							)  # Build model with some sample inputs
      with TemporaryDirectory() as tempdir:
        _snake_case   :				str														=	Path(UpperCamelCase							) / 'saved.model'
        tf.saved_model.save(UpperCamelCase      ,				UpperCamelCase      ,				signatures={'serving_default': model.serving}							)
        _snake_case   :				Optional[int]														=	tf.saved_model.load(UpperCamelCase							)
      _snake_case   :				List[str]														=	loaded_model.signatures['serving_default'](UpperCamelCase							)['output_0']
      # We may see small differences because the loaded model is compiled, so we need an epsilon for the test
      self.assertTrue(tf.reduce_all(out == loaded_output							)							)







  @slow
  def UpperCamelCase_ (      self      :					Optional[int]							):


    '''simple docstring'''





    for tf_tokenizer in self.tf_tokenizers:
      _snake_case   :				Optional[Any]														=	tf.convert_to_tensor([self.test_sentences[0]]							)
      _snake_case   :				Any														=	tf_tokenizer(UpperCamelCase							)  # Build model with some sample inputs

      _snake_case   :				Optional[Any]														=	tf_tokenizer.get_config()
      _snake_case   :				Tuple														=	TFGPTaTokenizer.from_config(UpperCamelCase							)
      _snake_case   :				Optional[Any]														=	model_from_config(UpperCamelCase							)

      for key in from_config_output.keys():
        self.assertTrue(tf.reduce_all(from_config_output[key] == out[key]							)							)







  @slow
  def UpperCamelCase_ (      self      :					Any							):


    '''simple docstring'''





    for tf_tokenizer in self.tf_tokenizers:
      # for the test to run
      _snake_case   :				Union[str, Any]														=	12_31_23

      for max_length in [3, 5, 10_24]:
        _snake_case   :				Union[str, Any]														=	tf.convert_to_tensor([self.test_sentences[0]]							)
        _snake_case   :				List[str]														=	tf_tokenizer(UpperCamelCase      ,				max_length=UpperCamelCase							)

        _snake_case   :				int														=	out['input_ids'].numpy().shape[1]

        assert out_length == max_length











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     lowerCamelCase_	(      lowerCAmelCase:	List[Any]    ,							lowerCAmelCase:	Optional[Any]    ,							lowerCAmelCase:	Dict    ,							lowerCAmelCase:	Union[str, Any]				)->		Optional[int]:
  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:
    _snake_case   :				Tuple														=	TOKENIZER_CLASSES
  else:
    _snake_case   :				Union[str, Any]														=	{tokenizer_name: getattr(lowerCAmelCase    ,							tokenizer_name + 'Fast'				)}

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

  for tokenizer_name in tokenizer_names:
    _snake_case   :				Dict														=	TOKENIZER_CLASSES[tokenizer_name]

    _snake_case   :				Optional[Any]														=	True
    if checkpoint_name is None:
      _snake_case   :				Union[str, Any]														=	list(tokenizer_class.max_model_input_sizes.keys()				)
    else:
      _snake_case   :				Optional[int]														=	[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
      _snake_case   :				str														=	tokenizer_class.from_pretrained(lowerCAmelCase    ,							force_download=lowerCAmelCase				)

      # 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:
        _snake_case							,					_snake_case   :				Tuple														=	checkpoint.split('/'				)
        _snake_case   :				int														=	os.path.join(lowerCAmelCase    ,							lowerCAmelCase				)
      elif add_prefix:
        _snake_case   :				Dict														=	checkpoint
        _snake_case   :				Optional[Any]														=	dump_path
      else:
        _snake_case   :				str														=	None
        _snake_case   :				Union[str, Any]														=	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]:
        _snake_case   :				Optional[Any]														=	list(tokenizer.pretrained_vocab_files_map.values()				)[0][checkpoint]
        _snake_case   :				Optional[int]														=	file_path.split(lowerCAmelCase				)[-1][0]
        if next_char == "/":
          _snake_case   :				Union[str, Any]														=	os.path.join(lowerCAmelCase    ,							lowerCAmelCase				)
          _snake_case   :				str														=	None

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

      _snake_case   :				Optional[int]														=	tokenizer.save_pretrained(
          lowerCAmelCase    ,							legacy_format=lowerCAmelCase    ,							filename_prefix=lowerCAmelCase				)
      logger.info(F"""=> File names {file_names}"""				)

      for file_name in file_names:
        if not file_name.endswith('tokenizer.json'				):
          os.remove(lowerCAmelCase				)
          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)




