





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


A       :   str					= get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
A       :   Union[str, Any]					= get_tests_dir('''fixtures/vocab.json''')
A       :   Tuple					= get_tests_dir('''fixtures''')


class     A     (unittest.TestCase       ):
			'''simple docstring'''




			__lowerCamelCase      :			List[Any]							     =       ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']






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







					"""simple docstring"""






					A__     =  0






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







					"""simple docstring"""






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






			def   a_			(     self  :       Dict  )      ->      Union[str, 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(__lowerCAmelCase  )
							processor.save_pretrained(__lowerCAmelCase  )

							A__     =  AutoProcessor.from_pretrained(__lowerCAmelCase  )

					self.assertIsInstance(__lowerCAmelCase				,      __lowerCAmelCase  )






			def   a_			(     self  :       Tuple  )      ->      Optional[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"""  )  )

							A__     =  AutoProcessor.from_pretrained(__lowerCAmelCase  )

					self.assertIsInstance(__lowerCAmelCase				,      __lowerCAmelCase  )






			def   a_			(     self  :       str  )      ->      Dict:







					"""simple docstring"""






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

							A__     =  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:
									A__     =  json.load(__lowerCAmelCase  )
									config_dict.pop("""processor_class"""  )

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

							A__     =  AutoProcessor.from_pretrained(__lowerCAmelCase  )

					self.assertIsInstance(__lowerCAmelCase				,      __lowerCAmelCase  )






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







					"""simple docstring"""






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

							A__     =  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:
									A__     =  json.load(__lowerCAmelCase  )
									config_dict.pop("""processor_class"""  )

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

							A__     =  AutoProcessor.from_pretrained(__lowerCAmelCase  )

					self.assertIsInstance(__lowerCAmelCase				,      __lowerCAmelCase  )






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







					"""simple docstring"""






					with tempfile.TemporaryDirectory() as tmpdirname:
							A__     =  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("""{}"""  )

							A__     =  AutoProcessor.from_pretrained(__lowerCAmelCase  )

					self.assertIsInstance(__lowerCAmelCase				,      __lowerCAmelCase  )






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







					"""simple docstring"""






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

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

					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=__lowerCAmelCase				,      use_fast=__lowerCAmelCase  )
							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   a_			(     self  :       Optional[int]  )      ->      Any:







					"""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
							A__     =  CustomFeatureExtractor.from_pretrained(__lowerCAmelCase  )

							with tempfile.TemporaryDirectory() as tmp_dir:
									A__     =  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]  )  )
									A__     =  CustomTokenizer(__lowerCAmelCase  )

							A__     =  CustomProcessor(__lowerCAmelCase				,      __lowerCAmelCase  )

							with tempfile.TemporaryDirectory() as tmp_dir:
									processor.save_pretrained(__lowerCAmelCase  )
									A__     =  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   a_			(     self  :       List[str]  )      ->      List[str]:







					"""simple docstring"""






					class     A     (SCREAMING_SNAKE_CASE       ):
								'''simple docstring'''




								__lowerCamelCase      :			Dict							     =       False


					class     A     (SCREAMING_SNAKE_CASE       ):
								'''simple docstring'''




								__lowerCamelCase      :			Union[str, Any]							     =       False


					class     A     (SCREAMING_SNAKE_CASE       ):
								'''simple docstring'''




								__lowerCamelCase      :			Union[str, Any]							     =       '''AutoFeatureExtractor'''
								__lowerCamelCase      :			List[Any]							     =       '''AutoTokenizer'''
								__lowerCamelCase      :			Optional[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.
							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=__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.
							A__     =  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   a_			(     self  :       List[str]  )      ->      Tuple:







					"""simple docstring"""






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







			def   a_			(     self  :       Tuple  )      ->      Any:







					"""simple docstring"""






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


@is_staging_test
class     A     (unittest.TestCase       ):
			'''simple docstring'''




			__lowerCamelCase      :			Union[str, Any]							     =       ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']






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







					"""simple docstring"""






					A__     =  TOKEN
					HfFolder.save_token(__lowerCAmelCase  )






			@classmethod
			def   a_			(     cls  :       str  )      ->      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   a_			(     self  :       Union[str, Any]  )      ->      List[Any]:







					"""simple docstring"""






					A__     =  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  )

							A__     =  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   a_			(     self  :       Any  )      ->      Any:







					"""simple docstring"""






					A__     =  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"""				,      )

							A__     =  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   a_			(     self  :       int  )      ->      List[Any]:







					"""simple docstring"""






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

					A__     =  CustomFeatureExtractor.from_pretrained(__lowerCAmelCase  )

					with tempfile.TemporaryDirectory() as tmp_dir:
							A__     =  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]  )  )
							A__     =  CustomTokenizer(__lowerCAmelCase  )

					A__     =  CustomProcessor(__lowerCAmelCase				,      __lowerCAmelCase  )

					with tempfile.TemporaryDirectory() as tmp_dir:
							create_repo(f'{USER}/test-dynamic-processor'				,      token=self._token  )
							A__     =  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:
									A__     =  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()

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










from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


A       :   Optional[int]					= {
    '''configuration_pix2struct''': [
        '''PIX2STRUCT_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''Pix2StructConfig''',
        '''Pix2StructTextConfig''',
        '''Pix2StructVisionConfig''',
    ],
    '''processing_pix2struct''': ['''Pix2StructProcessor'''],
}

try:
					if not is_vision_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					A       :   Optional[int]					= ['''Pix2StructImageProcessor''']


try:
					if not is_torch_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					A       :   Dict					= [
					    '''PIX2STRUCT_PRETRAINED_MODEL_ARCHIVE_LIST''',
					    '''Pix2StructPreTrainedModel''',
					    '''Pix2StructForConditionalGeneration''',
					    '''Pix2StructVisionModel''',
					    '''Pix2StructTextModel''',
					]

if TYPE_CHECKING:
					from .configuration_pixastruct import (
					    PIX2STRUCT_PRETRAINED_CONFIG_ARCHIVE_MAP,
					    PixaStructConfig,
					    PixaStructTextConfig,
					    PixaStructVisionConfig,
					)
					from .processing_pixastruct import PixaStructProcessor

					try:
										if not is_vision_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .image_processing_pixastruct import PixaStructImageProcessor

					try:
										if not is_torch_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_pixastruct import (
										    PIX2STRUCT_PRETRAINED_MODEL_ARCHIVE_LIST,
										    PixaStructForConditionalGeneration,
										    PixaStructPreTrainedModel,
										    PixaStructTextModel,
										    PixaStructVisionModel,
										)

else:
					import sys

					A       :   Any					= _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)



