






import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import PaddingStrategy, logging
from .tokenization_realm import RealmTokenizer


__lowerCamelCase   :     Dict 		= logging.get_logger(__name__)

__lowerCamelCase   :     List[str] 		= {'''vocab_file''': '''vocab.txt''', '''tokenizer_file''': '''tokenizer.json'''}

__lowerCamelCase   :     str 		= {
    '''vocab_file''': {
        '''google/realm-cc-news-pretrained-embedder''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-embedder/resolve/main/vocab.txt'''
        ),
        '''google/realm-cc-news-pretrained-encoder''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-encoder/resolve/main/vocab.txt'''
        ),
        '''google/realm-cc-news-pretrained-scorer''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-scorer/resolve/main/vocab.txt'''
        ),
        '''google/realm-cc-news-pretrained-openqa''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-openqa/aresolve/main/vocab.txt'''
        ),
        '''google/realm-orqa-nq-openqa''': '''https://huggingface.co/google/realm-orqa-nq-openqa/resolve/main/vocab.txt''',
        '''google/realm-orqa-nq-reader''': '''https://huggingface.co/google/realm-orqa-nq-reader/resolve/main/vocab.txt''',
        '''google/realm-orqa-wq-openqa''': '''https://huggingface.co/google/realm-orqa-wq-openqa/resolve/main/vocab.txt''',
        '''google/realm-orqa-wq-reader''': '''https://huggingface.co/google/realm-orqa-wq-reader/resolve/main/vocab.txt''',
    },
    '''tokenizer_file''': {
        '''google/realm-cc-news-pretrained-embedder''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-embedder/resolve/main/tokenizer.jsont'''
        ),
        '''google/realm-cc-news-pretrained-encoder''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-encoder/resolve/main/tokenizer.json'''
        ),
        '''google/realm-cc-news-pretrained-scorer''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-scorer/resolve/main/tokenizer.json'''
        ),
        '''google/realm-cc-news-pretrained-openqa''': (
            '''https://huggingface.co/google/realm-cc-news-pretrained-openqa/aresolve/main/tokenizer.json'''
        ),
        '''google/realm-orqa-nq-openqa''': (
            '''https://huggingface.co/google/realm-orqa-nq-openqa/resolve/main/tokenizer.json'''
        ),
        '''google/realm-orqa-nq-reader''': (
            '''https://huggingface.co/google/realm-orqa-nq-reader/resolve/main/tokenizer.json'''
        ),
        '''google/realm-orqa-wq-openqa''': (
            '''https://huggingface.co/google/realm-orqa-wq-openqa/resolve/main/tokenizer.json'''
        ),
        '''google/realm-orqa-wq-reader''': (
            '''https://huggingface.co/google/realm-orqa-wq-reader/resolve/main/tokenizer.json'''
        ),
    },
}

__lowerCamelCase   :     Union[str, Any] 		= {
    '''google/realm-cc-news-pretrained-embedder''': 512,
    '''google/realm-cc-news-pretrained-encoder''': 512,
    '''google/realm-cc-news-pretrained-scorer''': 512,
    '''google/realm-cc-news-pretrained-openqa''': 512,
    '''google/realm-orqa-nq-openqa''': 512,
    '''google/realm-orqa-nq-reader''': 512,
    '''google/realm-orqa-wq-openqa''': 512,
    '''google/realm-orqa-wq-reader''': 512,
}

__lowerCamelCase   :     List[Any] 		= {
    '''google/realm-cc-news-pretrained-embedder''': {'''do_lower_case''': True},
    '''google/realm-cc-news-pretrained-encoder''': {'''do_lower_case''': True},
    '''google/realm-cc-news-pretrained-scorer''': {'''do_lower_case''': True},
    '''google/realm-cc-news-pretrained-openqa''': {'''do_lower_case''': True},
    '''google/realm-orqa-nq-openqa''': {'''do_lower_case''': True},
    '''google/realm-orqa-nq-reader''': {'''do_lower_case''': True},
    '''google/realm-orqa-wq-openqa''': {'''do_lower_case''': True},
    '''google/realm-orqa-wq-reader''': {'''do_lower_case''': True},
}



