


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


SCREAMING_SNAKE_CASE       :      int												= get_tests_dir("fixtures/dummy_feature_extractor_config.json")
SCREAMING_SNAKE_CASE       :      Optional[Any]												= get_tests_dir("fixtures/vocab.json")
SCREAMING_SNAKE_CASE       :      int												= get_tests_dir("fixtures")






class _lowerCamelCase(						unittest.TestCase   ):
       lowercase_ :	Optional[int]     =			["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]
       def 			UpperCamelCase     (					self)       ->   Optional[int]:




           """simple docstring"""

           _lowercase       :							Optional[int]      		=		0
       def 			UpperCamelCase     (					self)       ->   List[str]:




           """simple docstring"""

           _lowercase       :							Optional[int]      		=		AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h')
           self.assertIsInstance(lowerCamelCase,      lowerCamelCase)
       def 			UpperCamelCase     (					self)       ->   Union[str, Any]:




           """simple docstring"""

           with tempfile.TemporaryDirectory() as tmpdirname:
               _lowercase       :							Union[str, Any]      		=		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[int]      		=		AutoProcessor.from_pretrained(lowerCamelCase)

           self.assertIsInstance(lowerCamelCase,      lowerCamelCase)
       def 			UpperCamelCase     (					self)       ->   Any:




           """simple docstring"""

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

               _lowercase       :							Optional[Any]      		=		AutoProcessor.from_pretrained(lowerCamelCase)

           self.assertIsInstance(lowerCamelCase,      lowerCamelCase)
       def 			UpperCamelCase     (					self)       ->   Optional[Any]:




           """simple docstring"""

           with tempfile.TemporaryDirectory() as tmpdirname:
               _lowercase       :							Any      		=		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 tokenizer
               with open(os.path.join(lowerCamelCase,      lowerCamelCase),      'r') as f:
                   _lowercase       :							Union[str, Any]      		=		json.load(lowerCamelCase)
                   config_dict.pop('processor_class')

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

               _lowercase       :							Any      		=		AutoProcessor.from_pretrained(lowerCamelCase)

           self.assertIsInstance(lowerCamelCase,      lowerCamelCase)
       def 			UpperCamelCase     (					self)       ->   int:




           """simple docstring"""

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

               _lowercase       :							Optional[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       :							Optional[Any]      		=		json.load(lowerCamelCase)
                   config_dict.pop('processor_class')

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

               _lowercase       :							Optional[Any]      		=		AutoProcessor.from_pretrained(lowerCamelCase)

           self.assertIsInstance(lowerCamelCase,      lowerCamelCase)
       def 			UpperCamelCase     (					self)       ->   Optional[Any]:




           """simple docstring"""

           with tempfile.TemporaryDirectory() as tmpdirname:
               _lowercase       :							str      		=		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       :							Optional[Any]      		=		AutoProcessor.from_pretrained(lowerCamelCase)

           self.assertIsInstance(lowerCamelCase,      lowerCamelCase)
       def 			UpperCamelCase     (					self)       ->   Optional[int]:




           """simple docstring"""

           with self.assertRaises(lowerCamelCase):
               _lowercase       :							Dict      		=		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       :							str      		=		AutoProcessor.from_pretrained(
                   'hf-internal-testing/test_dynamic_processor',      trust_remote_code=lowerCamelCase)

           _lowercase       :							Optional[Any]      		=		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       :							int      		=		processor.feature_extractor
           self.assertTrue(feature_extractor.special_attribute_present)
           self.assertEqual(feature_extractor.__class__.__name__,      'NewFeatureExtractor')

           _lowercase       :							Union[str, Any]      		=		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       :							int      		=		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)       ->   List[str]:




           """simple docstring"""

           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       :							List[str]      		=		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       :							int      		=		CustomProcessor(lowerCamelCase,      lowerCamelCase)

               with tempfile.TemporaryDirectory() as tmp_dir:
                   processor.save_pretrained(lowerCamelCase)
                   _lowercase       :							Optional[int]      		=		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:




           """simple docstring"""

           class _lowerCamelCase(						_a   ):
                  lowercase_ :	List[str]     =			False






           class _lowerCamelCase(						_a   ):
                  lowercase_ :	Optional[Any]     =			False






           class _lowerCamelCase(						_a   ):
                  lowercase_ :	Any     =			"""AutoFeatureExtractor"""
                  lowercase_ :	Tuple     =			"""AutoTokenizer"""
                  lowercase_ :	int     =			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       :							int      		=		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       :							int      		=		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)       ->   Union[str, Any]:




           """simple docstring"""

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




       def 			UpperCamelCase     (					self)       ->   int:




           """simple docstring"""

           _lowercase       :							List[str]      		=		AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-convnext')
           self.assertEqual(processor.__class__.__name__,      'ConvNextImageProcessor')






@is_staging_test
class _lowerCamelCase(						unittest.TestCase   ):
       lowercase_ :	Optional[int]     =			["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]
       @classmethod
       def 			UpperCamelCase     (					cls)       ->   Optional[int]:




           """simple docstring"""

           _lowercase       :							int      		=		TOKEN
           HfFolder.save_token(lowerCamelCase)
       @classmethod
       def 			UpperCamelCase     (					cls)       ->   Optional[int]:




           """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 			UpperCamelCase     (					self)       ->   Tuple:




           """simple docstring"""

           _lowercase       :							Tuple      		=		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       :							Tuple      		=		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)       ->   int:




           """simple docstring"""

           _lowercase       :							List[Any]      		=		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       :							Union[str, 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)       ->   Any:




           """simple docstring"""

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

           _lowercase       :							int      		=		CustomFeatureExtractor.from_pretrained(lowerCamelCase)

           with tempfile.TemporaryDirectory() as tmp_dir:
               _lowercase       :							Optional[int]      		=		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       :							Union[str, Any]      		=		CustomProcessor(lowerCamelCase,      lowerCamelCase)

           with tempfile.TemporaryDirectory() as tmp_dir:
               create_repo(F'''{USER}/test-dynamic-processor''',      token=self._token)
               _lowercase       :							Dict      		=		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       :							int      		=		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       :							List[str]      		=		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')






import random
from typing import Any



def 				UpperCamelCase_(      lowerCamelCase_ )    ->						list[Any]:
    for _ in range(len(lowerCamelCase_ ) ):
        _lowercase       :							Optional[int]      		=		random.randint(0							, len(lowerCamelCase_ ) - 1 )
        _lowercase       :							str      		=		random.randint(0							, len(lowerCamelCase_ ) - 1 )
        _lowercase		,				_lowercase       :							Optional[int]      		=		data[b], data[a]
    return data


if __name__ == "__main__":
 SCREAMING_SNAKE_CASE       :      str												= [0, 1, 2, 3, 4, 5, 6, 7]
 SCREAMING_SNAKE_CASE       :      int												= ["python", "says", "hello", "!"]
 print("Fisher-Yates Shuffle:")
 print("List", integers, strings)
 print("FY Shuffle", fisher_yates_shuffle(integers), fisher_yates_shuffle(strings))


