


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     UpperCamelCase_(			_A				:List[Any]					)->						Any:
			if not (is_datasets_available() and is_faiss_available() and is_psutil_available()):
						UpperCamelCase__   =    unittest.skip("test requires Datasets, Faiss, psutil"					)(_A					)
			return test_case







@require_distributed_retrieval
class     lowerCamelCase__		(							UpperCAmelCase ):






			"""simple docstring"""






			def 			snake_case__			(							self       ):

						'''simple docstring'''


						UpperCamelCase__   =    tempfile.mkdtemp()
						UpperCamelCase__   =    8

						# DPR tok
						UpperCamelCase__   =    [
						    "[UNK]",
						    "[CLS]",
						    "[SEP]",
						    "[PAD]",
						    "[MASK]",
						    "want",
						    "##want",
						    "##ed",
						    "wa",
						    "un",
						    "runn",
						    "##ing",
						    ",",
						    "low",
						    "lowest",
						]
						UpperCamelCase__   =    os.path.join(self.tmpdirname ,  "dpr_tokenizer"       )
						os.makedirs(snake_case ,  exist_ok=snake_case       )
						UpperCamelCase__   =    os.path.join(snake_case ,  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__   =    [
						    "l",
						    "o",
						    "w",
						    "e",
						    "r",
						    "s",
						    "t",
						    "i",
						    "d",
						    "n",
						    "\u0120",
						    "\u0120l",
						    "\u0120n",
						    "\u0120lo",
						    "\u0120low",
						    "er",
						    "\u0120lowest",
						    "\u0120newer",
						    "\u0120wider",
						    "<unk>",
						]
						UpperCamelCase__   =    dict(zip(snake_case ,  range(len(snake_case       )       )       )       )
						UpperCamelCase__   =    ["#version: 0.2", "\u0120 l", "\u0120l o", "\u0120lo w", "e r", ""]
						UpperCamelCase__   =    {"unk_token": "<unk>"}

						UpperCamelCase__   =    os.path.join(self.tmpdirname ,  "bart_tokenizer"       )
						os.makedirs(snake_case ,  exist_ok=snake_case       )
						UpperCamelCase__   =    os.path.join(snake_case ,  BART_VOCAB_FILES_NAMES["vocab_file"]       )
						UpperCamelCase__   =    os.path.join(snake_case ,  BART_VOCAB_FILES_NAMES["merges_file"]       )
						with open(self.vocab_file ,  "w" ,  encoding="utf-8"       ) as fp:
									fp.write(json.dumps(snake_case       ) + "\n"       )
						with open(self.merges_file ,  "w" ,  encoding="utf-8"       ) as fp:
									fp.write("\n".join(snake_case       )       )




			def 			snake_case__			(							self       ):

						'''simple docstring'''


						return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname ,  "dpr_tokenizer"       )       )




			def 			snake_case__			(							self       ):

						'''simple docstring'''


						return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname ,  "bart_tokenizer"       )       )




			def 			snake_case__			(							self       ):

						'''simple docstring'''


						shutil.rmtree(self.tmpdirname       )




			def 			snake_case__			(							self       ):

						'''simple docstring'''


						UpperCamelCase__   =    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 ,  snake_case ,  snake_case=12345       ):

						'''simple docstring'''


						UpperCamelCase__   =    self.get_dummy_dataset()
						UpperCamelCase__   =    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__   =    dataset
									UpperCamelCase__   =    RagPyTorchDistributedRetriever(
									    snake_case ,  question_encoder_tokenizer=self.get_dpr_tokenizer() ,  generator_tokenizer=self.get_bart_tokenizer() ,  )
									if init_retrieval:
												retriever.init_retrieval(snake_case       )
						return retriever




			def 			snake_case__			(							self ,  snake_case       ):

						'''simple docstring'''


						ray.init(local_mode=snake_case       )
						UpperCamelCase__   =    RagConfig(
						    retrieval_vector_size=self.retrieval_vector_size ,  question_encoder=DPRConfig().to_dict() ,  generator=BartConfig().to_dict() ,  )
						UpperCamelCase__   =    ray.remote(snake_case       )
						UpperCamelCase__   =    [remote_cls.remote() for _ in range(1       )]
						with patch("transformers.models.rag.retrieval_rag.load_dataset"       ) as mock_load_dataset:
									UpperCamelCase__   =    self.get_dummy_dataset()
									UpperCamelCase__   =    RagRayDistributedRetriever(
									    snake_case ,  question_encoder_tokenizer=self.get_dpr_tokenizer() ,  generator_tokenizer=self.get_bart_tokenizer() ,  retrieval_workers=snake_case ,  )
									if init_retrieval:
												retriever.init_retrieval()
						return retriever




			def 			snake_case__			(							self ,  snake_case ,  snake_case ,  snake_case=12345       ):

						'''simple docstring'''


						UpperCamelCase__   =    self.get_dummy_dataset()
						UpperCamelCase__   =    RagConfig(
						    retrieval_vector_size=self.retrieval_vector_size ,  question_encoder=DPRConfig().to_dict() ,  generator=BartConfig().to_dict() ,  index_name="custom" ,  )
						if from_disk:
									UpperCamelCase__   =    os.path.join(self.tmpdirname ,  "dataset"       )
									UpperCamelCase__   =    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__   =    RagPyTorchDistributedRetriever(
									    snake_case ,  question_encoder_tokenizer=self.get_dpr_tokenizer() ,  generator_tokenizer=self.get_bart_tokenizer() ,  )
						else:
									UpperCamelCase__   =    RagPyTorchDistributedRetriever(
									    snake_case ,  question_encoder_tokenizer=self.get_dpr_tokenizer() ,  generator_tokenizer=self.get_bart_tokenizer() ,  index=CustomHFIndex(config.retrieval_vector_size ,  snake_case       ) ,  )
						if init_retrieval:
									retriever.init_retrieval(snake_case       )
						return retriever




			def 			snake_case__			(							self ,  snake_case ,  snake_case       ):

						'''simple docstring'''


						ray.init(local_mode=snake_case       )
						UpperCamelCase__   =    self.get_dummy_dataset()
						UpperCamelCase__   =    RagConfig(
						    retrieval_vector_size=self.retrieval_vector_size ,  question_encoder=DPRConfig().to_dict() ,  generator=BartConfig().to_dict() ,  index_name="custom" ,  )
						UpperCamelCase__   =    ray.remote(snake_case       )
						UpperCamelCase__   =    [remote_cls.remote() for _ in range(1       )]
						if from_disk:
									UpperCamelCase__   =    os.path.join(self.tmpdirname ,  "dataset"       )
									UpperCamelCase__   =    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__   =    RagRayDistributedRetriever(
									    snake_case ,  question_encoder_tokenizer=self.get_dpr_tokenizer() ,  generator_tokenizer=self.get_bart_tokenizer() ,  retrieval_workers=snake_case ,  index=CustomHFIndex.load_from_disk(
									        vector_size=config.retrieval_vector_size ,  dataset_path=config.passages_path ,  index_path=config.index_path ,  ) ,  )
						else:
									UpperCamelCase__   =    RagRayDistributedRetriever(
									    snake_case ,  question_encoder_tokenizer=self.get_dpr_tokenizer() ,  generator_tokenizer=self.get_bart_tokenizer() ,  retrieval_workers=snake_case ,  index=CustomHFIndex(config.retrieval_vector_size ,  snake_case       ) ,  )
						if init_retrieval:
									retriever.init_retrieval()
						return retriever




			def 			snake_case__			(							self ,  snake_case ,  snake_case ,  snake_case       ):

						'''simple docstring'''


						UpperCamelCase__, UpperCamelCase__, UpperCamelCase__   =    retriever.retrieve(snake_case ,  n_docs=snake_case       )
						self.assertEqual(retrieved_doc_embeds.shape ,  (2, n_docs, self.retrieval_vector_size)       )
						self.assertEqual(len(snake_case       ) ,  2       )
						self.assertEqual(sorted(doc_dicts[0]       ) ,  ["embeddings", "id", "text", "title"]       )
						self.assertEqual(len(doc_dicts[0]["id"]       ) ,  snake_case       )
						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       ):

						'''simple docstring'''


						UpperCamelCase__   =    1
						UpperCamelCase__   =    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=snake_case       ) ,  snake_case ,  snake_case       )




			def 			snake_case__			(							self       ):

						'''simple docstring'''


						UpperCamelCase__   =    1
						UpperCamelCase__   =    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=snake_case ,  from_disk=snake_case       ) ,  snake_case ,  snake_case ,  )




			def 			snake_case__			(							self       ):

						'''simple docstring'''


						UpperCamelCase__   =    1
						UpperCamelCase__   =    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=snake_case ,  from_disk=snake_case       ) ,  snake_case ,  snake_case ,  )




			@require_ray
			def 			snake_case__			(							self       ):

						'''simple docstring'''


						UpperCamelCase__   =    1
						UpperCamelCase__   =    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=snake_case       ) ,  snake_case ,  snake_case       )
						ray.shutdown()




			@require_ray
			def 			snake_case__			(							self       ):

						'''simple docstring'''


						UpperCamelCase__   =    1
						UpperCamelCase__   =    np.array(
						    [np.ones(self.retrieval_vector_size       ), -np.ones(self.retrieval_vector_size       )] ,  dtype=np.floataa       )
						with self.assertRaises(snake_case       ):
									self.distributed_retriever_check(
									    self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=snake_case ,  from_disk=snake_case       ) ,  snake_case ,  snake_case ,  )
						ray.shutdown()







			@require_ray
			def 			snake_case__			(							self       ):

						'''simple docstring'''


						UpperCamelCase__   =    1
						UpperCamelCase__   =    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=snake_case ,  from_disk=snake_case       ) ,  snake_case ,  snake_case       )
						ray.shutdown()






