code: import json
  import os
  import shutil
  import sys
  import tempfile
  import unittest
  from unittest import TestCase
  from unittest.mock import patch
  
  import faiss
  import numpy as np
  from datasets import Dataset
  
  from transformers import BartConfig, BartTokenizer, DPRConfig, DPRQuestionEncoderTokenizer, RagConfig
  from transformers.file_utils import is_datasets_available, is_faiss_available, is_psutil_available, is_torch_available
  from transformers.integrations import is_ray_available
  from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
  from transformers.models.rag.retrieval_rag import CustomHFIndex, RagRetriever
  from transformers.models.roberta.tokenization_roberta import VOCAB_FILES_NAMES as BART_VOCAB_FILES_NAMES
  from transformers.testing_utils import require_ray
  
  
  sys.path.append(os.path.join(os.getcwd()))  # noqa: E402 # noqa: E402 # isort:skip
  
  if is_torch_available():
   from distributed_pytorch_retriever import RagPyTorchDistributedRetriever  # noqa: E402 # isort:skip
  else:
   UpperCAmelCase					=				None
  
  if is_ray_available():
   import ray  # noqa: E402 # isort:skip
   from distributed_ray_retriever import RagRayDistributedRetriever, RayRetriever  # noqa: E402 # isort:skip
  else:
   UpperCAmelCase					=				None
   UpperCAmelCase					=				None
   UpperCAmelCase					=				None
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	List[str]				):
         if not (is_datasets_available() and is_faiss_available() and is_psutil_available()):
                _UpperCamelCase     :       List[Any]  						=  unittest.skip('test requires Datasets, Faiss, psutil'				)(lowerCAmelCase__				)
         return test_case
  
  
  
  
  
  @require_distributed_retrieval
  class    __lowerCAmelCase  (       snake_case__						):
  
  
  
   '''simple docstring'''
  
   def 					__snake_case					(			self       :		Dict  ) ->							List[Any]:
          _UpperCamelCase     :       Optional[Any]  						=  tempfile.mkdtemp()
          _UpperCamelCase     :       Tuple  						=  8
  
          # DPR tok
          _UpperCamelCase     :       Any  						=  [
              '[UNK]',
              '[CLS]',
              '[SEP]',
              '[PAD]',
              '[MASK]',
              'want',
              '##want',
              '##ed',
              'wa',
              'un',
              'runn',
              '##ing',
              ',',
              'low',
              'lowest',
          ]
          _UpperCamelCase     :       List[str]  						=  os.path.join(self.tmpdirname       ,			'dpr_tokenizer'  )
          os.makedirs(lowerCamelCase       ,			exist_ok=lowerCamelCase  )
          _UpperCamelCase     :       int  						=  os.path.join(lowerCamelCase       ,			DPR_VOCAB_FILES_NAMES['vocab_file']  )
          with open(self.vocab_file       ,			'w'       ,			encoding='utf-8'  ) as vocab_writer:
                 vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]  )  )
  
          # BART tok
          _UpperCamelCase     :       Optional[Any]  						=  [
              'l',
              'o',
              'w',
              'e',
              'r',
              's',
              't',
              'i',
              'd',
              'n',
              '\u0120',
              '\u0120l',
              '\u0120n',
              '\u0120lo',
              '\u0120low',
              'er',
              '\u0120lowest',
              '\u0120newer',
              '\u0120wider',
              '<unk>',
          ]
          _UpperCamelCase     :       Any  						=  dict(zip(lowerCamelCase       ,			range(len(lowerCamelCase  )  )  )  )
          _UpperCamelCase     :       str  						=  ['#version: 0.2', '\u0120 l', '\u0120l o', '\u0120lo w', 'e r', '']
          _UpperCamelCase     :       Union[str, Any]  						=  {'unk_token': '<unk>'}
  
          _UpperCamelCase     :       int  						=  os.path.join(self.tmpdirname       ,			'bart_tokenizer'  )
          os.makedirs(lowerCamelCase       ,			exist_ok=lowerCamelCase  )
          _UpperCamelCase     :       List[Any]  						=  os.path.join(lowerCamelCase       ,			BART_VOCAB_FILES_NAMES['vocab_file']  )
          _UpperCamelCase     :       Tuple  						=  os.path.join(lowerCamelCase       ,			BART_VOCAB_FILES_NAMES['merges_file']  )
          with open(self.vocab_file       ,			'w'       ,			encoding='utf-8'  ) as fp:
                 fp.write(json.dumps(lowerCamelCase  ) + '\n'  )
          with open(self.merges_file       ,			'w'       ,			encoding='utf-8'  ) as fp:
                 fp.write('\n'.join(lowerCamelCase  )  )
  
   def 					__snake_case					(			self       :		List[Any]  ) ->							Tuple:
          return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname       ,			'dpr_tokenizer'  )  )
  
   def 					__snake_case					(			self       :		Union[str, Any]  ) ->							int:
          return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname       ,			'bart_tokenizer'  )  )
  
   def 					__snake_case					(			self       :		str  ) ->							List[Any]:
          shutil.rmtree(self.tmpdirname  )
  
   def 					__snake_case					(			self       :		Union[str, Any]  ) ->							Any:
          _UpperCamelCase     :       Optional[int]  						=  Dataset.from_dict(
              {
                  'id': ['0', '1'],
                  'text': ['foo', 'bar'],
                  'title': ['Foo', 'Bar'],
                  'embeddings': [np.ones(self.retrieval_vector_size  ), 2 * np.ones(self.retrieval_vector_size  )],
              }  )
          dataset.add_faiss_index('embeddings'       ,			string_factory='Flat'       ,			metric_type=faiss.METRIC_INNER_PRODUCT  )
          return dataset
  
   def 					__snake_case					(			self       :		Any       ,			lowerCamelCase       :		bool       ,			lowerCamelCase       :		Optional[Any]=1_2_3_4_5  ) ->							Union[str, Any]:
          _UpperCamelCase     :       Optional[Any]  						=  self.get_dummy_dataset()
          _UpperCamelCase     :       Optional[int]  						=  RagConfig(
              retrieval_vector_size=self.retrieval_vector_size       ,			question_encoder=DPRConfig().to_dict()       ,			generator=BartConfig().to_dict()       ,			)
          with patch('transformers.models.rag.retrieval_rag.load_dataset'  ) as mock_load_dataset:
                 _UpperCamelCase     :       Optional[Any]  						=  dataset
                 _UpperCamelCase     :       List[str]  						=  RagPyTorchDistributedRetriever(
                     lowerCamelCase       ,			question_encoder_tokenizer=self.get_dpr_tokenizer()       ,			generator_tokenizer=self.get_bart_tokenizer()       ,			)
                 if init_retrieval:
                        retriever.init_retrieval(lowerCamelCase  )
          return retriever
  
   def 					__snake_case					(			self       :		Dict       ,			lowerCamelCase       :		bool  ) ->							str:
          # Have to run in local mode because sys.path modifications at top of
          # file are not propogated to remote workers.
          # https://stackoverflow.com/questions/54338013/parallel-import-a-python-file-from-sibling-folder
          ray.init(local_mode=lowerCamelCase  )
          _UpperCamelCase     :       Optional[int]  						=  RagConfig(
              retrieval_vector_size=self.retrieval_vector_size       ,			question_encoder=DPRConfig().to_dict()       ,			generator=BartConfig().to_dict()       ,			)
          _UpperCamelCase     :       Any  						=  ray.remote(lowerCamelCase  )
          _UpperCamelCase     :       int  						=  [remote_cls.remote() for _ in range(1  )]
          with patch('transformers.models.rag.retrieval_rag.load_dataset'  ) as mock_load_dataset:
                 _UpperCamelCase     :       int  						=  self.get_dummy_dataset()
                 _UpperCamelCase     :       Tuple  						=  RagRayDistributedRetriever(
                     lowerCamelCase       ,			question_encoder_tokenizer=self.get_dpr_tokenizer()       ,			generator_tokenizer=self.get_bart_tokenizer()       ,			retrieval_workers=lowerCamelCase       ,			)
                 if init_retrieval:
                        retriever.init_retrieval()
          return retriever
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		bool       ,			lowerCamelCase       :		bool       ,			lowerCamelCase       :		List[Any]=1_2_3_4_5  ) ->							List[str]:
          _UpperCamelCase     :       int  						=  self.get_dummy_dataset()
          _UpperCamelCase     :       Any  						=  RagConfig(
              retrieval_vector_size=self.retrieval_vector_size       ,			question_encoder=DPRConfig().to_dict()       ,			generator=BartConfig().to_dict()       ,			index_name='custom'       ,			)
          if from_disk:
                 _UpperCamelCase     :       Any  						=  os.path.join(self.tmpdirname       ,			'dataset'  )
                 _UpperCamelCase     :       str  						=  os.path.join(self.tmpdirname       ,			'index.faiss'  )
                 dataset.get_index('embeddings'  ).save(os.path.join(self.tmpdirname       ,			'index.faiss'  )  )
                 dataset.drop_index('embeddings'  )
                 dataset.save_to_disk(os.path.join(self.tmpdirname       ,			'dataset'  )  )
                 del dataset
                 _UpperCamelCase     :       str  						=  RagPyTorchDistributedRetriever(
                     lowerCamelCase       ,			question_encoder_tokenizer=self.get_dpr_tokenizer()       ,			generator_tokenizer=self.get_bart_tokenizer()       ,			)
          else:
                 _UpperCamelCase     :       List[Any]  						=  RagPyTorchDistributedRetriever(
                     lowerCamelCase       ,			question_encoder_tokenizer=self.get_dpr_tokenizer()       ,			generator_tokenizer=self.get_bart_tokenizer()       ,			index=CustomHFIndex(config.retrieval_vector_size       ,			lowerCamelCase  )       ,			)
          if init_retrieval:
                 retriever.init_retrieval(lowerCamelCase  )
          return retriever
  
   def 					__snake_case					(			self       :		Dict       ,			lowerCamelCase       :		bool       ,			lowerCamelCase       :		bool  ) ->							Any:
          # Have to run in local mode because sys.path modifications at top of
          # file are not propogated to remote workers.
          # https://stackoverflow.com/questions/54338013/parallel-import-a-python-file-from-sibling-folder
          ray.init(local_mode=lowerCamelCase  )
          _UpperCamelCase     :       Dict  						=  self.get_dummy_dataset()
          _UpperCamelCase     :       Dict  						=  RagConfig(
              retrieval_vector_size=self.retrieval_vector_size       ,			question_encoder=DPRConfig().to_dict()       ,			generator=BartConfig().to_dict()       ,			index_name='custom'       ,			)
          _UpperCamelCase     :       Any  						=  ray.remote(lowerCamelCase  )
          _UpperCamelCase     :       List[str]  						=  [remote_cls.remote() for _ in range(1  )]
          if from_disk:
                 _UpperCamelCase     :       int  						=  os.path.join(self.tmpdirname       ,			'dataset'  )
                 _UpperCamelCase     :       Optional[int]  						=  os.path.join(self.tmpdirname       ,			'index.faiss'  )
                 dataset.get_index('embeddings'  ).save(os.path.join(self.tmpdirname       ,			'index.faiss'  )  )
                 dataset.drop_index('embeddings'  )
                 dataset.save_to_disk(os.path.join(self.tmpdirname       ,			'dataset'  )  )
                 del dataset
                 _UpperCamelCase     :       List[Any]  						=  RagRayDistributedRetriever(
                     lowerCamelCase       ,			question_encoder_tokenizer=self.get_dpr_tokenizer()       ,			generator_tokenizer=self.get_bart_tokenizer()       ,			retrieval_workers=lowerCamelCase       ,			index=CustomHFIndex.load_from_disk(
                         vector_size=config.retrieval_vector_size       ,			dataset_path=config.passages_path       ,			index_path=config.index_path       ,			)       ,			)
          else:
                 _UpperCamelCase     :       List[str]  						=  RagRayDistributedRetriever(
                     lowerCamelCase       ,			question_encoder_tokenizer=self.get_dpr_tokenizer()       ,			generator_tokenizer=self.get_bart_tokenizer()       ,			retrieval_workers=lowerCamelCase       ,			index=CustomHFIndex(config.retrieval_vector_size       ,			lowerCamelCase  )       ,			)
          if init_retrieval:
                 retriever.init_retrieval()
          return retriever
  
   def 					__snake_case					(			self       :		Any       ,			lowerCamelCase       :		RagRetriever       ,			lowerCamelCase       :		np.array       ,			lowerCamelCase       :		int  ) ->							List[Any]:
          _UpperCamelCase     :       List[Any]  						=  retriever.retrieve(lowerCamelCase       ,			n_docs=lowerCamelCase  )
          self.assertEqual(retrieved_doc_embeds.shape       ,			(2, n_docs, self.retrieval_vector_size)  )
          self.assertEqual(len(lowerCamelCase  )       ,			2  )
          self.assertEqual(sorted(doc_dicts[0]  )       ,			['embeddings', 'id', 'text', 'title']  )
          self.assertEqual(len(doc_dicts[0]['id']  )       ,			lowerCamelCase  )
          self.assertEqual(doc_dicts[0]['id'][0]       ,			'1'  )  # max inner product is reached with second doc
          self.assertEqual(doc_dicts[1]['id'][0]       ,			'0'  )  # max inner product is reached with first doc
          self.assertListEqual(doc_ids.tolist()       ,			[[1], [0]]  )
  
   def 					__snake_case					(			self       :		int  ) ->							Union[str, Any]:
          _UpperCamelCase     :       List[Any]  						=  1
          _UpperCamelCase     :       Union[str, Any]  						=  np.array(
              [np.ones(self.retrieval_vector_size  ), -np.ones(self.retrieval_vector_size  )]       ,			dtype=np.floataa  )
  
          self.distributed_retriever_check(
              self.get_dummy_pytorch_distributed_retriever(init_retrieval=lowerCamelCase  )       ,			lowerCamelCase       ,			lowerCamelCase  )
  
   def 					__snake_case					(			self       :		Optional[int]  ) ->							int:
          _UpperCamelCase     :       Optional[Any]  						=  1
          _UpperCamelCase     :       int  						=  np.array(
              [np.ones(self.retrieval_vector_size  ), -np.ones(self.retrieval_vector_size  )]       ,			dtype=np.floataa  )
  
          self.distributed_retriever_check(
              self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=lowerCamelCase       ,			from_disk=lowerCamelCase  )       ,			lowerCamelCase       ,			lowerCamelCase       ,			)
  
   def 					__snake_case					(			self       :		List[Any]  ) ->							List[Any]:
          _UpperCamelCase     :       List[str]  						=  1
          _UpperCamelCase     :       Union[str, Any]  						=  np.array(
              [np.ones(self.retrieval_vector_size  ), -np.ones(self.retrieval_vector_size  )]       ,			dtype=np.floataa  )
  
          self.distributed_retriever_check(
              self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=lowerCamelCase       ,			from_disk=lowerCamelCase  )       ,			lowerCamelCase       ,			lowerCamelCase       ,			)
  
   @require_ray
   def 					__snake_case					(			self       :		Optional[Any]  ) ->							Optional[Any]:
          _UpperCamelCase     :       Optional[int]  						=  1
          _UpperCamelCase     :       Tuple  						=  np.array(
              [np.ones(self.retrieval_vector_size  ), -np.ones(self.retrieval_vector_size  )]       ,			dtype=np.floataa  )
  
          self.distributed_retriever_check(
              self.get_dummy_ray_distributed_retriever(init_retrieval=lowerCamelCase  )       ,			lowerCamelCase       ,			lowerCamelCase  )
          ray.shutdown()
  
   @require_ray
   def 					__snake_case					(			self       :		List[str]  ) ->							Tuple:
          _UpperCamelCase     :       str  						=  1
          _UpperCamelCase     :       Optional[int]  						=  np.array(
              [np.ones(self.retrieval_vector_size  ), -np.ones(self.retrieval_vector_size  )]       ,			dtype=np.floataa  )
          with self.assertRaises(lowerCamelCase  ):
                 self.distributed_retriever_check(
                     self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=lowerCamelCase       ,			from_disk=lowerCamelCase  )       ,			lowerCamelCase       ,			lowerCamelCase       ,			)
          ray.shutdown()
  
   @require_ray
   def 					__snake_case					(			self       :		List[Any]  ) ->							Union[str, Any]:
          _UpperCamelCase     :       int  						=  1
          _UpperCamelCase     :       int  						=  np.array(
              [np.ones(self.retrieval_vector_size  ), -np.ones(self.retrieval_vector_size  )]       ,			dtype=np.floataa  )
  
          self.distributed_retriever_check(
              self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=lowerCamelCase       ,			from_disk=lowerCamelCase  )       ,			lowerCamelCase       ,			lowerCamelCase  )
          ray.shutdown()
  
  
