"""simple docstring"""





import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase_     			=						logging.get_logger(__name__)


lowerCAmelCase_     			=						{'vocab_file': 'vocab.json', 'merges_file': 'merges.txt'}

lowerCAmelCase_     			=						{
    'vocab_file': {
        'allenai/longformer-base-4096': 'https://huggingface.co/allenai/longformer-base-4096/resolve/main/vocab.json',
        'allenai/longformer-large-4096': (
            'https://huggingface.co/allenai/longformer-large-4096/resolve/main/vocab.json'
        ),
        'allenai/longformer-large-4096-finetuned-triviaqa': (
            'https://huggingface.co/allenai/longformer-large-4096-finetuned-triviaqa/resolve/main/vocab.json'
        ),
        'allenai/longformer-base-4096-extra.pos.embd.only': (
            'https://huggingface.co/allenai/longformer-base-4096-extra.pos.embd.only/resolve/main/vocab.json'
        ),
        'allenai/longformer-large-4096-extra.pos.embd.only': (
            'https://huggingface.co/allenai/longformer-large-4096-extra.pos.embd.only/resolve/main/vocab.json'
        ),
    },
    'merges_file': {
        'allenai/longformer-base-4096': 'https://huggingface.co/allenai/longformer-base-4096/resolve/main/merges.txt',
        'allenai/longformer-large-4096': (
            'https://huggingface.co/allenai/longformer-large-4096/resolve/main/merges.txt'
        ),
        'allenai/longformer-large-4096-finetuned-triviaqa': (
            'https://huggingface.co/allenai/longformer-large-4096-finetuned-triviaqa/resolve/main/merges.txt'
        ),
        'allenai/longformer-base-4096-extra.pos.embd.only': (
            'https://huggingface.co/allenai/longformer-base-4096-extra.pos.embd.only/resolve/main/merges.txt'
        ),
        'allenai/longformer-large-4096-extra.pos.embd.only': (
            'https://huggingface.co/allenai/longformer-large-4096-extra.pos.embd.only/resolve/main/merges.txt'
        ),
    },
}

lowerCAmelCase_     			=						{
    'allenai/longformer-base-4096': 4096,
    'allenai/longformer-large-4096': 4096,
    'allenai/longformer-large-4096-finetuned-triviaqa': 4096,
    'allenai/longformer-base-4096-extra.pos.embd.only': 4096,
    'allenai/longformer-large-4096-extra.pos.embd.only': 4096,
}



@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def   lowerCamelCase_()->     Union[str, Any]:
							_SCREAMING_SNAKE_CASE		:    List[str]      	=     (
							    list(range(ord("""!"""	)  ,						ord("""~"""	) + 1	)	) + list(range(ord("""¡"""	)  ,						ord("""¬"""	) + 1	)	) + list(range(ord("""®"""	)  ,						ord("""ÿ"""	) + 1	)	)
							)
							_SCREAMING_SNAKE_CASE		:    Dict      	=     bs[:]
							_SCREAMING_SNAKE_CASE		:    List[Any]      	=     0
							for b in range(2**8	):
														if b not in bs:
																					bs.append(__SCREAMING_SNAKE_CASE	)
																					cs.append(2**8 + n	)
																					n += 1
							_SCREAMING_SNAKE_CASE		:    Dict      	=     [chr(__SCREAMING_SNAKE_CASE	) for n in cs]
							return dict(zip(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE	)	)



def   lowerCamelCase_(__SCREAMING_SNAKE_CASE	)->     List[str]:
							_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     set()
							_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)	)
														_SCREAMING_SNAKE_CASE		:    List[Any]      	=     char
							return pairs




