




import os
from pickle import UnpicklingError
from typing import Dict, Tuple

import jax
import jax.numpy as jnp
import numpy as np
from flax.serialization import from_bytes
from flax.traverse_util import flatten_dict, unflatten_dict

import transformers

from .utils import logging


lowerCAmelCase_						=			logging.get_logger(__name__)






def  snake_case(    __magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__=False   )			->							Dict:


     try:
          import torch  # noqa: F401
     except ImportError:
          logger.error(
              '''Loading a PyTorch model in Flax, requires both PyTorch and Flax to be installed. Please see'''
              ''' https://pytorch.org/ and https://flax.readthedocs.io/en/latest/installation.html for installation'''
              ''' instructions.'''   )
          raise

     if not is_sharded:
          lowercase :       Any       =			os.path.abspath(__magic_name__   )
          logger.info(F"""Loading PyTorch weights from {pt_path}"""   )

          lowercase :       List[str]       =			torch.load(__magic_name__			,		map_location='''cpu'''   )
          logger.info(F"""PyTorch checkpoint contains {sum(t.numel() for t in pt_state_dict.values()   ):,} parameters."""   )

          lowercase :       int       =			convert_pytorch_state_dict_to_flax(__magic_name__			,		__magic_name__   )
     else:
          # model is sharded and pytorch_checkpoint_path already contains the list of .pt shard files
          lowercase :       Optional[Any]       =			convert_pytorch_sharded_state_dict_to_flax(__magic_name__			,		__magic_name__   )
     return flax_state_dict






def  snake_case(    __magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__			,		)			->							(Tuple[str], np.ndarray):


     def is_key_or_prefix_key_in_dict(__magic_name__   ) -> bool:
          return len(set(__magic_name__   ) & {key, (model_prefix,) + key}   ) > 0

     # layer norm
     lowercase :       str       =			pt_tuple_key[:-1] + ('''scale''',)
     if pt_tuple_key[-1] in ["weight", "gamma"] and is_key_or_prefix_key_in_dict(__magic_name__   ):
          return renamed_pt_tuple_key, pt_tensor

     # batch norm layer mean
     lowercase :       Dict       =			pt_tuple_key[:-1] + ('''mean''',)
     if pt_tuple_key[-1] == "running_mean" and not is_key_or_prefix_key_in_dict(__magic_name__   ):
          return renamed_pt_tuple_key, pt_tensor

     # batch norm layer var
     lowercase :       List[Any]       =			pt_tuple_key[:-1] + ('''var''',)
     if pt_tuple_key[-1] == "running_var" and not is_key_or_prefix_key_in_dict(__magic_name__   ):
          return renamed_pt_tuple_key, pt_tensor

     # embedding
     lowercase :       Optional[int]       =			pt_tuple_key[:-1] + ('''embedding''',)
     if pt_tuple_key[-1] == "weight" and is_key_or_prefix_key_in_dict(__magic_name__   ):
          return renamed_pt_tuple_key, pt_tensor

     # conv layer
     lowercase :       Optional[int]       =			pt_tuple_key[:-1] + ('''kernel''',)
     if pt_tuple_key[-1] == "weight" and pt_tensor.ndim == 4 and not is_key_or_prefix_key_in_dict(__magic_name__   ):
          lowercase :       Tuple       =			pt_tensor.transpose(2			,		3			,		1			,		0   )
          return renamed_pt_tuple_key, pt_tensor

     # linear layer
     lowercase :       Optional[Any]       =			pt_tuple_key[:-1] + ('''kernel''',)
     if pt_tuple_key[-1] == "weight" and not is_key_or_prefix_key_in_dict(__magic_name__   ):
          lowercase :       List[Any]       =			pt_tensor.T
          return renamed_pt_tuple_key, pt_tensor

     # old PyTorch layer norm weight
     lowercase :       str       =			pt_tuple_key[:-1] + ('''weight''',)
     if pt_tuple_key[-1] == "gamma":
          return renamed_pt_tuple_key, pt_tensor

     # old PyTorch layer norm bias
     lowercase :       Optional[Any]       =			pt_tuple_key[:-1] + ('''bias''',)
     if pt_tuple_key[-1] == "beta":
          return renamed_pt_tuple_key, pt_tensor

     # New `weight_norm` from https://github.com/huggingface/transformers/pull/24030
     lowercase :       List[str]       =			None
     if pt_tuple_key[-3::2] == ("parametrizations", "original0"):
          lowercase :       str       =			pt_tuple_key[-2] + '''_g'''
     elif pt_tuple_key[-3::2] == ("parametrizations", "original1"):
          lowercase :       List[Any]       =			pt_tuple_key[-2] + '''_v'''
     if name is not None:
          lowercase :       int       =			pt_tuple_key[:-3] + (name,)
          return renamed_pt_tuple_key, pt_tensor

     return pt_tuple_key, pt_tensor