class 				SCREAMING_SNAKE_CASE__		(							UpperCamelCase_		):





					"""simple docstring"""






					a_      =						VOCAB_FILES_NAMES
					a_      =						PRETRAINED_VOCAB_FILES_MAP
					a_      =						PRETRAINED_INIT_CONFIGURATION
					a_      =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					a_      =						RealmTokenizer



					def __init__(   self	:    List[str]				,						__A	:    List[Any]=None				,						__A	:    List[str]=None				,						__A	:    Union[str, Any]=True				,						__A	:    Optional[Any]="[UNK]"				,						__A	:    List[Any]="[SEP]"				,						__A	:    Optional[int]="[PAD]"				,						__A	:    Union[str, Any]="[CLS]"				,						__A	:    str="[MASK]"				,						__A	:    Tuple=True				,						__A	:    str=None				,						**__A	:    List[Any]				,						):
						super().__init__(
						    _SCREAMING_SNAKE_CASE				,						tokenizer_file=_SCREAMING_SNAKE_CASE				,						do_lower_case=_SCREAMING_SNAKE_CASE				,						unk_token=_SCREAMING_SNAKE_CASE				,						sep_token=_SCREAMING_SNAKE_CASE				,						pad_token=_SCREAMING_SNAKE_CASE				,						cls_token=_SCREAMING_SNAKE_CASE				,						mask_token=_SCREAMING_SNAKE_CASE				,						tokenize_chinese_chars=_SCREAMING_SNAKE_CASE				,						strip_accents=_SCREAMING_SNAKE_CASE				,						**_SCREAMING_SNAKE_CASE				,						)

						snake_case__					: Optional[int]  =    json.loads(self.backend_tokenizer.normalizer.__getstate__()			)
						if (
						    normalizer_state.get("lowercase"				,						_SCREAMING_SNAKE_CASE			) != do_lower_case
						    or normalizer_state.get("strip_accents"				,						_SCREAMING_SNAKE_CASE			) != strip_accents
						    or normalizer_state.get("handle_chinese_chars"				,						_SCREAMING_SNAKE_CASE			) != tokenize_chinese_chars
						):
							snake_case__					: Optional[int]  =    getattr(_SCREAMING_SNAKE_CASE				,						normalizer_state.pop("type"			)			)
							snake_case__					: Any  =    do_lower_case
							snake_case__					: Optional[Any]  =    strip_accents
							snake_case__					: Optional[Any]  =    tokenize_chinese_chars
							snake_case__					: List[str]  =    normalizer_class(**_SCREAMING_SNAKE_CASE			)

						snake_case__					: Optional[int]  =    do_lower_case



					def 			_lowercase	(   self	:    int				,						__A	:    int				,						**__A	:    str			):
						snake_case__					: int  =    PaddingStrategy.MAX_LENGTH

						snake_case__					: Tuple  =    text
						snake_case__					: int  =    kwargs.pop("text_pair"				,						_SCREAMING_SNAKE_CASE			)
						snake_case__					: Any  =    kwargs.pop("return_tensors"				,						_SCREAMING_SNAKE_CASE			)

						snake_case__					: List[str]  =    {
						    "input_ids": [],
						    "attention_mask": [],
						    "token_type_ids": [],
						}

						for idx, candidate_text in enumerate(_SCREAMING_SNAKE_CASE			):
							if batch_text_pair is not None:
								snake_case__					: List[str]  =    batch_text_pair[idx]
							else:
								snake_case__					: List[Any]  =    None

							snake_case__					: Optional[Any]  =    super().__call__(_SCREAMING_SNAKE_CASE				,						_SCREAMING_SNAKE_CASE				,						return_tensors=_SCREAMING_SNAKE_CASE				,						**_SCREAMING_SNAKE_CASE			)

							snake_case__					: Optional[int]  =    encoded_candidates.get("input_ids"			)
							snake_case__					: Tuple  =    encoded_candidates.get("attention_mask"			)
							snake_case__					: str  =    encoded_candidates.get("token_type_ids"			)

							if encoded_input_ids is not None:
								output_data["input_ids"].append(_SCREAMING_SNAKE_CASE			)
							if encoded_attention_mask is not None:
								output_data["attention_mask"].append(_SCREAMING_SNAKE_CASE			)
							if encoded_token_type_ids is not None:
								output_data["token_type_ids"].append(_SCREAMING_SNAKE_CASE			)

						snake_case__					: Any  =    {key: item for key, item in output_data.items() if len(_SCREAMING_SNAKE_CASE			) != 0}

						return BatchEncoding(_SCREAMING_SNAKE_CASE				,						tensor_type=_SCREAMING_SNAKE_CASE			)



					def 			_lowercase	(   self	:    Union[str, Any]				,						__A	:    int				,						__A	:    str=None			):
						snake_case__					: List[Any]  =    [self.cls_token_id] + token_ids_a + [self.sep_token_id]

						if token_ids_a:
							output += token_ids_a + [self.sep_token_id]

						return output



					def 			_lowercase	(   self	:    Union[str, Any]				,						__A	:    Tuple				,						__A	:    Union[str, Any] = None			):
						snake_case__					: Optional[int]  =    [self.sep_token_id]
						snake_case__					: Union[str, Any]  =    [self.cls_token_id]
						if token_ids_a is None:
							return len(cls + token_ids_a + sep			) * [0]
						return len(cls + token_ids_a + sep			) * [0] + len(token_ids_a + sep			) * [1]






					def 			_lowercase	(   self	:    str				,						__A	:    Optional[Any]				,						__A	:    Optional[int] = None			):
						snake_case__					: List[Any]  =    self._tokenizer.model.save(_SCREAMING_SNAKE_CASE				,						name=_SCREAMING_SNAKE_CASE			)
						return tuple(_SCREAMING_SNAKE_CASE			)







import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

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

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


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

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowerCAmelCase__    :Optional[Any]       =				get_tests_dir('''fixtures''')