class 			_snake_case		(     snake_case_			):







				"""simple docstring"""






				a      =    VOCAB_FILES_NAMES
				a      =    PRETRAINED_VOCAB_FILES_MAP
				a      =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				a      =    ['input_ids', 'attention_mask']





				def __init__(   self				:				Tuple   ,     _A				:				List[Any]   ,     _A				:				str   ,     _A				:				Any="replace"   ,     _A				:				List[str]="<s>"   ,     _A				:				Any="</s>"   ,     _A				:				Any="</s>"   ,     _A				:				Dict="<s>"   ,     _A				:				Union[str, Any]="<unk>"   ,     _A				:				int="<pad>"   ,     _A				:				Optional[int]="<mask>"   ,     _A				:				Tuple=False   ,     **_A				:				Dict   ,     ):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    str      	=     AddedToken(lowerCAmelCase__   ,     lstrip=lowerCAmelCase__   ,     rstrip=lowerCAmelCase__) if isinstance(lowerCAmelCase__   ,     lowerCAmelCase__) else bos_token
											_SCREAMING_SNAKE_CASE		:    Union[str, Any]      	=     AddedToken(lowerCAmelCase__   ,     lstrip=lowerCAmelCase__   ,     rstrip=lowerCAmelCase__) if isinstance(lowerCAmelCase__   ,     lowerCAmelCase__) else eos_token
											_SCREAMING_SNAKE_CASE		:    List[str]      	=     AddedToken(lowerCAmelCase__   ,     lstrip=lowerCAmelCase__   ,     rstrip=lowerCAmelCase__) if isinstance(lowerCAmelCase__   ,     lowerCAmelCase__) else sep_token
											_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     AddedToken(lowerCAmelCase__   ,     lstrip=lowerCAmelCase__   ,     rstrip=lowerCAmelCase__) if isinstance(lowerCAmelCase__   ,     lowerCAmelCase__) else cls_token
											_SCREAMING_SNAKE_CASE		:    Tuple      	=     AddedToken(lowerCAmelCase__   ,     lstrip=lowerCAmelCase__   ,     rstrip=lowerCAmelCase__) if isinstance(lowerCAmelCase__   ,     lowerCAmelCase__) else unk_token
											_SCREAMING_SNAKE_CASE		:    Tuple      	=     AddedToken(lowerCAmelCase__   ,     lstrip=lowerCAmelCase__   ,     rstrip=lowerCAmelCase__) if isinstance(lowerCAmelCase__   ,     lowerCAmelCase__) else pad_token

											# Mask token behave like a normal word, i.e. include the space before it
											_SCREAMING_SNAKE_CASE		:    List[Any]      	=     AddedToken(lowerCAmelCase__   ,     lstrip=lowerCAmelCase__   ,     rstrip=lowerCAmelCase__) if isinstance(lowerCAmelCase__   ,     lowerCAmelCase__) else mask_token

											super().__init__(
											    errors=lowerCAmelCase__   ,     bos_token=lowerCAmelCase__   ,     eos_token=lowerCAmelCase__   ,     unk_token=lowerCAmelCase__   ,     sep_token=lowerCAmelCase__   ,     cls_token=lowerCAmelCase__   ,     pad_token=lowerCAmelCase__   ,     mask_token=lowerCAmelCase__   ,     add_prefix_space=lowerCAmelCase__   ,     **lowerCAmelCase__   ,     )

											with open(lowerCAmelCase__   ,     encoding="""utf-8""") as vocab_handle:
																		_SCREAMING_SNAKE_CASE		:    Dict      	=     json.load(lowerCAmelCase__)
											_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     {v: k for k, v in self.encoder.items()}
											_SCREAMING_SNAKE_CASE		:    Tuple      	=     errors  # how to handle errors in decoding
											_SCREAMING_SNAKE_CASE		:    str      	=     bytes_to_unicode()
											_SCREAMING_SNAKE_CASE		:    int      	=     {v: k for k, v in self.byte_encoder.items()}
											with open(lowerCAmelCase__   ,     encoding="""utf-8""") as merges_handle:
																		_SCREAMING_SNAKE_CASE		:    Dict      	=     merges_handle.read().split("""\n""")[1:-1]
											_SCREAMING_SNAKE_CASE		:    Dict      	=     [tuple(merge.split()) for merge in bpe_merges]
											_SCREAMING_SNAKE_CASE		:    Any      	=     dict(zip(lowerCAmelCase__   ,     range(len(lowerCAmelCase__))))
											_SCREAMING_SNAKE_CASE		:    Dict      	=     {}
											_SCREAMING_SNAKE_CASE		:    str      	=     add_prefix_space

											# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
											_SCREAMING_SNAKE_CASE		:    List[str]      	=     re.compile(r"""\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+""")





				@property
				def 		_lowerCAmelCase		(   self				:				Optional[Any]):



											"""simple docstring"""







											return len(self.encoder)





				def 		_lowerCAmelCase		(   self				:				List[str]):



											"""simple docstring"""







											return dict(self.encoder   ,     **self.added_tokens_encoder)





				def 		_lowerCAmelCase		(   self				:				Optional[int]   ,     _A				:				List[str]):



											"""simple docstring"""







											if token in self.cache:
																		return self.cache[token]
											_SCREAMING_SNAKE_CASE		:    List[Any]      	=     tuple(lowerCAmelCase__)
											_SCREAMING_SNAKE_CASE		:    List[Any]      	=     get_pairs(lowerCAmelCase__)

											if not pairs:
																		return token

											while True:
																		_SCREAMING_SNAKE_CASE		:    Union[str, Any]      	=     min(lowerCAmelCase__   ,     key=lambda _A: self.bpe_ranks.get(lowerCAmelCase__   ,     float("""inf""")))
																		if bigram not in self.bpe_ranks:
																									break
																		_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     bigram
																		_SCREAMING_SNAKE_CASE		:    int      	=     []
																		_SCREAMING_SNAKE_CASE		:    Tuple      	=     0
																		while i < len(lowerCAmelCase__):
																									try:
																																_SCREAMING_SNAKE_CASE		:    int      	=     word.index(lowerCAmelCase__   ,     lowerCAmelCase__)
																									except ValueError:
																																new_word.extend(word[i:])
																																break
																									else:
																																new_word.extend(word[i:j])
																																_SCREAMING_SNAKE_CASE		:    List[str]      	=     j

																									if word[i] == first and i < len(lowerCAmelCase__) - 1 and word[i + 1] == second:
																																new_word.append(first + second)
																																i += 2
																									else:
																																new_word.append(word[i])
																																i += 1
																		_SCREAMING_SNAKE_CASE		:    Any      	=     tuple(lowerCAmelCase__)
																		_SCREAMING_SNAKE_CASE		:    List[Any]      	=     new_word
																		if len(lowerCAmelCase__) == 1:
																									break
																		else:
																									_SCREAMING_SNAKE_CASE		:    Union[str, Any]      	=     get_pairs(lowerCAmelCase__)
											_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     ''' '''.join(lowerCAmelCase__)
											_SCREAMING_SNAKE_CASE		:    str      	=     word
											return word





				def 		_lowerCAmelCase		(   self				:				List[Any]   ,     _A				:				Any):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    List[Any]      	=     []
											for token in re.findall(self.pat   ,     lowerCAmelCase__):
																		_SCREAMING_SNAKE_CASE		:    Any      	=     ''''''.join(
																		    self.byte_encoder[b] for b in token.encode("""utf-8"""))  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
																		bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCAmelCase__).split(""" """))
											return bpe_tokens





				def 		_lowerCAmelCase		(   self				:				str   ,     _A				:				Tuple):



											"""simple docstring"""







											return self.encoder.get(lowerCAmelCase__   ,     self.encoder.get(self.unk_token))





				def 		_lowerCAmelCase		(   self				:				Optional[Any]   ,     _A				:				Optional[Any]):



											"""simple docstring"""







											return self.decoder.get(lowerCAmelCase__)





				def 		_lowerCAmelCase		(   self				:				str   ,     _A				:				Any):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    List[str]      	=     ''''''.join(lowerCAmelCase__)
											_SCREAMING_SNAKE_CASE		:    Tuple      	=     bytearray([self.byte_decoder[c] for c in text]).decode("""utf-8"""   ,     errors=self.errors)
											return text





				def 		_lowerCAmelCase		(   self				:				int   ,     _A				:				str   ,     _A				:				Optional[str] = None):



											"""simple docstring"""







											if not os.path.isdir(lowerCAmelCase__):
																		logger.error(f"""Vocabulary path ({save_directory}) should be a directory""")
																		return
											_SCREAMING_SNAKE_CASE		:    int      	=     os.path.join(
											    lowerCAmelCase__   ,     (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""])
											_SCREAMING_SNAKE_CASE		:    Any      	=     os.path.join(
											    lowerCAmelCase__   ,     (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""])

											with open(lowerCAmelCase__   ,     """w"""   ,     encoding="""utf-8""") as f:
																		f.write(json.dumps(self.encoder   ,     indent=2   ,     sort_keys=lowerCAmelCase__   ,     ensure_ascii=lowerCAmelCase__) + """\n""")

											_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     0
											with open(lowerCAmelCase__   ,     """w"""   ,     encoding="""utf-8""") as writer:
																		writer.write("""#version: 0.2\n""")
																		for bpe_tokens, token_index in sorted(self.bpe_ranks.items()   ,     key=lambda _A: kv[1]):
																									if index != token_index:
																																logger.warning(
																																    f"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
																																    """ Please check that the tokenizer is not corrupted!""")
																																_SCREAMING_SNAKE_CASE		:    Optional[Any]      	=     token_index
																									writer.write(""" """.join(lowerCAmelCase__) + """\n""")
																									index += 1

											return vocab_file, merge_file





				def 		_lowerCAmelCase		(   self				:				List[str]   ,     _A				:				List[int]   ,     _A				:				Optional[List[int]] = None):



											"""simple docstring"""







											if token_ids_a is None:
																		return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
											_SCREAMING_SNAKE_CASE		:    Union[str, Any]      	=     [self.cls_token_id]
											_SCREAMING_SNAKE_CASE		:    Tuple      	=     [self.sep_token_id]
											return cls + token_ids_a + sep + sep + token_ids_a + sep





				def 		_lowerCAmelCase		(   self				:				str   ,     _A				:				List[int]   ,     _A				:				Optional[List[int]] = None   ,     _A				:				bool = False):



											"""simple docstring"""







											if already_has_special_tokens:
																		return super().get_special_tokens_mask(
																		    token_ids_a=lowerCAmelCase__   ,     token_ids_a=lowerCAmelCase__   ,     already_has_special_tokens=lowerCAmelCase__)

											if token_ids_a is None:
																		return [1] + ([0] * len(lowerCAmelCase__)) + [1]
											return [1] + ([0] * len(lowerCAmelCase__)) + [1, 1] + ([0] * len(lowerCAmelCase__)) + [1]





				def 		_lowerCAmelCase		(   self				:				str   ,     _A				:				List[int]   ,     _A				:				Optional[List[int]] = None):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    Any      	=     [self.sep_token_id]
											_SCREAMING_SNAKE_CASE		:    Optional[int]      	=     [self.cls_token_id]

											if token_ids_a is None:
																		return len(cls + token_ids_a + sep) * [0]
											return len(cls + token_ids_a + sep + sep + token_ids_a + sep) * [0]





				def 		_lowerCAmelCase		(   self				:				Optional[int]   ,     _A				:				List[Any]   ,     _A				:				List[str]=False   ,     **_A				:				Optional[int]):



											"""simple docstring"""







											_SCREAMING_SNAKE_CASE		:    List[str]      	=     kwargs.pop("""add_prefix_space"""   ,     self.add_prefix_space)
											if (is_split_into_words or add_prefix_space) and (len(lowerCAmelCase__) > 0 and not text[0].isspace()):
																		_SCREAMING_SNAKE_CASE		:    Any      	=     ''' ''' + text
											return (text, kwargs)







