
"""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


UpperCAmelCase        =				logging.get_logger(__name__)


UpperCAmelCase        =				{"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt"""}

UpperCAmelCase        =				{
    """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"""
        ),
    },
}

UpperCAmelCase        =				{
    """allenai/longformer-base-4096""": 4_0_9_6,
    """allenai/longformer-large-4096""": 4_0_9_6,
    """allenai/longformer-large-4096-finetuned-triviaqa""": 4_0_9_6,
    """allenai/longformer-base-4096-extra.pos.embd.only""": 4_0_9_6,
    """allenai/longformer-large-4096-extra.pos.embd.only""": 4_0_9_6,
}




@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def __magic_name__	(				)					->						Dict:

	'''simple docstring'''




	lowerCAmelCase												=  (
	    list(range(ord('''!'''			),					ord('''~'''			) + 1			)			) + list(range(ord('''¡'''			),					ord('''¬'''			) + 1			)			) + list(range(ord('''®'''			),					ord('''ÿ'''			) + 1			)			)
	)
	lowerCAmelCase												=  bs[:]
	lowerCAmelCase												=  0
	for b in range(2**8			):
		if b not in bs:
			bs.append(_lowerCamelCase			)
			cs.append(2**8 + n			)
			n += 1
	lowerCAmelCase												=  [chr(_lowerCamelCase			) for n in cs]
	return dict(zip(_lowerCamelCase,					_lowerCamelCase			)			)




def __magic_name__	(				_lowerCamelCase:		List[Any]			)					->						List[Any]:

	'''simple docstring'''




	lowerCAmelCase												=  set()
	lowerCAmelCase												=  word[0]
	for char in word[1:]:
		pairs.add((prev_char, char)			)
		lowerCAmelCase												=  char
	return pairs



