




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      :						Optional[Any]							=  get_tests_dir("""fixtures/dummy_feature_extractor_config.json""")
_lowerCamelCase      :						Union[str, Any]							=  get_tests_dir("""fixtures/vocab.json""")
_lowerCamelCase      :						List[Any]							=  get_tests_dir("""fixtures""")


class 							UpperCamelCase_       (					unittest.TestCase	):

    '''simple docstring'''






    UpperCAmelCase__				    = ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']


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





           '''simple docstring'''

           A__ 					=       0


    def 			SCREAMING_SNAKE_CASE				(     self :      Optional[int])		->int:





           '''simple docstring'''

           A__ 					=       AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h''')
           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      Optional[int])		->Any:





           '''simple docstring'''

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

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

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      str)		->Any:





           '''simple docstring'''

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

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      List[str])		->Any:





           '''simple docstring'''

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

                  A__ 					=       WavaVecaProcessor(UpperCAmelCase__   ,					UpperCAmelCase__)

                  # save in new folder
                  processor.save_pretrained(UpperCAmelCase__)

                  # drop `processor_class` in tokenizer
                  with open(os.path.join(UpperCAmelCase__   ,					UpperCAmelCase__)   ,					'''r''') as f:
                         A__ 					=       json.load(UpperCAmelCase__)
                         config_dict.pop('''processor_class''')

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

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      str)		->Optional[int]:





           '''simple docstring'''

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

                  A__ 					=       WavaVecaProcessor(UpperCAmelCase__   ,					UpperCAmelCase__)

                  # save in new folder
                  processor.save_pretrained(UpperCAmelCase__)

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

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

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      str)		->Optional[int]:





           '''simple docstring'''

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

                  A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)

           self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)


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





           '''simple docstring'''

           with self.assertRaises(UpperCAmelCase__):
                  A__ 					=       AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor''')
           # If remote code is disabled, we can't load this config.
           with self.assertRaises(UpperCAmelCase__):
                  A__ 					=       AutoProcessor.from_pretrained(
                      '''hf-internal-testing/test_dynamic_processor'''   ,					trust_remote_code=UpperCAmelCase__)

           A__ 					=       AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''   ,					trust_remote_code=UpperCAmelCase__)
           self.assertTrue(processor.special_attribute_present)
           self.assertEqual(processor.__class__.__name__   ,					'''NewProcessor''')

           A__ 					=       processor.feature_extractor
           self.assertTrue(feature_extractor.special_attribute_present)
           self.assertEqual(feature_extractor.__class__.__name__   ,					'''NewFeatureExtractor''')

           A__ 					=       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
                  A__ 					=       AutoProcessor.from_pretrained(
                      '''hf-internal-testing/test_dynamic_processor'''   ,					trust_remote_code=UpperCAmelCase__   ,					use_fast=UpperCAmelCase__)
                  A__ 					=       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 			SCREAMING_SNAKE_CASE				(     self :      Optional[Any])		->Union[str, Any]:





           '''simple docstring'''

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

                  # Now that the config is registered, it can be used as any other config with the auto-API
                  A__ 					=       CustomFeatureExtractor.from_pretrained(UpperCAmelCase__)

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

                  A__ 					=       CustomProcessor(UpperCAmelCase__   ,					UpperCAmelCase__)

                  with tempfile.TemporaryDirectory() as tmp_dir:
                         processor.save_pretrained(UpperCAmelCase__)
                         A__ 					=       AutoProcessor.from_pretrained(UpperCAmelCase__)
                         self.assertIsInstance(UpperCAmelCase__   ,					UpperCAmelCase__)

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





           '''simple docstring'''

           class 							UpperCamelCase_       (					UpperCAmelCase__	):

               '''simple docstring'''






               UpperCAmelCase__				    = False


           class 							UpperCamelCase_       (					UpperCAmelCase__	):

               '''simple docstring'''






               UpperCAmelCase__				    = False


           class 							UpperCamelCase_       (					UpperCAmelCase__	):

               '''simple docstring'''






               UpperCAmelCase__				    = '''AutoFeatureExtractor'''
               UpperCAmelCase__				    = '''AutoTokenizer'''
               UpperCAmelCase__				    = False

           try:
                  AutoConfig.register('''custom'''   ,					UpperCAmelCase__)
                  AutoFeatureExtractor.register(UpperCAmelCase__   ,					UpperCAmelCase__)
                  AutoTokenizer.register(UpperCAmelCase__   ,					slow_tokenizer_class=UpperCAmelCase__)
                  AutoProcessor.register(UpperCAmelCase__   ,					UpperCAmelCase__)
                  # If remote code is not set, the default is to use local classes.
                  A__ 					=       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.
                  A__ 					=       AutoProcessor.from_pretrained(
                      '''hf-internal-testing/test_dynamic_processor'''   ,					trust_remote_code=UpperCAmelCase__)
                  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.
                  A__ 					=       AutoProcessor.from_pretrained(
                      '''hf-internal-testing/test_dynamic_processor'''   ,					trust_remote_code=UpperCAmelCase__)
                  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 			SCREAMING_SNAKE_CASE				(     self :      Optional[int])		->Dict:





           '''simple docstring'''

           A__ 					=       AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert''')
           self.assertEqual(processor.__class__.__name__   ,					'''BertTokenizerFast''')





    def 			SCREAMING_SNAKE_CASE				(     self :      Any)		->Optional[Any]:





           '''simple docstring'''

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


@is_staging_test
class 							UpperCamelCase_       (					unittest.TestCase	):

    '''simple docstring'''






    UpperCAmelCase__				    = ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']


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





           '''simple docstring'''

           A__ 					=       TOKEN
           HfFolder.save_token(UpperCAmelCase__)


    @classmethod
    def 			SCREAMING_SNAKE_CASE				(     cls :      Union[str, Any])		->Any:





           '''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 			SCREAMING_SNAKE_CASE				(     self :      List[str])		->Any:





           '''simple docstring'''

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

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


    def 			SCREAMING_SNAKE_CASE				(     self :      List[Any])		->str:





           '''simple docstring'''

           A__ 					=       WavaVecaProcessor.from_pretrained(UpperCAmelCase__)

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

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





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





           '''simple docstring'''

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

           A__ 					=       CustomFeatureExtractor.from_pretrained(UpperCAmelCase__)

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

           A__ 					=       CustomProcessor(UpperCAmelCase__   ,					UpperCAmelCase__)

           with tempfile.TemporaryDirectory() as tmp_dir:
                  create_repo(f"""{USER}/test-dynamic-processor"""   ,					token=self._token)
                  A__ 					=       Repository(UpperCAmelCase__   ,					clone_from=f"""{USER}/test-dynamic-processor"""   ,					token=self._token)
                  processor.save_pretrained(UpperCAmelCase__)

                  # 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(UpperCAmelCase__   ,					'''tokenizer_config.json''')) as f:
                         A__ 					=       json.load(UpperCAmelCase__)
                  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(UpperCAmelCase__   ,					'''custom_feature_extraction.py''')))
                  self.assertTrue(os.path.isfile(os.path.join(UpperCAmelCase__   ,					'''custom_tokenization.py''')))
                  self.assertTrue(os.path.isfile(os.path.join(UpperCAmelCase__   ,					'''custom_processing.py''')))

                  repo.push_to_hub()

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







