


import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

from huggingface_hub import HfFolder, Repository, create_repo, delete_repo
from requests.exceptions import HTTPError

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoProcessor,
    AutoTokenizer,
    BertTokenizer,
    ProcessorMixin,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
)
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test
from transformers.tokenization_utils import TOKENIZER_CONFIG_FILE
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_tokenizers_available


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402
from test_module.custom_processing import CustomProcessor  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


snake_case							: Optional[int]      =   get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
snake_case							: List[Any]      =   get_tests_dir('''fixtures/vocab.json''')
snake_case							: Optional[Any]      =   get_tests_dir('''fixtures''')







class  snake_case_     (unittest.TestCase   ):
					UpperCAmelCase__      :					Tuple					   =    ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']
					def 				lowerCamelCase__(   self     :str   )      ->					Optional[Any]:
												a__    =  0
					def 				lowerCamelCase__(   self     :Tuple   )      ->					Union[str, Any]:
												a__    =  AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'   )
												self.assertIsInstance(__snake_case    ,__snake_case   )
					def 				lowerCamelCase__(   self     :Dict   )      ->					int:
												with tempfile.TemporaryDirectory() as tmpdirname:
																			a__    =  WavaVecaConfig()
																			a__    =  AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'   )

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

																			a__    =  AutoProcessor.from_pretrained(__snake_case   )

												self.assertIsInstance(__snake_case    ,__snake_case   )
					def 				lowerCamelCase__(   self     :Optional[Any]   )      ->					List[str]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																			# copy relevant files
																			copyfile(__snake_case    ,os.path.join(__snake_case    ,__snake_case   )   )
																			copyfile(__snake_case    ,os.path.join(__snake_case    ,'vocab.json'   )   )

																			a__    =  AutoProcessor.from_pretrained(__snake_case   )

												self.assertIsInstance(__snake_case    ,__snake_case   )
					def 				lowerCamelCase__(   self     :Dict   )      ->					List[Any]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																			a__    =  WavaVecaFeatureExtractor()
																			a__    =  AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'   )

																			a__    =  WavaVecaProcessor(__snake_case    ,__snake_case   )

																			# save in new folder
																			processor.save_pretrained(__snake_case   )

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

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

																			a__    =  AutoProcessor.from_pretrained(__snake_case   )

												self.assertIsInstance(__snake_case    ,__snake_case   )
					def 				lowerCamelCase__(   self     :Optional[int]   )      ->					Optional[Any]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																			a__    =  WavaVecaFeatureExtractor()
																			a__    =  AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'   )

																			a__    =  WavaVecaProcessor(__snake_case    ,__snake_case   )

																			# save in new folder
																			processor.save_pretrained(__snake_case   )

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

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

																			a__    =  AutoProcessor.from_pretrained(__snake_case   )

												self.assertIsInstance(__snake_case    ,__snake_case   )
					def 				lowerCamelCase__(   self     :Optional[int]   )      ->					Union[str, Any]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																			a__    =  WavaVecaConfig(processor_class='Wav2Vec2Processor'   )
																			model_config.save_pretrained(__snake_case   )
																			# copy relevant files
																			copyfile(__snake_case    ,os.path.join(__snake_case    ,'vocab.json'   )   )
																			# create emtpy sample processor
																			with open(os.path.join(__snake_case    ,__snake_case   )    ,'w'   ) as f:
																										f.write('{}'   )

																			a__    =  AutoProcessor.from_pretrained(__snake_case   )

												self.assertIsInstance(__snake_case    ,__snake_case   )
					def 				lowerCamelCase__(   self     :List[str]   )      ->					Optional[int]:
												# If remote code is not set, we will time out when asking whether to load the model.
												with self.assertRaises(__snake_case   ):
																			a__    =  AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'   )
												# If remote code is disabled, we can't load this config.
												with self.assertRaises(__snake_case   ):
																			a__    =  AutoProcessor.from_pretrained(
																			    'hf-internal-testing/test_dynamic_processor'    ,trust_remote_code=__snake_case   )

												a__    =  AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'    ,trust_remote_code=__snake_case   )
												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=__snake_case    ,use_fast=__snake_case   )
																			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 				lowerCamelCase__(   self     :Tuple   )      ->					Any:
												try:
																			AutoConfig.register('custom'    ,__snake_case   )
																			AutoFeatureExtractor.register(__snake_case    ,__snake_case   )
																			AutoTokenizer.register(__snake_case    ,slow_tokenizer_class=__snake_case   )
																			AutoProcessor.register(__snake_case    ,__snake_case   )
																			# Trying to register something existing in the Transformers library will raise an error
																			with self.assertRaises(__snake_case   ):
																										AutoProcessor.register(__snake_case    ,__snake_case   )

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

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

																			a__    =  CustomProcessor(__snake_case    ,__snake_case   )

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

												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 				lowerCamelCase__(   self     :str   )      ->					Dict:







												class  snake_case_     (lowerCamelCase_   ):
																	UpperCAmelCase__      :					str					   =    False







												class  snake_case_     (lowerCamelCase_   ):
																	UpperCAmelCase__      :					Tuple					   =    False







												class  snake_case_     (lowerCamelCase_   ):
																	UpperCAmelCase__      :					Dict					   =    '''AutoFeatureExtractor'''
																	UpperCAmelCase__      :					List[str]					   =    '''AutoTokenizer'''
																	UpperCAmelCase__      :					Optional[int]					   =    False

												try:
																			AutoConfig.register('custom'    ,__snake_case   )
																			AutoFeatureExtractor.register(__snake_case    ,__snake_case   )
																			AutoTokenizer.register(__snake_case    ,slow_tokenizer_class=__snake_case   )
																			AutoProcessor.register(__snake_case    ,__snake_case   )
																			# 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=__snake_case   )
																			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=__snake_case   )
																			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 				lowerCamelCase__(   self     :List[str]   )      ->					List[str]:
												a__    =  AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-bert'   )
												self.assertEqual(processor.__class__.__name__    ,'BertTokenizerFast'   )



					def 				lowerCamelCase__(   self     :Dict   )      ->					int:
												a__    =  AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-convnext'   )
												self.assertEqual(processor.__class__.__name__    ,'ConvNextImageProcessor'   )







