

from typing import List, Optional

from tokenizers import ByteLevelBPETokenizer

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_blenderbot_small import BlenderbotSmallTokenizer


_snake_case						: List[str]											=							logging.get_logger(__name__)

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

_snake_case						: Optional[int]											=							{
    'vocab_file': {
        'facebook/blenderbot_small-90M': 'https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/vocab.json'
    },
    'merges_file': {
        'facebook/blenderbot_small-90M': 'https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/merges.txt'
    },
    'tokenizer_config_file': {
        'facebook/blenderbot_small-90M': (
            'https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/tokenizer_config.json'
        )
    },
}

_snake_case						: Dict											=							{
    'facebook/blenderbot_small-90M': 5_12,
}





class __SCREAMING_SNAKE_CASE     (	__UpperCAmelCase		):
    SCREAMING_SNAKE_CASE__      =VOCAB_FILES_NAMES
    SCREAMING_SNAKE_CASE__      =PRETRAINED_VOCAB_FILES_MAP
    SCREAMING_SNAKE_CASE__      =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    SCREAMING_SNAKE_CASE__      =BlenderbotSmallTokenizer






    def __init__(       self,		_a=None,		_a=None,		_a="<|endoftext|>",		_a="<|endoftext|>",		_a="<|endoftext|>",		_a=False,		_a=True,		**_a,		)   ->					Any:
       super().__init__(
           ByteLevelBPETokenizer(
               vocab=lowerCamelCase__,		merges=lowerCamelCase__,		add_prefix_space=lowerCamelCase__,		trim_offsets=lowerCamelCase__,		),		bos_token=lowerCamelCase__,		eos_token=lowerCamelCase__,		unk_token=lowerCamelCase__,		**lowerCamelCase__,		)
       __SCREAMING_SNAKE_CASE       =				add_prefix_space






    def 						__lowerCAmelCase    (       self,		_a,		_a=None )   ->					Optional[int]:
       __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 						__lowerCAmelCase    (       self,		_a,		_a = None )   ->					List[int]:
       __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]





"""simple docstring"""


import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

from huggingface_hub import HfFolder, Repository, create_repo, delete_repo
from requests.exceptions import HTTPError

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoProcessor,
    AutoTokenizer,
    BertTokenizer,
    ProcessorMixin,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
)
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test
from transformers.tokenization_utils import TOKENIZER_CONFIG_FILE
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_tokenizers_available


sys.path.append(str(Path(__file__).parent.parent.parent.parent / 'utils'))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402
from test_module.custom_processing import CustomProcessor  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


__snake_case    					=					get_tests_dir('fixtures/dummy_feature_extractor_config.json')
__snake_case    					=					get_tests_dir('fixtures/vocab.json')
__snake_case    					=					get_tests_dir('fixtures')

