import concurrent.futures
import json
import os
import shutil
import tempfile
import unittest

from transformers import AutoTokenizer, PreTrainedTokenizerFast
from transformers.testing_utils import require_tokenizers

from ..test_tokenization_common import TokenizerTesterMixin

@require_tokenizers
class 						_SCREAMING_SNAKE_CASE				(  _lowercase   ,  unittest.TestCase				):






		"""simple docstring"""







		A_    			=   PreTrainedTokenizerFast
		A_    			=   False
		A_    			=   True
		A_    			=   '''tokenizer_file'''



		def 					_a					(      self     :					int ):






					"""simple docstring"""
					_snake_case														=   False  # because we don't have pretrained_vocab_files_map
					super().setUp()
					_snake_case														=   True

					_snake_case														=   ["robot-test/dummy-tokenizer-fast", "robot-test/dummy-tokenizer-wordlevel"]
					_snake_case														=   "SaulLu/dummy-tokenizer-bytelevel-bpe"

					# Inclusion of 2 tokenizers to test different types of models (Unigram and WordLevel for the moment)
					_snake_case														=   [(PreTrainedTokenizerFast, model_path, {}) for model_path in model_paths]

					_snake_case														=   PreTrainedTokenizerFast.from_pretrained(model_paths[0] )
					tokenizer.save_pretrained(self.tmpdirname )



		def 					_a					(      self     :					Union[str, Any] ):






					"""simple docstring"""
					# We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
					# model
					pass



		def 					_a					(      self     :					int ):






					"""simple docstring"""
					# We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
					# model
					pass



		def 					_a					(      self     :					str ):






					"""simple docstring"""
					# We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
					# model
					pass



		def 					_a					(      self     :					Union[str, Any] ):






					"""simple docstring"""
					# PreTrainedTokenizerFast doesn't have tokenizer_file in its signature
					pass



		def 					_a					(      self     :					Union[str, Any] ):






					"""simple docstring"""
					_snake_case														=   self.tmpdirname
					# Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
					for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
								with self.subTest(F'{tokenizer.__class__.__name__} ({pretrained_name})' ):
											try:
														_snake_case														=   tempfile.mkdtemp()
														_snake_case														=   self.rust_tokenizer_class.from_pretrained(__SCREAMING_SNAKE_CASE	,       **__SCREAMING_SNAKE_CASE )

														tokenizer.save_pretrained(self.tmpdirname )
														super().test_training_new_tokenizer()
											finally:
														# Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
														# is restored
														shutil.rmtree(self.tmpdirname )
														_snake_case														=   tmpdirname_orig



		def 					_a					(      self     :					Tuple ):






					"""simple docstring"""
					_snake_case														=   self.tmpdirname
					# Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
					for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
								with self.subTest(F'{tokenizer.__class__.__name__} ({pretrained_name})' ):
											try:
														_snake_case														=   tempfile.mkdtemp()
														_snake_case														=   self.rust_tokenizer_class.from_pretrained(__SCREAMING_SNAKE_CASE	,       **__SCREAMING_SNAKE_CASE )

														tokenizer.save_pretrained(self.tmpdirname )
														super().test_training_new_tokenizer_with_special_tokens_change()
											finally:
														# Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
														# is restored
														shutil.rmtree(self.tmpdirname )
														_snake_case														=   tmpdirname_orig







		def 					_a					(      self     :					List[Any] ):






					"""simple docstring"""
					_snake_case														=   self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name )

					_snake_case														=   ("a" for _ in range(1_000 ))
					_snake_case														=   tokenizer.train_new_from_iterator(text_iterator=__SCREAMING_SNAKE_CASE	,       length=1_000	,       vocab_size=50 )

					_snake_case														=   new_tokenizer.encode("a🤗" )
					self.assertEqual(__SCREAMING_SNAKE_CASE	,       [64, 172, 253, 97, 245] )

