





'''simple docstring'''


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       UpperCamelCase__  :







	def __init__(				self			:   Dict , lowerCamelCase			:   int , lowerCamelCase			:   List[str]=1_3 , lowerCamelCase			:   List[str]=7 , lowerCamelCase			:   str=True , lowerCamelCase			:   int=True , lowerCamelCase			:   Any=True , lowerCamelCase			:   Optional[int]=True , lowerCamelCase			:   List[Any]=9_9 , lowerCamelCase			:   Dict=3_2 , lowerCamelCase			:   List[Any]=2 , lowerCamelCase			:   Tuple=4 , lowerCamelCase			:   Dict=3_7 , lowerCamelCase			:   Tuple="gelu" , lowerCamelCase			:   str=0.1 , lowerCamelCase			:   List[str]=0.1 , lowerCamelCase			:   Optional[Any]=5_1_2 , lowerCamelCase			:   List[Any]=1_6 , lowerCamelCase			:   Any=2 , lowerCamelCase			:   Union[str, Any]=0.02 , lowerCamelCase			:   Tuple=3 , lowerCamelCase			:   int=4 , lowerCamelCase			:   Dict=None , lowerCamelCase			:   List[Any]=0 , ):




				'''simple docstring'''
				a__     = parent
				a__     = batch_size
				a__     = seq_length
				a__     = is_training
				a__     = use_input_mask
				a__     = use_token_type_ids
				a__     = use_labels
				a__     = vocab_size
				a__     = hidden_size
				a__     = num_hidden_layers
				a__     = num_attention_heads
				a__     = intermediate_size
				a__     = hidden_act
				a__     = hidden_dropout_prob
				a__     = attention_probs_dropout_prob
				a__     = max_position_embeddings
				a__     = type_vocab_size
				a__     = type_sequence_label_size
				a__     = initializer_range
				a__     = num_labels
				a__     = num_choices
				a__     = scope
				a__     = projection_dim







	def 						__a   (				self			:   List[str]		):




				'''simple docstring'''
				a__     = ids_tensor([self.batch_size, self.seq_length] , self.vocab_size		)

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

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

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

				a__     = 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=lowerCamelCase , initializer_range=self.initializer_range , )
				a__     = 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 						__a   (				self			:   List[str] , lowerCamelCase			:   List[str] , lowerCamelCase			:   Union[str, Any] , lowerCamelCase			:   List[Any] , lowerCamelCase			:   Union[str, Any] , lowerCamelCase			:   Dict , lowerCamelCase			:   Dict , lowerCamelCase			:   Tuple		):




				'''simple docstring'''
				a__     = TFDPRContextEncoder(config=lowerCamelCase		)
				a__     = model(lowerCamelCase , attention_mask=lowerCamelCase , token_type_ids=lowerCamelCase		)
				a__     = model(lowerCamelCase , token_type_ids=lowerCamelCase		)
				a__     = model(lowerCamelCase		)
				self.parent.assertEqual(result.pooler_output.shape , (self.batch_size, self.projection_dim or self.hidden_size)		)







	def 						__a   (				self			:   Union[str, Any] , lowerCamelCase			:   str , lowerCamelCase			:   List[str] , lowerCamelCase			:   str , lowerCamelCase			:   int , lowerCamelCase			:   List[Any] , lowerCamelCase			:   List[Any] , lowerCamelCase			:   str		):




				'''simple docstring'''
				a__     = TFDPRQuestionEncoder(config=lowerCamelCase		)
				a__     = model(lowerCamelCase , attention_mask=lowerCamelCase , token_type_ids=lowerCamelCase		)
				a__     = model(lowerCamelCase , token_type_ids=lowerCamelCase		)
				a__     = model(lowerCamelCase		)
				self.parent.assertEqual(result.pooler_output.shape , (self.batch_size, self.projection_dim or self.hidden_size)		)







	def 						__a   (				self			:   List[str] , lowerCamelCase			:   Dict , lowerCamelCase			:   List[Any] , lowerCamelCase			:   List[Any] , lowerCamelCase			:   Tuple , lowerCamelCase			:   List[str] , lowerCamelCase			:   Union[str, Any] , lowerCamelCase			:   Any		):




				'''simple docstring'''
				a__     = TFDPRReader(config=lowerCamelCase		)
				a__     = model(lowerCamelCase , attention_mask=lowerCamelCase		)

				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 						__a   (				self			:   List[Any]		):




				'''simple docstring'''
				a__     = self.prepare_config_and_inputs()
				(
				    (
				    a__
				)   ,				(
				    a__
				)   ,				(
				    a__
				)   ,				(
				    a__
				)   ,				(
				    a__
				)   ,				(
				    a__
				)   ,				(
				    a__
				)   ,				
				)     = config_and_inputs
				a__     = {"input_ids": input_ids}
				return config, inputs_dict