@is_staging_test
class  snake_case_     (unittest.TestCase   ):
					UpperCAmelCase__      :					Union[str, Any]					   =    ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']
					@classmethod
					def 				lowerCamelCase__(   cls     :Optional[int]   )      ->					Tuple:
												a__    =  TOKEN
												HfFolder.save_token(__snake_case   )
					@classmethod
					def 				lowerCamelCase__(   cls     :Union[str, Any]   )      ->					List[str]:
												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 				lowerCamelCase__(   self     :Union[str, Any]   )      ->					Any:
												a__    =  WavaVecaProcessor.from_pretrained(__snake_case   )
												with tempfile.TemporaryDirectory() as tmp_dir:
																			processor.save_pretrained(
																			    os.path.join(__snake_case    ,'test-processor'   )    ,push_to_hub=__snake_case    ,use_auth_token=self._token   )

																			a__    =  WavaVecaProcessor.from_pretrained(F'{USER}/test-processor'   )
																			for k, v in processor.feature_extractor.__dict__.items():
																										self.assertEqual(__snake_case    ,getattr(new_processor.feature_extractor    ,__snake_case   )   )
																			self.assertDictEqual(new_processor.tokenizer.get_vocab()    ,processor.tokenizer.get_vocab()   )
					def 				lowerCamelCase__(   self     :List[Any]   )      ->					List[Any]:
												a__    =  WavaVecaProcessor.from_pretrained(__snake_case   )

												with tempfile.TemporaryDirectory() as tmp_dir:
																			processor.save_pretrained(
																			    os.path.join(__snake_case    ,'test-processor-org'   )    ,push_to_hub=__snake_case    ,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(__snake_case    ,getattr(new_processor.feature_extractor    ,__snake_case   )   )
																			self.assertDictEqual(new_processor.tokenizer.get_vocab()    ,processor.tokenizer.get_vocab()   )



					def 				lowerCamelCase__(   self     :int   )      ->					int:
												CustomFeatureExtractor.register_for_auto_class()
												CustomTokenizer.register_for_auto_class()
												CustomProcessor.register_for_auto_class()

												a__    =  CustomFeatureExtractor.from_pretrained(__snake_case   )

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

												a__    =  CustomProcessor(__snake_case    ,__snake_case   )

												with tempfile.TemporaryDirectory() as tmp_dir:
																			create_repo(F'{USER}/test-dynamic-processor'    ,token=self._token   )
																			a__    =  Repository(__snake_case    ,clone_from=F'{USER}/test-dynamic-processor'    ,token=self._token   )
																			processor.save_pretrained(__snake_case   )

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

																			repo.push_to_hub()

												a__    =  AutoProcessor.from_pretrained(F'{USER}/test-dynamic-processor'    ,trust_remote_code=__snake_case   )
												# 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 List, Optional, Union

import numpy as np
import PIL.Image

from ...image_processing_utils import BaseImageProcessor, BatchFeature
from ...image_transforms import rescale, resize, to_channel_dimension_format
from ...image_utils import (
    ChannelDimension,
    PILImageResampling,
    get_image_size,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


snake_case							: Dict      =   logging.get_logger(__name__)







class  snake_case_     (lowerCamelCase_   ):
					UpperCAmelCase__      :					Dict					   =    ['''pixel_values''']
					def __init__(   self     :Optional[Any]    ,__snake_case     :bool = True    ,__snake_case     :int = 32    ,__snake_case     :Union[str, Any]=PILImageResampling.BILINEAR    ,__snake_case     :bool = True    ,**__snake_case     :Tuple    ,)      ->					None:
												a__    =  do_resize
												a__    =  do_rescale
												a__    =  size_divisor
												a__    =  resample
												super().__init__(**__snake_case   )
					def 				lowerCamelCase__(   self     :Union[str, Any]    ,__snake_case     :np.ndarray    ,__snake_case     :int    ,__snake_case     :Tuple    ,__snake_case     :Optional[ChannelDimension] = None    ,**__snake_case     :List[Any]   )      ->					np.ndarray:
												a__   ,      a__    =  get_image_size(__snake_case   )
												# Rounds the height and width down to the closest multiple of size_divisor
												a__    =  height // size_divisor * size_divisor
												a__    =  width // size_divisor * size_divisor
												a__    =  resize(__snake_case    ,(new_h, new_w)    ,resample=__snake_case    ,data_format=__snake_case    ,**__snake_case   )
												return image
					def 				lowerCamelCase__(   self     :List[str]    ,__snake_case     :np.ndarray    ,__snake_case     :float    ,__snake_case     :Optional[ChannelDimension] = None    ,**__snake_case     :str   )      ->					np.ndarray:
												return rescale(image=__snake_case    ,scale=__snake_case    ,data_format=__snake_case    ,**__snake_case   )



					def 				lowerCamelCase__(   self     :Tuple    ,__snake_case     :Union["PIL.Image.Image", TensorType, List["PIL.Image.Image"], List[TensorType]]    ,__snake_case     :Optional[bool] = None    ,__snake_case     :Optional[int] = None    ,__snake_case     :Union[str, Any]=None    ,__snake_case     :Optional[bool] = None    ,__snake_case     :Optional[Union[TensorType, str]] = None    ,__snake_case     :ChannelDimension = ChannelDimension.FIRST    ,**__snake_case     :List[Any]    ,)      ->					BatchFeature:
												a__    =  do_resize if do_resize is not None else self.do_resize
												a__    =  do_rescale if do_rescale is not None else self.do_rescale
												a__    =  size_divisor if size_divisor is not None else self.size_divisor
												a__    =  resample if resample is not None else self.resample

												if do_resize and size_divisor is None:
																			raise ValueError('size_divisor is required for resizing'   )

												a__    =  make_list_of_images(__snake_case   )

												if not valid_images(__snake_case   ):
																			raise ValueError('Invalid image(s)'   )

												# All transformations expect numpy arrays.
												a__    =  [to_numpy_array(__snake_case   ) for img in images]

												if do_resize:
																			a__    =  [self.resize(__snake_case    ,size_divisor=__snake_case    ,resample=__snake_case   ) for image in images]

												if do_rescale:
																			a__    =  [self.rescale(__snake_case    ,scale=1 / 2_55   ) for image in images]

												a__    =  [to_channel_dimension_format(__snake_case    ,__snake_case   ) for image in images]

												a__    =  {'pixel_values': images}
												return BatchFeature(data=__snake_case    ,tensor_type=__snake_case   )




