






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      UpperCamelCase_							(				UpperCamelCase ,       unittest.TestCase       ):
						lowercase									=  PreTrainedTokenizerFast
						lowercase									=  False
						lowercase									=  True
						lowercase									=  """tokenizer_file"""




						def      snake_case__(    self       )			->						Union[str, Any]:
													_a  :    Optional[int]			=						False  # because we don't have pretrained_vocab_files_map
													super().setUp()
													_a  :    Any			=						True

													_a  :    int			=						["robot-test/dummy-tokenizer-fast", "robot-test/dummy-tokenizer-wordlevel"]
													_a  :    Optional[int]			=						"SaulLu/dummy-tokenizer-bytelevel-bpe"

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

													_a  :    int			=						PreTrainedTokenizerFast.from_pretrained(model_paths[0]       )
													tokenizer.save_pretrained(self.tmpdirname       )




						def      snake_case__(    self       )			->						Union[str, Any]:
													# We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
													# model
													pass




						def      snake_case__(    self       )			->						Optional[int]:
													# We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
													# model
													pass




						def      snake_case__(    self       )			->						Any:
													# We disable this test for PreTrainedTokenizerFast because it is the only tokenizer that is not linked to any
													# model
													pass




						def      snake_case__(    self       )			->						List[Any]:
													# PreTrainedTokenizerFast doesn't have tokenizer_file in its signature
													pass




						def      snake_case__(    self       )			->						str:
													_a  :    int			=						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:
																																		_a  :    List[str]			=						tempfile.mkdtemp()
																																		_a  :    str			=						self.rust_tokenizer_class.from_pretrained(_lowerCamelCase		,	**_lowerCamelCase       )

																																		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       )
																																		_a  :    Dict			=						tmpdirname_orig




						def      snake_case__(    self       )			->						Tuple:
													_a  :    str			=						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:
																																		_a  :    Dict			=						tempfile.mkdtemp()
																																		_a  :    str			=						self.rust_tokenizer_class.from_pretrained(_lowerCamelCase		,	**_lowerCamelCase       )

																																		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       )
																																		_a  :    Tuple			=						tmpdirname_orig







						def      snake_case__(    self       )			->						Dict:
													_a  :    Tuple			=						self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name       )

													_a  :    Union[str, Any]			=						("a" for _ in range(1_000       ))
													_a  :    Union[str, Any]			=						tokenizer.train_new_from_iterator(text_iterator=_lowerCamelCase		,	length=1_000		,	vocab_size=50       )

													_a  :    List[Any]			=						new_tokenizer.encode('''a🤗'''       )
													self.assertEqual(_lowerCamelCase		,	[64, 172, 253, 97, 245]       )

@require_tokenizers
class      UpperCamelCase_							(				unittest.TestCase       ):




						def      snake_case__(    self       )			->						str:
													_a  :    List[str]			=						AutoTokenizer.from_pretrained('''bert-base-cased'''       )
													_a  :    List[Any]			=						json.loads(tokenizer._tokenizer.to_str()       )
													_a  :    int			=						len(_lowerCamelCase       )

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

																				# This should pick the new tokenizer file as the version of Transformers is > 4.0.0
																				_a  :    Tuple			=						AutoTokenizer.from_pretrained(_lowerCamelCase       )
																				self.assertEqual(len(_lowerCamelCase       )		,	len(_lowerCamelCase       ) + 1       )
																				_a  :    Tuple			=						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(_lowerCamelCase		,	'''tokenizer.4.0.0.json'''       )		,	os.path.join(_lowerCamelCase		,	'''tokenizer.42.0.0.json'''       )       )
																				_a  :    List[str]			=						["tokenizer.42.0.0.json"]
																				tokenizer.save_pretrained(_lowerCamelCase       )
																				_a  :    str			=						AutoTokenizer.from_pretrained(_lowerCamelCase       )
																				self.assertEqual(len(_lowerCamelCase       )		,	len(_lowerCamelCase       )       )
																				_a  :    Dict			=						json.loads(new_tokenizer._tokenizer.to_str()       )
																				self.assertNotIn('''huggingface'''		,	json_tokenizer['''model''']['''vocab''']       )







						def      snake_case__(    self       )			->						List[str]:
													# This repo has two tokenizer files, one for v4.0.0 and above with an added token, one for versions lower.
													_a  :    Optional[int]			=						"hf-internal-testing/test-two-tokenizers"

													# This should pick the new tokenizer file as the version of Transformers is > 4.0.0
													_a  :    Dict			=						AutoTokenizer.from_pretrained(_lowerCamelCase       )
													self.assertEqual(len(_lowerCamelCase       )		,	28_997       )
													_a  :    Optional[int]			=						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

													_a  :    Dict			=						"3.0.0"
													_a  :    int			=						old_transformers.models.auto.AutoTokenizer.from_pretrained(_lowerCamelCase       )
													self.assertEqual(len(_lowerCamelCase       )		,	28_996       )
													_a  :    int			=						json.loads(old_tokenizer._tokenizer.to_str()       )
													self.assertNotIn('''huggingface'''		,	json_tokenizer['''model''']['''vocab''']       )