def  snake_case(    __magic_name__			,		__magic_name__   )			->							Tuple:
     lowercase :       List[Any]       =			{k: v.numpy() for k, v in pt_state_dict.items()}

     lowercase :       Optional[int]       =			flax_model.base_model_prefix

     # use params dict if the model contains batch norm layers
     if "params" in flax_model.params:
          lowercase :       Any       =			flax_model.params['''params''']
     else:
          lowercase :       List[Any]       =			flax_model.params
     lowercase :       str       =			flatten_dict(__magic_name__   )

     # add batch_stats keys,values to dict
     if "batch_stats" in flax_model.params:
          lowercase :       Tuple       =			flatten_dict(flax_model.params['''batch_stats''']   )
          random_flax_state_dict.update(__magic_name__   )

     lowercase :       Optional[Any]       =			{}

     lowercase :       List[Any]       =			(model_prefix not in flax_model_params) and (
         model_prefix in {k.split('''.'''   )[0] for k in pt_state_dict.keys()}
     )
     lowercase :       Optional[Any]       =			(model_prefix in flax_model_params) and (
         model_prefix not in {k.split('''.'''   )[0] for k in pt_state_dict.keys()}
     )

     # Need to change some parameters name to match Flax names
     for pt_key, pt_tensor in pt_state_dict.items():
          lowercase :       int       =			tuple(pt_key.split('''.'''   )   )

          # remove base model prefix if necessary
          lowercase :       List[Any]       =			pt_tuple_key[0] == model_prefix
          if load_model_with_head_into_base_model and has_base_model_prefix:
               lowercase :       Optional[Any]       =			pt_tuple_key[1:]

          # Correctly rename weight parameters
          lowercase :       List[str]       =			rename_key_and_reshape_tensor(
              __magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__   )

          # add model prefix if necessary
          lowercase :       Dict       =			(model_prefix,) + flax_key in random_flax_state_dict
          if load_base_model_into_model_with_head and require_base_model_prefix:
               lowercase :       List[Any]       =			(model_prefix,) + flax_key

          if flax_key in random_flax_state_dict:
               if flax_tensor.shape != random_flax_state_dict[flax_key].shape:
                    raise ValueError(
                        F"""PyTorch checkpoint seems to be incorrect. Weight {pt_key} was expected to be of shape """
                        F"""{random_flax_state_dict[flax_key].shape}, but is {flax_tensor.shape}."""   )

        # add batch stats if the model contains batchnorm layers
          if "batch_stats" in flax_model.params:
               if "mean" in flax_key[-1] or "var" in flax_key[-1]:
                    lowercase :       str       =			jnp.asarray(__magic_name__   )
                    continue
               # remove num_batches_tracked key
               if "num_batches_tracked" in flax_key[-1]:
                    flax_state_dict.pop(__magic_name__			,		__magic_name__   )
                    continue

               # also add unexpected weight so that warning is thrown
               lowercase :       Any       =			jnp.asarray(__magic_name__   )

          else:
               # also add unexpected weight so that warning is thrown
               lowercase :       Tuple       =			jnp.asarray(__magic_name__   )

     return unflatten_dict(__magic_name__   )






