




'''simple docstring'''





import json
import os
from typing import Optional, Tuple

import regex as re

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


UpperCAmelCase_		:				List[Any]									=     logging.get_logger(__name__)

UpperCAmelCase_		:				Union[str, Any]									=     {
    "vocab_file": "vocab.json",
    "merges_file": "merges.txt",
}

UpperCAmelCase_		:				Dict									=     {
    "vocab_file": {"ctrl": "https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json"},
    "merges_file": {"ctrl": "https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt"},
}

UpperCAmelCase_		:				Any									=     {
    "ctrl": 256,
}

UpperCAmelCase_		:				int									=     {
    "Pregnancy": 168629,
    "Christianity": 7675,
    "Explain": 106423,
    "Fitness": 63440,
    "Saving": 63163,
    "Ask": 27171,
    "Ass": 95985,
    "Joke": 163509,
    "Questions": 45622,
    "Thoughts": 49605,
    "Retail": 52342,
    "Feminism": 164338,
    "Writing": 11992,
    "Atheism": 192263,
    "Netflix": 48616,
    "Computing": 39639,
    "Opinion": 43213,
    "Alone": 44967,
    "Funny": 58917,
    "Gaming": 40358,
    "Human": 4088,
    "India": 1331,
    "Joker": 77138,
    "Diet": 36206,
    "Legal": 11859,
    "Norman": 4939,
    "Tip": 72689,
    "Weight": 52343,
    "Movies": 46273,
    "Running": 23425,
    "Science": 2090,
    "Horror": 37793,
    "Confession": 60572,
    "Finance": 12250,
    "Politics": 16360,
    "Scary": 191985,
    "Support": 12654,
    "Technologies": 32516,
    "Teenage": 66160,
    "Event": 32769,
    "Learned": 67460,
    "Notion": 182770,
    "Wikipedia": 37583,
    "Books": 6665,
    "Extract": 76050,
    "Confessions": 102701,
    "Conspiracy": 75932,
    "Links": 63674,
    "Narcissus": 150425,
    "Relationship": 54766,
    "Relationships": 134796,
    "Reviews": 41671,
    "News": 4256,
    "Translation": 26820,
    "multilingual": 128406,
}




def 		lowerCAmelCase_   (	lowerCamelCase   ):
				__magic_name__	:	Optional[Any]								=set()
				__magic_name__	:	Dict								=word[0]
				for char in word[1:]:
								pairs.add((prev_char, char)   )
								__magic_name__	:	Union[str, Any]								=char

				__magic_name__	:	List[str]								=set(lowerCamelCase   )
				return pairs




