
'''simple docstring'''


import json
import os
import unittest

from transformers import BatchEncoding, LEDTokenizer, LEDTokenizerFast
from transformers.models.led.tokenization_led import VOCAB_FILES_NAMES
from transformers.testing_utils import require_tokenizers, require_torch
from transformers.utils import cached_property

from ...test_tokenization_common import TokenizerTesterMixin

@require_tokenizers
class 	__lowercase				(  _lowercase      ,     unittest.TestCase    ):
			lowerCamelCase			:		str	      =			LEDTokenizer
			lowerCamelCase			:		Optional[Any]	      =			LEDTokenizerFast
			lowerCamelCase			:		Any	      =			True

			def        UpperCAmelCase__ (self      ):
								super().setUp()
								lowerCamelCase_					:      List[str]     =		[
								    '''l''',
								    '''o''',
								    '''w''',
								    '''e''',
								    '''r''',
								    '''s''',
								    '''t''',
								    '''i''',
								    '''d''',
								    '''n''',
								    '''\u0120''',
								    '''\u0120l''',
								    '''\u0120n''',
								    '''\u0120lo''',
								    '''\u0120low''',
								    '''er''',
								    '''\u0120lowest''',
								    '''\u0120newer''',
								    '''\u0120wider''',
								    '''<unk>''',
								]
								lowerCamelCase_					:      List[str]     =		dict(zip(A			,   range(len(A      )      )      )      )
								lowerCamelCase_					:      int     =		['''#version: 0.2''', '''\u0120 l''', '''\u0120l o''', '''\u0120lo w''', '''e r''', '''''']
								lowerCamelCase_					:      Optional[Any]     =		{'''unk_token''': '''<unk>'''}

								lowerCamelCase_					:      List[Any]     =		os.path.join(self.tmpdirname			,   VOCAB_FILES_NAMES['''vocab_file''']      )
								lowerCamelCase_					:      Union[str, Any]     =		os.path.join(self.tmpdirname			,   VOCAB_FILES_NAMES['''merges_file''']      )
								with open(self.vocab_file			,   '''w'''			,   encoding='''utf-8'''      ) as fp:
													fp.write(json.dumps(A      ) + '''\n'''      )
								with open(self.merges_file			,   '''w'''			,   encoding='''utf-8'''      ) as fp:
													fp.write('''\n'''.join(A      )      )

			def        UpperCAmelCase__ (self			,   **A      ):
								kwargs.update(self.special_tokens_map      )
								return self.tokenizer_class.from_pretrained(self.tmpdirname			,   **A      )

			def        UpperCAmelCase__ (self			,   **A      ):
								kwargs.update(self.special_tokens_map      )
								return self.rust_tokenizer_class.from_pretrained(self.tmpdirname			,   **A      )

			def        UpperCAmelCase__ (self			,   A      ):
								return "lower newer", "lower newer"

			@cached_property
			def        UpperCAmelCase__ (self      ):
								return LEDTokenizer.from_pretrained('''allenai/led-base-16384'''      )

			@cached_property
			def        UpperCAmelCase__ (self      ):
								return LEDTokenizerFast.from_pretrained('''allenai/led-base-16384'''      )

			@require_torch
			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      str     =		['''A long paragraph for summarization.''', '''Another paragraph for summarization.''']
								lowerCamelCase_					:      List[Any]     =		[0, 2_5_0, 2_5_1, 1_7_8_1_8, 1_3, 3_9_1_8_6, 1_9_3_8, 4, 2]

								for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
													lowerCamelCase_					:      int     =		tokenizer(A			,   max_length=len(A      )			,   padding=A			,   return_tensors='''pt'''      )
													self.assertIsInstance(A			,   A      )

													self.assertEqual((2, 9)			,   batch.input_ids.shape      )
													self.assertEqual((2, 9)			,   batch.attention_mask.shape      )
													lowerCamelCase_					:      Dict     =		batch.input_ids.tolist()[0]
													self.assertListEqual(A			,   A      )

			@require_torch
			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      Tuple     =		['''A long paragraph for summarization.''', '''Another paragraph for summarization.''']
								for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
													lowerCamelCase_					:      int     =		tokenizer(A			,   padding=A			,   return_tensors='''pt'''      )
													self.assertIn('''input_ids'''			,   A      )
													self.assertIn('''attention_mask'''			,   A      )
													self.assertNotIn('''labels'''			,   A      )
													self.assertNotIn('''decoder_attention_mask'''			,   A      )

			@require_torch
			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      List[Any]     =		[
								    '''Summary of the text.''',
								    '''Another summary.''',
								]
								for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
													lowerCamelCase_					:      Dict     =		tokenizer(text_target=A			,   max_length=3_2			,   padding='''max_length'''			,   return_tensors='''pt'''      )
													self.assertEqual(3_2			,   targets['''input_ids'''].shape[1]      )

			@require_torch
			def        UpperCAmelCase__ (self      ):
								for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
													lowerCamelCase_					:      Optional[Any]     =		tokenizer(
													    ['''I am a small frog''' * 1_0_2_4, '''I am a small frog''']			,   padding=A			,   truncation=A			,   return_tensors='''pt'''      )
													self.assertIsInstance(A			,   A      )
													self.assertEqual(batch.input_ids.shape			,   (2, 5_1_2_2)      )

			@require_torch
			def        UpperCAmelCase__ (self      ):
								lowerCamelCase_					:      int     =		['''A long paragraph for summarization.''']
								lowerCamelCase_					:      Optional[int]     =		[
								    '''Summary of the text.''',
								]
								for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
													lowerCamelCase_					:      str     =		tokenizer(A			,   return_tensors='''pt'''      )
													lowerCamelCase_					:      List[Any]     =		tokenizer(text_target=A			,   return_tensors='''pt'''      )
													lowerCamelCase_					:      List[Any]     =		inputs['''input_ids''']
													lowerCamelCase_					:      Tuple     =		targets['''input_ids''']
													self.assertTrue((input_ids[:, 0] == tokenizer.bos_token_id).all().item()      )
													self.assertTrue((labels[:, 0] == tokenizer.bos_token_id).all().item()      )
													self.assertTrue((input_ids[:, -1] == tokenizer.eos_token_id).all().item()      )
													self.assertTrue((labels[:, -1] == tokenizer.eos_token_id).all().item()      )

			@require_torch
			def        UpperCAmelCase__ (self      ):
								for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
													lowerCamelCase_					:      Tuple     =		['''Summary of the text.''', '''Another summary.''']
													lowerCamelCase_					:      Union[str, Any]     =		[[0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, -1, -1]]

													lowerCamelCase_					:      int     =		tokenizer(A			,   padding=A      )
													lowerCamelCase_					:      int     =		[[0] * len(A      ) for x in encoded_output['''input_ids''']]
													lowerCamelCase_					:      int     =		tokenizer.pad(A      )
													self.assertSequenceEqual(outputs['''global_attention_mask''']			,   A      )

			def        UpperCAmelCase__ (self      ):
								pass






			def        UpperCAmelCase__ (self      ):
								for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
													with self.subTest(F"""{tokenizer.__class__.__name__} ({pretrained_name})"""      ):
																		lowerCamelCase_					:      List[str]     =		self.rust_tokenizer_class.from_pretrained(A			,   **A      )
																		lowerCamelCase_					:      Any     =		self.tokenizer_class.from_pretrained(A			,   **A      )
																		lowerCamelCase_					:      Any     =		'''A, <mask> AllenNLP sentence.'''
																		lowerCamelCase_					:      Union[str, Any]     =		tokenizer_r.encode_plus(A			,   add_special_tokens=A			,   return_token_type_ids=A      )
																		lowerCamelCase_					:      Union[str, Any]     =		tokenizer_p.encode_plus(A			,   add_special_tokens=A			,   return_token_type_ids=A      )
																		self.assertEqual(sum(tokens_r['''token_type_ids''']      )			,   sum(tokens_p['''token_type_ids''']      )      )
																		self.assertEqual(
																		    sum(tokens_r['''attention_mask''']      ) / len(tokens_r['''attention_mask''']      )			,   sum(tokens_p['''attention_mask''']      ) / len(tokens_p['''attention_mask''']      )			,   )

																		lowerCamelCase_					:      Dict     =		tokenizer_r.convert_ids_to_tokens(tokens_r['''input_ids''']      )
																		lowerCamelCase_					:      Optional[int]     =		tokenizer_p.convert_ids_to_tokens(tokens_p['''input_ids''']      )
																		self.assertSequenceEqual(tokens_p['''input_ids''']			,   [0, 2_5_0, 6, 5_0_2_6_4, 3_8_2_3, 4_8_7, 2_1_9_9_2, 3_6_4_5, 4, 2]      )
																		self.assertSequenceEqual(tokens_r['''input_ids''']			,   [0, 2_5_0, 6, 5_0_2_6_4, 3_8_2_3, 4_8_7, 2_1_9_9_2, 3_6_4_5, 4, 2]      )

																		self.assertSequenceEqual(
																		    A			,   ['''<s>''', '''A''', ''',''', '''<mask>''', '''ĠAllen''', '''N''', '''LP''', '''Ġsentence''', '''.''', '''</s>''']      )
																		self.assertSequenceEqual(
																		    A			,   ['''<s>''', '''A''', ''',''', '''<mask>''', '''ĠAllen''', '''N''', '''LP''', '''Ġsentence''', '''.''', '''</s>''']      )




