


import json
import os
import re
import shutil
import tempfile
import unittest
from typing import Tuple

from transformers import AddedToken, BatchEncoding, PerceiverTokenizer
from transformers.utils import cached_property, is_tf_available, is_torch_available

from ...test_tokenization_common import TokenizerTesterMixin


if is_torch_available():
						_snake_case    		=       'pt'
elif is_tf_available():
						_snake_case    		=       'tf'
else:
						_snake_case    		=       'jax'







class        lowerCamelCase__	(							_a					,	unittest.TestCase      ):






			"""simple docstring"""






			_UpperCamelCase		:					List[str]    =		PerceiverTokenizer
			_UpperCamelCase		:					Optional[Any]    =		False




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						super().setUp()
						UpperCamelCase__					=       PerceiverTokenizer()
						tokenizer.save_pretrained(self.tmpdirname     )




			@cached_property
			def 				snake_case__				(							self     ):

						'''simple docstring'''


						return PerceiverTokenizer.from_pretrained("deepmind/language-perceiver"     )




			def 				snake_case__				(							self      ,						**snake_case     ):

						'''simple docstring'''


						return self.tokenizer_class.from_pretrained(self.tmpdirname      ,						**_A     )




			def 				snake_case__				(							self      ,						snake_case      ,						snake_case=False      ,						snake_case=20      ,						snake_case=5     ):

						'''simple docstring'''


						UpperCamelCase__					=       []
						for i in range(len(_A     )     ):
									try:
												UpperCamelCase__					=       tokenizer.decode([i]      ,						clean_up_tokenization_spaces=_A     )
									except UnicodeDecodeError:
												pass
									toks.append((i, tok)     )

						UpperCamelCase__					=       list(filter(lambda snake_case     : re.match(R"^[ a-zA-Z]+$"      ,						t[1]     )      ,						_A     )     )
						UpperCamelCase__					=       list(filter(lambda snake_case     : [t[0]] == tokenizer.encode(t[1]      ,						add_special_tokens=_A     )      ,						_A     )     )
						if max_length is not None and len(_A     ) > max_length:
									UpperCamelCase__					=       toks[:max_length]
						if min_length is not None and len(_A     ) < min_length and len(_A     ) > 0:
									while len(_A     ) < min_length:
												UpperCamelCase__					=       toks + toks
        # toks_str = [t[1] for t in toks]
						UpperCamelCase__					=       [t[0] for t in toks]

						# Ensure consistency
						UpperCamelCase__					=       tokenizer.decode(_A      ,						clean_up_tokenization_spaces=_A     )
						if " " not in output_txt and len(_A     ) > 1:
									UpperCamelCase__					=       (
									    tokenizer.decode([toks_ids[0]]      ,						clean_up_tokenization_spaces=_A     )
									    + " "
									    + tokenizer.decode(toks_ids[1:]      ,						clean_up_tokenization_spaces=_A     )
									)
						if with_prefix_space:
									UpperCamelCase__					=       " " + output_txt
						UpperCamelCase__					=       tokenizer.encode(_A      ,						add_special_tokens=_A     )
						return output_txt, output_ids




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						UpperCamelCase__					=       self.perceiver_tokenizer
						UpperCamelCase__					=       "Unicode €."
						UpperCamelCase__					=       tokenizer(_A     )
						UpperCamelCase__					=       [4, 91, 116, 111, 105, 117, 106, 107, 38, 232, 136, 178, 52, 5]
						self.assertEqual(encoded["input_ids"]      ,						_A     )

						# decoding
						UpperCamelCase__					=       tokenizer.decode(_A     )
						self.assertEqual(_A      ,						"[CLS]Unicode €.[SEP]"     )

						UpperCamelCase__					=       tokenizer("e è é ê ë"     )
						UpperCamelCase__					=       [4, 107, 38, 201, 174, 38, 201, 175, 38, 201, 176, 38, 201, 177, 5]
						self.assertEqual(encoded["input_ids"]      ,						_A     )
						# decoding
						UpperCamelCase__					=       tokenizer.decode(_A     )
						self.assertEqual(_A      ,						"[CLS]e è é ê ë[SEP]"     )

						# encode/decode, but with `encode` instead of `__call__`
						self.assertEqual(tokenizer.decode(tokenizer.encode("e è é ê ë"     )     )      ,						"[CLS]e è é ê ë[SEP]"     )




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						UpperCamelCase__					=       self.perceiver_tokenizer
						UpperCamelCase__					=       ["A long paragraph for summarization.", "Another paragraph for summarization."]
						# fmt: off
						UpperCamelCase__					=       [4, 71, 38, 114, 117, 116, 109, 38, 118, 103, 120, 103, 109, 120, 103, 118, 110, 38, 108, 117, 120, 38, 121, 123, 115, 115, 103, 120, 111, 128, 103, 122, 111, 117, 116, 52, 5, 0]
						# fmt: on
						UpperCamelCase__					=       tokenizer(_A      ,						padding=_A      ,						return_tensors=_A     )
						self.assertIsInstance(_A      ,						_A     )

						if FRAMEWORK != "jax":
									UpperCamelCase__					=       list(batch.input_ids.numpy()[0]     )
						else:
									UpperCamelCase__					=       list(batch.input_ids.tolist()[0]     )

						self.assertListEqual(_A      ,						_A     )

						self.assertEqual((2, 38)      ,						batch.input_ids.shape     )
						self.assertEqual((2, 38)      ,						batch.attention_mask.shape     )




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						UpperCamelCase__					=       self.perceiver_tokenizer
						UpperCamelCase__					=       ["A long paragraph for summarization.", "Another paragraph for summarization."]
						UpperCamelCase__					=       tokenizer(_A      ,						padding=_A      ,						return_tensors=_A     )
						# check if input_ids are returned and no decoder_input_ids
						self.assertIn("input_ids"      ,						_A     )
						self.assertIn("attention_mask"      ,						_A     )
						self.assertNotIn("decoder_input_ids"      ,						_A     )
						self.assertNotIn("decoder_attention_mask"      ,						_A     )




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						UpperCamelCase__					=       self.perceiver_tokenizer
						UpperCamelCase__					=       [
						    "Summary of the text.",
						    "Another summary.",
						]
						UpperCamelCase__					=       tokenizer(
						    text_target=_A      ,						max_length=32      ,						padding="max_length"      ,						truncation=_A      ,						return_tensors=_A     )
						self.assertEqual(32      ,						targets["input_ids"].shape[1]     )




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						UpperCamelCase__					=       self.get_tokenizers()
						for tokenizer in tokenizers:
									with self.subTest(F'''{tokenizer.__class__.__name__}'''     ):
												self.assertNotEqual(tokenizer.model_max_length      ,						42     )

        # Now let's start the test
						UpperCamelCase__					=       self.get_tokenizers()
						for tokenizer in tokenizers:
									with self.subTest(F'''{tokenizer.__class__.__name__}'''     ):
												# Isolate this from the other tests because we save additional tokens/etc
												UpperCamelCase__					=       tempfile.mkdtemp()

												UpperCamelCase__					=       " He is very happy, UNwant\u00E9d,running"
												UpperCamelCase__					=       tokenizer.encode(_A      ,						add_special_tokens=_A     )
												tokenizer.save_pretrained(_A     )

												UpperCamelCase__					=       tokenizer.__class__.from_pretrained(_A     )
												UpperCamelCase__					=       after_tokenizer.encode(_A      ,						add_special_tokens=_A     )
												self.assertListEqual(_A      ,						_A     )

												shutil.rmtree(_A     )

						UpperCamelCase__					=       self.get_tokenizers(model_max_length=42     )
						for tokenizer in tokenizers:
									with self.subTest(F'''{tokenizer.__class__.__name__}'''     ):
												# Isolate this from the other tests because we save additional tokens/etc
												UpperCamelCase__					=       tempfile.mkdtemp()

												UpperCamelCase__					=       " He is very happy, UNwant\u00E9d,running"
												tokenizer.add_tokens(["bim", "bambam"]     )
												UpperCamelCase__					=       tokenizer.additional_special_tokens
												additional_special_tokens.append("new_additional_special_token"     )
												tokenizer.add_special_tokens({"additional_special_tokens": additional_special_tokens}     )
												UpperCamelCase__					=       tokenizer.encode(_A      ,						add_special_tokens=_A     )
												tokenizer.save_pretrained(_A     )

												UpperCamelCase__					=       tokenizer.__class__.from_pretrained(_A     )
												UpperCamelCase__					=       after_tokenizer.encode(_A      ,						add_special_tokens=_A     )
												self.assertListEqual(_A      ,						_A     )
												self.assertIn("new_additional_special_token"      ,						after_tokenizer.additional_special_tokens     )
												self.assertEqual(after_tokenizer.model_max_length      ,						42     )

												UpperCamelCase__					=       tokenizer.__class__.from_pretrained(_A      ,						model_max_length=43     )
												self.assertEqual(tokenizer.model_max_length      ,						43     )

												shutil.rmtree(_A     )




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						UpperCamelCase__					=       []
						if self.test_slow_tokenizer:
									tokenizer_list.append((self.tokenizer_class, self.get_tokenizer())     )

						if self.test_rust_tokenizer:
									tokenizer_list.append((self.rust_tokenizer_class, self.get_rust_tokenizer())     )

						for tokenizer_class, tokenizer_utils in tokenizer_list:
									with tempfile.TemporaryDirectory() as tmp_dir:
												tokenizer_utils.save_pretrained(_A     )

												with open(os.path.join(_A      ,						"special_tokens_map.json"     )      ,						encoding="utf-8"     ) as json_file:
															UpperCamelCase__					=       json.load(_A     )

												with open(os.path.join(_A      ,						"tokenizer_config.json"     )      ,						encoding="utf-8"     ) as json_file:
															UpperCamelCase__					=       json.load(_A     )

												UpperCamelCase__					=       [F'''<extra_id_{i}>''' for i in range(125     )]

												UpperCamelCase__					=       added_tokens_extra_ids + [
												    "an_additional_special_token"
												]
												UpperCamelCase__					=       added_tokens_extra_ids + [
												    "an_additional_special_token"
												]

												with open(os.path.join(_A      ,						"special_tokens_map.json"     )      ,						"w"      ,						encoding="utf-8"     ) as outfile:
															json.dump(_A      ,						_A     )
												with open(os.path.join(_A      ,						"tokenizer_config.json"     )      ,						"w"      ,						encoding="utf-8"     ) as outfile:
															json.dump(_A      ,						_A     )

												# the following checks allow us to verify that our test works as expected, i.e. that the tokenizer takes
												# into account the new value of additional_special_tokens given in the "tokenizer_config.json" and
												# "special_tokens_map.json" files
												UpperCamelCase__					=       tokenizer_class.from_pretrained(
												    _A      ,						)
												self.assertIn(
												    "an_additional_special_token"      ,						tokenizer_without_change_in_init.additional_special_tokens     )
												self.assertEqual(
												    ["an_additional_special_token"]      ,						tokenizer_without_change_in_init.convert_ids_to_tokens(
												        tokenizer_without_change_in_init.convert_tokens_to_ids(["an_additional_special_token"]     )     )      ,						)

												# Now we test that we can change the value of additional_special_tokens in the from_pretrained
												UpperCamelCase__					=       added_tokens_extra_ids + [AddedToken("a_new_additional_special_token"      ,						lstrip=_A     )]
												UpperCamelCase__					=       tokenizer_class.from_pretrained(
												    _A      ,						additional_special_tokens=_A      ,						)

												self.assertIn("a_new_additional_special_token"      ,						tokenizer.additional_special_tokens     )
												self.assertEqual(
												    ["a_new_additional_special_token"]      ,						tokenizer.convert_ids_to_tokens(
												        tokenizer.convert_tokens_to_ids(["a_new_additional_special_token"]     )     )      ,						)




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						UpperCamelCase__					=       self.perceiver_tokenizer
						self.assertEqual(tokenizer.decode([178]     )      ,						"�"     )




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						pass




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						pass




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						pass




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						pass







			def 				snake_case__				(							self     ):

						'''simple docstring'''


						UpperCamelCase__					=       self.get_tokenizers(fast=_A      ,						do_lower_case=_A     )
						for tokenizer in tokenizers:
									with self.subTest(F'''{tokenizer.__class__.__name__}'''     ):
												UpperCamelCase__					=       ["[CLS]", "t", "h", "i", "s", " ", "i", "s", " ", "a", " ", "t", "e", "s", "t", "[SEP]"]
												UpperCamelCase__					=       tokenizer.convert_tokens_to_string(_A     )

												self.assertIsInstance(_A      ,						_A     )