def  snake_case(    __magic_name__			,		__magic_name__   )			->							int:
     import torch

     # Load the index
     lowercase :       Optional[int]       =			{}
     for shard_file in shard_filenames:
          # load using msgpack utils
          lowercase :       Any       =			torch.load(__magic_name__   )
          lowercase :       Dict       =			{k: v.numpy() for k, v in pt_state_dict.items()}

          lowercase :       Optional[int]       =			flax_model.base_model_prefix

          # use params dict if the model contains batch norm layers and then add batch_stats keys,values to dict
          if "batch_stats" in flax_model.params:
               lowercase :       Tuple       =			flax_model.params['''params''']

               lowercase :       List[str]       =			flatten_dict(__magic_name__   )
               random_flax_state_dict.update(flatten_dict(flax_model.params['''batch_stats''']   )   )
          else:
               lowercase :       Tuple       =			flax_model.params
               lowercase :       Optional[int]       =			flatten_dict(__magic_name__   )

          lowercase :       int       =			(model_prefix not in flax_model_params) and (
              model_prefix in {k.split('''.'''   )[0] for k in pt_state_dict.keys()}
          )
          lowercase :       List[Any]       =			(model_prefix in flax_model_params) and (
              model_prefix not in {k.split('''.'''   )[0] for k in pt_state_dict.keys()}
          )
          # Need to change some parameters name to match Flax names
          for pt_key, pt_tensor in pt_state_dict.items():
               lowercase :       Dict       =			tuple(pt_key.split('''.'''   )   )

               # remove base model prefix if necessary
               lowercase :       str       =			pt_tuple_key[0] == model_prefix
               if load_model_with_head_into_base_model and has_base_model_prefix:
                    lowercase :       Union[str, Any]       =			pt_tuple_key[1:]

               # Correctly rename weight parameters
               lowercase :       str       =			rename_key_and_reshape_tensor(
                   __magic_name__			,		__magic_name__			,		__magic_name__			,		__magic_name__   )
               # add model prefix if necessary
               lowercase :       Optional[int]       =			(model_prefix,) + flax_key in random_flax_state_dict
               if load_base_model_into_model_with_head and require_base_model_prefix:
                    lowercase :       List[str]       =			(model_prefix,) + flax_key

               if flax_key in random_flax_state_dict:
                    if flax_tensor.shape != random_flax_state_dict[flax_key].shape:
                         raise ValueError(
                             F"""PyTorch checkpoint seems to be incorrect. Weight {pt_key} was expected to be of shape """
                             F"""{random_flax_state_dict[flax_key].shape}, but is {flax_tensor.shape}."""   )

            # add batch stats if the model contains batchnorm layers
               if "batch_stats" in flax_model.params:
                    if "mean" in flax_key[-1]:
                         lowercase :       List[Any]       =			jnp.asarray(__magic_name__   )
                         continue
                    if "var" in flax_key[-1]:
                         lowercase :       Union[str, Any]       =			jnp.asarray(__magic_name__   )
                         continue
                    # remove num_batches_tracked key
                    if "num_batches_tracked" in flax_key[-1]:
                         flax_state_dict.pop(__magic_name__			,		__magic_name__   )
                         continue

                    # also add unexpected weight so that warning is thrown
                    lowercase :       str       =			jnp.asarray(__magic_name__   )

               else:
                    # also add unexpected weight so that warning is thrown
                    lowercase :       Optional[int]       =			jnp.asarray(__magic_name__   )
     return unflatten_dict(__magic_name__   )






def  snake_case(    __magic_name__			,		__magic_name__   )			->							Optional[Any]:
     lowercase :       str       =			os.path.abspath(__magic_name__   )
     logger.info(F"""Loading Flax weights from {flax_checkpoint_path}"""   )

     # import correct flax class
     lowercase :       Union[str, Any]       =			getattr(__magic_name__			,		'''Flax''' + model.__class__.__name__   )

     # load flax weight dict
     with open(__magic_name__			,		'''rb'''   ) as state_f:
          try:
               lowercase :       List[Any]       =			from_bytes(__magic_name__			,		state_f.read()   )
          except UnpicklingError:
               raise EnvironmentError(F"""Unable to convert {flax_checkpoint_path} to Flax deserializable object. """   )

     return load_flax_weights_in_pytorch_model(__magic_name__			,		__magic_name__   )






