



'''simple docstring'''
import json
import os
from functools import lru_cache
from typing import Dict, List, Optional, Tuple, Union

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


lowercase      			=	logging.get_logger(__name__)


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

# See all LED models at https://huggingface.co/models?filter=LED
lowercase      			=	{
    '''vocab_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json''',
    },
    '''merges_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt''',
    },
    '''tokenizer_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json''',
    },
}

lowercase      			=	{
    '''allenai/led-base-16384''': 1_6384,
}



@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def        __A   (							):







					"""simple docstring"""






					__SCREAMING_SNAKE_CASE				:    List[Any]						    =							(
					    list(range(ord("!"	)					,   ord("~"	) + 1	)	) + list(range(ord("¡"	)					,   ord("¬"	) + 1	)	) + list(range(ord("®"	)					,   ord("ÿ"	) + 1	)	)
					)
					__SCREAMING_SNAKE_CASE				:    int						    =							bs[:]
					__SCREAMING_SNAKE_CASE				:    Any						    =							0
					for b in range(2**8	):
										if b not in bs:
															bs.append(__lowerCamelCase	)
															cs.append(2**8 + n	)
															n += 1
					__SCREAMING_SNAKE_CASE				:    Tuple						    =							[chr(__lowerCamelCase	) for n in cs]
					return dict(zip(__lowerCamelCase					,   __lowerCamelCase	)	)



def        __A   (							_SCREAMING_SNAKE_CASE     :  List[str]	):







					"""simple docstring"""






					__SCREAMING_SNAKE_CASE				:    Tuple						    =							set()
					__SCREAMING_SNAKE_CASE				:    int						    =							word[0]
					for char in word[1:]:
										pairs.add((prev_char, char)	)
										__SCREAMING_SNAKE_CASE				:    List[Any]						    =							char
					return pairs






