




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


A							:				Tuple			       =	logging.get_logger(__name__)


A							:				Optional[Any]			       =	{"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt"""}

# See all MVP models at https://huggingface.co/models?filter=mvp
A							:				Dict			       =	{
    """vocab_file""": {
        """RUCAIBox/mvp""": """https://huggingface.co/RUCAIBox/mvp/resolve/main/vocab.json""",
    },
    """added_tokens.json""": {
        """RUCAIBox/mvp""": """https://huggingface.co/RUCAIBox/mvp/resolve/main/added_tokens.json""",
    },
    """merges_file""": {
        """RUCAIBox/mvp""": """https://huggingface.co/RUCAIBox/mvp/resolve/main/merges.txt""",
    },
}

A							:				Any			       =	{
    """RUCAIBox/mvp""": 1_024,
}
@lru_cache()
def     snake_case_  (       ):



    """simple docstring"""




    __lowercase          =						(
        list(range(ord("""!"""      )   ,ord("""~"""      ) + 1      )      ) + list(range(ord("""¡"""      )   ,ord("""¬"""      ) + 1      )      ) + list(range(ord("""®"""      )   ,ord("""ÿ"""      ) + 1      )      )
    )
    __lowercase          =						bs[:]
    __lowercase          =						0
    for b in range(2**8      ):
        if b not in bs:
            bs.append(SCREAMING_SNAKE_CASE_      )
            cs.append(2**8 + n      )
            n += 1
    __lowercase          =						[chr(SCREAMING_SNAKE_CASE_      ) for n in cs]
    return dict(zip(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_      )      )
def     snake_case_  (       a__		:     Any      ):



    """simple docstring"""




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