def  snake_case(    __magic_name__			,		__magic_name__   )			->							str:


     try:
          import torch  # noqa: F401
     except ImportError:
          logger.error(
              '''Loading a Flax weights in PyTorch, requires both PyTorch and Flax to be installed. Please see'''
              ''' https://pytorch.org/ and https://flax.readthedocs.io/en/latest/installation.html for installation'''
              ''' instructions.'''   )
          raise

     # check if we have bf16 weights
     lowercase :       str       =			flatten_dict(jax.tree_util.tree_map(lambda __magic_name__   : x.dtype == jnp.bfloataa			,		__magic_name__   )   ).values()
     if any(__magic_name__   ):
          # convert all weights to fp32 if the are bf16 since torch.from_numpy can-not handle bf16
          # and bf16 is not fully supported in PT yet.
          logger.warning(
              '''Found ``bfloat16`` weights in Flax model. Casting all ``bfloat16`` weights to ``float32`` '''
              '''before loading those in PyTorch model.'''   )
          lowercase :       Dict       =			jax.tree_util.tree_map(
              lambda __magic_name__   : params.astype(np.floataa   ) if params.dtype == jnp.bfloataa else params			,		__magic_name__   )

     lowercase :       Tuple       =			flatten_dict(__magic_name__   )
     lowercase :       List[Any]       =			pt_model.state_dict()

     lowercase :       Optional[int]       =			(pt_model.base_model_prefix in flax_state) and (
         pt_model.base_model_prefix not in {k.split('''.'''   )[0] for k in pt_model_dict.keys()}
     )
     lowercase :       Dict       =			(pt_model.base_model_prefix not in flax_state) and (
         pt_model.base_model_prefix in {k.split('''.'''   )[0] for k in pt_model_dict.keys()}
     )

     # keep track of unexpected & missing keys
     lowercase :       Optional[int]       =			[]
     lowercase :       List[str]       =			set(pt_model_dict.keys()   )

     for flax_key_tuple, flax_tensor in flax_state_dict.items():
          lowercase :       Dict       =			flax_key_tuple[0] == pt_model.base_model_prefix
          lowercase :       Any       =			'''.'''.join((pt_model.base_model_prefix,) + flax_key_tuple   ) in pt_model_dict

          # adapt flax_key to prepare for loading from/to base model only
          if load_model_with_head_into_base_model and has_base_model_prefix:
               lowercase :       Tuple       =			flax_key_tuple[1:]
          elif load_base_model_into_model_with_head and require_base_model_prefix:
               lowercase :       Optional[int]       =			(pt_model.base_model_prefix,) + flax_key_tuple

          # rename flax weights to PyTorch format
          if flax_key_tuple[-1] == "kernel" and flax_tensor.ndim == 4 and ".".join(__magic_name__   ) not in pt_model_dict:
               # conv layer
               lowercase :       Dict       =			flax_key_tuple[:-1] + ('''weight''',)
               lowercase :       Dict       =			jnp.transpose(__magic_name__			,		(3, 2, 0, 1)   )
          elif flax_key_tuple[-1] == "kernel" and ".".join(__magic_name__   ) not in pt_model_dict:
               # linear layer
               lowercase :       str       =			flax_key_tuple[:-1] + ('''weight''',)
               lowercase :       List[str]       =			flax_tensor.T
          elif flax_key_tuple[-1] in ["scale", "embedding"]:
               lowercase :       List[Any]       =			flax_key_tuple[:-1] + ('''weight''',)

          # adding batch stats from flax batch norm to pt
          elif "mean" in flax_key_tuple[-1]:
               lowercase :       List[str]       =			flax_key_tuple[:-1] + ('''running_mean''',)
          elif "var" in flax_key_tuple[-1]:
               lowercase :       List[str]       =			flax_key_tuple[:-1] + ('''running_var''',)

          if "batch_stats" in flax_state:
               lowercase :       List[Any]       =			'''.'''.join(flax_key_tuple[1:]   )  # Remove the params/batch_stats header
          else:
               lowercase :       int       =			'''.'''.join(__magic_name__   )

          # We also need to look at `pt_model_dict` and see if there are keys requiring further transformation.
          lowercase :       str       =			{}
          # New `weight_norm` from https://github.com/huggingface/transformers/pull/24030
          for key in pt_model_dict:
               lowercase :       Optional[Any]       =			key.split('''.'''   )
               lowercase :       Union[str, Any]       =			None
               if key_components[-3::2] == ["parametrizations", "original0"]:
                    lowercase :       Tuple       =			key_components[-2] + '''_g'''
               elif key_components[-3::2] == ["parametrizations", "original1"]:
                    lowercase :       List[str]       =			key_components[-2] + '''_v'''
               if name is not None:
                    lowercase :       Dict       =			key_components[:-3] + [name]
                    lowercase :       int       =			'''.'''.join(__magic_name__   )
                    lowercase :       int       =			key

          if flax_key in special_pt_names:
               lowercase :       str       =			special_pt_names[flax_key]

          if flax_key in pt_model_dict:
               if flax_tensor.shape != pt_model_dict[flax_key].shape:
                    raise ValueError(
                        F"""Flax checkpoint seems to be incorrect. Weight {flax_key_tuple} was expected """
                        F"""to be of shape {pt_model_dict[flax_key].shape}, but is {flax_tensor.shape}."""   )
               else:
                    # add weight to pytorch dict
                    lowercase :       str       =			np.asarray(__magic_name__   ) if not isinstance(__magic_name__			,		np.ndarray   ) else flax_tensor
                    lowercase :       Tuple       =			torch.from_numpy(__magic_name__   )
                    # remove from missing keys
                    missing_keys.remove(__magic_name__   )
          else:
               # weight is not expected by PyTorch model
               unexpected_keys.append(__magic_name__   )

     pt_model.load_state_dict(__magic_name__   )

     # re-transform missing_keys to list
     lowercase :       int       =			list(__magic_name__   )

     if len(__magic_name__   ) > 0:
          logger.warning(
              '''Some weights of the Flax model were not used when initializing the PyTorch model'''
              F""" {pt_model.__class__.__name__}: {unexpected_keys}\n- This IS expected if you are initializing"""
              F""" {pt_model.__class__.__name__} from a Flax model trained on another task or with another architecture"""
              ''' (e.g. initializing a BertForSequenceClassification model from a FlaxBertForPreTraining model).\n- This'''
              F""" IS NOT expected if you are initializing {pt_model.__class__.__name__} from a Flax model that you expect"""
              ''' to be exactly identical (e.g. initializing a BertForSequenceClassification model from a'''
              ''' FlaxBertForSequenceClassification model).'''   )
     else:
          logger.warning(F"""All Flax model weights were used when initializing {pt_model.__class__.__name__}.\n"""   )
     if len(__magic_name__   ) > 0:
          logger.warning(
              F"""Some weights of {pt_model.__class__.__name__} were not initialized from the Flax model and are newly"""
              F""" initialized: {missing_keys}\nYou should probably TRAIN this model on a down-stream task to be able to"""
              ''' use it for predictions and inference.'''   )
     else:
          logger.warning(
              F"""All the weights of {pt_model.__class__.__name__} were initialized from the Flax model.\n"""
              '''If your task is similar to the task the model of the checkpoint was trained on, '''
              F"""you can already use {pt_model.__class__.__name__} for predictions without further training."""   )

     return pt_model





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