'''simple docstring'''


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


__lowercase							:   Union[str, Any]					=     logging.get_logger(__name__)




def lowercase_	(   )     ->							Union[str, Any]:
					'''simple docstring'''

					lowerCamelCase_					:      Optional[Any]     =		os.getenv('''SM_HP_MP_PARAMETERS'''			,       '''{}'''				)
					try:
										# Parse it and check the field "partitions" is included, it is required for model parallel.
										lowerCamelCase_					:      Optional[Any]     =		json.loads(_lowercase				)
										if "partitions" not in smp_options:
															return False
					except json.JSONDecodeError:
										return False

					# Get the sagemaker specific framework parameters from mpi_options variable.
					lowerCamelCase_					:      List[str]     =		os.getenv('''SM_FRAMEWORK_PARAMS'''			,       '''{}'''				)
					try:
										# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
										lowerCamelCase_					:      List[str]     =		json.loads(_lowercase				)
										if not mpi_options.get('''sagemaker_mpi_enabled'''			,       _lowercase				):
															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 	__lowercase				(  _lowercase    ):
			lowerCamelCase			:		str	      =			field(
			    default=""      ,     metadata={"help": "Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer"}      ,     )

			def        UpperCAmelCase__ (self      ):
								super().__post_init__()
								warnings.warn(
								    '''`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use '''
								    '''`TrainingArguments` instead.'''			,   A			,   )

			@cached_property
			def        UpperCAmelCase__ (self      ):
								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:
													lowerCamelCase_					:      Optional[int]     =		torch.device('''cpu'''      )
													lowerCamelCase_					:      Dict     =		0
								elif is_sagemaker_model_parallel_available():
													lowerCamelCase_					:      int     =		smp.local_rank()
													lowerCamelCase_					:      str     =		torch.device('''cuda'''			,   A      )
													lowerCamelCase_					:      Tuple     =		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      )
													lowerCamelCase_					:      List[Any]     =		int(os.getenv('''SMDATAPARALLEL_LOCAL_RANK'''      )      )
													lowerCamelCase_					:      Union[str, Any]     =		torch.device('''cuda'''			,   self.local_rank      )
													lowerCamelCase_					:      List[str]     =		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
													lowerCamelCase_					:      List[str]     =		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.
													lowerCamelCase_					:      Any     =		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      )
													lowerCamelCase_					:      Any     =		torch.device('''cuda'''			,   self.local_rank      )
													lowerCamelCase_					:      List[str]     =		1

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

								return device

			@property
			def        UpperCAmelCase__ (self      ):
								if is_sagemaker_model_parallel_available():
													return smp.dp_size()

								return super().world_size

			@property
			def        UpperCAmelCase__ (self      ):
								return not is_sagemaker_model_parallel_available()






			@property
			def        UpperCAmelCase__ (self      ):
								return False