import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


__UpperCamelCase    		=       logging.get_logger(__name__)
def        UpperCamelCase_(				)->		Any:
			# Get the sagemaker specific mp parameters from smp_options variable.
			UpperCamelCase__					=       os.getenv("SM_HP_MP_PARAMETERS"				,     "{}"  )
			try:
						# Parse it and check the field "partitions" is included, it is required for model parallel.
						UpperCamelCase__					=       json.loads(_A  )
						if "partitions" not in smp_options:
									return False
			except json.JSONDecodeError:
						return False

			# Get the sagemaker specific framework parameters from mpi_options variable.
			UpperCamelCase__					=       os.getenv("SM_FRAMEWORK_PARAMS"				,     "{}"  )
			try:
						# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
						UpperCamelCase__					=       json.loads(_A  )
						if not mpi_options.get("sagemaker_mpi_enabled"				,     _A  ):
									return False
			except json.JSONDecodeError:
						return False
			# Lastly, check if the `smdistributed` module is present.
			return importlib.util.find_spec("smdistributed"  ) is not None


if is_sagemaker_model_parallel_available():
						import smdistributed.modelparallel.torch as smp

						smp.init()







@dataclass
class        lowerCamelCase__	(							UpperCAmelCase      ):






			"""simple docstring"""






			_UpperCamelCase		:					str    =		field(
			    default=''					,	metadata={'help': 'Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer'}					,	)




			def 				snake_case__				(							self     ):

						'''simple docstring'''


						super().__post_init__()
						warnings.warn(
						    "`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use "
						    "`TrainingArguments` instead."      ,						snake_case      ,						)




			@cached_property
			def 				snake_case__				(							self     ):

						'''simple docstring'''


						logger.info("PyTorch: setting up devices"     )
						if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
									logger.warning(
									    "torch.distributed process group is initialized, but local_rank == -1. "
									    "In order to use Torch DDP, launch your script with `python -m torch.distributed.launch"     )
						if self.no_cuda:
									UpperCamelCase__					=       torch.device("cpu"     )
									UpperCamelCase__					=       0
						elif is_sagemaker_model_parallel_available():
									UpperCamelCase__					=       smp.local_rank()
									UpperCamelCase__					=       torch.device("cuda"      ,						snake_case     )
									UpperCamelCase__					=       1
						elif is_sagemaker_dp_enabled():
									import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

									torch.distributed.init_process_group(backend="smddp"      ,						timeout=self.ddp_timeout_delta     )
									UpperCamelCase__					=       int(os.getenv("SMDATAPARALLEL_LOCAL_RANK"     )     )
									UpperCamelCase__					=       torch.device("cuda"      ,						self.local_rank     )
									UpperCamelCase__					=       1
						elif self.local_rank == -1:
									# if n_gpu is > 1 we'll use nn.DataParallel.
									# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
									# Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
									# trigger an error that a device index is missing. Index 0 takes into account the
									# GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
									# will use the first GPU in that env, i.e. GPU#1
									UpperCamelCase__					=       torch.device("cuda:0" if torch.cuda.is_available() else "cpu"     )
									# Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
									# the default value.
									UpperCamelCase__					=       torch.cuda.device_count()
						else:
									# Here, we'll use torch.distributed.
									# Initializes the distributed backend which will take care of synchronizing nodes/GPUs
									if not torch.distributed.is_initialized():
												torch.distributed.init_process_group(backend="nccl"      ,						timeout=self.ddp_timeout_delta     )
									UpperCamelCase__					=       torch.device("cuda"      ,						self.local_rank     )
									UpperCamelCase__					=       1

						if device.type == "cuda":
									torch.cuda.set_device(snake_case     )

						return device




			@property
			def 				snake_case__				(							self     ):

						'''simple docstring'''


						if is_sagemaker_model_parallel_available():
									return smp.dp_size()

						return super().world_size




			@property
			def 				snake_case__				(							self     ):

						'''simple docstring'''


						return not is_sagemaker_model_parallel_available()







			@property
			def 				snake_case__				(							self     ):

						'''simple docstring'''


						return False