@require_tokenizers
class      UpperCamelCase_							(				unittest.TestCase       ):




						def      snake_case__(    self       )			->						Tuple:
													# See https://github.com/huggingface/transformers/pull/12550
													# and https://github.com/huggingface/tokenizers/issues/537
													_a  :    str			=						PreTrainedTokenizerFast.from_pretrained('''robot-test/dummy-tokenizer-wordlevel'''       )
													_a  :    List[str]			=						"The Matrix is a 1999 science fiction action film."

													with concurrent.futures.ThreadPoolExecutor() as executor:
																				_a  :    List[str]			=						[executor.submit(self.fetch		,	_lowerCamelCase		,	_lowerCamelCase       ) for i in range(10       )]
																				_a  :    Optional[int]			=						[future.result() for future in futures]
																				self.assertEqual(_lowerCamelCase		,	[[1, 10, 0, 8, 0, 18, 0, 0, 0, 2] for i in range(10       )]       )







						def      snake_case__(    self		,	lowercase		,	lowercase       )			->						List[str]:
													return tokenizer.encode(_lowerCamelCase		,	truncation='''longest_first'''		,	padding='''longest'''       )

import torch

from accelerate import Accelerator, DistributedType
class    _UpperCamelCase:

						def __enter__(					self		:			str			):
													if self.enabled:
																				_UpperCAmelCase  :			int	       =		self.model.no_sync()
																				self.model_sync_obj.__enter__()

													return self

						def __exit__(					self		:			str	,						_lowerCamelCase		:			Union[str, Any]	,						_lowerCamelCase		:			Tuple	,						_lowerCamelCase		:			Optional[Any]			):
													if self.enabled:
																				# Average all models on exit
																				self._sync_and_avg_model_params()
																				self.model_sync_obj.__exit__(_lowerCamelCase	,						_lowerCamelCase	,						_lowerCamelCase			)

						def __init__(					self		:			str	,						_lowerCamelCase		:			Accelerator	,						_lowerCamelCase		:			torch.nn.Module	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			bool = True			):
													if accelerator.distributed_type not in [
													    DistributedType.NO,
													    DistributedType.MULTI_CPU,
													    DistributedType.MULTI_GPU,
													]:
																				raise NotImplementedError("LocalSGD is supported only for CPUs and GPUs (no DeepSpeed or MegatronLM)"			)
													_UpperCAmelCase  :			Any	       =		enabled and accelerator.distributed_type != DistributedType.NO
													_UpperCAmelCase  :			str	       =		0
													if self.enabled:
																				_UpperCAmelCase  :			Tuple	       =		accelerator
																				_UpperCAmelCase  :			List[str]	       =		model
																				_UpperCAmelCase  :			Dict	       =		local_sgd_steps

						def 						a__ (					self		:			Optional[int]			):
													self.num_steps += 1
													if not self.enabled:
																				return

													if self.num_steps % self.local_sgd_steps == 0:
																				self._sync_and_avg_model_params()






						def 						a__ (					self		:			Tuple			):
													self.accelerator.wait_for_everyone()
													with self.accelerator.autocast():
																				for param in self.model.parameters():
																											_UpperCAmelCase  :			Optional[Any]	       =		self.accelerator.reduce(param.data	,						reduction="mean"			)