import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

import numpy as np
import regex as re

from ...utils import is_tf_available, is_torch_available, logging


if TYPE_CHECKING:
						if is_torch_available():
												import torch
						if is_tf_available():
												import tensorflow as tf

from ...tokenization_utils import AddedToken, PreTrainedTokenizer


__UpperCamelCase      						=    logging.get_logger(__name__)

__UpperCamelCase      						=    {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

__UpperCamelCase      						=    {
    'vocab_file': {
        'Salesforce/codegen-350M-mono': 'https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/vocab.json',
    },
    'merges_file': {
        'Salesforce/codegen-350M-mono': 'https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/merges.txt',
    },
}

__UpperCamelCase      						=    {
    'Salesforce/codegen-350M-mono': 2_0_4_8,
}
@lru_cache()
def     UpperCamelCase_(			)->						Tuple:
			UpperCamelCase__   =    (
			    list(range(ord("!"					)			,       ord("~"					) + 1					)					) + list(range(ord("¡"					)			,       ord("¬"					) + 1					)					) + list(range(ord("®"					)			,       ord("ÿ"					) + 1					)					)
			)
			UpperCamelCase__   =    bs[:]
			UpperCamelCase__   =    0
			for b in range(2**8					):
						if b not in bs:
									bs.append(_A					)
									cs.append(2**8 + n					)
									n += 1
			UpperCamelCase__   =    [chr(_A					) for n in cs]
			return dict(zip(_A			,       _A					)					)
def     UpperCamelCase_(			_A				:Tuple					)->						Optional[Any]:
			UpperCamelCase__   =    set()
			UpperCamelCase__   =    word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)					)
						UpperCamelCase__   =    char
			return pairs