lowerCAmelCase_						=			get_tests_dir('fixtures/dummy_feature_extractor_config.json')
lowerCAmelCase_						=			get_tests_dir('fixtures/vocab.json')
lowerCAmelCase_						=			get_tests_dir('fixtures')
class 						_A (  unittest.TestCase				):
  _UpperCamelCase       :      int			=      ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']






  def  __a      (  self		:     Union[str, Any]				)		->   Tuple:






       """simple docstring"""







       lowercase :       Union[str, Any]       =			0






  def  __a      (  self		:     Union[str, Any]				)		->   Union[str, Any]:






       """simple docstring"""







       lowercase :       Union[str, Any]       =			AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''				)
       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Any				)		->   int:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       str       =			WavaVecaConfig()
            lowercase :       Tuple       =			AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''				)

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

            lowercase :       Optional[Any]       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Tuple				)		->   Union[str, Any]:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            # copy relevant files
            copyfile(_A				,						os.path.join(_A				,						_A				)				)
            copyfile(_A				,						os.path.join(_A				,						'''vocab.json'''				)				)

            lowercase :       Any       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Tuple				)		->   Any:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       List[Any]       =			WavaVecaFeatureExtractor()
            lowercase :       Optional[int]       =			AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''				)

            lowercase :       Union[str, Any]       =			WavaVecaProcessor(_A				,						_A				)

            # save in new folder
            processor.save_pretrained(_A				)

            # drop `processor_class` in tokenizer
            with open(os.path.join(_A				,						_A				)				,						'''r'''				) as f:
                 lowercase :       Union[str, Any]       =			json.load(_A				)
                 config_dict.pop('''processor_class'''				)

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

            lowercase :       int       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     str				)		->   List[str]:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       int       =			WavaVecaFeatureExtractor()
            lowercase :       Dict       =			AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''				)

            lowercase :       Optional[int]       =			WavaVecaProcessor(_A				,						_A				)

            # save in new folder
            processor.save_pretrained(_A				)

            # drop `processor_class` in feature extractor
            with open(os.path.join(_A				,						_A				)				,						'''r'''				) as f:
                 lowercase :       int       =			json.load(_A				)
                 config_dict.pop('''processor_class'''				)

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

            lowercase :       Any       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     List[Any]				)		->   Tuple:






       """simple docstring"""







       with tempfile.TemporaryDirectory() as tmpdirname:
            lowercase :       Dict       =			WavaVecaConfig(processor_class='''Wav2Vec2Processor'''				)
            model_config.save_pretrained(_A				)
            # copy relevant files
            copyfile(_A				,						os.path.join(_A				,						'''vocab.json'''				)				)
            # create emtpy sample processor
            with open(os.path.join(_A				,						_A				)				,						'''w'''				) as f:
                 f.write('''{}'''				)

            lowercase :       Tuple       =			AutoProcessor.from_pretrained(_A				)

       self.assertIsInstance(_A				,						_A				)






  def  __a      (  self		:     Union[str, Any]				)		->   Tuple:






       """simple docstring"""







       with self.assertRaises(_A				):
            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(_A				):
            lowercase :       List[str]       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				)

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

       lowercase :       Union[str, Any]       =			processor.feature_extractor
       self.assertTrue(feature_extractor.special_attribute_present				)
       self.assertEqual(feature_extractor.__class__.__name__				,						'''NewFeatureExtractor'''				)

       lowercase :       Optional[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 :       Any       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				,						use_fast=_A				)
            lowercase :       Any       =			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  __a      (  self		:     Dict				)		->   Optional[Any]:






       """simple docstring"""







       try:
            AutoConfig.register('''custom'''				,						_A				)
            AutoFeatureExtractor.register(_A				,						_A				)
            AutoTokenizer.register(_A				,						slow_tokenizer_class=_A				)
            AutoProcessor.register(_A				,						_A				)
            # Trying to register something existing in the Transformers library will raise an error
            with self.assertRaises(_A				):
                 AutoProcessor.register(_A				,						_A				)

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

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

            lowercase :       Union[str, Any]       =			CustomProcessor(_A				,						_A				)

            with tempfile.TemporaryDirectory() as tmp_dir:
                 processor.save_pretrained(_A				)
                 lowercase :       Dict       =			AutoProcessor.from_pretrained(_A				)
                 self.assertIsInstance(_A				,						_A				)

       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  __a      (  self		:     str				)		->   Any:






       """simple docstring"""







       class 						_A (  _lowerCamelCase				):
         _UpperCamelCase       :      int			=      False


       class 						_A (  _lowerCamelCase				):
         _UpperCamelCase       :      Union[str, Any]			=      False


       class 						_A (  _lowerCamelCase				):
         _UpperCamelCase       :      Tuple			=      '''AutoFeatureExtractor'''
         _UpperCamelCase       :      Any			=      '''AutoTokenizer'''
         _UpperCamelCase       :      Any			=      False

       try:
            AutoConfig.register('''custom'''				,						_A				)
            AutoFeatureExtractor.register(_A				,						_A				)
            AutoTokenizer.register(_A				,						slow_tokenizer_class=_A				)
            AutoProcessor.register(_A				,						_A				)
            # If remote code is not set, the default is to use local classes.
            lowercase :       List[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 :       Any       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				)
            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 :       Union[str, Any]       =			AutoProcessor.from_pretrained(
                '''hf-internal-testing/test_dynamic_processor'''				,						trust_remote_code=_A				)
            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  __a      (  self		:     Tuple				)		->   Dict:






       """simple docstring"""







       lowercase :       Optional[int]       =			AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert'''				)
       self.assertEqual(processor.__class__.__name__				,						'''BertTokenizerFast'''				)






  def  __a      (  self		:     Any				)		->   int:






       """simple docstring"""







       lowercase :       Any       =			AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-convnext'''				)
       self.assertEqual(processor.__class__.__name__				,						'''ConvNextImageProcessor'''				)


