





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


snake_case        =       logging.get_logger(__name__)


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

# See all BART models at https://huggingface.co/models?filter=bart
snake_case        =       {
    """vocab_file""": {
        """facebook/bart-base""": """https://huggingface.co/facebook/bart-base/resolve/main/vocab.json""",
        """facebook/bart-large""": """https://huggingface.co/facebook/bart-large/resolve/main/vocab.json""",
        """facebook/bart-large-mnli""": """https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json""",
        """facebook/bart-large-cnn""": """https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json""",
        """facebook/bart-large-xsum""": """https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json""",
        """yjernite/bart_eli5""": """https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """facebook/bart-base""": """https://huggingface.co/facebook/bart-base/resolve/main/merges.txt""",
        """facebook/bart-large""": """https://huggingface.co/facebook/bart-large/resolve/main/merges.txt""",
        """facebook/bart-large-mnli""": """https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt""",
        """facebook/bart-large-cnn""": """https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt""",
        """facebook/bart-large-xsum""": """https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt""",
        """yjernite/bart_eli5""": """https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt""",
    },
}

snake_case        =       {
    """facebook/bart-base""": 1_0_2_4,
    """facebook/bart-large""": 1_0_2_4,
    """facebook/bart-large-mnli""": 1_0_2_4,
    """facebook/bart-large-cnn""": 1_0_2_4,
    """facebook/bart-large-xsum""": 1_0_2_4,
    """yjernite/bart_eli5""": 1_0_2_4,
}

@lru_cache()
def 							SCREAMING_SNAKE_CASE__       (  )      ->						Any:
  _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(a__     )
      cs.append(2**8 + n     )
      n += 1
  _lowercase							= [chr(a__     ) for n in cs]
  return dict(zip(a__			,							a__     )     )

def 							SCREAMING_SNAKE_CASE__       (  snake_case__  :Tuple     )      ->						Any:
  _lowercase							= set()
  _lowercase							= word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)     )
    _lowercase							= char
  return pairs







