



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__     =	5_0358
lowerCamelCase__     =	5_0362






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





							def   UpperCamelCase_   (				self						:						Tuple		):

											'''simple docstring'''
											__a					       =						"""openai/whisper-small.en"""
											__a					       =						tempfile.mkdtemp()





							def   UpperCamelCase_   (				self						:						Union[str, Any] ,			**__lowercase						:						Tuple		):

											'''simple docstring'''
											return WhisperTokenizer.from_pretrained(self.checkpoint ,			**snake_case_		)





							def   UpperCamelCase_   (				self						:						List[str] ,			**__lowercase						:						str		):

											'''simple docstring'''
											return WhisperFeatureExtractor.from_pretrained(self.checkpoint ,			**snake_case_		)





							def   UpperCamelCase_   (				self						:						str		):

											'''simple docstring'''
											shutil.rmtree(self.tmpdirname		)





							def   UpperCamelCase_   (				self						:						Tuple		):

											'''simple docstring'''
											__a					       =						self.get_tokenizer()
											__a					       =						self.get_feature_extractor()

											__a					       =						WhisperProcessor(tokenizer=snake_case_ ,			feature_extractor=snake_case_		)

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

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

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





							def   UpperCamelCase_   (				self						:						int		):

											'''simple docstring'''
											__a					       =						WhisperProcessor(tokenizer=self.get_tokenizer() ,			feature_extractor=self.get_feature_extractor()		)
											processor.save_pretrained(self.tmpdirname		)

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

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

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

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





							def   UpperCamelCase_   (				self						:						Union[str, Any]		):

											'''simple docstring'''
											__a					       =						self.get_feature_extractor()
											__a					       =						self.get_tokenizer()

											__a					       =						WhisperProcessor(tokenizer=snake_case_ ,			feature_extractor=snake_case_		)

											__a					       =						floats_list((3, 1000)		)

											__a					       =						feature_extractor(snake_case_ ,			return_tensors="""np"""		)
											__a					       =						processor(snake_case_ ,			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]		):

											'''simple docstring'''
											__a					       =						self.get_feature_extractor()
											__a					       =						self.get_tokenizer()

											__a					       =						WhisperProcessor(tokenizer=snake_case_ ,			feature_extractor=snake_case_		)

											__a					       =						"""This is a test string"""

											__a					       =						processor(text=snake_case_		)

											__a					       =						tokenizer(snake_case_		)

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





							def   UpperCamelCase_   (				self						:						Union[str, Any]		):

											'''simple docstring'''
											__a					       =						self.get_feature_extractor()
											__a					       =						self.get_tokenizer()

											__a					       =						WhisperProcessor(tokenizer=snake_case_ ,			feature_extractor=snake_case_		)

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

											__a					       =						processor.batch_decode(snake_case_		)
											__a					       =						tokenizer.batch_decode(snake_case_		)

											self.assertListEqual(snake_case_ ,			snake_case_		)





							def   UpperCamelCase_   (				self						:						List[str]		):

											'''simple docstring'''
											__a					       =						self.get_feature_extractor()
											__a					       =						self.get_tokenizer()

											__a					       =						WhisperProcessor(tokenizer=snake_case_ ,			feature_extractor=snake_case_		)

											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						:						Tuple		):

											'''simple docstring'''
											__a					       =						self.get_feature_extractor()
											__a					       =						self.get_tokenizer()

											__a					       =						WhisperProcessor(tokenizer=snake_case_ ,			feature_extractor=snake_case_		)
											__a					       =						processor.get_decoder_prompt_ids(task="""transcribe""" ,			no_timestamps=snake_case_		)

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

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





							def   UpperCamelCase_   (				self						:						Any		):

											'''simple docstring'''
											__a					       =						WhisperProcessor(tokenizer=self.get_tokenizer() ,			feature_extractor=self.get_feature_extractor()		)
											__a					       =						processor.get_prompt_ids("""Mr. Quilter"""		)
											__a					       =						processor.tokenizer.decode(snake_case_		)

											self.assertListEqual(prompt_ids.tolist() ,			[50360, 1770, 13, 2264, 346, 353]		)
											self.assertEqual(snake_case_ ,			"""<|startofprev|> Mr. Quilter"""		)





							def   UpperCamelCase_   (				self						:						List[Any]		):

											'''simple docstring'''
											__a					       =						WhisperProcessor(tokenizer=self.get_tokenizer() ,			feature_extractor=self.get_feature_extractor()		)
											__a					       =						processor.get_prompt_ids(""""""		)
											__a					       =						processor.tokenizer.decode(snake_case_		)

											self.assertListEqual(prompt_ids.tolist() ,			[50360, 220]		)
											self.assertEqual(snake_case_ ,			"""<|startofprev|> """		)







							def   UpperCamelCase_   (				self						:						Optional[int]		):

											'''simple docstring'''
											__a					       =						WhisperProcessor(tokenizer=self.get_tokenizer() ,			feature_extractor=self.get_feature_extractor()		)

											def _test_prompt_error_raised_helper(__lowercase						:						List[str] ,			__lowercase						:						int		):
															with pytest.raises(snake_case_		) as excinfo:
																			processor.get_prompt_ids(snake_case_		)
															__a					       =						F"Encountered text in the prompt corresponding to disallowed special token: {special_token}."
															self.assertEqual(snake_case_ ,			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|>"""		)





#
# Sort large text files in a minimum amount of memory
#
import argparse
import os






class 				SCREAMING_SNAKE_CASE     :
							__lowerCamelCase		:		Tuple      ='block_{0}.dat'





							def __init__(				self						:						int ,			__lowercase						:						Dict		):

											'''simple docstring'''
											__a					       =						filename
											__a					       =						[]





							def   UpperCamelCase_   (				self						:						int ,			__lowercase						:						Dict ,			__lowercase						:						Optional[Any]		):

											'''simple docstring'''
											__a					       =						self.BLOCK_FILENAME_FORMAT.format(__lowercase		)
											with open(__lowercase ,			"""w"""		) as file:
															file.write(__lowercase		)
											self.block_filenames.append(__lowercase		)





							def   UpperCamelCase_   (				self						:						List[str]		):

											'''simple docstring'''
											return self.block_filenames





							def   UpperCamelCase_   (				self						:						Any ,			__lowercase						:						str ,			__lowercase						:						Tuple=None		):

											'''simple docstring'''
											__a					       =						0
											with open(self.filename		) as file:
															while True:
																			__a					       =						file.readlines(__lowercase		)

																			if lines == []:
																							break

																			if sort_key is None:
																							lines.sort()
																			else:
																							lines.sort(key=__lowercase		)

																			self.write_block("""""".join(__lowercase		) ,			__lowercase		)
																			i += 1







							def   UpperCamelCase_   (				self						:						List[Any]		):

											'''simple docstring'''
											map(os.remove ,			self.block_filenames		)






class 				SCREAMING_SNAKE_CASE     :





							def   UpperCamelCase_   (				self						:						Dict ,			__lowercase						:						Optional[Any]		):

											'''simple docstring'''
											__a					       =						-1
											__a					       =						None

											for i in range(len(__lowercase		)		):
															if min_str is None or choices[i] < min_str:
																			__a					       =						i

											return min_index






class 				SCREAMING_SNAKE_CASE     :





							def __init__(				self						:						Optional[int] ,			__lowercase						:						Optional[Any]		):

											'''simple docstring'''
											__a					       =						files
											__a					       =						set()
											__a					       =						len(__lowercase		)
											__a					       =						{i: None for i in range(self.num_buffers		)}





							def   UpperCamelCase_   (				self						:						int		):

											'''simple docstring'''
											return {
											    i: self.buffers[i] for i in range(self.num_buffers		) if i not in self.empty
											}





							def   UpperCamelCase_   (				self						:						List[str]		):

											'''simple docstring'''
											for i in range(self.num_buffers		):
															if self.buffers[i] is None and i not in self.empty:
																			__a					       =						self.files[i].readline()

																			if self.buffers[i] == "":
																							self.empty.add(__lowercase		)
																							self.files[i].close()

											if len(self.empty		) == self.num_buffers:
															return False

											return True







							def   UpperCamelCase_   (				self						:						List[str] ,			__lowercase						:						List[Any]		):

											'''simple docstring'''
											__a					       =						self.buffers[index]
											__a					       =						None

											return value






class 				SCREAMING_SNAKE_CASE     :





							def __init__(				self						:						Optional[int] ,			__lowercase						:						Any		):

											'''simple docstring'''
											__a					       =						merge_strategy





							def   UpperCamelCase_   (				self						:						str ,			__lowercase						:						Union[str, Any] ,			__lowercase						:						List[str] ,			__lowercase						:						Dict		):

											'''simple docstring'''
											__a					       =						FilesArray(self.get_file_handles(__lowercase ,			__lowercase		)		)
											with open(__lowercase ,			"""w""" ,			__lowercase		) as outfile:
															while buffers.refresh():
																			__a					       =						self.merge_strategy.select(buffers.get_dict()		)
																			outfile.write(buffers.unshift(__lowercase		)		)







							def   UpperCamelCase_   (				self						:						List[Any] ,			__lowercase						:						str ,			__lowercase						:						Dict		):

											'''simple docstring'''
											__a					       =						{}

											for i in range(len(__lowercase		)		):
															__a					       =						open(filenames[i] ,			"""r""" ,			__lowercase		)  # noqa: UP015

											return files






class 				SCREAMING_SNAKE_CASE     :





							def __init__(				self						:						str ,			__lowercase						:						Tuple		):

											'''simple docstring'''
											__a					       =						block_size





							def   UpperCamelCase_   (				self						:						Tuple ,			__lowercase						:						str ,			__lowercase						:						Union[str, Any]=None		):

											'''simple docstring'''
											__a					       =						self.get_number_blocks(__lowercase ,			self.block_size		)
											__a					       =						FileSplitter(__lowercase		)
											splitter.split(self.block_size ,			__lowercase		)

											__a					       =						FileMerger(NWayMerge()		)
											__a					       =						self.block_size / (num_blocks + 1)
											merger.merge(splitter.get_block_filenames() ,			filename + """.out""" ,			__lowercase		)

											splitter.cleanup()







							def   UpperCamelCase_   (				self						:						str ,			__lowercase						:						Optional[Any] ,			__lowercase						:						Dict		):

											'''simple docstring'''
											return (os.stat(__lowercase		).st_size / block_size) + 1



def 					lowerCAmelCase__  (							_SCREAMING_SNAKE_CASE    :     Tuple      ):
				"""simple docstring"""







				if string[-1].lower() == "k":
								return int(string[:-1]      ) * 1024
				elif string[-1].lower() == "m":
								return int(string[:-1]      ) * 1024 * 1024
				elif string[-1].lower() == "g":
								return int(string[:-1]      ) * 1024 * 1024 * 1024
				else:
								return int(_SCREAMING_SNAKE_CASE      )

def 					lowerCAmelCase__  (							):
				"""simple docstring"""







				__a					       =						argparse.ArgumentParser()
				parser.add_argument(
				    """-m"""			,       """--mem"""			,       help="""amount of memory to use for sorting"""			,       default="""100M"""      )
				parser.add_argument(
				    """filename"""			,       metavar="""<filename>"""			,       nargs=1			,       help="""name of file to sort"""      )
				__a					       =						parser.parse_args()

				__a					       =						ExternalSort(parse_memory(args.mem      )      )
				sorter.sort(args.filename[0]      )


if __name__ == "__main__":
						main()