'''simple docstring'''



import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
							from sagemaker import Session, TrainingJobAnalytics
							from sagemaker.huggingface import HuggingFace
@pytest.mark.skipif(
    literal_eval(os.getenv('TEST_SAGEMAKER'     ,					'False'      )      ) is not True     ,					reason='Skipping test because should only be run when releasing minor transformers version'     ,					)
@pytest.mark.usefixtures('sm_env'      )
@parameterized_class(
    [
        {
            'framework': 'pytorch',
            'script': 'run_glue_model_parallelism.py',
            'model_name_or_path': 'roberta-large',
            'instance_type': 'ml.p3dn.24xlarge',
            'results': {'train_runtime': 1_6_0_0, 'eval_accuracy': 0.3, 'eval_loss': 1.2},
        },
        {
            'framework': 'pytorch',
            'script': 'run_glue.py',
            'model_name_or_path': 'roberta-large',
            'instance_type': 'ml.p3dn.24xlarge',
            'results': {'train_runtime': 1_6_0_0, 'eval_accuracy': 0.3, 'eval_loss': 1.2},
        },
    ]      )
class 						lowerCamelCase_				(       unittest.TestCase      ):







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

													"""simple docstring"""
													if self.framework == "pytorch":
																				subprocess.run(
																				    F"""cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py""".split()					,     encoding='''utf-8'''					,     check=lowerCAmelCase__					,     )
													assert hasattr(self					,     '''env'''					)







						def        __lowercase     (	self :   Optional[int]					,     lowerCAmelCase__ :   List[Any]					):

													"""simple docstring"""
													# configuration for running training on smdistributed Model Parallel
													SCREAMING_SNAKE_CASE       :   Tuple         =      {
													    '''enabled''': True,
													    '''processes_per_host''': 8,
													}
													SCREAMING_SNAKE_CASE       :   Optional[int]         =      {
													    '''enabled''': True,
													    '''parameters''': {
													        '''microbatches''': 4,
													        '''placement_strategy''': '''spread''',
													        '''pipeline''': '''interleaved''',
													        '''optimize''': '''speed''',
													        '''partitions''': 4,
													        '''ddp''': True,
													    },
													}

													SCREAMING_SNAKE_CASE       :   str         =      {'''smdistributed''': {'''modelparallel''': smp_options}, '''mpi''': mpi_options}

													SCREAMING_SNAKE_CASE       :   Optional[Any]         =      '''trainer''' if self.script == '''run_glue.py''' else '''smtrainer'''
													# creates estimator
													return HuggingFace(
													    entry_point=self.script					,     source_dir=self.env.test_path					,     role=self.env.role					,     image_uri=self.env.image_uri					,     base_job_name=F"""{self.env.base_job_name}-{instance_count}-smp-{name_extension}"""					,     instance_count=lowerCAmelCase__					,     instance_type=self.instance_type					,     debugger_hook_config=lowerCAmelCase__					,     hyperparameters={
													        **self.env.hyperparameters,
													        '''model_name_or_path''': self.model_name_or_path,
													        '''max_steps''': 5_00,
													    }					,     metric_definitions=self.env.metric_definitions					,     distribution=lowerCAmelCase__					,     py_version='''py36'''					,     )







						def        __lowercase     (	self :   Dict					,     lowerCAmelCase__ :   int					):

													"""simple docstring"""
													TrainingJobAnalytics(lowerCAmelCase__					).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv"""					)







						@parameterized.expand([(1,)]					)
						def        __lowercase     (	self :   Dict					,     lowerCAmelCase__ :   List[Any]					):

													"""simple docstring"""
													# create estimator
													SCREAMING_SNAKE_CASE       :   Any         =      self.create_estimator(lowerCAmelCase__					)

													# run training
													estimator.fit()

													# result dataframe
													SCREAMING_SNAKE_CASE       :   Optional[Any]         =      TrainingJobAnalytics(estimator.latest_training_job.name					).dataframe()

													# extract kpis
													SCREAMING_SNAKE_CASE       :   int         =      list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value''']					)
													SCREAMING_SNAKE_CASE       :   Tuple         =      list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value''']					)
													# get train time from SageMaker job, this includes starting, preprocessing, stopping
													SCREAMING_SNAKE_CASE       :   int         =      (
													    Session().describe_training_job(estimator.latest_training_job.name					).get('''TrainingTimeInSeconds'''					,     99_99_99					)
													)

													# assert kpis
													assert train_runtime <= self.results["train_runtime"]
													assert all(t >= self.results['''eval_accuracy'''] for t in eval_accuracy					)
													assert all(t <= self.results['''eval_loss'''] for t in eval_loss					)

													# dump tests result into json file to share in PR
													with open(F"""{estimator.latest_training_job.name}.json"""					,     '''w'''					) as outfile:
																				json.dump({'''train_time''': train_runtime, '''eval_accuracy''': eval_accuracy, '''eval_loss''': eval_loss}					,     lowerCAmelCase__					)