class 			lowercase					( lowercase__		):
			lowercase        =     VOCAB_FILES_NAMES
			lowercase        =     PRETRAINED_VOCAB_FILES_MAP
			lowercase        =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			lowercase        =     ['''input_ids''', '''attention_mask''']






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





				"""simple docstring"""


				lowerCAmelCase												=  AddedToken(SCREAMING_SNAKE_CASE_      ,lstrip=SCREAMING_SNAKE_CASE_      ,rstrip=SCREAMING_SNAKE_CASE_			) if isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) else bos_token
				lowerCAmelCase												=  AddedToken(SCREAMING_SNAKE_CASE_      ,lstrip=SCREAMING_SNAKE_CASE_      ,rstrip=SCREAMING_SNAKE_CASE_			) if isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) else eos_token
				lowerCAmelCase												=  AddedToken(SCREAMING_SNAKE_CASE_      ,lstrip=SCREAMING_SNAKE_CASE_      ,rstrip=SCREAMING_SNAKE_CASE_			) if isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) else sep_token
				lowerCAmelCase												=  AddedToken(SCREAMING_SNAKE_CASE_      ,lstrip=SCREAMING_SNAKE_CASE_      ,rstrip=SCREAMING_SNAKE_CASE_			) if isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) else cls_token
				lowerCAmelCase												=  AddedToken(SCREAMING_SNAKE_CASE_      ,lstrip=SCREAMING_SNAKE_CASE_      ,rstrip=SCREAMING_SNAKE_CASE_			) if isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) else unk_token
				lowerCAmelCase												=  AddedToken(SCREAMING_SNAKE_CASE_      ,lstrip=SCREAMING_SNAKE_CASE_      ,rstrip=SCREAMING_SNAKE_CASE_			) if isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) else pad_token

				# Mask token behave like a normal word, i.e. include the space before it
				lowerCAmelCase												=  AddedToken(SCREAMING_SNAKE_CASE_      ,lstrip=SCREAMING_SNAKE_CASE_      ,rstrip=SCREAMING_SNAKE_CASE_			) if isinstance(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			) else mask_token

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

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

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






			@property
			def 					UpperCAmelCase							(self  :	int			)       ->	Any:





				"""simple docstring"""


				return len(self.encoder			)






			def 					UpperCAmelCase							(self  :	Tuple			)       ->	Dict:





				"""simple docstring"""


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






			def 					UpperCAmelCase							(self  :	Union[str, Any]      ,SCREAMING_SNAKE_CASE_  :	Union[str, Any]			)       ->	Optional[Any]:





				"""simple docstring"""


				if token in self.cache:
					return self.cache[token]
				lowerCAmelCase												=  tuple(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  get_pairs(SCREAMING_SNAKE_CASE_			)

				if not pairs:
					return token

				while True:
					lowerCAmelCase												=  min(SCREAMING_SNAKE_CASE_      ,key=lambda SCREAMING_SNAKE_CASE_			: self.bpe_ranks.get(SCREAMING_SNAKE_CASE_      ,float('''inf'''			)			)			)
					if bigram not in self.bpe_ranks:
						break
					lowerCAmelCase							,       lowerCAmelCase												=  bigram
					lowerCAmelCase												=  []
					lowerCAmelCase												=  0
					while i < len(SCREAMING_SNAKE_CASE_			):
						try:
							lowerCAmelCase												=  word.index(SCREAMING_SNAKE_CASE_      ,SCREAMING_SNAKE_CASE_			)
						except ValueError:
							new_word.extend(word[i:]			)
							break
						else:
							new_word.extend(word[i:j]			)
							lowerCAmelCase												=  j

						if word[i] == first and i < len(SCREAMING_SNAKE_CASE_			) - 1 and word[i + 1] == second:
							new_word.append(first + second			)
							i += 2
						else:
							new_word.append(word[i]			)
							i += 1
					lowerCAmelCase												=  tuple(SCREAMING_SNAKE_CASE_			)
					lowerCAmelCase												=  new_word
					if len(SCREAMING_SNAKE_CASE_			) == 1:
						break
					else:
						lowerCAmelCase												=  get_pairs(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  ''' '''.join(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  word
				return word






			def 					UpperCAmelCase							(self  :	Optional[int]      ,SCREAMING_SNAKE_CASE_  :	Optional[Any]			)       ->	List[Any]:





				"""simple docstring"""


				lowerCAmelCase												=  []
				for token in re.findall(self.pat      ,SCREAMING_SNAKE_CASE_			):
					lowerCAmelCase												=  ''''''.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(SCREAMING_SNAKE_CASE_			).split(''' '''			)			)
				return bpe_tokens






			def 					UpperCAmelCase							(self  :	Optional[int]      ,SCREAMING_SNAKE_CASE_  :	Dict			)       ->	Dict:





				"""simple docstring"""


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






			def 					UpperCAmelCase							(self  :	Optional[int]      ,SCREAMING_SNAKE_CASE_  :	Optional[int]			)       ->	Union[str, Any]:





				"""simple docstring"""


				return self.decoder.get(SCREAMING_SNAKE_CASE_			)






			def 					UpperCAmelCase							(self  :	List[str]      ,SCREAMING_SNAKE_CASE_  :	Optional[Any]			)       ->	Any:





				"""simple docstring"""


				lowerCAmelCase												=  ''''''.join(SCREAMING_SNAKE_CASE_			)
				lowerCAmelCase												=  bytearray([self.byte_decoder[c] for c in text]			).decode('''utf-8'''      ,errors=self.errors			)
				return text






			def 					UpperCAmelCase							(self  :	List[str]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	Optional[str] = None			)       ->	Tuple[str]:





				"""simple docstring"""


				if not os.path.isdir(SCREAMING_SNAKE_CASE_			):
					logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""			)
					return
				lowerCAmelCase												=  os.path.join(
				    SCREAMING_SNAKE_CASE_      ,(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']			)
				lowerCAmelCase												=  os.path.join(
				    SCREAMING_SNAKE_CASE_      ,(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']			)

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

				lowerCAmelCase												=  0
				with open(SCREAMING_SNAKE_CASE_      ,'''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 SCREAMING_SNAKE_CASE_			: 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!'''			)
							lowerCAmelCase												=  token_index
						writer.write(''' '''.join(SCREAMING_SNAKE_CASE_			) + '''\n'''			)
						index += 1

				return vocab_file, merge_file






			def 					UpperCAmelCase							(self  :	List[str]      ,SCREAMING_SNAKE_CASE_  :	List[int]      ,SCREAMING_SNAKE_CASE_  :	Optional[List[int]] = None			)       ->	List[int]:





				"""simple docstring"""


				if token_ids_a is None:
					return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
				lowerCAmelCase												=  [self.cls_token_id]
				lowerCAmelCase												=  [self.sep_token_id]
				return cls + token_ids_a + sep + sep + token_ids_a + sep






			def 					UpperCAmelCase							(self  :	List[str]      ,SCREAMING_SNAKE_CASE_  :	List[int]      ,SCREAMING_SNAKE_CASE_  :	Optional[List[int]] = None      ,SCREAMING_SNAKE_CASE_  :	bool = False			)       ->	List[int]:





				"""simple docstring"""


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

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






			def 					UpperCAmelCase							(self  :	Optional[int]      ,SCREAMING_SNAKE_CASE_  :	List[int]      ,SCREAMING_SNAKE_CASE_  :	Optional[List[int]] = None			)       ->	List[int]:





				"""simple docstring"""


				lowerCAmelCase												=  [self.sep_token_id]
				lowerCAmelCase												=  [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 					UpperCAmelCase							(self  :	str      ,SCREAMING_SNAKE_CASE_  :	Any      ,SCREAMING_SNAKE_CASE_  :	Optional[int]=False      ,**SCREAMING_SNAKE_CASE_  :	List[Any]			)       ->	Any:





				"""simple docstring"""


				lowerCAmelCase												=  kwargs.pop('''add_prefix_space'''      ,self.add_prefix_space			)
				if (is_split_into_words or add_prefix_space) and (len(SCREAMING_SNAKE_CASE_			) > 0 and not text[0].isspace()):
					lowerCAmelCase												=  ''' ''' + text
				return (text, kwargs)




"""simple docstring"""






from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.testing_utils import (
    TestCasePlus,
    execute_subprocess_async,
    get_torch_dist_unique_port,
    require_torch_multi_gpu,
    require_torch_neuroncore,
)
from transformers.training_args import ParallelMode
from transformers.utils import logging


UpperCAmelCase        =				logging.get_logger(__name__)


if is_torch_available():
					import torch
					from torch import nn
					from torch.utils.data import Dataset

					from transformers import Trainer



					class 			lowercase					( lowercase__		):






								def __init__(self  :	int      ,SCREAMING_SNAKE_CASE_  :	int = 101			)       ->	List[str]:





									"""simple docstring"""


									lowerCAmelCase												=  length






								def __len__(self  :	Any			)       ->	List[Any]:





									"""simple docstring"""


									return self.length






								def __getitem__(self  :	str      ,SCREAMING_SNAKE_CASE_  :	Union[str, Any]			)       ->	int:





									"""simple docstring"""


									return i





					class 			lowercase					:






								def __call__(self  :	List[Any]      ,SCREAMING_SNAKE_CASE_  :	List[Any]			)       ->	Optional[Any]:





									"""simple docstring"""


									return {"input_ids": torch.tensor(SCREAMING_SNAKE_CASE_			), "labels": torch.tensor(SCREAMING_SNAKE_CASE_			)}





					class 			lowercase					( nn.Module		):






								def __init__(self  :	List[str]			)       ->	List[Any]:





									"""simple docstring"""


									super().__init__()
									# Add some (unused) params otherwise DDP will complain.
									lowerCAmelCase												=  nn.Linear(120      ,80			)






								def 					UpperCAmelCase							(self  :	Optional[int]      ,SCREAMING_SNAKE_CASE_  :	str      ,SCREAMING_SNAKE_CASE_  :	Optional[int]=None			)       ->	Union[str, Any]:





									"""simple docstring"""


									if labels is not None:
										return torch.tensor(0.0      ,device=input_ids.device			), input_ids
									else:
										return input_ids



class 			lowercase					( lowercase__		):






			@require_torch_neuroncore
			def 					UpperCAmelCase							(self  :	Union[str, Any]			)       ->	Union[str, Any]:





				"""simple docstring"""


				lowerCAmelCase												=  F"""--nproc_per_node=2
            --master_port={get_torch_dist_unique_port()}
            {self.test_file_dir}/test_trainer_distributed.py
        """.split()
				lowerCAmelCase												=  self.get_auto_remove_tmp_dir()
				lowerCAmelCase												=  F"""--output_dir {output_dir}""".split()
				lowerCAmelCase												=  ['''torchrun'''] + distributed_args + args
				execute_subprocess_async(SCREAMING_SNAKE_CASE_      ,env=self.get_env()			)
				# successful return here == success - any errors would have caused an error in the sub-call





class 			lowercase					( lowercase__		):






			@require_torch_multi_gpu
			def 					UpperCAmelCase							(self  :	List[str]			)       ->	List[str]:





				"""simple docstring"""


				lowerCAmelCase												=  F"""--nproc_per_node={torch.cuda.device_count()}
            --master_port={get_torch_dist_unique_port()}
            {self.test_file_dir}/test_trainer_distributed.py
        """.split()
				lowerCAmelCase												=  self.get_auto_remove_tmp_dir()
				lowerCAmelCase												=  F"""--output_dir {output_dir}""".split()
				lowerCAmelCase												=  ['''torchrun'''] + distributed_args + args
				execute_subprocess_async(SCREAMING_SNAKE_CASE_      ,env=self.get_env()			)
				# successful return here == success - any errors would have caused an error in the sub-call


if __name__ == "__main__":
					# The script below is meant to be run under torch.distributed, on a machine with multiple GPUs:
					#
					# PYTHONPATH="src" python -m torch.distributed.run --nproc_per_node 2 --output_dir output_dir ./tests/test_trainer_distributed.py

					UpperCAmelCase        =				HfArgumentParser((TrainingArguments,))
					UpperCAmelCase        =				parser.parse_args_into_dataclasses()[0]

					logger.warning(
					    f"""Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}, """
					    f"""distributed training: {training_args.parallel_mode != ParallelMode.NOT_DISTRIBUTED}"""
					)

					# Essentially, what we want to verify in the distributed case is that we get all samples back,
					# in the right order. (this is crucial for prediction for instance)
					for dataset_length in [1_0_1, 4_0, 7]:
										UpperCAmelCase        =				DummyDataset(dataset_length)




										def __magic_name__	(				_lowerCamelCase:		EvalPrediction			)					->						Dict:

											'''simple docstring'''




											lowerCAmelCase												=  list(range(len(_lowerCamelCase			)			)			)
											lowerCAmelCase												=  p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
											if not success and training_args.local_rank == 0:
												logger.warning(
												    '''Predictions and/or labels do not match expected results:\n  - predictions: '''
												    F"""{p.predictions.tolist()}\n  - labels: {p.label_ids.tolist()}\n  - expected: {sequential}"""			)
											return {"success": success}

										UpperCAmelCase        =				Trainer(
										    model=DummyModel(),
										    args=training_args,
										    data_collator=DummyDataCollator(),
										    eval_dataset=dataset,
										    compute_metrics=compute_metrics,
										)
										UpperCAmelCase        =				trainer.evaluate()
										logger.info(metrics)
										if metrics["eval_success"] is not True:
															logger.error(metrics)
															exit(1)

										UpperCAmelCase        =				trainer.predict(dataset)
										logger.info(p.metrics)
										if p.metrics["test_success"] is not True:
															logger.error(p.metrics)
															exit(1)

										UpperCAmelCase        =				2

										UpperCAmelCase        =				trainer.evaluate()
										logger.info(metrics)
										if metrics["eval_success"] is not True:
															logger.error(metrics)
															exit(1)

										UpperCAmelCase        =				trainer.predict(dataset)
										logger.info(p.metrics)
										if p.metrics["test_success"] is not True:
															logger.error(p.metrics)
															exit(1)



										UpperCAmelCase        =				None