class 	SCREAMING_SNAKE_CASE(    __A					):
       snake_case_			:					str   = VOCAB_FILES_NAMES
       snake_case_			:					List[str]   = PRETRAINED_VOCAB_FILES_MAP
       snake_case_			:					Tuple   = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       snake_case_			:					List[Any]   = ["""input_ids""", """attention_mask"""]

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

           """simple docstring"""



           __lowercase          =						AddedToken(lowerCamelCase__ ,		lstrip=lowerCamelCase__ ,		rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__ ,		lowerCamelCase__					) else bos_token
           __lowercase          =						AddedToken(lowerCamelCase__ ,		lstrip=lowerCamelCase__ ,		rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__ ,		lowerCamelCase__					) else eos_token
           __lowercase          =						AddedToken(lowerCamelCase__ ,		lstrip=lowerCamelCase__ ,		rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__ ,		lowerCamelCase__					) else sep_token
           __lowercase          =						AddedToken(lowerCamelCase__ ,		lstrip=lowerCamelCase__ ,		rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__ ,		lowerCamelCase__					) else cls_token
           __lowercase          =						AddedToken(lowerCamelCase__ ,		lstrip=lowerCamelCase__ ,		rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__ ,		lowerCamelCase__					) else unk_token
           __lowercase          =						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
           __lowercase          =						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:
               __lowercase          =						json.load(lowerCamelCase__					)
           __lowercase          =						{v: k for k, v in self.encoder.items()}
           __lowercase          =						errors  # how to handle errors in decoding
           __lowercase          =						bytes_to_unicode()
           __lowercase          =						{v: k for k, v in self.byte_encoder.items()}
           with open(lowerCamelCase__ ,		encoding="""utf-8"""					) as merges_handle:
               __lowercase          =						merges_handle.read().split("""\n"""					)[1:-1]
           __lowercase          =						[tuple(merge.split()					) for merge in bpe_merges]
           __lowercase          =						dict(zip(lowerCamelCase__ ,		range(len(lowerCamelCase__					)					)					)					)
           __lowercase          =						{}
           __lowercase          =						add_prefix_space

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

       @property
       def    snake_case__      (   self					)   -> Union[str, Any]:

           """simple docstring"""



           return len(self.encoder					)

       def    snake_case__      (   self					)   -> int:

           """simple docstring"""



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

       def    snake_case__      (   self ,		lowerCamelCase__					)   -> Dict:

           """simple docstring"""



           if token in self.cache:
               return self.cache[token]
           __lowercase          =						tuple(lowerCamelCase__					)
           __lowercase          =						get_pairs(lowerCamelCase__					)

           if not pairs:
               return token

           while True:
               __lowercase          =						min(lowerCamelCase__ ,		key=lambda lowerCamelCase__					: self.bpe_ranks.get(lowerCamelCase__ ,		float("""inf"""					)					)					)
               if bigram not in self.bpe_ranks:
                   break
               __lowercase    ,__lowercase          =						bigram
               __lowercase          =						[]
               __lowercase          =						0
               while i < len(lowerCamelCase__					):
                   try:
                       __lowercase          =						word.index(lowerCamelCase__ ,		lowerCamelCase__					)
                   except ValueError:
                       new_word.extend(word[i:]					)
                       break
                   else:
                       new_word.extend(word[i:j]					)
                       __lowercase          =						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
               __lowercase          =						tuple(lowerCamelCase__					)
               __lowercase          =						new_word
               if len(lowerCamelCase__					) == 1:
                   break
               else:
                   __lowercase          =						get_pairs(lowerCamelCase__					)
           __lowercase          =						""" """.join(lowerCamelCase__					)
           __lowercase          =						word
           return word

       def    snake_case__      (   self ,		lowerCamelCase__					)   -> Tuple:

           """simple docstring"""



           __lowercase          =						[]
           for token in re.findall(self.pat ,		lowerCamelCase__					):
               __lowercase          =						"""""".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    snake_case__      (   self ,		lowerCamelCase__					)   -> Any:

           """simple docstring"""



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

       def    snake_case__      (   self ,		lowerCamelCase__					)   -> Dict:

           """simple docstring"""



           return self.decoder.get(lowerCamelCase__					)

       def    snake_case__      (   self ,		lowerCamelCase__					)   -> Any:

           """simple docstring"""



           __lowercase          =						"""""".join(lowerCamelCase__					)
           __lowercase          =						bytearray([self.byte_decoder[c] for c in text]					).decode("""utf-8""" ,		errors=self.errors					)
           return text

       def    snake_case__      (   self ,		lowerCamelCase__ ,		lowerCamelCase__ = None					)   -> Tuple[str]:

           """simple docstring"""



           if not os.path.isdir(lowerCamelCase__					):
               logger.error(F'Vocabulary path ({save_directory}) should be a directory'					)
               return
           __lowercase          =						os.path.join(
               lowerCamelCase__ ,		(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]					)
           __lowercase          =						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"""					)

           __lowercase          =						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 lowerCamelCase__					: 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!"""					)
                       __lowercase          =						token_index
                   writer.write(""" """.join(lowerCamelCase__					) + """\n"""					)
                   index += 1

           return vocab_file, merge_file

       def    snake_case__      (   self ,		lowerCamelCase__ ,		lowerCamelCase__ = None					)   -> List[int]:

           """simple docstring"""



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

       def    snake_case__      (   self ,		lowerCamelCase__ ,		lowerCamelCase__ = None ,		lowerCamelCase__ = False					)   -> List[int]:

           """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    snake_case__      (   self ,		lowerCamelCase__ ,		lowerCamelCase__ = None					)   -> List[int]:

           """simple docstring"""



           __lowercase          =						[self.sep_token_id]
           __lowercase          =						[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    snake_case__      (   self ,		lowerCamelCase__ ,		lowerCamelCase__=False ,		**lowerCamelCase__					)   -> List[str]:

           """simple docstring"""



           __lowercase          =						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()):
               __lowercase          =						""" """ + text
           return (text, kwargs)










'''simple docstring'''
from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


A							:				Tuple			     =	{
    """configuration_dpr""": ["""DPR_PRETRAINED_CONFIG_ARCHIVE_MAP""", """DPRConfig"""],
    """tokenization_dpr""": [
        """DPRContextEncoderTokenizer""",
        """DPRQuestionEncoderTokenizer""",
        """DPRReaderOutput""",
        """DPRReaderTokenizer""",
    ],
}


try:
  if not is_tokenizers_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  A							:				str			     =	[
      """DPRContextEncoderTokenizerFast""",
      """DPRQuestionEncoderTokenizerFast""",
      """DPRReaderTokenizerFast""",
  ]

try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  A							:				Tuple			     =	[
      """DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """DPRContextEncoder""",
      """DPRPretrainedContextEncoder""",
      """DPRPreTrainedModel""",
      """DPRPretrainedQuestionEncoder""",
      """DPRPretrainedReader""",
      """DPRQuestionEncoder""",
      """DPRReader""",
  ]

try:
  if not is_tf_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  A							:				Tuple			     =	[
      """TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """TFDPRContextEncoder""",
      """TFDPRPretrainedContextEncoder""",
      """TFDPRPretrainedQuestionEncoder""",
      """TFDPRPretrainedReader""",
      """TFDPRQuestionEncoder""",
      """TFDPRReader""",
  ]


if TYPE_CHECKING:
  from .configuration_dpr import DPR_PRETRAINED_CONFIG_ARCHIVE_MAP, DPRConfig
  from .tokenization_dpr import (
      DPRContextEncoderTokenizer,
      DPRQuestionEncoderTokenizer,
      DPRReaderOutput,
      DPRReaderTokenizer,
  )

  try:
    if not is_tokenizers_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .tokenization_dpr_fast import (
        DPRContextEncoderTokenizerFast,
        DPRQuestionEncoderTokenizerFast,
        DPRReaderTokenizerFast,
    )

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_dpr import (
        DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
        DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
        DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST,
        DPRContextEncoder,
        DPRPretrainedContextEncoder,
        DPRPreTrainedModel,
        DPRPretrainedQuestionEncoder,
        DPRPretrainedReader,
        DPRQuestionEncoder,
        DPRReader,
    )

  try:
    if not is_tf_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_tf_dpr import (
        TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
        TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
        TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST,
        TFDPRContextEncoder,
        TFDPRPretrainedContextEncoder,
        TFDPRPretrainedQuestionEncoder,
        TFDPRPretrainedReader,
        TFDPRQuestionEncoder,
        TFDPRReader,
    )

else:
  import sys

  A							:				int			     =	_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)