@require_tokenizers
class 						_SCREAMING_SNAKE_CASE				(  unittest.TestCase				):






		"""simple docstring"""







		def 					_a					(      self     :					Union[str, Any] ):






					"""simple docstring"""
					_snake_case														=   AutoTokenizer.from_pretrained("bert-base-cased" )
					_snake_case														=   json.loads(tokenizer._tokenizer.to_str() )
					_snake_case														=   len(__SCREAMING_SNAKE_CASE )

					with tempfile.TemporaryDirectory() as tmp_dir:
								# Hack to save this in the tokenizer_config.json
								_snake_case														=   ["tokenizer.4.0.0.json"]
								tokenizer.save_pretrained(__SCREAMING_SNAKE_CASE )
								json.dump(__SCREAMING_SNAKE_CASE	,       open(os.path.join(__SCREAMING_SNAKE_CASE	,       "tokenizer.4.0.0.json" )	,       "w" ) )

								# This should pick the new tokenizer file as the version of Transformers is > 4.0.0
								_snake_case														=   AutoTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE )
								self.assertEqual(len(__SCREAMING_SNAKE_CASE )	,       len(__SCREAMING_SNAKE_CASE ) + 1 )
								_snake_case														=   json.loads(new_tokenizer._tokenizer.to_str() )
								self.assertIn("huggingface"	,       json_tokenizer["model"]["vocab"] )

								# Will need to be adjusted if we reach v42 and this test is still here.
								# Should pick the old tokenizer file as the version of Transformers is < 4.0.0
								shutil.move(os.path.join(__SCREAMING_SNAKE_CASE	,       "tokenizer.4.0.0.json" )	,       os.path.join(__SCREAMING_SNAKE_CASE	,       "tokenizer.42.0.0.json" ) )
								_snake_case														=   ["tokenizer.42.0.0.json"]
								tokenizer.save_pretrained(__SCREAMING_SNAKE_CASE )
								_snake_case														=   AutoTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE )
								self.assertEqual(len(__SCREAMING_SNAKE_CASE )	,       len(__SCREAMING_SNAKE_CASE ) )
								_snake_case														=   json.loads(new_tokenizer._tokenizer.to_str() )
								self.assertNotIn("huggingface"	,       json_tokenizer["model"]["vocab"] )







		def 					_a					(      self     :					int ):






					"""simple docstring"""
					# This repo has two tokenizer files, one for v4.0.0 and above with an added token, one for versions lower.
					_snake_case														=   "hf-internal-testing/test-two-tokenizers"

					# This should pick the new tokenizer file as the version of Transformers is > 4.0.0
					_snake_case														=   AutoTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE )
					self.assertEqual(len(__SCREAMING_SNAKE_CASE )	,       28_997 )
					_snake_case														=   json.loads(tokenizer._tokenizer.to_str() )
					self.assertIn("huggingface"	,       json_tokenizer["model"]["vocab"] )

					# Testing an older version by monkey-patching the version in the module it's used.
					import transformers as old_transformers

					_snake_case														=   "3.0.0"
					_snake_case														=   old_transformers.models.auto.AutoTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE )
					self.assertEqual(len(__SCREAMING_SNAKE_CASE )	,       28_996 )
					_snake_case														=   json.loads(old_tokenizer._tokenizer.to_str() )
					self.assertNotIn("huggingface"	,       json_tokenizer["model"]["vocab"] )

@require_tokenizers
class 						_SCREAMING_SNAKE_CASE				(  unittest.TestCase				):






		"""simple docstring"""







		def 					_a					(      self     :					Optional[Any] ):






					"""simple docstring"""
					# See https://github.com/huggingface/transformers/pull/12550
					# and https://github.com/huggingface/tokenizers/issues/537
					_snake_case														=   PreTrainedTokenizerFast.from_pretrained("robot-test/dummy-tokenizer-wordlevel" )
					_snake_case														=   "The Matrix is a 1999 science fiction action film."

					with concurrent.futures.ThreadPoolExecutor() as executor:
								_snake_case														=   [executor.submit(self.fetch	,       __SCREAMING_SNAKE_CASE	,       __SCREAMING_SNAKE_CASE ) for i in range(10 )]
								_snake_case														=   [future.result() for future in futures]
								self.assertEqual(__SCREAMING_SNAKE_CASE	,       [[1, 10, 0, 8, 0, 18, 0, 0, 0, 2] for i in range(10 )] )







		def 					_a					(      self     :					Optional[Any]	,       __SCREAMING_SNAKE_CASE     :					Optional[Any]	,       __SCREAMING_SNAKE_CASE     :					Union[str, Any] ):






					"""simple docstring"""
					return tokenizer.encode(__SCREAMING_SNAKE_CASE	,       truncation="longest_first"	,       padding="longest" )


import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
			import torch

if is_vision_available():
			from PIL import Image

			from transformers import ViltImageProcessor