class 		__A      (			UpperCAmelCase_			):
						UpperCamelCase     						=					VOCAB_FILES_NAMES
						UpperCamelCase     						=					PRETRAINED_VOCAB_FILES_MAP
						UpperCamelCase     						=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						UpperCamelCase     						=					CONTROL_CODES

						def __init__(		self   :Optional[Any]    , __snake_case   :Union[str, Any]    , __snake_case   :Optional[Any]    , __snake_case   :Optional[Any]="<unk>"    , **__snake_case   :Optional[int] ):
										'''simple docstring'''

										super().__init__(unk_token=_lowercase    , **_lowercase )

										with open(_lowercase    , encoding="""utf-8""" ) as vocab_handle:
														__magic_name__	:	List[Any]								=json.load(_lowercase )
										__magic_name__	:	int								={v: k for k, v in self.encoder.items()}
										with open(_lowercase    , encoding="""utf-8""" ) as merges_handle:
														__magic_name__	:	List[Any]								=merges_handle.read().split("""\n""" )[1:-1]
										__magic_name__	:	List[Any]								=[tuple(merge.split() ) for merge in merges]
										__magic_name__	:	List[str]								=dict(zip(_lowercase    , range(len(_lowercase ) ) ) )
										__magic_name__	:	Union[str, Any]								={}

						@property
						def     A__						(		self   :List[Any] ):
										'''simple docstring'''

										return len(self.encoder )

						def     A__						(		self   :str ):
										'''simple docstring'''

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

						def     A__						(		self   :List[str]    , __snake_case   :Any ):
										'''simple docstring'''

										if token in self.cache:
														return self.cache[token]
										__magic_name__	:	List[Any]								=tuple(_lowercase )
										__magic_name__	:	str								=tuple(list(word[:-1] ) + [word[-1] + """</w>"""] )
										__magic_name__	:	Union[str, Any]								=get_pairs(_lowercase )

										if not pairs:
														return token

										while True:
														__magic_name__	:	Tuple								=min(_lowercase    , key=lambda __snake_case : self.bpe_ranks.get(_lowercase    , float("""inf""" ) ) )
														if bigram not in self.bpe_ranks:
																		break
														__magic_name__	:	Optional[Any]								=bigram
														__magic_name__	:	Dict								=[]
														__magic_name__	:	Tuple								=0
														while i < len(_lowercase ):
																		try:
																						__magic_name__	:	List[Any]								=word.index(_lowercase    , _lowercase )
																		except ValueError:
																						new_word.extend(word[i:] )
																						break
																		else:
																						new_word.extend(word[i:j] )
																						__magic_name__	:	List[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
														__magic_name__	:	List[str]								=tuple(_lowercase )
														__magic_name__	:	Tuple								=new_word
														if len(_lowercase ) == 1:
																		break
														else:
																		__magic_name__	:	Optional[Any]								=get_pairs(_lowercase )
										__magic_name__	:	Union[str, Any]								='@@ '.join(_lowercase )
										__magic_name__	:	int								=word[:-4]
										__magic_name__	:	Any								=word
										return word

						def     A__						(		self   :List[str]    , __snake_case   :int ):
										'''simple docstring'''

										__magic_name__	:	Optional[Any]								=[]

										__magic_name__	:	Optional[int]								=re.findall(r"""\S+\n?"""    , _lowercase )

										for token in words:
														split_tokens.extend(list(self.bpe(_lowercase ).split(""" """ ) ) )
										return split_tokens

						def     A__						(		self   :Tuple    , __snake_case   :List[Any] ):
										'''simple docstring'''

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

						def     A__						(		self   :Optional[Any]    , __snake_case   :int ):
										'''simple docstring'''

										return self.decoder.get(_lowercase    , self.unk_token )

						def     A__						(		self   :List[str]    , __snake_case   :Optional[int] ):
										'''simple docstring'''

										__magic_name__	:	Union[str, Any]								=' '.join(_lowercase ).replace("""@@ """    , """""" ).strip()
										return out_string


						def     A__						(		self   :str    , __snake_case   :Dict    , __snake_case   :Optional[Any] = None ):
										'''simple docstring'''

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

										__magic_name__	:	Optional[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 __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!""" )
																						__magic_name__	:	Optional[Any]								=token_index
																		writer.write(""" """.join(_lowercase ) + """\n""" )
																		index += 1

										return vocab_file, merge_file

						# def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
						#     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
						#     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
						#     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
						#     return ''.join(tokens_generated_so_far)










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      :

						def __init__(		self   :Optional[int]    , __snake_case   :Optional[Any]    , __snake_case   :List[Any]=13    , __snake_case   :str=7    , __snake_case   :Tuple=True    , __snake_case   :Dict=True    , __snake_case   :Optional[Any]=True    , __snake_case   :str=True    , __snake_case   :int=99    , __snake_case   :int=32    , __snake_case   :Dict=2    , __snake_case   :Optional[Any]=4    , __snake_case   :Dict=37    , __snake_case   :Optional[int]="gelu"    , __snake_case   :Tuple=0.1    , __snake_case   :Tuple=0.1    , __snake_case   :int=5_12    , __snake_case   :int=16    , __snake_case   :int=2    , __snake_case   :Optional[int]=0.02    , __snake_case   :Union[str, Any]=3    , __snake_case   :Any=4    , __snake_case   :str=None    , __snake_case   :str=0    , ):
										'''simple docstring'''

										__magic_name__	:	Optional[int]								=parent
										__magic_name__	:	int								=batch_size
										__magic_name__	:	Any								=seq_length
										__magic_name__	:	List[str]								=is_training
										__magic_name__	:	Any								=use_input_mask
										__magic_name__	:	Union[str, Any]								=use_token_type_ids
										__magic_name__	:	Union[str, Any]								=use_labels
										__magic_name__	:	Optional[Any]								=vocab_size
										__magic_name__	:	Optional[Any]								=hidden_size
										__magic_name__	:	Optional[Any]								=num_hidden_layers
										__magic_name__	:	Optional[Any]								=num_attention_heads
										__magic_name__	:	Tuple								=intermediate_size
										__magic_name__	:	Tuple								=hidden_act
										__magic_name__	:	Tuple								=hidden_dropout_prob
										__magic_name__	:	Any								=attention_probs_dropout_prob
										__magic_name__	:	Union[str, Any]								=max_position_embeddings
										__magic_name__	:	int								=type_vocab_size
										__magic_name__	:	Optional[Any]								=type_sequence_label_size
										__magic_name__	:	Union[str, Any]								=initializer_range
										__magic_name__	:	Optional[Any]								=num_labels
										__magic_name__	:	Any								=num_choices
										__magic_name__	:	Optional[Any]								=scope
										__magic_name__	:	str								=projection_dim

						def     A__						(		self   :Dict ):
										'''simple docstring'''

										__magic_name__	:	Optional[int]								=ids_tensor([self.batch_size, self.seq_length]    , self.vocab_size )

										__magic_name__	:	Union[str, Any]								=None
										if self.use_input_mask:
														# follow test_modeling_tf_ctrl.py
														__magic_name__	:	List[Any]								=random_attention_mask([self.batch_size, self.seq_length] )

										__magic_name__	:	Optional[int]								=None
										if self.use_token_type_ids:
														__magic_name__	:	str								=ids_tensor([self.batch_size, self.seq_length]    , self.type_vocab_size )

										__magic_name__	:	List[str]								=None
										__magic_name__	:	List[Any]								=None
										__magic_name__	:	List[str]								=None
										if self.use_labels:
														__magic_name__	:	str								=ids_tensor([self.batch_size]    , self.type_sequence_label_size )
														__magic_name__	:	Optional[int]								=ids_tensor([self.batch_size, self.seq_length]    , self.num_labels )
														__magic_name__	:	Union[str, Any]								=ids_tensor([self.batch_size]    , self.num_choices )

										__magic_name__	:	Optional[int]								=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=__snake_case    , initializer_range=self.initializer_range    , )
										__magic_name__	:	Dict								=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[Any]    , __snake_case   :List[Any]    , __snake_case   :List[str]    , __snake_case   :Any    , __snake_case   :Any    , __snake_case   :Optional[Any]    , __snake_case   :Optional[Any]    , __snake_case   :List[str] ):
										'''simple docstring'''

										__magic_name__	:	List[Any]								=TFDPRContextEncoder(config=__snake_case )
										__magic_name__	:	Union[str, Any]								=model(__snake_case    , attention_mask=__snake_case    , token_type_ids=__snake_case )
										__magic_name__	:	str								=model(__snake_case    , token_type_ids=__snake_case )
										__magic_name__	:	Optional[int]								=model(__snake_case )
										self.parent.assertEqual(result.pooler_output.shape    , (self.batch_size, self.projection_dim or self.hidden_size) )

						def     A__						(		self   :int    , __snake_case   :str    , __snake_case   :Tuple    , __snake_case   :Optional[Any]    , __snake_case   :Optional[Any]    , __snake_case   :Tuple    , __snake_case   :List[Any]    , __snake_case   :int ):
										'''simple docstring'''

										__magic_name__	:	int								=TFDPRQuestionEncoder(config=__snake_case )
										__magic_name__	:	Union[str, Any]								=model(__snake_case    , attention_mask=__snake_case    , token_type_ids=__snake_case )
										__magic_name__	:	List[str]								=model(__snake_case    , token_type_ids=__snake_case )
										__magic_name__	:	Union[str, Any]								=model(__snake_case )
										self.parent.assertEqual(result.pooler_output.shape    , (self.batch_size, self.projection_dim or self.hidden_size) )

						def     A__						(		self   :List[Any]    , __snake_case   :List[str]    , __snake_case   :Optional[int]    , __snake_case   :str    , __snake_case   :Any    , __snake_case   :str    , __snake_case   :Optional[Any]    , __snake_case   :Dict ):
										'''simple docstring'''

										__magic_name__	:	Optional[int]								=TFDPRReader(config=__snake_case )
										__magic_name__	:	List[str]								=model(__snake_case    , attention_mask=__snake_case )

										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   :Optional[Any] ):
										'''simple docstring'''

										__magic_name__	:	str								=self.prepare_config_and_inputs()
										(
										    (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, 
										)	:	Union[str, Any]								=config_and_inputs
										__magic_name__	:	List[str]								={"""input_ids""": input_ids}
										return config, inputs_dict




@require_tf
class 		__A      (			UpperCamelCase__					,						UpperCamelCase__					,						unittest.TestCase			):
						UpperCamelCase     						=					(
						    (
						        TFDPRContextEncoder,
						        TFDPRQuestionEncoder,
						        TFDPRReader,
						    )
						    if is_tf_available()
						    else ()
						)
						UpperCamelCase     						=					{"""feature-extraction""": TFDPRQuestionEncoder} if is_tf_available() else {}

						UpperCamelCase     						=					False
						UpperCamelCase     						=					False
						UpperCamelCase     						=					False
						UpperCamelCase     						=					False
						UpperCamelCase     						=					False

						def     A__						(		self   :Union[str, Any] ):
										'''simple docstring'''

										__magic_name__	:	Union[str, Any]								=TFDPRModelTester(self )
										__magic_name__	:	Optional[Any]								=ConfigTester(self    , config_class=__snake_case    , hidden_size=37 )

						def     A__						(		self   :Dict ):
										'''simple docstring'''

										self.config_tester.run_common_tests()

						def     A__						(		self   :Tuple ):
										'''simple docstring'''

										__magic_name__	:	str								=self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_dpr_context_encoder(*__snake_case )

						def     A__						(		self   :List[Any] ):
										'''simple docstring'''

										__magic_name__	:	int								=self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_dpr_question_encoder(*__snake_case )

						def     A__						(		self   :Any ):
										'''simple docstring'''

										__magic_name__	:	Optional[int]								=self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_dpr_reader(*__snake_case )


						@slow
						def     A__						(		self   :Tuple ):
										'''simple docstring'''

										for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														__magic_name__	:	Dict								=TFDPRContextEncoder.from_pretrained(__snake_case )
														self.assertIsNotNone(__snake_case )

										for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														__magic_name__	:	Dict								=TFDPRContextEncoder.from_pretrained(__snake_case )
														self.assertIsNotNone(__snake_case )

										for model_name in TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														__magic_name__	:	Union[str, Any]								=TFDPRQuestionEncoder.from_pretrained(__snake_case )
														self.assertIsNotNone(__snake_case )

										for model_name in TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														__magic_name__	:	List[str]								=TFDPRReader.from_pretrained(__snake_case )
														self.assertIsNotNone(__snake_case )




@require_tf
class 		__A      (			unittest.TestCase			):

						@slow
						def     A__						(		self   :Optional[Any] ):
										'''simple docstring'''

										__magic_name__	:	int								=TFDPRQuestionEncoder.from_pretrained("""facebook/dpr-question_encoder-single-nq-base""" )

										__magic_name__	:	int								=tf.constant(
										    [[1_01, 75_92, 10_10, 20_03, 20_26, 38_99, 1_01_40, 10_29, 1_02]] )  # [CLS] hello, is my dog cute? [SEP]
										__magic_name__	:	Optional[Any]								=model(__snake_case )[0]  # embedding shape = (1, 768)
										# compare the actual values for a slice.
										__magic_name__	:	Tuple								=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 ) )