@is_staging_test
class 						_A (  unittest.TestCase				):
  _UpperCamelCase       :      int			=      ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']






  @classmethod
  def  __a      (  cls		:     Tuple				)		->   List[Any]:






       """simple docstring"""







       lowercase :       str       =			TOKEN
       HfFolder.save_token(_A				)






  @classmethod
  def  __a      (  cls		:     List[Any]				)		->   Dict:






       """simple docstring"""







       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  __a      (  self		:     Union[str, Any]				)		->   Tuple:






       """simple docstring"""







       lowercase :       Any       =			WavaVecaProcessor.from_pretrained(_A				)
       with tempfile.TemporaryDirectory() as tmp_dir:
            processor.save_pretrained(
                os.path.join(_A				,						'''test-processor'''				)				,						push_to_hub=_A				,						use_auth_token=self._token				)

            lowercase :       Any       =			WavaVecaProcessor.from_pretrained(f"""{USER}/test-processor"""				)
            for k, v in processor.feature_extractor.__dict__.items():
                 self.assertEqual(_A				,						getattr(new_processor.feature_extractor				,						_A				)				)
            self.assertDictEqual(new_processor.tokenizer.get_vocab()				,						processor.tokenizer.get_vocab()				)






  def  __a      (  self		:     Optional[Any]				)		->   str:






       """simple docstring"""







       lowercase :       int       =			WavaVecaProcessor.from_pretrained(_A				)

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

            lowercase :       Dict       =			WavaVecaProcessor.from_pretrained('''valid_org/test-processor-org'''				)
            for k, v in processor.feature_extractor.__dict__.items():
                 self.assertEqual(_A				,						getattr(new_processor.feature_extractor				,						_A				)				)
            self.assertDictEqual(new_processor.tokenizer.get_vocab()				,						processor.tokenizer.get_vocab()				)






  def  __a      (  self		:     str				)		->   Dict:






       """simple docstring"""







       CustomFeatureExtractor.register_for_auto_class()
       CustomTokenizer.register_for_auto_class()
       CustomProcessor.register_for_auto_class()

       lowercase :       List[str]       =			CustomFeatureExtractor.from_pretrained(_A				)

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

       lowercase :       Optional[Any]       =			CustomProcessor(_A				,						_A				)

       with tempfile.TemporaryDirectory() as tmp_dir:
            create_repo(f"""{USER}/test-dynamic-processor"""				,						token=self._token				)
            lowercase :       List[str]       =			Repository(_A				,						clone_from=f"""{USER}/test-dynamic-processor"""				,						token=self._token				)
            processor.save_pretrained(_A				)

            # 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(_A				,						'''tokenizer_config.json'''				)				) as f:
                 lowercase :       Optional[Any]       =			json.load(_A				)
            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(_A				,						'''custom_feature_extraction.py'''				)				)				)
            self.assertTrue(os.path.isfile(os.path.join(_A				,						'''custom_tokenization.py'''				)				)				)
            self.assertTrue(os.path.isfile(os.path.join(_A				,						'''custom_processing.py'''				)				)				)

            repo.push_to_hub()

       lowercase :       Tuple       =			AutoProcessor.from_pretrained(f"""{USER}/test-dynamic-processor"""				,						trust_remote_code=_A				)
       # 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'''				)