

import argparse
import gc
import json
import os
import shutil
import warnings

import torch

from transformers import LlamaConfig, LlamaForCausalLM, LlamaTokenizer


try:
     from transformers import LlamaTokenizerFast
except ImportError as e:
     warnings.warn(e)
     warnings.warn(
         'The converted tokenizer will be the `slow` tokenizer. To use the fast, update your `tokenizers` library and re-run the tokenizer conversion'
     )
     _snake_case						: Any											=							None
_snake_case						: List[Any]											=							{
    '7B': 1_10_08,
    '13B': 1_38_24,
    '30B': 1_79_20,
    '65B': 2_20_16,
    '70B': 2_86_72,
}
_snake_case						: int											=							{
    '7B': 1,
    '7Bf': 1,
    '13B': 2,
    '13Bf': 2,
    '30B': 4,
    '65B': 8,
    '70B': 8,
    '70Bf': 8,
}





def    _A							(						__snake_case  :Any   , __snake_case  :Dict=1   , __snake_case  :Union[str, Any]=256  )   -> Any:



   """simple docstring"""


   return multiple_of * ((int(ffn_dim_multiplier * int(8 * n / 3  )  ) + multiple_of - 1) // multiple_of)





def    _A							(						__snake_case  :Union[str, Any]  )   -> Tuple:



   """simple docstring"""


   with open(lowerCamelCase__   , "r"  ) as f:
      return json.load(lowerCamelCase__  )





def    _A							(						__snake_case  :List[Any]   , __snake_case  :Optional[int]  )   -> Dict:



   """simple docstring"""


   with open(lowerCamelCase__   , "w"  ) as f:
      json.dump(lowerCamelCase__   , lowerCamelCase__  )





def    _A							(						__snake_case  :int   , __snake_case  :Any   , __snake_case  :Tuple   , __snake_case  :str=True  )   -> str:



   """simple docstring"""


   os.makedirs(lowerCamelCase__   , exist_ok=lowerCamelCase__  )
   __SCREAMING_SNAKE_CASE       =				os.path.join(lowerCamelCase__   , "tmp"  )
   os.makedirs(lowerCamelCase__   , exist_ok=lowerCamelCase__  )

   __SCREAMING_SNAKE_CASE       =				read_json(os.path.join(lowerCamelCase__   , "params.json"  )  )
   __SCREAMING_SNAKE_CASE       =				NUM_SHARDS[model_size]
   __SCREAMING_SNAKE_CASE       =				params["""n_layers"""]
   __SCREAMING_SNAKE_CASE       =				params["""n_heads"""]
   __SCREAMING_SNAKE_CASE       =				n_heads // num_shards
   __SCREAMING_SNAKE_CASE       =				params["""dim"""]
   __SCREAMING_SNAKE_CASE       =				dim // n_heads
   __SCREAMING_SNAKE_CASE       =				1_0000.0
   __SCREAMING_SNAKE_CASE       =				1.0 / (base ** (torch.arange(0   , lowerCamelCase__   , 2  ).float() / dims_per_head))

   if "n_kv_heads" in params:
      __SCREAMING_SNAKE_CASE       =				params["""n_kv_heads"""]  # for GQA / MQA
      __SCREAMING_SNAKE_CASE       =				n_heads_per_shard // num_key_value_heads
      __SCREAMING_SNAKE_CASE       =				dim // num_key_value_heads
   else:  # compatibility with other checkpoints
      __SCREAMING_SNAKE_CASE       =				n_heads
      __SCREAMING_SNAKE_CASE       =				n_heads_per_shard
      __SCREAMING_SNAKE_CASE       =				dim

   # permute for sliced rotary
   def permute(__snake_case  :Tuple   , __snake_case  :str=n_heads   , __snake_case  :List[str]=dim   , __snake_case  :Tuple=dim  ):
      return w.view(lowerCamelCase__   , dima // n_heads // 2   , 2   , lowerCamelCase__  ).transpose(1   , 2  ).reshape(lowerCamelCase__   , lowerCamelCase__  )

   print(f'''Fetching all parameters from the checkpoint at {input_base_path}.'''  )
   # Load weights
   if model_size == "7B":
      # Not sharded
      # (The sharded implementation would also work, but this is simpler.)
      __SCREAMING_SNAKE_CASE       =				torch.load(os.path.join(lowerCamelCase__   , "consolidated.00.pth"  )   , map_location="cpu"  )
   else:
      # Sharded
      __SCREAMING_SNAKE_CASE       =				[
          torch.load(os.path.join(lowerCamelCase__   , f'''consolidated.{i:02d}.pth'''  )   , map_location="cpu"  )
          for i in range(lowerCamelCase__  )
      ]
   __SCREAMING_SNAKE_CASE       =				0
   __SCREAMING_SNAKE_CASE       =				{"""weight_map""": {}}
   for layer_i in range(lowerCamelCase__  ):
      __SCREAMING_SNAKE_CASE       =				f'''pytorch_model-{layer_i + 1}-of-{n_layers + 1}.bin'''
      if model_size == "7B":
         # Unsharded
         __SCREAMING_SNAKE_CASE       =				{
             f'''model.layers.{layer_i}.self_attn.q_proj.weight''': permute(
                 loaded[f'''layers.{layer_i}.attention.wq.weight''']  ),
             f'''model.layers.{layer_i}.self_attn.k_proj.weight''': permute(
                 loaded[f'''layers.{layer_i}.attention.wk.weight''']  ),
             f'''model.layers.{layer_i}.self_attn.v_proj.weight''': loaded[f'''layers.{layer_i}.attention.wv.weight'''],
             f'''model.layers.{layer_i}.self_attn.o_proj.weight''': loaded[f'''layers.{layer_i}.attention.wo.weight'''],
             f'''model.layers.{layer_i}.mlp.gate_proj.weight''': loaded[f'''layers.{layer_i}.feed_forward.w1.weight'''],
             f'''model.layers.{layer_i}.mlp.down_proj.weight''': loaded[f'''layers.{layer_i}.feed_forward.w2.weight'''],
             f'''model.layers.{layer_i}.mlp.up_proj.weight''': loaded[f'''layers.{layer_i}.feed_forward.w3.weight'''],
             f'''model.layers.{layer_i}.input_layernorm.weight''': loaded[f'''layers.{layer_i}.attention_norm.weight'''],
             f'''model.layers.{layer_i}.post_attention_layernorm.weight''': loaded[f'''layers.{layer_i}.ffn_norm.weight'''],
         }
      else:
         # Sharded
         # Note that attention.w{q,k,v,o}, feed_fordward.w[1,2,3], attention_norm.weight and ffn_norm.weight share
         # the same storage object, saving attention_norm and ffn_norm will save other weights too, which is
         # redundant as other weights will be stitched from multiple shards. To avoid that, they are cloned.

         __SCREAMING_SNAKE_CASE       =				{
             f'''model.layers.{layer_i}.input_layernorm.weight''': loaded[0][
                 f'''layers.{layer_i}.attention_norm.weight'''
             ].clone(),
             f'''model.layers.{layer_i}.post_attention_layernorm.weight''': loaded[0][
                 f'''layers.{layer_i}.ffn_norm.weight'''
             ].clone(),
         }
         __SCREAMING_SNAKE_CASE       =				permute(
             torch.cat(
                 [
                     loaded[i][f'''layers.{layer_i}.attention.wq.weight'''].view(lowerCamelCase__   , lowerCamelCase__   , lowerCamelCase__  )
                     for i in range(lowerCamelCase__  )
                 ]   , dim=0   , ).reshape(lowerCamelCase__   , lowerCamelCase__  )  )
         __SCREAMING_SNAKE_CASE       =				permute(
             torch.cat(
                 [
                     loaded[i][f'''layers.{layer_i}.attention.wk.weight'''].view(
                         lowerCamelCase__   , lowerCamelCase__   , lowerCamelCase__  )
                     for i in range(lowerCamelCase__  )
                 ]   , dim=0   , ).reshape(lowerCamelCase__   , lowerCamelCase__  )   , lowerCamelCase__   , lowerCamelCase__   , lowerCamelCase__   , )
         __SCREAMING_SNAKE_CASE       =				torch.cat(
             [
                 loaded[i][f'''layers.{layer_i}.attention.wv.weight'''].view(
                     lowerCamelCase__   , lowerCamelCase__   , lowerCamelCase__  )
                 for i in range(lowerCamelCase__  )
             ]   , dim=0   , ).reshape(lowerCamelCase__   , lowerCamelCase__  )

         __SCREAMING_SNAKE_CASE       =				torch.cat(
             [loaded[i][f'''layers.{layer_i}.attention.wo.weight'''] for i in range(lowerCamelCase__  )]   , dim=1  )
         __SCREAMING_SNAKE_CASE       =				torch.cat(
             [loaded[i][f'''layers.{layer_i}.feed_forward.w1.weight'''] for i in range(lowerCamelCase__  )]   , dim=0  )
         __SCREAMING_SNAKE_CASE       =				torch.cat(
             [loaded[i][f'''layers.{layer_i}.feed_forward.w2.weight'''] for i in range(lowerCamelCase__  )]   , dim=1  )
         __SCREAMING_SNAKE_CASE       =				torch.cat(
             [loaded[i][f'''layers.{layer_i}.feed_forward.w3.weight'''] for i in range(lowerCamelCase__  )]   , dim=0  )

      __SCREAMING_SNAKE_CASE       =				inv_freq
      for k, v in state_dict.items():
         __SCREAMING_SNAKE_CASE       =				filename
         param_count += v.numel()
      torch.save(lowerCamelCase__   , os.path.join(lowerCamelCase__   , lowerCamelCase__  )  )

   __SCREAMING_SNAKE_CASE       =				f'''pytorch_model-{n_layers + 1}-of-{n_layers + 1}.bin'''
   if model_size == "7B":
      # Unsharded
      __SCREAMING_SNAKE_CASE       =				{
          """model.embed_tokens.weight""": loaded["""tok_embeddings.weight"""],
          """model.norm.weight""": loaded["""norm.weight"""],
          """lm_head.weight""": loaded["""output.weight"""],
      }
   else:
      __SCREAMING_SNAKE_CASE       =				{
          """model.norm.weight""": loaded[0]["""norm.weight"""],
          """model.embed_tokens.weight""": torch.cat(
              [loaded[i]["tok_embeddings.weight"] for i in range(lowerCamelCase__  )]   , dim=1  ),
          """lm_head.weight""": torch.cat([loaded[i]["output.weight"] for i in range(lowerCamelCase__  )]   , dim=0  ),
      }

   for k, v in state_dict.items():
      __SCREAMING_SNAKE_CASE       =				filename
      param_count += v.numel()
   torch.save(lowerCamelCase__   , os.path.join(lowerCamelCase__   , lowerCamelCase__  )  )

   # Write configs
   __SCREAMING_SNAKE_CASE       =				{"""total_size""": param_count * 2}
   write_json(lowerCamelCase__   , os.path.join(lowerCamelCase__   , "pytorch_model.bin.index.json"  )  )
   __SCREAMING_SNAKE_CASE       =				params["""ffn_dim_multiplier"""] if """ffn_dim_multiplier""" in params else 1
   __SCREAMING_SNAKE_CASE       =				params["""multiple_of"""] if """multiple_of""" in params else 256
   __SCREAMING_SNAKE_CASE       =				LlamaConfig(
       hidden_size=lowerCamelCase__   , intermediate_size=compute_intermediate_size(lowerCamelCase__   , lowerCamelCase__   , lowerCamelCase__  )   , num_attention_heads=params["n_heads"]   , num_hidden_layers=params["n_layers"]   , rms_norm_eps=params["norm_eps"]   , num_key_value_heads=lowerCamelCase__   , )
   config.save_pretrained(lowerCamelCase__  )

   # Make space so we can load the model properly now.
   del state_dict
   del loaded
   gc.collect()

   print("Loading the checkpoint in a Llama model."  )
   __SCREAMING_SNAKE_CASE       =				LlamaForCausalLM.from_pretrained(lowerCamelCase__   , torch_dtype=torch.floataa   , low_cpu_mem_usage=lowerCamelCase__  )
   # Avoid saving this as part of the config.
   del model.config._name_or_path

   print("Saving in the Transformers format."  )
   model.save_pretrained(lowerCamelCase__   , safe_serialization=lowerCamelCase__  )
   shutil.rmtree(lowerCamelCase__  )





def    _A							(						__snake_case  :Optional[int]   , __snake_case  :Any  )   -> List[Any]:



   """simple docstring"""


   __SCREAMING_SNAKE_CASE       =				LlamaTokenizer if LlamaTokenizerFast is None else LlamaTokenizerFast
   print(f'''Saving a {tokenizer_class.__name__} to {tokenizer_path}.'''  )
   __SCREAMING_SNAKE_CASE       =				tokenizer_class(lowerCamelCase__  )
   tokenizer.save_pretrained(lowerCamelCase__  )





def    _A							(						)   -> Optional[Any]:



   """simple docstring"""


   __SCREAMING_SNAKE_CASE       =				argparse.ArgumentParser()
   parser.add_argument(
       "--input_dir"   , help="Location of LLaMA weights, which contains tokenizer.model and model folders"   , )
   parser.add_argument(
       "--model_size"   , choices=["7B", "7Bf", "13B", "13Bf", "30B", "65B", "70B", "70Bf", "tokenizer_only"]   , )
   parser.add_argument(
       "--output_dir"   , help="Location to write HF model and tokenizer"   , )
   parser.add_argument("--safe_serialization"   , type=lowerCamelCase__   , help="Whether or not to save using `safetensors`."  )
   __SCREAMING_SNAKE_CASE       =				parser.parse_args()
   if args.model_size != "tokenizer_only":
      write_model(
          model_path=args.output_dir   , input_base_path=os.path.join(args.input_dir   , args.model_size  )   , model_size=args.model_size   , safe_serialization=args.safe_serialization   , )
   __SCREAMING_SNAKE_CASE       =				os.path.join(args.input_dir   , "tokenizer.model"  )
   write_tokenizer(args.output_dir   , lowerCamelCase__  )


if __name__ == "__main__":
     main()





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