






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

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

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


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

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
	from test_module.custom_tokenization_fast import CustomTokenizerFast



class 							SCREAMING_SNAKE_CASE__						(    unittest.TestCase						):





					"""simple docstring"""






					def 	_lowercase					( self					:  Optional[int]	):
						# A mock response for an HTTP head request to emulate server down
						snake_case__			:				Tuple       =  mock.Mock()
						snake_case__			:				str       =  5_0_0
						snake_case__			:				Optional[Any]       =  {}
						snake_case__			:				int       =  HTTPError
						snake_case__			:				Dict       =  {}

						# Download this model to make sure it's in the cache.
						snake_case__			:				Dict       =  BertTokenizer.from_pretrained("hf-internal-testing/tiny-random-bert"	)

						# Under the mock environment we get a 500 error when trying to reach the tokenizer.
						with mock.patch("requests.Session.request"							,		return_value=__A	) as mock_head:
							snake_case__			:				List[Any]       =  BertTokenizer.from_pretrained("hf-internal-testing/tiny-random-bert"	)
							# This check we did call the fake head request
							mock_head.assert_called()



					@require_tokenizers
					def 	_lowercase					( self					:  List[Any]	):
						# A mock response for an HTTP head request to emulate server down
						snake_case__			:				List[Any]       =  mock.Mock()
						snake_case__			:				Any       =  5_0_0
						snake_case__			:				List[str]       =  {}
						snake_case__			:				int       =  HTTPError
						snake_case__			:				List[Any]       =  {}

						# Download this model to make sure it's in the cache.
						snake_case__			:				Optional[Any]       =  GPTaTokenizerFast.from_pretrained("gpt2"	)

						# Under the mock environment we get a 500 error when trying to reach the tokenizer.
						with mock.patch("requests.Session.request"							,		return_value=__A	) as mock_head:
							snake_case__			:				Any       =  GPTaTokenizerFast.from_pretrained("gpt2"	)
							# This check we did call the fake head request
							mock_head.assert_called()



					def 	_lowercase					( self					:  str	):
						# This test is for deprecated behavior and can be removed in v5
						try:
							snake_case__			:				List[str]       =  tempfile.mktemp()
							with open(__A							,		"wb"	) as f:
								http_get("https://huggingface.co/albert-base-v1/resolve/main/spiece.model"							,		__A	)

							snake_case__			:				int       =  AlbertTokenizer.from_pretrained(__A	)
						finally:
							os.remove(__A	)

						# Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
						# the current folder and have the right name.
						if os.path.isfile("tokenizer.json"	):
							# We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
							return
						try:
							with open("tokenizer.json"							,		"wb"	) as f:
								http_get("https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json"							,		__A	)
							snake_case__			:				int       =  AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-gpt2"	)
							# The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
							self.assertEqual(tokenizer.vocab_size							,		1_0_0_0	)
							# Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

						finally:
							os.remove("tokenizer.json"	)






					def 	_lowercase					( self					:  Union[str, Any]	):
						# This test is for deprecated behavior and can be removed in v5
						snake_case__			:				Optional[int]       =  AlbertTokenizer.from_pretrained("https://huggingface.co/albert-base-v1/resolve/main/spiece.model"	)