class 				_SCREAMING_SNAKE_CASE				(							unittest.TestCase  ):
   """simple docstring"""



   _a   :       int									=    ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']


   def 					UpperCAmelCase__(						self						)      ->							Any:
    lowercase__       :       Tuple				=						0


   def 					UpperCAmelCase__(						self						)      ->							Optional[int]:
    lowercase__       :       List[str]				=						AutoProcessor.from_pretrained("""facebook/wav2vec2-base-960h"""						)
    self.assertIsInstance(lowerCamelCase__ ,     lowerCamelCase__						)


   def 					UpperCAmelCase__(						self						)      ->							Any:
    with tempfile.TemporaryDirectory() as tmpdirname:
     lowercase__       :       str				=						WavaVecaConfig()
     lowercase__       :       int				=						AutoProcessor.from_pretrained("""facebook/wav2vec2-base-960h"""						)

     # save in new folder
     model_config.save_pretrained(lowerCamelCase__						)
     processor.save_pretrained(lowerCamelCase__						)

     lowercase__       :       Optional[Any]				=						AutoProcessor.from_pretrained(lowerCamelCase__						)

    self.assertIsInstance(lowerCamelCase__ ,     lowerCamelCase__						)


   def 					UpperCAmelCase__(						self						)      ->							Any:
    with tempfile.TemporaryDirectory() as tmpdirname:
     # copy relevant files
     copyfile(lowerCamelCase__ ,     os.path.join(lowerCamelCase__ ,     lowerCamelCase__						)						)
     copyfile(lowerCamelCase__ ,     os.path.join(lowerCamelCase__ ,     """vocab.json"""						)						)

     lowercase__       :       Tuple				=						AutoProcessor.from_pretrained(lowerCamelCase__						)

    self.assertIsInstance(lowerCamelCase__ ,     lowerCamelCase__						)


   def 					UpperCAmelCase__(						self						)      ->							Union[str, Any]:
    with tempfile.TemporaryDirectory() as tmpdirname:
     lowercase__       :       Any				=						WavaVecaFeatureExtractor()
     lowercase__       :       Dict				=						AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h"""						)

     lowercase__       :       int				=						WavaVecaProcessor(lowerCamelCase__ ,     lowerCamelCase__						)

     # save in new folder
     processor.save_pretrained(lowerCamelCase__						)

     # drop `processor_class` in tokenizer
     with open(os.path.join(lowerCamelCase__ ,     lowerCamelCase__						) ,     """r"""						) as f:
      lowercase__       :       str				=						json.load(lowerCamelCase__						)
      config_dict.pop("""processor_class"""						)

     with open(os.path.join(lowerCamelCase__ ,     lowerCamelCase__						) ,     """w"""						) as f:
      f.write(json.dumps(lowerCamelCase__						)						)

     lowercase__       :       Dict				=						AutoProcessor.from_pretrained(lowerCamelCase__						)

    self.assertIsInstance(lowerCamelCase__ ,     lowerCamelCase__						)


   def 					UpperCAmelCase__(						self						)      ->							Any:
    with tempfile.TemporaryDirectory() as tmpdirname:
     lowercase__       :       Dict				=						WavaVecaFeatureExtractor()
     lowercase__       :       int				=						AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h"""						)

     lowercase__       :       List[Any]				=						WavaVecaProcessor(lowerCamelCase__ ,     lowerCamelCase__						)

     # save in new folder
     processor.save_pretrained(lowerCamelCase__						)

     # drop `processor_class` in feature extractor
     with open(os.path.join(lowerCamelCase__ ,     lowerCamelCase__						) ,     """r"""						) as f:
      lowercase__       :       List[str]				=						json.load(lowerCamelCase__						)
      config_dict.pop("""processor_class"""						)

     with open(os.path.join(lowerCamelCase__ ,     lowerCamelCase__						) ,     """w"""						) as f:
      f.write(json.dumps(lowerCamelCase__						)						)

     lowercase__       :       int				=						AutoProcessor.from_pretrained(lowerCamelCase__						)

    self.assertIsInstance(lowerCamelCase__ ,     lowerCamelCase__						)


   def 					UpperCAmelCase__(						self						)      ->							Dict:
    with tempfile.TemporaryDirectory() as tmpdirname:
     lowercase__       :       List[Any]				=						WavaVecaConfig(processor_class="""Wav2Vec2Processor"""						)
     model_config.save_pretrained(lowerCamelCase__						)
     # copy relevant files
     copyfile(lowerCamelCase__ ,     os.path.join(lowerCamelCase__ ,     """vocab.json"""						)						)
     # create emtpy sample processor
     with open(os.path.join(lowerCamelCase__ ,     lowerCamelCase__						) ,     """w"""						) as f:
      f.write("""{}"""						)

     lowercase__       :       Union[str, Any]				=						AutoProcessor.from_pretrained(lowerCamelCase__						)

    self.assertIsInstance(lowerCamelCase__ ,     lowerCamelCase__						)


   def 					UpperCAmelCase__(						self						)      ->							Any:
    # If remote code is not set, we will time out when asking whether to load the model.
    with self.assertRaises(lowerCamelCase__						):
     lowercase__       :       Optional[Any]				=						AutoProcessor.from_pretrained("""hf-internal-testing/test_dynamic_processor"""						)
    # If remote code is disabled, we can't load this config.
    with self.assertRaises(lowerCamelCase__						):
     lowercase__       :       Any				=						AutoProcessor.from_pretrained(
         """hf-internal-testing/test_dynamic_processor""" ,     trust_remote_code=lowerCamelCase__						)

    lowercase__       :       str				=						AutoProcessor.from_pretrained("""hf-internal-testing/test_dynamic_processor""" ,     trust_remote_code=lowerCamelCase__						)
    self.assertTrue(processor.special_attribute_present						)
    self.assertEqual(processor.__class__.__name__ ,     """NewProcessor"""						)

    lowercase__       :       Dict				=						processor.feature_extractor
    self.assertTrue(feature_extractor.special_attribute_present						)
    self.assertEqual(feature_extractor.__class__.__name__ ,     """NewFeatureExtractor"""						)

    lowercase__       :       int				=						processor.tokenizer
    self.assertTrue(tokenizer.special_attribute_present						)
    if is_tokenizers_available():
     self.assertEqual(tokenizer.__class__.__name__ ,     """NewTokenizerFast"""						)

     # Test we can also load the slow version
     lowercase__       :       str				=						AutoProcessor.from_pretrained(
         """hf-internal-testing/test_dynamic_processor""" ,     trust_remote_code=lowerCamelCase__ ,     use_fast=lowerCamelCase__						)
     lowercase__       :       Tuple				=						new_processor.tokenizer
     self.assertTrue(new_tokenizer.special_attribute_present						)
     self.assertEqual(new_tokenizer.__class__.__name__ ,     """NewTokenizer"""						)
    else:
     self.assertEqual(tokenizer.__class__.__name__ ,     """NewTokenizer"""						)


   def 					UpperCAmelCase__(						self						)      ->							str:
    try:
     AutoConfig.register("""custom""" ,     lowerCamelCase__						)
     AutoFeatureExtractor.register(lowerCamelCase__ ,     lowerCamelCase__						)
     AutoTokenizer.register(lowerCamelCase__ ,     slow_tokenizer_class=lowerCamelCase__						)
     AutoProcessor.register(lowerCamelCase__ ,     lowerCamelCase__						)
     # Trying to register something existing in the Transformers library will raise an error
     with self.assertRaises(lowerCamelCase__						):
      AutoProcessor.register(lowerCamelCase__ ,     lowerCamelCase__						)

     # Now that the config is registered, it can be used as any other config with the auto-API
     lowercase__       :       Union[str, Any]				=						CustomFeatureExtractor.from_pretrained(lowerCamelCase__						)

     with tempfile.TemporaryDirectory() as tmp_dir:
      lowercase__       :       Any				=						os.path.join(lowerCamelCase__ ,     """vocab.txt"""						)
      with open(lowerCamelCase__ ,     """w""" ,     encoding="""utf-8"""						) as vocab_writer:
       vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
      lowercase__       :       Tuple				=						CustomTokenizer(lowerCamelCase__						)

     lowercase__       :       Dict				=						CustomProcessor(lowerCamelCase__ ,     lowerCamelCase__						)

     with tempfile.TemporaryDirectory() as tmp_dir:
      processor.save_pretrained(lowerCamelCase__						)
      lowercase__       :       str				=						AutoProcessor.from_pretrained(lowerCamelCase__						)
      self.assertIsInstance(lowerCamelCase__ ,     lowerCamelCase__						)

    finally:
     if "custom" in CONFIG_MAPPING._extra_content:
      del CONFIG_MAPPING._extra_content["custom"]
     if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
      del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]
     if CustomConfig in TOKENIZER_MAPPING._extra_content:
      del TOKENIZER_MAPPING._extra_content[CustomConfig]
     if CustomConfig in PROCESSOR_MAPPING._extra_content:
      del PROCESSOR_MAPPING._extra_content[CustomConfig]


   def 					UpperCAmelCase__(						self						)      ->							Dict:

    class 				_SCREAMING_SNAKE_CASE				(							__UpperCAmelCase  ):
       """simple docstring"""



       _a   :       int									=    False

    class 				_SCREAMING_SNAKE_CASE				(							__UpperCAmelCase  ):
       """simple docstring"""



       _a   :       Union[str, Any]									=    False

    class 				_SCREAMING_SNAKE_CASE				(							__UpperCAmelCase  ):
       """simple docstring"""



       _a   :       Union[str, Any]									=    '''AutoFeatureExtractor'''
       _a   :       List[str]									=    '''AutoTokenizer'''
       _a   :       Optional[Any]									=    False

    try:
     AutoConfig.register("""custom""" ,     lowerCamelCase__						)
     AutoFeatureExtractor.register(lowerCamelCase__ ,     lowerCamelCase__						)
     AutoTokenizer.register(lowerCamelCase__ ,     slow_tokenizer_class=lowerCamelCase__						)
     AutoProcessor.register(lowerCamelCase__ ,     lowerCamelCase__						)
     # If remote code is not set, the default is to use local classes.
     lowercase__       :       Union[str, Any]				=						AutoProcessor.from_pretrained("""hf-internal-testing/test_dynamic_processor"""						)
     self.assertEqual(processor.__class__.__name__ ,     """NewProcessor"""						)
     self.assertFalse(processor.special_attribute_present						)
     self.assertFalse(processor.feature_extractor.special_attribute_present						)
     self.assertFalse(processor.tokenizer.special_attribute_present						)

     # If remote code is disabled, we load the local ones.
     lowercase__       :       Tuple				=						AutoProcessor.from_pretrained(
         """hf-internal-testing/test_dynamic_processor""" ,     trust_remote_code=lowerCamelCase__						)
     self.assertEqual(processor.__class__.__name__ ,     """NewProcessor"""						)
     self.assertFalse(processor.special_attribute_present						)
     self.assertFalse(processor.feature_extractor.special_attribute_present						)
     self.assertFalse(processor.tokenizer.special_attribute_present						)

     # If remote is enabled, we load from the Hub.
     lowercase__       :       Any				=						AutoProcessor.from_pretrained(
         """hf-internal-testing/test_dynamic_processor""" ,     trust_remote_code=lowerCamelCase__						)
     self.assertEqual(processor.__class__.__name__ ,     """NewProcessor"""						)
     self.assertTrue(processor.special_attribute_present						)
     self.assertTrue(processor.feature_extractor.special_attribute_present						)
     self.assertTrue(processor.tokenizer.special_attribute_present						)

    finally:
     if "custom" in CONFIG_MAPPING._extra_content:
      del CONFIG_MAPPING._extra_content["custom"]
     if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
      del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]
     if CustomConfig in TOKENIZER_MAPPING._extra_content:
      del TOKENIZER_MAPPING._extra_content[CustomConfig]
     if CustomConfig in PROCESSOR_MAPPING._extra_content:
      del PROCESSOR_MAPPING._extra_content[CustomConfig]


   def 					UpperCAmelCase__(						self						)      ->							int:
    lowercase__       :       Tuple				=						AutoProcessor.from_pretrained("""hf-internal-testing/tiny-random-bert"""						)
    self.assertEqual(processor.__class__.__name__ ,     """BertTokenizerFast"""						)







   def 					UpperCAmelCase__(						self						)      ->							Any:
    lowercase__       :       Tuple				=						AutoProcessor.from_pretrained("""hf-internal-testing/tiny-random-convnext"""						)
    self.assertEqual(processor.__class__.__name__ ,     """ConvNextImageProcessor"""						)