class     lowerCamelCase__		(							UpperCAmelCase ):






			"""simple docstring"""






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




			def __init__(							self ,  snake_case ,  snake_case ,  snake_case="replace" ,  snake_case="<|endoftext|>" ,  snake_case="<|endoftext|>" ,  snake_case="<|endoftext|>" ,  snake_case=None ,  snake_case=False ,  snake_case=False ,  **snake_case ,  ):

						'''simple docstring'''


						UpperCamelCase__   =    AddedToken(snake_case ,  lstrip=snake_case ,  rstrip=snake_case       ) if isinstance(snake_case ,  snake_case       ) else bos_token
						UpperCamelCase__   =    AddedToken(snake_case ,  lstrip=snake_case ,  rstrip=snake_case       ) if isinstance(snake_case ,  snake_case       ) else eos_token
						UpperCamelCase__   =    AddedToken(snake_case ,  lstrip=snake_case ,  rstrip=snake_case       ) if isinstance(snake_case ,  snake_case       ) else unk_token
						UpperCamelCase__   =    AddedToken(snake_case ,  lstrip=snake_case ,  rstrip=snake_case       ) if isinstance(snake_case ,  snake_case       ) else pad_token
						super().__init__(
						    errors=snake_case ,  unk_token=snake_case ,  bos_token=snake_case ,  eos_token=snake_case ,  pad_token=snake_case ,  add_prefix_space=snake_case ,  add_bos_token=snake_case ,  **snake_case ,  )
						UpperCamelCase__   =    add_bos_token

						with open(snake_case ,  encoding="utf-8"       ) as vocab_handle:
									UpperCamelCase__   =    json.load(snake_case       )
						UpperCamelCase__   =    {v: k for k, v in self.encoder.items()}
						UpperCamelCase__   =    errors  # how to handle errors in decoding
						UpperCamelCase__   =    bytes_to_unicode()
						UpperCamelCase__   =    {v: k for k, v in self.byte_encoder.items()}
						with open(snake_case ,  encoding="utf-8"       ) as merges_handle:
									UpperCamelCase__   =    merges_handle.read().split("\n"       )[1:-1]
						UpperCamelCase__   =    [tuple(merge.split()       ) for merge in bpe_merges]
						UpperCamelCase__   =    dict(zip(snake_case ,  range(len(snake_case       )       )       )       )
						UpperCamelCase__   =    {}
						UpperCamelCase__   =    add_prefix_space

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




			@property
			def 			snake_case__			(							self       ):

						'''simple docstring'''


						return len(self.encoder       )




			def 			snake_case__			(							self       ):

						'''simple docstring'''


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




			def 			snake_case__			(							self ,  snake_case       ):

						'''simple docstring'''


						if token in self.cache:
									return self.cache[token]
						UpperCamelCase__   =    tuple(snake_case       )
						UpperCamelCase__   =    get_pairs(snake_case       )

						if not pairs:
									return token

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

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




			def 			snake_case__			(							self ,  snake_case ,  snake_case=None       ):

						'''simple docstring'''


						if self.add_bos_token:
									UpperCamelCase__   =    [self.bos_token_id]
						else:
									UpperCamelCase__   =    []

						UpperCamelCase__   =    bos_token_ids + token_ids_a

						if token_ids_a is None:
									return output

						return output + bos_token_ids + token_ids_a




			def 			snake_case__			(							self ,  snake_case       ):

						'''simple docstring'''


						UpperCamelCase__   =    []
						for token in re.findall(self.pat ,  snake_case       ):
									UpperCamelCase__   =    "".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(snake_case       ).split(" "       )       )
						return bpe_tokens




			def 			snake_case__			(							self ,  snake_case       ):

						'''simple docstring'''


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




			def 			snake_case__			(							self ,  snake_case       ):

						'''simple docstring'''


						return self.decoder.get(snake_case       )




			def 			snake_case__			(							self ,  snake_case       ):

						'''simple docstring'''


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




			def 			snake_case__			(							self ,  snake_case ,  snake_case = None       ):

						'''simple docstring'''


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

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

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

						return vocab_file, merge_file




			def 			snake_case__			(							self ,  snake_case ,  snake_case=False ,  **snake_case       ):

						'''simple docstring'''


						UpperCamelCase__   =    kwargs.pop("add_prefix_space" ,  self.add_prefix_space       )
						if is_split_into_words or add_prefix_space:
									UpperCamelCase__   =    " " + text
						return (text, kwargs)




			def 			snake_case__			(							self ,  snake_case ,  snake_case = False ,  snake_case = None ,  snake_case = None ,  **snake_case ,  ):

						'''simple docstring'''


						UpperCamelCase__   =    super()._decode(
						    token_ids=snake_case ,  skip_special_tokens=snake_case ,  clean_up_tokenization_spaces=snake_case ,  **snake_case ,  )

						if truncate_before_pattern is not None and len(snake_case       ) > 0:
									UpperCamelCase__   =    self.truncate(snake_case ,  snake_case       )

						return decoded_text







			def 			snake_case__			(							self ,  snake_case ,  snake_case       ):

						'''simple docstring'''


						def find_re(snake_case ,  snake_case ,  snake_case       ):
									UpperCamelCase__   =    pattern.search(snake_case ,  snake_case       )
									return m.start() if m else -1

						UpperCamelCase__   =    [re.compile(snake_case ,  re.MULTILINE       ) for pattern in truncate_before_pattern]

						UpperCamelCase__   =    list(re.finditer("^print" ,  snake_case ,  re.MULTILINE       )       )

						if len(snake_case       ) > 1:
									UpperCamelCase__   =    completion[: prints[1].start()]

						UpperCamelCase__   =    list(re.finditer("^def" ,  snake_case ,  re.MULTILINE       )       )

						if len(snake_case       ) > 1:
									UpperCamelCase__   =    completion[: defs[1].start()]

						UpperCamelCase__   =    0

						UpperCamelCase__   =    [
						    pos for pos in [find_re(snake_case ,  snake_case ,  snake_case       ) for terminal in terminals] if pos != -1
						]

						if len(snake_case       ) > 0:
									return completion[: min(snake_case       )]
						else:
									return completion