'''simple docstring'''
from typing import List

import numpy as np






def A__       (	UpperCAmelCase_				):
						_UpperCamelCase   :						Any												=  {key: len(UpperCAmelCase_				) for key, value in gen_kwargs.items() if isinstance(UpperCAmelCase_    ,    UpperCAmelCase_				)}
						if len(set(lists_lengths.values()				)				) > 1:
												raise RuntimeError(
												    (
												        'Sharding is ambiguous for this dataset: '
												        + 'we found several data sources lists of different lengths, and we don\'t know over which list we should parallelize:\n'
												        + '\n'.join(f'\t- key {key} has length {length}' for key, length in lists_lengths.items()				)
												        + '\nTo fix this, check the \'gen_kwargs\' and make sure to use lists only for data sources, '
												        + 'and use tuples otherwise. In the end there should only be one single list, or several lists with the same length.'
												    )				)
						_UpperCamelCase   :						Tuple												=  max(lists_lengths.values()    ,    default=0				)
						return max(1    ,    UpperCAmelCase_				)






def A__       (	UpperCAmelCase_    ,    UpperCAmelCase_				):
						_UpperCamelCase   :						Dict												=  []
						for group_idx in range(UpperCAmelCase_				):
												_UpperCamelCase   :						int												=  num_shards // max_num_jobs + (group_idx < (num_shards % max_num_jobs))
												if num_shards_to_add == 0:
																		break
												_UpperCamelCase   :						Tuple												=  shards_indices_per_group[-1].stop if shards_indices_per_group else 0
												_UpperCamelCase   :						Tuple												=  range(UpperCAmelCase_    ,    start + num_shards_to_add				)
												shards_indices_per_group.append(UpperCAmelCase_				)
						return shards_indices_per_group






def A__       (	UpperCAmelCase_    ,    UpperCAmelCase_				):
						_UpperCamelCase   :						Tuple												=  _number_of_shards_in_gen_kwargs(UpperCAmelCase_				)
						if num_shards == 1:
												return [dict(UpperCAmelCase_				)]
						else:
												_UpperCamelCase   :						str												=  _distribute_shards(num_shards=UpperCAmelCase_    ,    max_num_jobs=UpperCAmelCase_				)
												return [
												    {
												        key: [value[shard_idx] for shard_idx in shard_indices_per_group[group_idx]]
												        if isinstance(UpperCAmelCase_    ,    UpperCAmelCase_				)
												        else value
												        for key, value in gen_kwargs.items()
												    }
												    for group_idx in range(len(UpperCAmelCase_				)				)
												]






def A__       (	UpperCAmelCase_				):
						return {
						    key: [value for gen_kwargs in gen_kwargs_list for value in gen_kwargs[key]]
						    if isinstance(gen_kwargs_list[0][key]    ,    UpperCAmelCase_				)
						    else gen_kwargs_list[0][key]
						    for key in gen_kwargs_list[0]
						}






def A__       (	UpperCAmelCase_    ,    UpperCAmelCase_				):
						_UpperCamelCase   :						int												=  {len(UpperCAmelCase_				) for value in gen_kwargs.values() if isinstance(UpperCAmelCase_    ,    UpperCAmelCase_				)}
						_UpperCamelCase   :						Union[str, Any]												=  {}
						for size in list_sizes:
												_UpperCamelCase   :						str												=  list(range(UpperCAmelCase_				)				)
												rng.shuffle(indices_per_size[size]				)
						# Now let's copy the gen_kwargs and shuffle the lists based on their sizes
						_UpperCamelCase   :						Union[str, Any]												=  dict(UpperCAmelCase_				)
						for key, value in shuffled_kwargs.items():
												if isinstance(UpperCAmelCase_    ,    UpperCAmelCase_				):
																		_UpperCamelCase   :						Dict												=  [value[i] for i in indices_per_size[len(UpperCAmelCase_				)]]
						return shuffled_kwargs