class 						_SCREAMING_SNAKE_CASE				(  unittest.TestCase				):






		"""simple docstring"""







		def __init__(      self     :					Optional[int]	,       __SCREAMING_SNAKE_CASE     :					List[str]	,       __SCREAMING_SNAKE_CASE     :					Union[str, Any]=7	,       __SCREAMING_SNAKE_CASE     :					Dict=3	,       __SCREAMING_SNAKE_CASE     :					Any=18	,       __SCREAMING_SNAKE_CASE     :					Optional[int]=30	,       __SCREAMING_SNAKE_CASE     :					int=400	,       __SCREAMING_SNAKE_CASE     :					int=True	,       __SCREAMING_SNAKE_CASE     :					Optional[int]=None	,       __SCREAMING_SNAKE_CASE     :					Any=2	,       __SCREAMING_SNAKE_CASE     :					Any=True	,       __SCREAMING_SNAKE_CASE     :					int=[0.5, 0.5, 0.5]	,       __SCREAMING_SNAKE_CASE     :					List[str]=[0.5, 0.5, 0.5]	,       ):






					"""simple docstring"""
					_snake_case														=   size if size is not None else {"shortest_edge": 30}
					_snake_case														=   parent
					_snake_case														=   batch_size
					_snake_case														=   num_channels
					_snake_case														=   image_size
					_snake_case														=   min_resolution
					_snake_case														=   max_resolution
					_snake_case														=   do_resize
					_snake_case														=   size
					_snake_case														=   size_divisor
					_snake_case														=   do_normalize
					_snake_case														=   image_mean
					_snake_case														=   image_std



		def 					_a					(      self     :					Optional[int] ):






					"""simple docstring"""
					return {
					    "image_mean": self.image_mean,
					    "image_std": self.image_std,
					    "do_normalize": self.do_normalize,
					    "do_resize": self.do_resize,
					    "size": self.size,
					    "size_divisor": self.size_divisor,
					}







		def 					_a					(      self     :					Any	,       __SCREAMING_SNAKE_CASE     :					List[str]	,       __SCREAMING_SNAKE_CASE     :					List[Any]=False ):






					"""simple docstring"""
					if not batched:
								_snake_case														=   self.size["shortest_edge"]
								_snake_case														=   image_inputs[0]
								if isinstance(__SCREAMING_SNAKE_CASE	,       Image.Image ):
											_snake_case							,			_snake_case														=   image.size
								else:
											_snake_case							,			_snake_case														=   image.shape[1], image.shape[2]
								_snake_case														=   size / min(__SCREAMING_SNAKE_CASE	,       __SCREAMING_SNAKE_CASE )
								if h < w:
											_snake_case							,			_snake_case														=   size, scale * w
								else:
											_snake_case							,			_snake_case														=   scale * h, size

								_snake_case														=   int((1_333 / 800) * size )
								if max(__SCREAMING_SNAKE_CASE	,       __SCREAMING_SNAKE_CASE ) > max_size:
											_snake_case														=   max_size / max(__SCREAMING_SNAKE_CASE	,       __SCREAMING_SNAKE_CASE )
											_snake_case														=   newh * scale
											_snake_case														=   neww * scale

								_snake_case							,			_snake_case														=   int(newh + 0.5 ), int(neww + 0.5 )
								_snake_case							,			_snake_case														=   (
								    newh // self.size_divisor * self.size_divisor,
								    neww // self.size_divisor * self.size_divisor,
								)

					else:
								_snake_case														=   []
								for image in image_inputs:
											_snake_case							,			_snake_case														=   self.get_expected_values([image] )
											expected_values.append((expected_height, expected_width) )
								_snake_case														=   max(__SCREAMING_SNAKE_CASE	,       key=lambda __SCREAMING_SNAKE_CASE : item[0] )[0]
								_snake_case														=   max(__SCREAMING_SNAKE_CASE	,       key=lambda __SCREAMING_SNAKE_CASE : item[1] )[1]

					return expected_height, expected_width