@require_tf
class       UpperCamelCase__  (		__lowerCAmelCase	,__lowerCAmelCase	,unittest.TestCase     ):
	lowerCAmelCase__					:			Any								=						(
	    (
	        TFDPRContextEncoder,
	        TFDPRQuestionEncoder,
	        TFDPRReader,
	    )
	    if is_tf_available()
	    else ()
	)
	lowerCAmelCase__					:			List[str]								=						{"feature-extraction": TFDPRQuestionEncoder} if is_tf_available() else {}

	lowerCAmelCase__					:			Optional[Any]								=						False
	lowerCAmelCase__					:			str								=						False
	lowerCAmelCase__					:			Optional[int]								=						False
	lowerCAmelCase__					:			Union[str, Any]								=						False
	lowerCAmelCase__					:			Tuple								=						False







	def 						__a   (				self			:   Any		):




				'''simple docstring'''
				a__     = TFDPRModelTester(self		)
				a__     = ConfigTester(self , config_class=lowerCamelCase , hidden_size=3_7		)







	def 						__a   (				self			:   Optional[int]		):




				'''simple docstring'''
				self.config_tester.run_common_tests()







	def 						__a   (				self			:   int		):




				'''simple docstring'''
				a__     = self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_dpr_context_encoder(*lowerCamelCase		)







	def 						__a   (				self			:   int		):




				'''simple docstring'''
				a__     = self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_dpr_question_encoder(*lowerCamelCase		)







	def 						__a   (				self			:   str		):




				'''simple docstring'''
				a__     = self.model_tester.prepare_config_and_inputs()
				self.model_tester.create_and_check_dpr_reader(*lowerCamelCase		)







	@slow
	def 						__a   (				self			:   Any		):




				'''simple docstring'''
				for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
							a__     = TFDPRContextEncoder.from_pretrained(lowerCamelCase		)
							self.assertIsNotNone(lowerCamelCase		)

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

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

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






@require_tf
class       UpperCamelCase__  (		unittest.TestCase     ):







	@slow
	def 						__a   (				self			:   int		):




				'''simple docstring'''
				a__     = TFDPRQuestionEncoder.from_pretrained("facebook/dpr-question_encoder-single-nq-base"		)

				a__     = tf.constant(
				    [[1_0_1, 7_5_9_2, 1_0_1_0, 2_0_0_3, 2_0_2_6, 3_8_9_9, 1_0_1_4_0, 1_0_2_9, 1_0_2]]		)  # [CLS] hello, is my dog cute? [SEP]
				a__     = model(lowerCamelCase		)[0]  # embedding shape = (1, 768)
				# compare the actual values for a slice.
				a__     = 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[:, :1_0].numpy() , expected_slice.numpy() , atol=1e-4		)		)










'''simple docstring'''


import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase_   :   str       =					logging.get_logger(__name__)

lowerCAmelCase_   :   Optional[Any]       =					"▁"

lowerCAmelCase_   :   Union[str, Any]       =					{"vocab_file": "sentencepiece.bpe.model"}

lowerCAmelCase_   :   str       =					{
    "vocab_file": {
        "facebook/xglm-564M": "https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model",
    }
}

lowerCAmelCase_   :   Optional[int]       =					{
    "facebook/xglm-564M": 2048,
}