@is_staging_test
class 							SCREAMING_SNAKE_CASE__						(    unittest.TestCase						):





					"""simple docstring"""






					a_     =		["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]



					@classmethod
					def 	_lowercase					( cls					:  Tuple	):
						snake_case__			:				int       =  TOKEN
						HfFolder.save_token(__A	)



					@classmethod
					def 	_lowercase					( cls					:  Optional[Any]	):
						try:
							delete_repo(token=cls._token							,		repo_id="test-tokenizer"	)
						except HTTPError:
							pass

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

						try:
							delete_repo(token=cls._token							,		repo_id="test-dynamic-tokenizer"	)
						except HTTPError:
							pass



					def 	_lowercase					( self					:  Any	):
						with tempfile.TemporaryDirectory() as tmp_dir:
							snake_case__			:				str       =  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]	)	)
							snake_case__			:				Optional[Any]       =  BertTokenizer(__A	)

						tokenizer.push_to_hub("test-tokenizer"							,		use_auth_token=self._token	)
						snake_case__			:				List[Any]       =  BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''	)
						self.assertDictEqual(new_tokenizer.vocab							,		tokenizer.vocab	)

						# Reset repo
						delete_repo(token=self._token							,		repo_id="test-tokenizer"	)

						# Push to hub via save_pretrained
						with tempfile.TemporaryDirectory() as tmp_dir:
							tokenizer.save_pretrained(__A							,		repo_id="test-tokenizer"							,		push_to_hub=__A							,		use_auth_token=self._token	)

						snake_case__			:				List[Any]       =  BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''	)
						self.assertDictEqual(new_tokenizer.vocab							,		tokenizer.vocab	)



					def 	_lowercase					( self					:  Tuple	):
						with tempfile.TemporaryDirectory() as tmp_dir:
							snake_case__			:				Union[str, Any]       =  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]	)	)
							snake_case__			:				Tuple       =  BertTokenizer(__A	)

						tokenizer.push_to_hub("valid_org/test-tokenizer-org"							,		use_auth_token=self._token	)
						snake_case__			:				Optional[Any]       =  BertTokenizer.from_pretrained("valid_org/test-tokenizer-org"	)
						self.assertDictEqual(new_tokenizer.vocab							,		tokenizer.vocab	)

						# Reset repo
						delete_repo(token=self._token							,		repo_id="valid_org/test-tokenizer-org"	)

						# Push to hub via save_pretrained
						with tempfile.TemporaryDirectory() as tmp_dir:
							tokenizer.save_pretrained(
							    __A							,		repo_id="valid_org/test-tokenizer-org"							,		push_to_hub=__A							,		use_auth_token=self._token	)

						snake_case__			:				int       =  BertTokenizer.from_pretrained("valid_org/test-tokenizer-org"	)
						self.assertDictEqual(new_tokenizer.vocab							,		tokenizer.vocab	)






					@require_tokenizers
					def 	_lowercase					( self					:  Tuple	):
						CustomTokenizer.register_for_auto_class()
						with tempfile.TemporaryDirectory() as tmp_dir:
							snake_case__			:				List[Any]       =  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]	)	)
							snake_case__			:				int       =  CustomTokenizer(__A	)

						# No fast custom tokenizer
						tokenizer.push_to_hub("test-dynamic-tokenizer"							,		use_auth_token=self._token	)

						snake_case__			:				Optional[int]       =  AutoTokenizer.from_pretrained(f'''{USER}/test-dynamic-tokenizer'''							,		trust_remote_code=__A	)
						# Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
						self.assertEqual(tokenizer.__class__.__name__							,		"CustomTokenizer"	)

						# Fast and slow custom tokenizer
						CustomTokenizerFast.register_for_auto_class()
						with tempfile.TemporaryDirectory() as tmp_dir:
							snake_case__			:				Union[str, Any]       =  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]	)	)

							snake_case__			:				Optional[Any]       =  BertTokenizerFast.from_pretrained(__A	)
							bert_tokenizer.save_pretrained(__A	)
							snake_case__			:				Dict       =  CustomTokenizerFast.from_pretrained(__A	)

						tokenizer.push_to_hub("test-dynamic-tokenizer"							,		use_auth_token=self._token	)

						snake_case__			:				Dict       =  AutoTokenizer.from_pretrained(f'''{USER}/test-dynamic-tokenizer'''							,		trust_remote_code=__A	)
						# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
						self.assertEqual(tokenizer.__class__.__name__							,		"CustomTokenizerFast"	)
						snake_case__			:				str       =  AutoTokenizer.from_pretrained(
						    f'''{USER}/test-dynamic-tokenizer'''							,		use_fast=__A							,		trust_remote_code=__A	)
						# Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
						self.assertEqual(tokenizer.__class__.__name__							,		"CustomTokenizer"	)



class 							SCREAMING_SNAKE_CASE__						(    unittest.TestCase						):





					"""simple docstring"""






					def 	_lowercase					( self					:  Optional[Any]	):
						snake_case__			:				List[Any]       =  Trie()
						trie.add("Hello 友達"	)
						self.assertEqual(trie.data							,		{"H": {"e": {"l": {"l": {"o": {" ": {"友": {"達": {"": 1}}}}}}}}}	)
						trie.add("Hello"	)
						trie.data
						self.assertEqual(trie.data							,		{"H": {"e": {"l": {"l": {"o": {"": 1, " ": {"友": {"達": {"": 1}}}}}}}}}	)



					def 	_lowercase					( self					:  List[Any]	):
						snake_case__			:				int       =  Trie()
						self.assertEqual(trie.split("[CLS] This is a extra_id_100"	)							,		["[CLS] This is a extra_id_100"]	)
						trie.add("[CLS]"	)
						trie.add("extra_id_1"	)
						trie.add("extra_id_100"	)
						self.assertEqual(trie.split("[CLS] This is a extra_id_100"	)							,		["[CLS]", " This is a ", "extra_id_100"]	)



					def 	_lowercase					( self					:  List[str]	):
						snake_case__			:				List[str]       =  Trie()
						trie.add("A"	)
						self.assertEqual(trie.split("ABC"	)							,		["A", "BC"]	)
						self.assertEqual(trie.split("BCA"	)							,		["BC", "A"]	)



					def 	_lowercase					( self					:  Union[str, Any]	):
						snake_case__			:				Optional[int]       =  Trie()
						trie.add("TOKEN]"	)
						trie.add("[SPECIAL_TOKEN]"	)
						self.assertEqual(trie.split("This is something [SPECIAL_TOKEN]"	)							,		["This is something ", "[SPECIAL_TOKEN]"]	)



					def 	_lowercase					( self					:  Any	):
						snake_case__			:				Optional[int]       =  Trie()
						trie.add("A"	)
						trie.add("P"	)
						trie.add("[SPECIAL_TOKEN]"	)
						self.assertEqual(trie.split("This is something [SPECIAL_TOKEN]"	)							,		["This is something ", "[SPECIAL_TOKEN]"]	)



					def 	_lowercase					( self					:  int	):
						snake_case__			:				Union[str, Any]       =  Trie()
						trie.add("AB"	)
						trie.add("B"	)
						trie.add("C"	)
						self.assertEqual(trie.split("ABC"	)							,		["AB", "C"]	)



					def 	_lowercase					( self					:  Union[str, Any]	):
						snake_case__			:				str       =  Trie()
						trie.add("ABC"	)
						trie.add("B"	)
						trie.add("CD"	)
						self.assertEqual(trie.split("ABCD"	)							,		["ABC", "D"]	)






					def 	_lowercase					( self					:  Any	):
						# Even if the offsets are wrong, we necessarily output correct string
						# parts.
						snake_case__			:				Tuple       =  Trie()
						snake_case__			:				List[Any]       =  trie.cut_text("ABC"							,		[0, 0, 2, 1, 2, 3]	)
						self.assertEqual(__A							,		["AB", "C"]	)











