





'''simple docstring'''

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


lowercase_					= get_tests_dir("""fixtures/dummy_feature_extractor_config.json""")
lowercase_					= get_tests_dir("""fixtures/vocab.json""")
lowercase_					= get_tests_dir("""fixtures""")


class 					a_     (				unittest.TestCase       ):

    '''simple docstring'''
    UpperCamelCase     = ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']


    def 			snake_case_(		self       )		->   Tuple:
        _SCREAMING_SNAKE_CASE									=			0


    def 			snake_case_(		self       )		->   Union[str, Any]:
        _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained("""facebook/wav2vec2-base-960h"""       )
        self.assertIsInstance(A ,		A       )


    def 			snake_case_(		self       )		->   List[Any]:
        with tempfile.TemporaryDirectory() as tmpdirname:
            _SCREAMING_SNAKE_CASE									=			WavaVecaConfig()
            _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained("""facebook/wav2vec2-base-960h"""       )

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

            _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained(A       )

        self.assertIsInstance(A ,		A       )


    def 			snake_case_(		self       )		->   Any:
        with tempfile.TemporaryDirectory() as tmpdirname:
            # copy relevant files
            copyfile(A ,		os.path.join(A ,		A       )       )
            copyfile(A ,		os.path.join(A ,		"""vocab.json"""       )       )

            _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained(A       )

        self.assertIsInstance(A ,		A       )


    def 			snake_case_(		self       )		->   Optional[int]:
        with tempfile.TemporaryDirectory() as tmpdirname:
            _SCREAMING_SNAKE_CASE									=			WavaVecaFeatureExtractor()
            _SCREAMING_SNAKE_CASE									=			AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h"""       )

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

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

            _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained(A       )

        self.assertIsInstance(A ,		A       )


    def 			snake_case_(		self       )		->   Tuple:
        with tempfile.TemporaryDirectory() as tmpdirname:
            _SCREAMING_SNAKE_CASE									=			WavaVecaFeatureExtractor()
            _SCREAMING_SNAKE_CASE									=			AutoTokenizer.from_pretrained("""facebook/wav2vec2-base-960h"""       )

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

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

            _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained(A       )

        self.assertIsInstance(A ,		A       )


    def 			snake_case_(		self       )		->   Optional[int]:
        with tempfile.TemporaryDirectory() as tmpdirname:
            _SCREAMING_SNAKE_CASE									=			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("""{}"""       )

            _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained(A       )

        self.assertIsInstance(A ,		A       )


    def 			snake_case_(		self       )		->   Optional[Any]:
        # If remote code is not set, we will time out when asking whether to load the model.
        with self.assertRaises(A       ):
            _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained("""hf-internal-testing/test_dynamic_processor"""       )
        # If remote code is disabled, we can't load this config.
        with self.assertRaises(A       ):
            _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained(
                """hf-internal-testing/test_dynamic_processor""" ,		trust_remote_code=A       )

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

        _SCREAMING_SNAKE_CASE									=			processor.feature_extractor
        self.assertTrue(feature_extractor.special_attribute_present       )
        self.assertEqual(feature_extractor.__class__.__name__ ,		"""NewFeatureExtractor"""       )

        _SCREAMING_SNAKE_CASE									=			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
            _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained(
                """hf-internal-testing/test_dynamic_processor""" ,		trust_remote_code=A ,		use_fast=A       )
            _SCREAMING_SNAKE_CASE									=			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 			snake_case_(		self       )		->   List[Any]:
        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
            _SCREAMING_SNAKE_CASE									=			CustomFeatureExtractor.from_pretrained(A       )

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

            _SCREAMING_SNAKE_CASE									=			CustomProcessor(A ,		A       )

            with tempfile.TemporaryDirectory() as tmp_dir:
                processor.save_pretrained(A       )
                _SCREAMING_SNAKE_CASE									=			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 			snake_case_(		self       )		->   str:


        class 					a_     (				snake_case_       ):

            '''simple docstring'''
            UpperCamelCase     = False




        class 					a_     (				snake_case_       ):

            '''simple docstring'''
            UpperCamelCase     = False




        class 					a_     (				snake_case_       ):

            '''simple docstring'''
            UpperCamelCase     = '''AutoFeatureExtractor'''
            UpperCamelCase     = '''AutoTokenizer'''
            UpperCamelCase     = 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.
            _SCREAMING_SNAKE_CASE									=			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.
            _SCREAMING_SNAKE_CASE									=			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.
            _SCREAMING_SNAKE_CASE									=			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 			snake_case_(		self       )		->   str:
        _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained("""hf-internal-testing/tiny-random-bert"""       )
        self.assertEqual(processor.__class__.__name__ ,		"""BertTokenizerFast"""       )





    def 			snake_case_(		self       )		->   str:
        _SCREAMING_SNAKE_CASE									=			AutoProcessor.from_pretrained("""hf-internal-testing/tiny-random-convnext"""       )
        self.assertEqual(processor.__class__.__name__ ,		"""ConvNextImageProcessor"""       )




@is_staging_test
class 					a_     (				unittest.TestCase       ):

    '''simple docstring'''
    UpperCamelCase     = ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']


    @classmethod
    def 			snake_case_(		cls       )		->   List[str]:
        _SCREAMING_SNAKE_CASE									=			TOKEN
        HfFolder.save_token(A       )


    @classmethod
    def 			snake_case_(		cls       )		->   int:
        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 			snake_case_(		self       )		->   str:
        _SCREAMING_SNAKE_CASE									=			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       )

            _SCREAMING_SNAKE_CASE									=			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 			snake_case_(		self       )		->   Optional[int]:
        _SCREAMING_SNAKE_CASE									=			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""" ,		)

            _SCREAMING_SNAKE_CASE									=			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 			snake_case_(		self       )		->   Optional[Any]:
        CustomFeatureExtractor.register_for_auto_class()
        CustomTokenizer.register_for_auto_class()
        CustomProcessor.register_for_auto_class()

        _SCREAMING_SNAKE_CASE									=			CustomFeatureExtractor.from_pretrained(A       )

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

        _SCREAMING_SNAKE_CASE									=			CustomProcessor(A ,		A       )

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

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










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