@require_torch
@require_vision
class 						_SCREAMING_SNAKE_CASE				(  _lowercase   ,  unittest.TestCase				):






		"""simple docstring"""







		A_    			=   ViltImageProcessor if is_vision_available() else None



		def 					_a					(      self     :					Tuple ):






					"""simple docstring"""
					_snake_case														=   ViltImageProcessingTester(self )



		@property
		def 					_a					(      self     :					List[Any] ):






					"""simple docstring"""
					return self.image_processor_tester.prepare_image_processor_dict()



		def 					_a					(      self     :					Optional[int] ):






					"""simple docstring"""
					_snake_case														=   self.image_processing_class(**self.image_processor_dict )
					self.assertTrue(hasattr(__SCREAMING_SNAKE_CASE	,       "image_mean" ) )
					self.assertTrue(hasattr(__SCREAMING_SNAKE_CASE	,       "image_std" ) )
					self.assertTrue(hasattr(__SCREAMING_SNAKE_CASE	,       "do_normalize" ) )
					self.assertTrue(hasattr(__SCREAMING_SNAKE_CASE	,       "do_resize" ) )
					self.assertTrue(hasattr(__SCREAMING_SNAKE_CASE	,       "size" ) )
					self.assertTrue(hasattr(__SCREAMING_SNAKE_CASE	,       "size_divisor" ) )



		def 					_a					(      self     :					Optional[Any] ):






					"""simple docstring"""
					_snake_case														=   self.image_processing_class.from_dict(self.image_processor_dict )
					self.assertEqual(image_processor.size	,       {"shortest_edge": 30} )

					_snake_case														=   self.image_processing_class.from_dict(self.image_processor_dict	,       size=42 )
					self.assertEqual(image_processor.size	,       {"shortest_edge": 42} )



		def 					_a					(      self     :					Dict ):






					"""simple docstring"""
					pass



		def 					_a					(      self     :					List[Any] ):






					"""simple docstring"""
					# Initialize image_processing
					_snake_case														=   self.image_processing_class(**self.image_processor_dict )
					# create random PIL images
					_snake_case														=   prepare_image_inputs(self.image_processor_tester	,       equal_resolution=__SCREAMING_SNAKE_CASE )
					for image in image_inputs:
								self.assertIsInstance(__SCREAMING_SNAKE_CASE	,       Image.Image )

					# Test not batched input
					_snake_case														=   image_processing(image_inputs[0]	,       return_tensors="pt" ).pixel_values

					_snake_case							,			_snake_case														=   self.image_processor_tester.get_expected_values(__SCREAMING_SNAKE_CASE )
					self.assertEqual(
					    encoded_images.shape	,       (1, self.image_processor_tester.num_channels, expected_height, expected_width)	,       )

					# Test batched
					_snake_case														=   image_processing(__SCREAMING_SNAKE_CASE	,       return_tensors="pt" ).pixel_values

					_snake_case							,			_snake_case														=   self.image_processor_tester.get_expected_values(__SCREAMING_SNAKE_CASE	,       batched=__SCREAMING_SNAKE_CASE )
					self.assertEqual(
					    encoded_images.shape	,       (
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_channels,
					        expected_height,
					        expected_width,
					    )	,       )



		def 					_a					(      self     :					List[Any] ):






					"""simple docstring"""
					# Initialize image_processing
					_snake_case														=   self.image_processing_class(**self.image_processor_dict )
					# create random numpy tensors
					_snake_case														=   prepare_image_inputs(self.image_processor_tester	,       equal_resolution=__SCREAMING_SNAKE_CASE	,       numpify=__SCREAMING_SNAKE_CASE )
					for image in image_inputs:
								self.assertIsInstance(__SCREAMING_SNAKE_CASE	,       np.ndarray )

					# Test not batched input
					_snake_case														=   image_processing(image_inputs[0]	,       return_tensors="pt" ).pixel_values

					_snake_case							,			_snake_case														=   self.image_processor_tester.get_expected_values(__SCREAMING_SNAKE_CASE )
					self.assertEqual(
					    encoded_images.shape	,       (1, self.image_processor_tester.num_channels, expected_height, expected_width)	,       )

					# Test batched
					_snake_case														=   image_processing(__SCREAMING_SNAKE_CASE	,       return_tensors="pt" ).pixel_values

					_snake_case							,			_snake_case														=   self.image_processor_tester.get_expected_values(__SCREAMING_SNAKE_CASE	,       batched=__SCREAMING_SNAKE_CASE )
					self.assertEqual(
					    encoded_images.shape	,       (
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_channels,
					        expected_height,
					        expected_width,
					    )	,       )







		def 					_a					(      self     :					Dict ):






					"""simple docstring"""
					# Initialize image_processing
					_snake_case														=   self.image_processing_class(**self.image_processor_dict )
					# create random PyTorch tensors
					_snake_case														=   prepare_image_inputs(self.image_processor_tester	,       equal_resolution=__SCREAMING_SNAKE_CASE	,       torchify=__SCREAMING_SNAKE_CASE )
					for image in image_inputs:
								self.assertIsInstance(__SCREAMING_SNAKE_CASE	,       torch.Tensor )

					# Test not batched input
					_snake_case														=   image_processing(image_inputs[0]	,       return_tensors="pt" ).pixel_values

					_snake_case							,			_snake_case														=   self.image_processor_tester.get_expected_values(__SCREAMING_SNAKE_CASE )
					self.assertEqual(
					    encoded_images.shape	,       (1, self.image_processor_tester.num_channels, expected_height, expected_width)	,       )

					# Test batched
					_snake_case														=   image_processing(__SCREAMING_SNAKE_CASE	,       return_tensors="pt" ).pixel_values

					_snake_case							,			_snake_case														=   self.image_processor_tester.get_expected_values(__SCREAMING_SNAKE_CASE	,       batched=__SCREAMING_SNAKE_CASE )
					self.assertEqual(
					    encoded_images.shape	,       (
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_channels,
					        expected_height,
					        expected_width,
					    )	,       )