class       UpperCamelCase__  (		__lowerCAmelCase     ):
	lowerCAmelCase__					:			int								=						VOCAB_FILES_NAMES
	lowerCAmelCase__					:			Optional[int]								=						PRETRAINED_VOCAB_FILES_MAP
	lowerCAmelCase__					:			Tuple								=						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	lowerCAmelCase__					:			Optional[int]								=						["input_ids", "attention_mask"]







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




				'''simple docstring'''
				a__     = {} if sp_model_kwargs is None else sp_model_kwargs

				# Compatibility with the original tokenizer
				a__     = 7
				a__     = [F'''<madeupword{i}>''' for i in range(self.num_madeup_words		)]

				a__     = kwargs.get("additional_special_tokens" , []		)
				kwargs["additional_special_tokens"] += [
				    word for word in madeup_words if word not in kwargs["additional_special_tokens"]
				]

				super().__init__(
				    bos_token=lowerCamelCase , eos_token=lowerCamelCase , unk_token=lowerCamelCase , sep_token=lowerCamelCase , cls_token=lowerCamelCase , pad_token=lowerCamelCase , sp_model_kwargs=self.sp_model_kwargs , **lowerCamelCase , )

				a__     = spm.SentencePieceProcessor(**self.sp_model_kwargs		)
				self.sp_model.Load(str(lowerCamelCase		)		)
				a__     = vocab_file

				# Original fairseq vocab and spm vocab must be "aligned":
				# Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
				# -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
				# fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
				# spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

				# The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
				a__     = 1

				# Mimic fairseq token-to-id alignment for the first 4 token
				a__     = {"<s>": 0, "<pad>": 1, "</s>": 2, "<unk>": 3}

				a__     = len(self.sp_model		)
				a__     = {F'''<madeupword{i}>''': sp_size + i + self.fairseq_offset for i in range(self.num_madeup_words		)}
				self.fairseq_tokens_to_ids.update(lowerCamelCase		)

				a__     = {v: k for k, v in self.fairseq_tokens_to_ids.items()}







	def __getstate__(				self			:   Optional[Any]		):




				'''simple docstring'''
				a__     = self.__dict__.copy()
				a__     = None
				a__     = self.sp_model.serialized_model_proto()
				return state







	def __setstate__(				self			:   Dict , lowerCamelCase			:   List[Any]		):




				'''simple docstring'''
				a__     = d

				# for backward compatibility
				if not hasattr(self , "sp_model_kwargs"		):
							a__     = {}

				a__     = spm.SentencePieceProcessor(**self.sp_model_kwargs		)
				self.sp_model.LoadFromSerializedProto(self.sp_model_proto		)







	def 						__a   (				self			:   Union[str, Any] , lowerCamelCase			:   List[int] , lowerCamelCase			:   Optional[List[int]] = None		):




				'''simple docstring'''

				if token_ids_a is None:
							return [self.sep_token_id] + token_ids_a
				a__     = [self.sep_token_id]
				return sep + token_ids_a + sep + sep + token_ids_a







	def 						__a   (				self			:   int , lowerCamelCase			:   List[int] , lowerCamelCase			:   Optional[List[int]] = None , lowerCamelCase			:   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		))
				return [1] + ([0] * len(lowerCamelCase		)) + [1, 1] + ([0] * len(lowerCamelCase		))







	def 						__a   (				self			:   Dict , lowerCamelCase			:   List[int] , lowerCamelCase			:   Optional[List[int]] = None		):




				'''simple docstring'''
				a__     = [self.sep_token_id]

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







	@property
	def 						__a   (				self			:   List[str]		):




				'''simple docstring'''
				return len(self.sp_model		) + self.fairseq_offset + self.num_madeup_words







	def 						__a   (				self			:   Dict		):




				'''simple docstring'''
				a__     = {self.convert_ids_to_tokens(lowerCamelCase		): i for i in range(self.vocab_size		)}
				vocab.update(self.added_tokens_encoder		)
				return vocab







	def 						__a   (				self			:   int , lowerCamelCase			:   str		):




				'''simple docstring'''
				return self.sp_model.encode(lowerCamelCase , out_type=lowerCamelCase		)







	def 						__a   (				self			:   List[str] , lowerCamelCase			:   Union[str, Any]		):




				'''simple docstring'''
				if token in self.fairseq_tokens_to_ids:
							return self.fairseq_tokens_to_ids[token]
				a__     = self.sp_model.PieceToId(lowerCamelCase		)

				# Need to return unknown token if the SP model returned 0
				return spm_id + self.fairseq_offset if spm_id else self.unk_token_id







	def 						__a   (				self			:   Any , lowerCamelCase			:   List[str]		):




				'''simple docstring'''
				if index in self.fairseq_ids_to_tokens:
							return self.fairseq_ids_to_tokens[index]
				return self.sp_model.IdToPiece(index - self.fairseq_offset		)







	def 						__a   (				self			:   int , lowerCamelCase			:   Optional[Any]		):




				'''simple docstring'''
				a__     = "".join(lowerCamelCase		).replace(lowerCamelCase , " "		).strip()
				return out_string







	def 						__a   (				self			:   int , lowerCamelCase			:   str , lowerCamelCase			:   Optional[str] = None		):




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

				if os.path.abspath(self.vocab_file		) != os.path.abspath(lowerCamelCase		) and os.path.isfile(self.vocab_file		):
							copyfile(self.vocab_file , lowerCamelCase		)
				elif not os.path.isfile(self.vocab_file		):
							with open(lowerCamelCase , "wb"		) as fi:
										a__     = self.sp_model.serialized_model_proto()
										fi.write(lowerCamelCase		)

				return (out_vocab_file,)