class 				A_				(      __lowercase	):







       """simple docstring"""





       SCREAMING_SNAKE_CASE_      : Optional[Any]      =  VOCAB_FILES_NAMES
       SCREAMING_SNAKE_CASE_      : List[Any]      =  PRETRAINED_VOCAB_FILES_MAP
       SCREAMING_SNAKE_CASE_      : Optional[int]      =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       SCREAMING_SNAKE_CASE_      : Optional[int]      =  ['input_ids', 'attention_mask']







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

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

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

         with open(__A						,encoding='utf-8'  ) as vocab_handle:
           _lowercase							= json.load(__A  )
         _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(__A						,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(__A						,range(len(__A  )  )  )  )
         _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    __UpperCAmelCase (    self	:	int  )						->					str:
         return len(self.encoder  )







       def    __UpperCAmelCase (    self	:	List[str]  )						->					List[str]:
         return dict(self.encoder						,**self.added_tokens_encoder  )







       def    __UpperCAmelCase (    self	:	Tuple						,__A	:	List[str]  )						->					List[Any]:
         if token in self.cache:
           return self.cache[token]
         _lowercase							= tuple(__A  )
         _lowercase							= get_pairs(__A  )

         if not pairs:
           return token

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

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







       def    __UpperCAmelCase (    self	:	int						,__A	:	Optional[Any]  )						->					Optional[Any]:
         _lowercase							= []
         for token in re.findall(self.pat						,__A  ):
           _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(__A  ).split(' '  )  )
         return bpe_tokens







       def    __UpperCAmelCase (    self	:	str						,__A	:	str  )						->					Optional[Any]:
         return self.encoder.get(__A						,self.encoder.get(self.unk_token  )  )







       def    __UpperCAmelCase (    self	:	Optional[Any]						,__A	:	Union[str, Any]  )						->					str:
         return self.decoder.get(__A  )







       def    __UpperCAmelCase (    self	:	Tuple						,__A	:	Union[str, Any]  )						->					List[Any]:
         _lowercase							= ''.join(__A  )
         _lowercase							= bytearray([self.byte_decoder[c] for c in text]  ).decode('utf-8'						,errors=self.errors  )
         return text







       def    __UpperCAmelCase (    self	:	int						,__A	:	List[str]						,__A	:	List[Any] = None  )						->					Tuple[str]:
         if not os.path.isdir(__A  ):
           logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""  )
           return
         _lowercase							= os.path.join(
             __A						,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']  )
         _lowercase							= os.path.join(
             __A						,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']  )

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

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

         return vocab_file, merge_file







       def    __UpperCAmelCase (    self	:	Optional[Any]						,__A	:	Tuple						,__A	:	Dict = None  )						->					List[int]:
         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    __UpperCAmelCase (    self	:	Dict						,__A	:	str						,__A	:	str = None						,__A	:	int = False  )						->					List[int]:
         if already_has_special_tokens:
           return super().get_special_tokens_mask(
               token_ids_a=__A						,token_ids_a=__A						,already_has_special_tokens=__A  )

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







       def    __UpperCAmelCase (    self	:	Any						,__A	:	Tuple						,__A	:	Optional[int] = None  )						->					List[int]:
         _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    __UpperCAmelCase (    self	:	List[Any]						,__A	:	Any						,__A	:	List[str]=False						,**__A	:	List[Any]  )						->					str:
         _lowercase							= kwargs.pop('add_prefix_space'						,self.add_prefix_space  )
         if (is_split_into_words or add_prefix_space) and (len(__A  ) > 0 and not text[0].isspace()):
           _lowercase							= ' ' + text
         return (text, kwargs)






from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
      import numpy
      import tensorflow as tf

      from transformers import (
          TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
          TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
          TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST,
          BertConfig,
          DPRConfig,
          TFDPRContextEncoder,
          TFDPRQuestionEncoder,
          TFDPRReader,
      )





class 				A_				:







       """simple docstring"""





       def __init__(    self	:	Union[str, Any]						,__A	:	Tuple						,__A	:	Dict=13						,__A	:	Union[str, Any]=7						,__A	:	Optional[Any]=True						,__A	:	Dict=True						,__A	:	List[str]=True						,__A	:	int=True						,__A	:	str=99						,__A	:	Tuple=32						,__A	:	List[Any]=2						,__A	:	Dict=4						,__A	:	List[str]=37						,__A	:	Dict="gelu"						,__A	:	Optional[Any]=0.1						,__A	:	Dict=0.1						,__A	:	Dict=512						,__A	:	Any=16						,__A	:	Tuple=2						,__A	:	List[Any]=0.02						,__A	:	Any=3						,__A	:	Tuple=4						,__A	:	Dict=None						,__A	:	List[Any]=0						,)						->					Optional[Any]:
         _lowercase							= parent
         _lowercase							= batch_size
         _lowercase							= seq_length
         _lowercase							= is_training
         _lowercase							= use_input_mask
         _lowercase							= use_token_type_ids
         _lowercase							= use_labels
         _lowercase							= vocab_size
         _lowercase							= hidden_size
         _lowercase							= num_hidden_layers
         _lowercase							= num_attention_heads
         _lowercase							= intermediate_size
         _lowercase							= hidden_act
         _lowercase							= hidden_dropout_prob
         _lowercase							= attention_probs_dropout_prob
         _lowercase							= max_position_embeddings
         _lowercase							= type_vocab_size
         _lowercase							= type_sequence_label_size
         _lowercase							= initializer_range
         _lowercase							= num_labels
         _lowercase							= num_choices
         _lowercase							= scope
         _lowercase							= projection_dim







       def    __UpperCAmelCase (    self	:	List[Any]  )						->					List[str]:
         _lowercase							= ids_tensor([self.batch_size, self.seq_length]						,self.vocab_size  )

         _lowercase							= None
         if self.use_input_mask:
           # follow test_modeling_tf_ctrl.py
           _lowercase							= random_attention_mask([self.batch_size, self.seq_length]  )

         _lowercase							= None
         if self.use_token_type_ids:
           _lowercase							= ids_tensor([self.batch_size, self.seq_length]						,self.type_vocab_size  )

         _lowercase							= None
         _lowercase							= None
         _lowercase							= None
         if self.use_labels:
           _lowercase							= ids_tensor([self.batch_size]						,self.type_sequence_label_size  )
           _lowercase							= ids_tensor([self.batch_size, self.seq_length]						,self.num_labels  )
           _lowercase							= ids_tensor([self.batch_size]						,self.num_choices  )

         _lowercase							= BertConfig(
             vocab_size=self.vocab_size						,hidden_size=self.hidden_size						,num_hidden_layers=self.num_hidden_layers						,num_attention_heads=self.num_attention_heads						,intermediate_size=self.intermediate_size						,hidden_act=self.hidden_act						,hidden_dropout_prob=self.hidden_dropout_prob						,attention_probs_dropout_prob=self.attention_probs_dropout_prob						,max_position_embeddings=self.max_position_embeddings						,type_vocab_size=self.type_vocab_size						,is_decoder=__A						,initializer_range=self.initializer_range						,)
         _lowercase							= DPRConfig(projection_dim=self.projection_dim						,**config.to_dict()  )

         return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels







       def    __UpperCAmelCase (    self	:	List[Any]						,__A	:	Dict						,__A	:	str						,__A	:	Any						,__A	:	Dict						,__A	:	Tuple						,__A	:	Tuple						,__A	:	int  )						->					List[str]:
         _lowercase							= TFDPRContextEncoder(config=__A  )
         _lowercase							= model(__A						,attention_mask=__A						,token_type_ids=__A  )
         _lowercase							= model(__A						,token_type_ids=__A  )
         _lowercase							= model(__A  )
         self.parent.assertEqual(result.pooler_output.shape						,(self.batch_size, self.projection_dim or self.hidden_size)  )







       def    __UpperCAmelCase (    self	:	List[Any]						,__A	:	Optional[Any]						,__A	:	Tuple						,__A	:	Optional[Any]						,__A	:	Union[str, Any]						,__A	:	Tuple						,__A	:	Tuple						,__A	:	Any  )						->					str:
         _lowercase							= TFDPRQuestionEncoder(config=__A  )
         _lowercase							= model(__A						,attention_mask=__A						,token_type_ids=__A  )
         _lowercase							= model(__A						,token_type_ids=__A  )
         _lowercase							= model(__A  )
         self.parent.assertEqual(result.pooler_output.shape						,(self.batch_size, self.projection_dim or self.hidden_size)  )







       def    __UpperCAmelCase (    self	:	int						,__A	:	List[str]						,__A	:	Optional[int]						,__A	:	List[str]						,__A	:	str						,__A	:	str						,__A	:	int						,__A	:	int  )						->					Optional[Any]:
         _lowercase							= TFDPRReader(config=__A  )
         _lowercase							= model(__A						,attention_mask=__A  )

         self.parent.assertEqual(result.start_logits.shape						,(self.batch_size, self.seq_length)  )
         self.parent.assertEqual(result.end_logits.shape						,(self.batch_size, self.seq_length)  )
         self.parent.assertEqual(result.relevance_logits.shape						,(self.batch_size,)  )







       def    __UpperCAmelCase (    self	:	str  )						->					str:
         _lowercase							= self.prepare_config_and_inputs()
         (
             (
             _lowercase
         )						,		(
             _lowercase
         )						,		(
             _lowercase
         )						,		(
             _lowercase
         )						,		(
             _lowercase
         )						,		(
             _lowercase
         )						,		(
             _lowercase
         )						,		
         )							= config_and_inputs
         _lowercase							= {'input_ids': input_ids}
         return config, inputs_dict





@require_tf
class 				A_				(      UpperCAmelCase					,	UpperCAmelCase					,	unittest.TestCase	):







       """simple docstring"""





       SCREAMING_SNAKE_CASE_      : Union[str, Any]      =  (
           (
               TFDPRContextEncoder,
               TFDPRQuestionEncoder,
               TFDPRReader,
           )
           if is_tf_available()
           else ()
       )
       SCREAMING_SNAKE_CASE_      : Optional[Any]      =  {'''feature-extraction''': TFDPRQuestionEncoder} if is_tf_available() else {}

       SCREAMING_SNAKE_CASE_      : List[Any]      =  False
       SCREAMING_SNAKE_CASE_      : Dict      =  False
       SCREAMING_SNAKE_CASE_      : int      =  False
       SCREAMING_SNAKE_CASE_      : List[Any]      =  False
       SCREAMING_SNAKE_CASE_      : Optional[int]      =  False







       def    __UpperCAmelCase (    self	:	List[str]  )						->					Optional[int]:
         _lowercase							= TFDPRModelTester(self  )
         _lowercase							= ConfigTester(self						,config_class=__A						,hidden_size=37  )







       def    __UpperCAmelCase (    self	:	Any  )						->					List[str]:
         self.config_tester.run_common_tests()







       def    __UpperCAmelCase (    self	:	Dict  )						->					Optional[int]:
         _lowercase							= self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_dpr_context_encoder(*__A  )







       def    __UpperCAmelCase (    self	:	Optional[Any]  )						->					Optional[Any]:
         _lowercase							= self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_dpr_question_encoder(*__A  )







       def    __UpperCAmelCase (    self	:	Dict  )						->					List[Any]:
         _lowercase							= self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_dpr_reader(*__A  )







       @slow
       def    __UpperCAmelCase (    self	:	Optional[Any]  )						->					str:
         for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           _lowercase							= TFDPRContextEncoder.from_pretrained(__A  )
           self.assertIsNotNone(__A  )

         for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           _lowercase							= TFDPRContextEncoder.from_pretrained(__A  )
           self.assertIsNotNone(__A  )

         for model_name in TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           _lowercase							= TFDPRQuestionEncoder.from_pretrained(__A  )
           self.assertIsNotNone(__A  )

         for model_name in TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           _lowercase							= TFDPRReader.from_pretrained(__A  )
           self.assertIsNotNone(__A  )





@require_tf
class 				A_				(      unittest.TestCase	):







       """simple docstring"""





       @slow
       def    __UpperCAmelCase (    self	:	List[Any]  )						->					int:
         _lowercase							= TFDPRQuestionEncoder.from_pretrained('facebook/dpr-question_encoder-single-nq-base'  )

         _lowercase							= tf.constant(
             [[101, 7592, 1010, 2003, 2026, 3899, 1_0140, 1029, 102]]  )  # [CLS] hello, is my dog cute? [SEP]
         _lowercase							= model(__A  )[0]  # embedding shape = (1, 768)
         # compare the actual values for a slice.
         _lowercase							= tf.constant(
             [
                 [
                     0.03236253,
                     0.12753335,
                     0.16818509,
                     0.00279786,
                     0.3896933,
                     0.24264945,
                     0.2178971,
                     -0.02335227,
                     -0.08481959,
                     -0.14324117,
                 ]
             ]  )
         self.assertTrue(numpy.allclose(output[:, :10].numpy()						,expected_slice.numpy()						,atol=1e-4  )  )