class 						__a (							unittest.TestCase     ):
		def       UpperCAmelCase__							(     self		)					->			Any:




									"""simple docstring"""





									_UpperCAmelCase											=     mock.Mock()
									_UpperCAmelCase											=     500
									_UpperCAmelCase											=     {}
									_UpperCAmelCase											=     HTTPError
									_UpperCAmelCase											=     {}

									# Download this model to make sure it's in the cache.
									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'		)
									# Under the mock environment we get a 500 error when trying to reach the model.
									with mock.patch('requests.Session.request'     ,       return_value=_SCREAMING_SNAKE_CASE		) as mock_head:
																_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'		)
																# This check we did call the fake head request
																mock_head.assert_called()






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




									"""simple docstring"""





									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(
									    'https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'		)



@is_staging_test
class 						__a (							unittest.TestCase     ):
		@classmethod
		def       UpperCAmelCase__							(     cls		)					->			Tuple:




									"""simple docstring"""





									_UpperCAmelCase											=     TOKEN
									HfFolder.save_token(_SCREAMING_SNAKE_CASE		)
		@classmethod
		def       UpperCAmelCase__							(     cls		)					->			Tuple:




									"""simple docstring"""





									try:
																delete_repo(token=cls._token     ,       repo_id='test-feature-extractor'		)
									except HTTPError:
																pass

									try:
																delete_repo(token=cls._token     ,       repo_id='valid_org/test-feature-extractor-org'		)
									except HTTPError:
																pass

									try:
																delete_repo(token=cls._token     ,       repo_id='test-dynamic-feature-extractor'		)
									except HTTPError:
																pass
		def       UpperCAmelCase__							(     self		)					->			Optional[int]:




									"""simple docstring"""





									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(_SCREAMING_SNAKE_CASE		)
									feature_extractor.push_to_hub('test-feature-extractor'     ,       use_auth_token=self._token		)

									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor'''		)
									for k, v in feature_extractor.__dict__.items():
																self.assertEqual(_SCREAMING_SNAKE_CASE     ,       getattr(_SCREAMING_SNAKE_CASE     ,       _SCREAMING_SNAKE_CASE		)		)

									# Reset repo
									delete_repo(token=self._token     ,       repo_id='test-feature-extractor'		)

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
																feature_extractor.save_pretrained(
																    _SCREAMING_SNAKE_CASE     ,       repo_id='test-feature-extractor'     ,       push_to_hub=_SCREAMING_SNAKE_CASE     ,       use_auth_token=self._token		)

									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(f'''{USER}/test-feature-extractor'''		)
									for k, v in feature_extractor.__dict__.items():
																self.assertEqual(_SCREAMING_SNAKE_CASE     ,       getattr(_SCREAMING_SNAKE_CASE     ,       _SCREAMING_SNAKE_CASE		)		)
		def       UpperCAmelCase__							(     self		)					->			Optional[Any]:




									"""simple docstring"""





									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained(_SCREAMING_SNAKE_CASE		)
									feature_extractor.push_to_hub('valid_org/test-feature-extractor'     ,       use_auth_token=self._token		)

									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor'		)
									for k, v in feature_extractor.__dict__.items():
																self.assertEqual(_SCREAMING_SNAKE_CASE     ,       getattr(_SCREAMING_SNAKE_CASE     ,       _SCREAMING_SNAKE_CASE		)		)

									# Reset repo
									delete_repo(token=self._token     ,       repo_id='valid_org/test-feature-extractor'		)

									# Push to hub via save_pretrained
									with tempfile.TemporaryDirectory() as tmp_dir:
																feature_extractor.save_pretrained(
																    _SCREAMING_SNAKE_CASE     ,       repo_id='valid_org/test-feature-extractor-org'     ,       push_to_hub=_SCREAMING_SNAKE_CASE     ,       use_auth_token=self._token		)

									_UpperCAmelCase											=     WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor-org'		)
									for k, v in feature_extractor.__dict__.items():
																self.assertEqual(_SCREAMING_SNAKE_CASE     ,       getattr(_SCREAMING_SNAKE_CASE     ,       _SCREAMING_SNAKE_CASE		)		)






		def       UpperCAmelCase__							(     self		)					->			Tuple:




									"""simple docstring"""





									CustomFeatureExtractor.register_for_auto_class()
									_UpperCAmelCase											=     CustomFeatureExtractor.from_pretrained(_SCREAMING_SNAKE_CASE		)

									feature_extractor.push_to_hub('test-dynamic-feature-extractor'     ,       use_auth_token=self._token		)

									# This has added the proper auto_map field to the config
									self.assertDictEqual(
									    feature_extractor.auto_map     ,       {'AutoFeatureExtractor': 'custom_feature_extraction.CustomFeatureExtractor'}     ,       )

									_UpperCAmelCase											=     AutoFeatureExtractor.from_pretrained(
									    f'''{USER}/test-dynamic-feature-extractor'''     ,       trust_remote_code=_SCREAMING_SNAKE_CASE		)
									# Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
									self.assertEqual(new_feature_extractor.__class__.__name__     ,       'CustomFeatureExtractor'		)