style_context: import json
  import os
  import re
  import unicodedata
  from typing import Optional, Tuple
  
  from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
  from ...utils import logging
  
  
  UpperCAmelCase					=				logging.get_logger(__name__)
  
  UpperCAmelCase					=				{
      '''vocab_file''': '''vocab.json''',
      '''merges_file''': '''merges.txt''',
  }
  
  UpperCAmelCase					=				{
      '''vocab_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/vocab.json'''},
      '''merges_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/merges.txt'''},
  }
  
  UpperCAmelCase					=				{
      '''openai-gpt''': 5_12,
  }
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	int				):
         _UpperCamelCase     :       Optional[Any]  						=  text.strip()
         if not text:
                return []
         _UpperCamelCase     :       Optional[Any]  						=  text.split()
         return tokens
  
  
  
  
  class    __lowerCAmelCase  (       a_						):
  
  
  
   '''simple docstring'''
  
   def __init__(			self       :		str       ,			lowerCamelCase       :		Union[str, Any]=True       ,			lowerCamelCase       :		Optional[Any]=None       ,			lowerCamelCase       :		Dict=True       ,			lowerCamelCase       :		List[Any]=None       ,			lowerCamelCase       :		Optional[Any]=True       ,			) ->							Optional[Any]:
          if never_split is None:
                 _UpperCamelCase     :       Union[str, Any]  						=  []
          _UpperCamelCase     :       List[Any]  						=  do_lower_case
          _UpperCamelCase     :       List[Any]  						=  set(lowerCamelCase  )
          _UpperCamelCase     :       int  						=  tokenize_chinese_chars
          _UpperCamelCase     :       Dict  						=  strip_accents
          _UpperCamelCase     :       List[Any]  						=  do_split_on_punc
  
   def 					__snake_case					(			self       :		int       ,			lowerCamelCase       :		List[str]       ,			lowerCamelCase       :		List[Any]=None  ) ->							List[str]:
          _UpperCamelCase     :       Any  						=  self.never_split.union(set(lowerCamelCase  )  ) if never_split else self.never_split
          _UpperCamelCase     :       int  						=  self._clean_text(lowerCamelCase  )
  
          # This was added on November 1st, 2018 for the multilingual and Chinese
          # models. This is also applied to the English models now, but it doesn't
          # matter since the English models were not trained on any Chinese data
          # and generally don't have any Chinese data in them (there are Chinese
          # characters in the vocabulary because Wikipedia does have some Chinese
          # words in the English Wikipedia.).
          if self.tokenize_chinese_chars:
                 _UpperCamelCase     :       Optional[int]  						=  self._tokenize_chinese_chars(lowerCamelCase  )
          # prevents treating the same character with different unicode codepoints as different characters
          _UpperCamelCase     :       int  						=  unicodedata.normalize('NFC'       ,			lowerCamelCase  )
          _UpperCamelCase     :       Tuple  						=  whitespace_tokenize(lowerCamelCase  )
          _UpperCamelCase     :       Dict  						=  []
          for token in orig_tokens:
                 if token not in never_split:
                        if self.do_lower_case:
                               _UpperCamelCase     :       Tuple  						=  token.lower()
                               if self.strip_accents is not False:
                                      _UpperCamelCase     :       List[Any]  						=  self._run_strip_accents(lowerCamelCase  )
                        elif self.strip_accents:
                               _UpperCamelCase     :       List[str]  						=  self._run_strip_accents(lowerCamelCase  )
                 split_tokens.extend(self._run_split_on_punc(lowerCamelCase       ,			lowerCamelCase  )  )
  
          _UpperCamelCase     :       str  						=  whitespace_tokenize(' '.join(lowerCamelCase  )  )
          return output_tokens
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		Union[str, Any]  ) ->							str:
          _UpperCamelCase     :       Optional[Any]  						=  unicodedata.normalize('NFD'       ,			lowerCamelCase  )
          _UpperCamelCase     :       Dict  						=  []
          for char in text:
                 _UpperCamelCase     :       Tuple  						=  unicodedata.category(lowerCamelCase  )
                 if cat == "Mn":
                        continue
                 output.append(lowerCamelCase  )
          return "".join(lowerCamelCase  )
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		List[Any]       ,			lowerCamelCase       :		int=None  ) ->							List[Any]:
          if not self.do_split_on_punc or (never_split is not None and text in never_split):
                 return [text]
          _UpperCamelCase     :       Optional[int]  						=  list(lowerCamelCase  )
          _UpperCamelCase     :       Dict  						=  0
          _UpperCamelCase     :       Union[str, Any]  						=  True
          _UpperCamelCase     :       int  						=  []
          while i < len(lowerCamelCase  ):
                 _UpperCamelCase     :       List[str]  						=  chars[i]
                 if _is_punctuation(lowerCamelCase  ):
                        output.append([char]  )
                        _UpperCamelCase     :       str  						=  True
                 else:
                        if start_new_word:
                               output.append([]  )
                        _UpperCamelCase     :       List[str]  						=  False
                        output[-1].append(lowerCamelCase  )
                 i += 1
  
          return ["".join(lowerCamelCase  ) for x in output]
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		int  ) ->							Optional[Any]:
          _UpperCamelCase     :       Dict  						=  []
          for char in text:
                 _UpperCamelCase     :       Optional[int]  						=  ord(lowerCamelCase  )
                 if self._is_chinese_char(lowerCamelCase  ):
                        output.append(' '  )
                        output.append(lowerCamelCase  )
                        output.append(' '  )
                 else:
                        output.append(lowerCamelCase  )
          return "".join(lowerCamelCase  )
  
   def 					__snake_case					(			self       :		Union[str, Any]       ,			lowerCamelCase       :		int  ) ->							Optional[Any]:
          # This defines a "chinese character" as anything in the CJK Unicode block:
          #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
          #
          # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
          # despite its name. The modern Korean Hangul alphabet is a different block,
          # as is Japanese Hiragana and Katakana. Those alphabets are used to write
          # space-separated words, so they are not treated specially and handled
          # like the all of the other languages.
          if (
              (cp >= 0x4_E00 and cp <= 0x9_FFF)
              or (cp >= 0x3_400 and cp <= 0x4_DBF)  #
              or (cp >= 0x20_000 and cp <= 0x2A_6DF)  #
              or (cp >= 0x2A_700 and cp <= 0x2B_73F)  #
              or (cp >= 0x2B_740 and cp <= 0x2B_81F)  #
              or (cp >= 0x2B_820 and cp <= 0x2C_EAF)  #
              or (cp >= 0xF_900 and cp <= 0xF_AFF)
              or (cp >= 0x2F_800 and cp <= 0x2F_A1F)  #
          ):  #
                 return True
  
          return False
  
   def 					__snake_case					(			self       :		Any       ,			lowerCamelCase       :		int  ) ->							Dict:
          _UpperCamelCase     :       int  						=  []
          for char in text:
                 _UpperCamelCase     :       str  						=  ord(lowerCamelCase  )
                 if cp == 0 or cp == 0xF_FFD or _is_control(lowerCamelCase  ):
                        continue
                 if _is_whitespace(lowerCamelCase  ):
                        output.append(' '  )
                 else:
                        output.append(lowerCamelCase  )
          return "".join(lowerCamelCase  )
  
  
  
  
  
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	Optional[Any]				):
         _UpperCamelCase     :       List[Any]  						=  set()
         _UpperCamelCase     :       Tuple  						=  word[0]
         for char in word[1:]:
                pairs.add((prev_char, char)				)
                _UpperCamelCase     :       Tuple  						=  char
         return pairs
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	Dict				):
         _UpperCamelCase     :       Union[str, Any]  						=  text.replace('—'		,						'-'				)
         _UpperCamelCase     :       int  						=  text.replace('–'		,						'-'				)
         _UpperCamelCase     :       Dict  						=  text.replace('―'		,						'-'				)
         _UpperCamelCase     :       Any  						=  text.replace('…'		,						'...'				)
         _UpperCamelCase     :       Dict  						=  text.replace('´'		,						'\''				)
         _UpperCamelCase     :       Union[str, Any]  						=  re.sub(R'(-+|~+|!+|"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)'		,						R' \1 '		,						lowercase_				)
         _UpperCamelCase     :       Dict  						=  re.sub(R'\s*\n\s*'		,						' \n '		,						lowercase_				)
         _UpperCamelCase     :       Union[str, Any]  						=  re.sub(R'[^\S\n]+'		,						' '		,						lowercase_				)
         return text.strip()
  
  
  
  
  
  class    __lowerCAmelCase  (       a_						):
  
  
  
   '''simple docstring'''
   _lowerCamelCase						:	int				    = VOCAB_FILES_NAMES
   _lowerCamelCase						:	Any				    = PRETRAINED_VOCAB_FILES_MAP
   _lowerCamelCase						:	Optional[int]				    = PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   _lowerCamelCase						:	List[str]				    = ['input_ids', 'attention_mask']
  
   def __init__(			self       :		str       ,			lowerCamelCase       :		Optional[Any]       ,			lowerCamelCase       :		Tuple       ,			lowerCamelCase       :		Union[str, Any]="<unk>"       ,			**lowerCamelCase       :		Any  ) ->							Dict:
          super().__init__(unk_token=lowerCamelCase       ,			**lowerCamelCase  )
  
          try:
                 import ftfy
                 from spacy.lang.en import English
  
                 _UpperCamelCase     :       Any  						=  English()
                 _UpperCamelCase     :       Any  						=  _nlp.tokenizer
                 _UpperCamelCase     :       Dict  						=  ftfy.fix_text
          except ImportError:
                 logger.warning('ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.'  )
                 _UpperCamelCase     :       int  						=  BasicTokenizer(do_lower_case=lowerCamelCase  )
                 _UpperCamelCase     :       List[Any]  						=  None
  
          with open(lowerCamelCase       ,			encoding='utf-8'  ) as vocab_handle:
                 _UpperCamelCase     :       Tuple  						=  json.load(lowerCamelCase  )
          _UpperCamelCase     :       List[Any]  						=  {v: k for k, v in self.encoder.items()}
          with open(lowerCamelCase       ,			encoding='utf-8'  ) as merges_handle:
                 _UpperCamelCase     :       List[str]  						=  merges_handle.read().split('\n'  )[1:-1]
          _UpperCamelCase     :       Optional[int]  						=  [tuple(merge.split()  ) for merge in merges]
          _UpperCamelCase     :       Dict  						=  dict(zip(lowerCamelCase       ,			range(len(lowerCamelCase  )  )  )  )
          _UpperCamelCase     :       Optional[Any]  						=  {}
  
   @property
   def 					__snake_case					(			self       :		str  ) ->							int:
          return True
  
   @property
   def 					__snake_case					(			self       :		int  ) ->							Dict:
          return len(self.encoder  )
  
   def 					__snake_case					(			self       :		Optional[Any]  ) ->							Union[str, Any]:
          return dict(self.encoder       ,			**self.added_tokens_encoder  )
  
   def 					__snake_case					(			self       :		Union[str, Any]       ,			lowerCamelCase       :		Optional[int]  ) ->							Union[str, Any]:
          _UpperCamelCase     :       int  						=  tuple(token[:-1]  ) + (token[-1] + '</w>',)
          if token in self.cache:
                 return self.cache[token]
          _UpperCamelCase     :       Dict  						=  get_pairs(lowerCamelCase  )
  
          if not pairs:
                 return token + "</w>"
  
          while True:
                 _UpperCamelCase     :       List[str]  						=  min(lowerCamelCase       ,			key=lambda lowerCamelCase  : self.bpe_ranks.get(lowerCamelCase       ,			float('inf'  )  )  )
                 if bigram not in self.bpe_ranks:
                        break
                 _UpperCamelCase						,      _UpperCamelCase     :       Any  						=  bigram
                 _UpperCamelCase     :       Optional[Any]  						=  []
                 _UpperCamelCase     :       Optional[int]  						=  0
                 while i < len(lowerCamelCase  ):
                        try:
                               _UpperCamelCase     :       int  						=  word.index(lowerCamelCase       ,			lowerCamelCase  )
                        except ValueError:
                               new_word.extend(word[i:]  )
                               break
                        else:
                               new_word.extend(word[i:j]  )
                               _UpperCamelCase     :       Union[str, Any]  						=  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
                 _UpperCamelCase     :       str  						=  tuple(lowerCamelCase  )
                 _UpperCamelCase     :       Optional[int]  						=  new_word
                 if len(lowerCamelCase  ) == 1:
                        break
                 else:
                        _UpperCamelCase     :       List[Any]  						=  get_pairs(lowerCamelCase  )
          _UpperCamelCase     :       Any  						=  ' '.join(lowerCamelCase  )
          if word == "\n  </w>":
                 _UpperCamelCase     :       Union[str, Any]  						=  '\n</w>'
          _UpperCamelCase     :       Tuple  						=  word
          return word
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		Optional[Any]  ) ->							List[Any]:
          _UpperCamelCase     :       Optional[int]  						=  []
          if self.fix_text is None:
                 # Using BERT's BasicTokenizer
                 _UpperCamelCase     :       int  						=  self.nlp.tokenize(lowerCamelCase  )
                 for token in text:
                        split_tokens.extend(list(self.bpe(lowerCamelCase  ).split(' '  )  )  )
          else:
                 # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
                 _UpperCamelCase     :       Optional[Any]  						=  self.nlp(text_standardize(self.fix_text(lowerCamelCase  )  )  )
                 for token in text:
                        split_tokens.extend(list(self.bpe(token.text.lower()  ).split(' '  )  )  )
          return split_tokens
  
   def 					__snake_case					(			self       :		int       ,			lowerCamelCase       :		Optional[int]  ) ->							int:
          return self.encoder.get(lowerCamelCase       ,			self.encoder.get(self.unk_token  )  )
  
   def 					__snake_case					(			self       :		Dict       ,			lowerCamelCase       :		List[str]  ) ->							str:
          return self.decoder.get(lowerCamelCase       ,			self.unk_token  )
  
   def 					__snake_case					(			self       :		Union[str, Any]       ,			lowerCamelCase       :		Dict  ) ->							Tuple:
          _UpperCamelCase     :       str  						=  ''.join(lowerCamelCase  ).replace('</w>'       ,			' '  ).strip()
          return out_string
  
   def 					__snake_case					(			self       :		List[str]       ,			lowerCamelCase       :		str       ,			lowerCamelCase       :		Optional[str] = None  ) ->							Tuple[str]:
          if not os.path.isdir(lowerCamelCase  ):
                 logger.error(f"Vocabulary path ({save_directory}) should be a directory"  )
                 return
          _UpperCamelCase     :       str  						=  os.path.join(
              lowerCamelCase       ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']  )
          _UpperCamelCase     :       List[str]  						=  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'  )
  
          _UpperCamelCase     :       List[Any]  						=  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!'  )
                               _UpperCamelCase     :       str  						=  token_index
                        writer.write(' '.join(lowerCamelCase  ) + '\n'  )
                        index += 1
  
          return vocab_file, merge_file
  
  
  
  