

import shutil
import tempfile
import unittest

import pytest

from transformers import WhisperTokenizer, is_speech_available
from transformers.testing_utils import require_sentencepiece, require_torch, require_torchaudio

from .test_feature_extraction_whisper import floats_list


if is_speech_available():
      from transformers import WhisperFeatureExtractor, WhisperProcessor


lowerCAmelCase__							=			50358
lowerCAmelCase__							=			50362






@require_torch
@require_torchaudio
@require_sentencepiece
class     __magic_name__     (							unittest.TestCase     ):


 def _UpperCamelCase						(    self		:			List[Any]     )     ->   List[str]:
   UpperCAmelCase          =	"openai/whisper-small.en"
   UpperCAmelCase          =	tempfile.mkdtemp()


 def _UpperCamelCase						(    self		:			Optional[Any]      ,    **lowerCAmelCase__		:			Optional[int]     )     ->   Dict:
   return WhisperTokenizer.from_pretrained(self.checkpoint      ,    **_a     )


 def _UpperCamelCase						(    self		:			str      ,    **lowerCAmelCase__		:			Tuple     )     ->   Any:
   return WhisperFeatureExtractor.from_pretrained(self.checkpoint      ,    **_a     )


 def _UpperCamelCase						(    self		:			Union[str, Any]     )     ->   Optional[Any]:
   shutil.rmtree(self.tmpdirname     )


 def _UpperCamelCase						(    self		:			int     )     ->   Union[str, Any]:
   UpperCAmelCase          =	self.get_tokenizer()
   UpperCAmelCase          =	self.get_feature_extractor()

   UpperCAmelCase          =	WhisperProcessor(tokenizer=_a      ,    feature_extractor=_a     )

   processor.save_pretrained(self.tmpdirname     )
   UpperCAmelCase          =	WhisperProcessor.from_pretrained(self.tmpdirname     )

   self.assertEqual(processor.tokenizer.get_vocab()      ,    tokenizer.get_vocab()     )
   self.assertIsInstance(processor.tokenizer      ,    _a     )

   self.assertEqual(processor.feature_extractor.to_json_string()      ,    feature_extractor.to_json_string()     )
   self.assertIsInstance(processor.feature_extractor      ,    _a     )


 def _UpperCamelCase						(    self		:			Dict     )     ->   Union[str, Any]:
   UpperCAmelCase          =	WhisperProcessor(tokenizer=self.get_tokenizer()      ,    feature_extractor=self.get_feature_extractor()     )
   processor.save_pretrained(self.tmpdirname     )

   UpperCAmelCase          =	self.get_tokenizer(bos_token="(BOS)"      ,    eos_token="(EOS)"     )
   UpperCAmelCase          =	self.get_feature_extractor(do_normalize=_a      ,    padding_value=1.0     )

   UpperCAmelCase          =	WhisperProcessor.from_pretrained(
       self.tmpdirname      ,    bos_token="(BOS)"      ,    eos_token="(EOS)"      ,    do_normalize=_a      ,    padding_value=1.0     )

   self.assertEqual(processor.tokenizer.get_vocab()      ,    tokenizer_add_kwargs.get_vocab()     )
   self.assertIsInstance(processor.tokenizer      ,    _a     )

   self.assertEqual(processor.feature_extractor.to_json_string()      ,    feature_extractor_add_kwargs.to_json_string()     )
   self.assertIsInstance(processor.feature_extractor      ,    _a     )


 def _UpperCamelCase						(    self		:			Tuple     )     ->   int:
   UpperCAmelCase          =	self.get_feature_extractor()
   UpperCAmelCase          =	self.get_tokenizer()

   UpperCAmelCase          =	WhisperProcessor(tokenizer=_a      ,    feature_extractor=_a     )

   UpperCAmelCase          =	floats_list((3, 1_0_0_0)     )

   UpperCAmelCase          =	feature_extractor(_a      ,    return_tensors="np"     )
   UpperCAmelCase          =	processor(_a      ,    return_tensors="np"     )

   for key in input_feat_extract.keys():
     self.assertAlmostEqual(input_feat_extract[key].sum()      ,    input_processor[key].sum()      ,    delta=1e-2     )


 def _UpperCamelCase						(    self		:			Optional[Any]     )     ->   str:
   UpperCAmelCase          =	self.get_feature_extractor()
   UpperCAmelCase          =	self.get_tokenizer()

   UpperCAmelCase          =	WhisperProcessor(tokenizer=_a      ,    feature_extractor=_a     )

   UpperCAmelCase          =	"This is a test string"

   UpperCAmelCase          =	processor(text=_a     )

   UpperCAmelCase          =	tokenizer(_a     )

   for key in encoded_tok.keys():
     self.assertListEqual(encoded_tok[key]      ,    encoded_processor[key]     )


 def _UpperCamelCase						(    self		:			Any     )     ->   Optional[int]:
   UpperCAmelCase          =	self.get_feature_extractor()
   UpperCAmelCase          =	self.get_tokenizer()

   UpperCAmelCase          =	WhisperProcessor(tokenizer=_a      ,    feature_extractor=_a     )

   UpperCAmelCase          =	[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

   UpperCAmelCase          =	processor.batch_decode(_a     )
   UpperCAmelCase          =	tokenizer.batch_decode(_a     )

   self.assertListEqual(_a      ,    _a     )


 def _UpperCamelCase						(    self		:			List[str]     )     ->   List[Any]:
   UpperCAmelCase          =	self.get_feature_extractor()
   UpperCAmelCase          =	self.get_tokenizer()

   UpperCAmelCase          =	WhisperProcessor(tokenizer=_a      ,    feature_extractor=_a     )

   self.assertListEqual(
       processor.model_input_names      ,    feature_extractor.model_input_names      ,    msg="`processor` and `feature_extractor` model input names do not match"      ,    )


 def _UpperCamelCase						(    self		:			List[Any]     )     ->   str:
   UpperCAmelCase          =	self.get_feature_extractor()
   UpperCAmelCase          =	self.get_tokenizer()

   UpperCAmelCase          =	WhisperProcessor(tokenizer=_a      ,    feature_extractor=_a     )
   UpperCAmelCase          =	processor.get_decoder_prompt_ids(task="transcribe"      ,    no_timestamps=_a     )

   self.assertIsInstance(_a      ,    _a     )
   for ids in forced_decoder_ids:
     self.assertIsInstance(_a      ,    (list, tuple)     )

   UpperCAmelCase          =	[TRANSCRIBE, NOTIMESTAMPS]
   self.assertListEqual([ids[-1] for ids in forced_decoder_ids]      ,    _a     )


 def _UpperCamelCase						(    self		:			str     )     ->   Dict:
   UpperCAmelCase          =	WhisperProcessor(tokenizer=self.get_tokenizer()      ,    feature_extractor=self.get_feature_extractor()     )
   UpperCAmelCase          =	processor.get_prompt_ids("Mr. Quilter"     )
   UpperCAmelCase          =	processor.tokenizer.decode(_a     )

   self.assertListEqual(prompt_ids.tolist()      ,    [5_0_3_6_0, 1_7_7_0, 1_3, 2_2_6_4, 3_4_6, 3_5_3]     )
   self.assertEqual(_a      ,    "<|startofprev|> Mr. Quilter"     )


 def _UpperCamelCase						(    self		:			List[str]     )     ->   List[str]:
   UpperCAmelCase          =	WhisperProcessor(tokenizer=self.get_tokenizer()      ,    feature_extractor=self.get_feature_extractor()     )
   UpperCAmelCase          =	processor.get_prompt_ids(""     )
   UpperCAmelCase          =	processor.tokenizer.decode(_a     )

   self.assertListEqual(prompt_ids.tolist()      ,    [5_0_3_6_0, 2_2_0]     )
   self.assertEqual(_a      ,    "<|startofprev|> "     )


 def _UpperCamelCase						(    self		:			Any     )     ->   Dict:
   UpperCAmelCase          =	WhisperProcessor(tokenizer=self.get_tokenizer()      ,    feature_extractor=self.get_feature_extractor()     )

   def _test_prompt_error_raised_helper(lowerCAmelCase__		:			Union[str, Any]      ,    lowerCAmelCase__		:			List[str]     ):
     with pytest.raises(_a     ) as excinfo:
       processor.get_prompt_ids(_a     )
     UpperCAmelCase          =	f"Encountered text in the prompt corresponding to disallowed special token: {special_token}."
     self.assertEqual(_a      ,    str(excinfo.value     )     )

   _test_prompt_error_raised_helper("<|startofprev|> test"      ,    "<|startofprev|>"     )
   _test_prompt_error_raised_helper("test <|notimestamps|>"      ,    "<|notimestamps|>"     )
   _test_prompt_error_raised_helper("test <|zh|> test <|transcribe|>"      ,    "<|zh|>"     )








import logging
import random

import ray

from transformers import RagConfig, RagRetriever, RagTokenizer
from transformers.models.rag.retrieval_rag import CustomHFIndex


__UpperCAmelCase    	=  logging.getLogger(__name__)


class 						lowerCamelCase__			:


						def __init__(			self    :	Optional[Any]      ):
													a__:							Tuple									=False


						def 	_lowerCamelCase      (			self    :	Dict   ,							_a    :	Any   ,							_a    :	Union[str, Any]   ,							_a    :	List[str]   ,							_a    :	Optional[int]      ):
													if not self.initialized:
																				a__:							Any									=RagRetriever(
																				    _a   ,							question_encoder_tokenizer=_a   ,							generator_tokenizer=_a   ,							index=_a   ,							init_retrieval=_a   ,							)
																				a__:							str									=True


						def 	_lowerCamelCase      (			self    :	List[str]      ):
													self.retriever.index.init_index()


						def 	_lowerCamelCase      (			self    :	Optional[int]      ):
													# delete the old self.retriever object before assigning the new index
													del self.retriever
													a__:							Dict									=False







						def 	_lowerCamelCase      (			self    :	str   ,							_a    :	List[Any]   ,							_a    :	Any      ):
													a__				,      a__:							List[str]									=self.retriever._main_retrieve(_a   ,							_a      )
													a__:							List[str]									=self.retriever.index.get_doc_dicts(_a      )
													return doc_ids, retrieved_doc_embeds, doc_dicts


class 						lowerCamelCase__			(		_a ):


						def __init__(			self    :	Optional[int]   ,							_a    :	List[Any]   ,							_a    :	Any   ,							_a    :	Optional[Any]   ,							_a    :	Union[str, Any]   ,							_a    :	str=None      ):
													if index is not None and index.is_initialized() and len(_a      ) > 0:
																				raise ValueError(
																				    "When using Ray for distributed fine-tuning, "
																				    "you'll need to provide the paths instead, "
																				    "as the dataset and the index are loaded "
																				    "separately. More info in examples/rag/use_own_knowledge_dataset.py "      )

													super().__init__(
													    _a   ,							question_encoder_tokenizer=_a   ,							generator_tokenizer=_a   ,							index=_a   ,							init_retrieval=_a   ,							)

													a__:							Optional[Any]									=retrieval_workers
													a__:							str									=question_encoder_tokenizer
													a__:							Optional[int]									=generator_tokenizer
													if len(self.retrieval_workers      ) > 0:
																				ray.get(
																				    [
																				        worker.create_rag_retriever.remote(_a   ,							_a   ,							_a   ,							_a      )
																				        for worker in self.retrieval_workers
																				    ]      )


						def 	_lowerCamelCase      (			self    :	List[Any]      ):
													logger.info("initializing retrieval"      )

													if len(self.retrieval_workers      ) > 0:
																				ray.get([worker.init_retrieval.remote() for worker in self.retrieval_workers]      )
													else:
																				# Non-distributed training. Load index into this same process.
																				self.index.init_index()


						def 	_lowerCamelCase      (			self    :	List[Any]   ,							_a    :	Union[str, Any]   ,							_a    :	Union[str, Any]      ):
													if len(self.retrieval_workers      ) > 0:
																				# Select a random retrieval actor.
																				a__:							Tuple									=self.retrieval_workers[random.randint(0   ,							len(self.retrieval_workers      ) - 1      )]
																				a__				,      a__				,      a__:							Union[str, Any]									=ray.get(
																				    random_worker.retrieve.remote(_a   ,							_a      )      )
													else:
																				a__				,      a__:							List[Any]									=self._main_retrieve(_a   ,							_a      )
																				a__:							List[str]									=self.index.get_doc_dicts(_a      )
													return retrieved_doc_embeds, doc_ids, doc_dicts


						@classmethod
						def 	_lowerCamelCase      (			cls    :	str   ,							_a    :	Union[str, Any]   ,							_a    :	Any=None   ,							**_a    :	Dict      ):
													return super(_a   ,							cls      ).get_tokenizers(_a   ,							_a   ,							**_a      )


						@classmethod
						def 	_lowerCamelCase      (			cls    :	Dict   ,							_a    :	Optional[int]   ,							_a    :	Tuple   ,							_a    :	int=None   ,							**_a    :	List[str]      ):
													a__:							Union[str, Any]									=kwargs.pop("config"   ,							_a      ) or RagConfig.from_pretrained(_a   ,							**_a      )
													a__:							List[Any]									=RagTokenizer.from_pretrained(_a   ,							config=_a      )
													a__:							str									=rag_tokenizer.question_encoder
													a__:							Any									=rag_tokenizer.generator

													if indexed_dataset is not None:
																				a__:							Dict									="custom"
																				a__:							Dict									=CustomHFIndex(config.retrieval_vector_size   ,							_a      )
													else:
																				a__:							List[str]									=cls._build_index(_a      )

													return cls(
													    _a   ,							question_encoder_tokenizer=_a   ,							generator_tokenizer=_a   ,							retrieval_workers=_a   ,							index=_a   ,							)







						def 	_lowerCamelCase      (			self    :	Union[str, Any]      ):
													logger.info("re-loading the new dataset with embeddings"      )
													# access from the training loop

													ray.get([worker.clear_object.remote() for worker in self.retrieval_workers]      )

													# build the index object again
													a__:							str									=self._build_index(self.config      )

													ray.get(
													    [
													        worker.create_rag_retriever.remote(
													            self.config   ,							self.question_encoder_tokenizer   ,							self.generator_tokenizer   ,							_a      )
													        for worker in self.retrieval_workers
													    ]      )