@is_staging_test
class 				_SCREAMING_SNAKE_CASE				(							unittest.TestCase  ):
   """simple docstring"""



   _a   :       int									=    ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']


   @classmethod
   def 					UpperCAmelCase__(						cls						)      ->							int:
    lowercase__       :       str				=						TOKEN
    HfFolder.save_token(lowerCamelCase__						)


   @classmethod
   def 					UpperCAmelCase__(						cls						)      ->							Dict:
    try:
     delete_repo(token=cls._token ,     repo_id="""test-processor"""						)
    except HTTPError:
     pass

    try:
     delete_repo(token=cls._token ,     repo_id="""valid_org/test-processor-org"""						)
    except HTTPError:
     pass

    try:
     delete_repo(token=cls._token ,     repo_id="""test-dynamic-processor"""						)
    except HTTPError:
     pass


   def 					UpperCAmelCase__(						self						)      ->							int:
    lowercase__       :       List[str]				=						WavaVecaProcessor.from_pretrained(lowerCamelCase__						)
    with tempfile.TemporaryDirectory() as tmp_dir:
     processor.save_pretrained(
         os.path.join(lowerCamelCase__ ,     """test-processor"""						) ,     push_to_hub=lowerCamelCase__ ,     use_auth_token=self._token						)

     lowercase__       :       Optional[Any]				=						WavaVecaProcessor.from_pretrained(F'''{USER}/test-processor'''						)
     for k, v in processor.feature_extractor.__dict__.items():
      self.assertEqual(lowerCamelCase__ ,     getattr(new_processor.feature_extractor ,     lowerCamelCase__						)						)
     self.assertDictEqual(new_processor.tokenizer.get_vocab() ,     processor.tokenizer.get_vocab()						)


   def 					UpperCAmelCase__(						self						)      ->							Optional[Any]:
    lowercase__       :       int				=						WavaVecaProcessor.from_pretrained(lowerCamelCase__						)

    with tempfile.TemporaryDirectory() as tmp_dir:
     processor.save_pretrained(
         os.path.join(lowerCamelCase__ ,     """test-processor-org"""						) ,     push_to_hub=lowerCamelCase__ ,     use_auth_token=self._token ,     organization="""valid_org""" ,     )

     lowercase__       :       Optional[Any]				=						WavaVecaProcessor.from_pretrained("""valid_org/test-processor-org"""						)
     for k, v in processor.feature_extractor.__dict__.items():
      self.assertEqual(lowerCamelCase__ ,     getattr(new_processor.feature_extractor ,     lowerCamelCase__						)						)
     self.assertDictEqual(new_processor.tokenizer.get_vocab() ,     processor.tokenizer.get_vocab()						)







   def 					UpperCAmelCase__(						self						)      ->							int:
    CustomFeatureExtractor.register_for_auto_class()
    CustomTokenizer.register_for_auto_class()
    CustomProcessor.register_for_auto_class()

    lowercase__       :       List[str]				=						CustomFeatureExtractor.from_pretrained(lowerCamelCase__						)

    with tempfile.TemporaryDirectory() as tmp_dir:
     lowercase__       :       Tuple				=						os.path.join(lowerCamelCase__ ,     """vocab.txt"""						)
     with open(lowerCamelCase__ ,     """w""" ,     encoding="""utf-8"""						) as vocab_writer:
      vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]						)						)
     lowercase__       :       Optional[int]				=						CustomTokenizer(lowerCamelCase__						)

    lowercase__       :       List[Any]				=						CustomProcessor(lowerCamelCase__ ,     lowerCamelCase__						)

    with tempfile.TemporaryDirectory() as tmp_dir:
     create_repo(F'''{USER}/test-dynamic-processor''' ,     token=self._token						)
     lowercase__       :       Tuple				=						Repository(lowerCamelCase__ ,     clone_from=F'''{USER}/test-dynamic-processor''' ,     token=self._token						)
     processor.save_pretrained(lowerCamelCase__						)

     # This has added the proper auto_map field to the feature extractor config
     self.assertDictEqual(
         processor.feature_extractor.auto_map ,     {
             """AutoFeatureExtractor""": """custom_feature_extraction.CustomFeatureExtractor""",
             """AutoProcessor""": """custom_processing.CustomProcessor""",
         } ,     )

     # This has added the proper auto_map field to the tokenizer config
     with open(os.path.join(lowerCamelCase__ ,     """tokenizer_config.json"""						)						) as f:
      lowercase__       :       List[Any]				=						json.load(lowerCamelCase__						)
     self.assertDictEqual(
         tokenizer_config["""auto_map"""] ,     {
             """AutoTokenizer""": ["""custom_tokenization.CustomTokenizer""", None],
             """AutoProcessor""": """custom_processing.CustomProcessor""",
         } ,     )

     # The code has been copied from fixtures
     self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase__ ,     """custom_feature_extraction.py"""						)						)						)
     self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase__ ,     """custom_tokenization.py"""						)						)						)
     self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase__ ,     """custom_processing.py"""						)						)						)

     repo.push_to_hub()

    lowercase__       :       int				=						AutoProcessor.from_pretrained(F'''{USER}/test-dynamic-processor''' ,     trust_remote_code=lowerCamelCase__						)
    # Can't make an isinstance check because the new_processor is from the CustomProcessor class of a dynamic module
    self.assertEqual(new_processor.__class__.__name__ ,     """CustomProcessor"""						)