UpperCAmelCase_     							= logging.get_logger(__name__)






class     lowercase__       (	__lowerCamelCase				):

   '''simple docstring'''


   a						:							str   		=		["pixel_values"]




   def __init__(    self,		__magic_name__ = True,		__magic_name__ = 32,		__magic_name__=PILImageResampling.BILINEAR,		__magic_name__ = True,		**__magic_name__,		) ->      None:






       """simple docstring"""
       UpperCamelCase__							:				int          =      do_resize
       UpperCamelCase__							:				Tuple          =      do_rescale
       UpperCamelCase__							:				Any          =      size_divisor
       UpperCamelCase__							:				List[Any]          =      resample
       super().__init__(**__magic_name__							)




   def       UpperCamelCase__  (    self,		__magic_name__,		__magic_name__,		__magic_name__,		__magic_name__ = None,		**__magic_name__							) ->      np.ndarray:






       """simple docstring"""
       UpperCamelCase__   ,UpperCamelCase__							:				List[Any]          =      get_image_size(__magic_name__							)
       # Rounds the height and width down to the closest multiple of size_divisor
       UpperCamelCase__							:				Any          =      height // size_divisor * size_divisor
       UpperCamelCase__							:				Optional[int]          =      width // size_divisor * size_divisor
       UpperCamelCase__							:				str          =      resize(__magic_name__,		(new_h, new_w),		resample=__magic_name__,		data_format=__magic_name__,		**__magic_name__							)
       return image




   def       UpperCamelCase__  (    self,		__magic_name__,		__magic_name__,		__magic_name__ = None,		**__magic_name__							) ->      np.ndarray:






       """simple docstring"""
       return rescale(image=__magic_name__,		scale=__magic_name__,		data_format=__magic_name__,		**__magic_name__							)




   def       UpperCamelCase__  (    self,		__magic_name__,		__magic_name__ = None,		__magic_name__ = None,		__magic_name__=None,		__magic_name__ = None,		__magic_name__ = None,		__magic_name__ = ChannelDimension.FIRST,		**__magic_name__,		) ->      BatchFeature:






       """simple docstring"""
       UpperCamelCase__							:				str          =      do_resize if do_resize is not None else self.do_resize
       UpperCamelCase__							:				Union[str, Any]          =      do_rescale if do_rescale is not None else self.do_rescale
       UpperCamelCase__							:				Any          =      size_divisor if size_divisor is not None else self.size_divisor
       UpperCamelCase__							:				str          =      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'''							)

       UpperCamelCase__							:				Union[str, Any]          =      make_list_of_images(__magic_name__							)

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

       # All transformations expect numpy arrays.
       UpperCamelCase__							:				Optional[Any]          =      [to_numpy_array(__magic_name__							) for img in images]

       if do_resize:
           UpperCamelCase__							:				str          =      [self.resize(__magic_name__,		size_divisor=__magic_name__,		resample=__magic_name__							) for image in images]

       if do_rescale:
           UpperCamelCase__							:				List[Any]          =      [self.rescale(__magic_name__,		scale=1 / 255							) for image in images]

       UpperCamelCase__							:				Optional[Any]          =      [to_channel_dimension_format(__magic_name__,		__magic_name__							) for image in images]

       UpperCamelCase__							:				Tuple          =      {'''pixel_values''': images}
       return BatchFeature(data=__magic_name__,		tensor_type=__magic_name__							)