import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType



class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					a_     =		["image_processor", "tokenizer"]
					a_     =		"ViltImageProcessor"
					a_     =		("BertTokenizer", "BertTokenizerFast")



					def __init__( self					:  Optional[int]							,		__A					:  Optional[int]=None							,		__A					:  Optional[Any]=None							,		**__A					:  int	):
						snake_case__			:				Union[str, Any]       =  None
						if "feature_extractor" in kwargs:
							warnings.warn(
							    "The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"
							    " instead."							,		__A							,		)
							snake_case__			:				Tuple       =  kwargs.pop("feature_extractor"	)

						snake_case__			:				Any       =  image_processor if image_processor is not None else feature_extractor
						if image_processor is None:
							raise ValueError("You need to specify an `image_processor`."	)
						if tokenizer is None:
							raise ValueError("You need to specify a `tokenizer`."	)

						super().__init__(__A							,		__A	)
						snake_case__			:				Tuple       =  self.image_processor



					def __call__( self					:  List[Any]							,		__A					:  int							,		__A					:  Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None							,		__A					:  bool = True							,		__A					:  Union[bool, str, PaddingStrategy] = False							,		__A					:  Union[bool, str, TruncationStrategy] = None							,		__A					:  Optional[int] = None							,		__A					:  int = 0							,		__A					:  Optional[int] = None							,		__A					:  Optional[bool] = None							,		__A					:  Optional[bool] = None							,		__A					:  bool = False							,		__A					:  bool = False							,		__A					:  bool = False							,		__A					:  bool = False							,		__A					:  bool = True							,		__A					:  Optional[Union[str, TensorType]] = None							,		**__A					:  List[Any]							,		):
						snake_case__			:				Optional[int]       =  self.tokenizer(
						    text=__A							,		add_special_tokens=__A							,		padding=__A							,		truncation=__A							,		max_length=__A							,		stride=__A							,		pad_to_multiple_of=__A							,		return_token_type_ids=__A							,		return_attention_mask=__A							,		return_overflowing_tokens=__A							,		return_special_tokens_mask=__A							,		return_offsets_mapping=__A							,		return_length=__A							,		verbose=__A							,		return_tensors=__A							,		**__A							,		)
						# add pixel_values + pixel_mask
						snake_case__			:				Optional[Any]       =  self.image_processor(__A							,		return_tensors=__A	)
						encoding.update(__A	)

						return encoding



					def 	_lowercase					( self					:  Optional[Any]							,		*__A					:  List[str]							,		**__A					:  Optional[int]	):
						return self.tokenizer.batch_decode(*__A							,		**__A	)



					def 	_lowercase					( self					:  Dict							,		*__A					:  str							,		**__A					:  str	):
						return self.tokenizer.decode(*__A							,		**__A	)



					@property
					def 	_lowercase					( self					:  str	):
						snake_case__			:				Optional[Any]       =  self.tokenizer.model_input_names
						snake_case__			:				Any       =  self.image_processor.model_input_names
						return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names	)	)



					@property
					def 	_lowercase					( self					:  List[Any]	):
						warnings.warn(
						    "`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."							,		__A							,		)
						return self.image_processor_class






					@property
					def 	_lowercase					( self					:  str	):
						warnings.warn(
						    "`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."							,		__A							,		)
						return self.image_processor