class    __lowerCamelCase							( _SCREAMING_SNAKE_CASE   ):







	'''simple docstring'''
	snake_case__		:   Optional[int]       =				VOCAB_FILES_NAMES
	snake_case__		:   Union[str, Any]       =				PRETRAINED_VOCAB_FILES_MAP
	snake_case__		:   Tuple       =				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	snake_case__		:   List[str]       =				['''input_ids''', '''attention_mask''']






	def __init__(      self      ,     a__      ,     a__      ,     a__="replace"      ,     a__="<s>"      ,     a__="</s>"      ,     a__="</s>"      ,     a__="<s>"      ,     a__="<unk>"      ,     a__="<pad>"      ,     a__="<mask>"      ,     a__=False      ,     **a__      ,     ):
						__SCREAMING_SNAKE_CASE				:    Any						    =							AddedToken(_lowercase      ,     lstrip=_lowercase      ,     rstrip=_lowercase					) if isinstance(_lowercase      ,     _lowercase					) else bos_token
						__SCREAMING_SNAKE_CASE				:    int						    =							AddedToken(_lowercase      ,     lstrip=_lowercase      ,     rstrip=_lowercase					) if isinstance(_lowercase      ,     _lowercase					) else eos_token
						__SCREAMING_SNAKE_CASE				:    Dict						    =							AddedToken(_lowercase      ,     lstrip=_lowercase      ,     rstrip=_lowercase					) if isinstance(_lowercase      ,     _lowercase					) else sep_token
						__SCREAMING_SNAKE_CASE				:    Optional[int]						    =							AddedToken(_lowercase      ,     lstrip=_lowercase      ,     rstrip=_lowercase					) if isinstance(_lowercase      ,     _lowercase					) else cls_token
						__SCREAMING_SNAKE_CASE				:    int						    =							AddedToken(_lowercase      ,     lstrip=_lowercase      ,     rstrip=_lowercase					) if isinstance(_lowercase      ,     _lowercase					) else unk_token
						__SCREAMING_SNAKE_CASE				:    str						    =							AddedToken(_lowercase      ,     lstrip=_lowercase      ,     rstrip=_lowercase					) if isinstance(_lowercase      ,     _lowercase					) else pad_token

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

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

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

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






	@property
	# Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
	def  a_					(      self					):
						return len(self.encoder					)






	def  a_					(      self					):
						return dict(self.encoder      ,     **self.added_tokens_encoder					)






	def  a_					(      self      ,     a__					):



						if token in self.cache:
											return self.cache[token]
						__SCREAMING_SNAKE_CASE				:    List[str]						    =							tuple(_lowercase					)
						__SCREAMING_SNAKE_CASE				:    Any						    =							get_pairs(_lowercase					)

						if not pairs:
											return token

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

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






	def  a_					(      self      ,     a__					):
						__SCREAMING_SNAKE_CASE				:    Tuple						    =							[]
						for token in re.findall(self.pat      ,     _lowercase					):
											__SCREAMING_SNAKE_CASE				:    Optional[int]						    =							"".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(_lowercase					).split(" "					)					)
						return bpe_tokens






	def  a_					(      self      ,     a__					):
						return self.encoder.get(_lowercase      ,     self.encoder.get(self.unk_token					)					)






	def  a_					(      self      ,     a__					):
						return self.decoder.get(_lowercase					)






	def  a_					(      self      ,     a__					):
						__SCREAMING_SNAKE_CASE				:    Tuple						    =							"".join(_lowercase					)
						__SCREAMING_SNAKE_CASE				:    Any						    =							bytearray([self.byte_decoder[c] for c in text]					).decode("utf-8"      ,     errors=self.errors					)
						return text






	def  a_					(      self      ,     a__      ,     a__ = None					):



						if not os.path.isdir(_lowercase					):
											logger.error(f'Vocabulary path ({save_directory}) should be a directory'					)
											return
						__SCREAMING_SNAKE_CASE				:    Union[str, Any]						    =							os.path.join(
						    _lowercase      ,     (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]					)
						__SCREAMING_SNAKE_CASE				:    Tuple						    =							os.path.join(
						    _lowercase      ,     (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]					)

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

						__SCREAMING_SNAKE_CASE				:    Union[str, Any]						    =							0
						with open(_lowercase      ,     "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				:    Dict						    =							token_index
																writer.write(" ".join(_lowercase					) + "\n"					)
																index += 1

						return vocab_file, merge_file






	def  a_					(      self      ,     a__      ,     a__ = None					):



						if token_ids_a is None:
											return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
						__SCREAMING_SNAKE_CASE				:    Dict						    =							[self.cls_token_id]
						__SCREAMING_SNAKE_CASE				:    List[str]						    =							[self.sep_token_id]
						return cls + token_ids_a + sep + sep + token_ids_a + sep






	def  a_					(      self      ,     a__      ,     a__ = None      ,     a__ = False					):



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

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






	def  a_					(      self      ,     a__      ,     a__ = None					):
						__SCREAMING_SNAKE_CASE				:    List[str]						    =							[self.sep_token_id]
						__SCREAMING_SNAKE_CASE				:    List[Any]						    =							[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  a_					(      self      ,     a__      ,     a__=False      ,     **a__					):
						__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(_lowercase					) > 0 and not text[0].isspace()):
											__SCREAMING_SNAKE_CASE				:    int						    =							" " + text
						return (text, kwargs)






	def  a_					(      self      ,     a__      ,     a__ = None      ,     a__ = PaddingStrategy.DO_NOT_PAD      ,     a__ = None      ,     a__ = None      ,     ):
						__SCREAMING_SNAKE_CASE				:    Union[str, Any]						    =							super()._pad(
						    encoded_inputs=_lowercase      ,     max_length=_lowercase      ,     padding_strategy=_lowercase      ,     pad_to_multiple_of=_lowercase      ,     return_attention_mask=_lowercase      ,     )

						# Load from model defaults
						if return_attention_mask is None:
											__SCREAMING_SNAKE_CASE				:    Optional[Any]						    =							"attention_mask" in self.model_input_names

						if return_attention_mask and "global_attention_mask" in encoded_inputs:
											__SCREAMING_SNAKE_CASE				:    Tuple						    =							encoded_inputs[self.model_input_names[0]]
											# `global_attention_mask` need to have the same length as other (sequential) inputs.
											__SCREAMING_SNAKE_CASE				:    str						    =							len(encoded_inputs["global_attention_mask"]					) != len(_lowercase					)

											if needs_to_be_padded:
																__SCREAMING_SNAKE_CASE				:    Any						    =							len(_lowercase					) - len(encoded_inputs["global_attention_mask"]					)

																if self.padding_side == "right":
																					# Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
																					__SCREAMING_SNAKE_CASE				:    str						    =							(
																					    encoded_inputs["global_attention_mask"] + [-1] * difference
																					)
																elif self.padding_side == "left":
																					__SCREAMING_SNAKE_CASE				:    Dict						    =							[-1] * difference + encoded_inputs[
																					    "global_attention_mask"
																					]
																else:
																					raise ValueError("Invalid padding strategy:" + str(self.padding_side					)					)

						return encoded_inputs




'''simple docstring'''



import os
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional, Union

import torch
from filelock import FileLock
from torch.utils.data import Dataset

from ...models.auto.modeling_auto import MODEL_FOR_QUESTION_ANSWERING_MAPPING
from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging
from ..processors.squad import SquadFeatures, SquadVaProcessor, SquadVaProcessor, squad_convert_examples_to_features


_lowercase          =						logging.get_logger(__name__)

_lowercase          =						list(MODEL_FOR_QUESTION_ANSWERING_MAPPING.keys())
_lowercase          =						tuple(conf.model_type for conf in MODEL_CONFIG_CLASSES)




@dataclass
class     UpperCAmelCase_	:






						'''simple docstring'''






						_lowercase						: str										=  field(
						    default=_SCREAMING_SNAKE_CASE			,				metadata={'''help''': '''Model type selected in the list: ''' + ''', '''.join(_SCREAMING_SNAKE_CASE					)}					)
						_lowercase						: str										=  field(
						    default=_SCREAMING_SNAKE_CASE			,				metadata={'''help''': '''The input data dir. Should contain the .json files for the SQuAD task.'''}					)
						_lowercase						: int										=  field(
						    default=1_2_8			,				metadata={
						        '''help''': (
						            '''The maximum total input sequence length after tokenization. Sequences longer '''
						            '''than this will be truncated, sequences shorter will be padded.'''
						        )
						    }			,				)
						_lowercase						: int										=  field(
						    default=1_2_8			,				metadata={'''help''': '''When splitting up a long document into chunks, how much stride to take between chunks.'''}			,				)
						_lowercase						: int										=  field(
						    default=6_4			,				metadata={
						        '''help''': (
						            '''The maximum number of tokens for the question. Questions longer than this will '''
						            '''be truncated to this length.'''
						        )
						    }			,				)
						_lowercase						: int										=  field(
						    default=3_0			,				metadata={
						        '''help''': (
						            '''The maximum length of an answer that can be generated. This is needed because the start '''
						            '''and end predictions are not conditioned on one another.'''
						        )
						    }			,				)
						_lowercase						: bool										=  field(
						    default=_SCREAMING_SNAKE_CASE			,				metadata={'''help''': '''Overwrite the cached training and evaluation sets'''}					)
						_lowercase						: bool										=  field(
						    default=_SCREAMING_SNAKE_CASE			,				metadata={'''help''': '''If true, the SQuAD examples contain some that do not have an answer.'''}					)
						_lowercase						: float										=  field(
						    default=0.0			,				metadata={'''help''': '''If null_score - best_non_null is greater than the threshold predict null.'''}					)
						_lowercase						: int										=  field(
						    default=2_0			,				metadata={'''help''': '''If null_score - best_non_null is greater than the threshold predict null.'''}					)
						_lowercase						: int										=  field(
						    default=0			,				metadata={
						        '''help''': (
						            '''language id of input for language-specific xlm models (see'''
						            ''' tokenization_xlm.PRETRAINED_INIT_CONFIGURATION)'''
						        )
						    }			,				)
						_lowercase						: int										=  field(default=1			,				metadata={'''help''': '''multiple threads for converting example to features'''}					)






class     UpperCAmelCase_	(						_SCREAMING_SNAKE_CASE					):






						'''simple docstring'''






						_lowercase						: str										=  '''train'''
						_lowercase						: Union[str, Any]										=  '''dev'''






class     UpperCAmelCase_	(						_SCREAMING_SNAKE_CASE					):






						'''simple docstring'''






						_lowercase						: SquadDataTrainingArguments
						_lowercase						: List[SquadFeatures]
						_lowercase						: Split
						_lowercase						: bool




						def __init__(   self       ,  _lowercase       ,  _lowercase       ,  _lowercase = None       ,  _lowercase = Split.train       ,  _lowercase = False       ,  _lowercase = None       ,  _lowercase = "pt"       ,  ):

												"""simple docstring"""



												_lowerCAmelCase    		=					args
												_lowerCAmelCase    		=					is_language_sensitive
												_lowerCAmelCase    		=					SquadVaProcessor() if args.version_2_with_negative else SquadVaProcessor()
												if isinstance(_lowercase       ,  _lowercase ):
																		try:
																								_lowerCAmelCase    		=					Split[mode]
																		except KeyError:
																								raise KeyError("""mode is not a valid split name""" )
												_lowerCAmelCase    		=					mode
												# Load data features from cache or dataset file
												_lowerCAmelCase    		=					"""v2""" if args.version_2_with_negative else """v1"""
												_lowerCAmelCase    		=					os.path.join(
												    cache_dir if cache_dir is not None else args.data_dir       ,  F'cached_{mode.value}_{tokenizer.__class__.__name__}_{args.max_seq_length}_{version_tag}'       ,  )

												# Make sure only the first process in distributed training processes the dataset,
												# and the others will use the cache.
												_lowerCAmelCase    		=					cached_features_file + """.lock"""
												with FileLock(_lowercase ):
																		if os.path.exists(_lowercase ) and not args.overwrite_cache:
																								_lowerCAmelCase    		=					time.time()
																								_lowerCAmelCase    		=					torch.load(_lowercase )

																								# Legacy cache files have only features, while new cache files
																								# will have dataset and examples also.
																								_lowerCAmelCase    		=					self.old_features["""features"""]
																								_lowerCAmelCase    		=					self.old_features.get("""dataset"""       ,  _lowercase )
																								_lowerCAmelCase    		=					self.old_features.get("""examples"""       ,  _lowercase )
																								logger.info(
																								    F'Loading features from cached file {cached_features_file} [took %.3f s]'       ,  time.time() - start )

																								if self.dataset is None or self.examples is None:
																														logger.warning(
																														    F'Deleting cached file {cached_features_file} will allow dataset and examples to be cached in'
																														    """ future run""" )
																		else:
																								if mode == Split.dev:
																														_lowerCAmelCase    		=					self.processor.get_dev_examples(args.data_dir )
																								else:
																														_lowerCAmelCase    		=					self.processor.get_train_examples(args.data_dir )

																								_lowerCAmelCase		,       _lowerCAmelCase    		=					squad_convert_examples_to_features(
																								    examples=self.examples       ,  tokenizer=_lowercase       ,  max_seq_length=args.max_seq_length       ,  doc_stride=args.doc_stride       ,  max_query_length=args.max_query_length       ,  is_training=mode == Split.train       ,  threads=args.threads       ,  return_dataset=_lowercase       ,  )

																								_lowerCAmelCase    		=					time.time()
																								torch.save(
																								    {"""features""": self.features, """dataset""": self.dataset, """examples""": self.examples}       ,  _lowercase       ,  )
																								# ^ This seems to take a lot of time so I want to investigate why and how we can improve.
																								logger.info(
																								    F'Saving features into cached file {cached_features_file} [took {time.time() - start:.3f} s]' )




						def __len__(   self ):

												"""simple docstring"""



												return len(self.features )




						def __getitem__(   self       ,  _lowercase ):

												"""simple docstring"""



												_lowerCAmelCase    		=					self.features[i]

												_lowerCAmelCase    		=					torch.tensor(feature.input_ids       ,  dtype=torch.long )
												_lowerCAmelCase    		=					torch.tensor(feature.attention_mask       ,  dtype=torch.long )
												_lowerCAmelCase    		=					torch.tensor(feature.token_type_ids       ,  dtype=torch.long )
												_lowerCAmelCase    		=					torch.tensor(feature.cls_index       ,  dtype=torch.long )
												_lowerCAmelCase    		=					torch.tensor(feature.p_mask       ,  dtype=torch.float )
												_lowerCAmelCase    		=					torch.tensor(feature.is_impossible       ,  dtype=torch.float )

												_lowerCAmelCase    		=					{
												    """input_ids""": input_ids,
												    """attention_mask""": attention_mask,
												    """token_type_ids""": token_type_ids,
												}

												if self.args.model_type in ["xlm", "roberta", "distilbert", "camembert"]:
																		del inputs["token_type_ids"]

												if self.args.model_type in ["xlnet", "xlm"]:
																		inputs.update({"""cls_index""": cls_index, """p_mask""": p_mask} )
																		if self.args.version_2_with_negative:
																								inputs.update({"""is_impossible""": is_impossible} )
																		if self.is_language_sensitive:
																								inputs.update({"""langs""": (torch.ones(input_ids.shape       ,  dtype=torch.intaa ) * self.args.lang_id)} )

												if self.mode == Split.train:
																		_lowerCAmelCase    		=					torch.tensor(feature.start_position       ,  dtype=torch.long )
																		_lowerCAmelCase    		=					torch.tensor(feature.end_position       ,  dtype=torch.long )
																		inputs.update({"""start_positions""": start_positions, """end_positions""": end_positions} )

												return inputs
