



"""simple docstring"""







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


__snake_case	:				Optional[Any]  					=			50_358
__snake_case	:				Tuple  					=			50_362

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





					"""simple docstring"""






					def 						A__  (					self :							int ):
										A__				=   '''openai/whisper-small.en'''
										A__				=   tempfile.mkdtemp()





					def 						A__  (					self :							Optional[int]	,      **_lowerCamelCase :							List[str] ):
										return WhisperTokenizer.from_pretrained(self.checkpoint	,      **__SCREAMING_SNAKE_CASE )





					def 						A__  (					self :							Any	,      **_lowerCamelCase :							List[Any] ):
										return WhisperFeatureExtractor.from_pretrained(self.checkpoint	,      **__SCREAMING_SNAKE_CASE )





					def 						A__  (					self :							List[str] ):
										shutil.rmtree(self.tmpdirname )





					def 						A__  (					self :							Union[str, Any] ):
										A__				=   self.get_tokenizer()
										A__				=   self.get_feature_extractor()

										A__				=   WhisperProcessor(tokenizer=__SCREAMING_SNAKE_CASE	,      feature_extractor=__SCREAMING_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	,      __SCREAMING_SNAKE_CASE )

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





					def 						A__  (					self :							List[str] ):
										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=__SCREAMING_SNAKE_CASE	,      padding_value=1.0 )

										A__				=   WhisperProcessor.from_pretrained(
										    self.tmpdirname	,      bos_token='''(BOS)'''	,      eos_token='''(EOS)'''	,      do_normalize=__SCREAMING_SNAKE_CASE	,      padding_value=1.0 )

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

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





					def 						A__  (					self :							Optional[Any] ):
										A__				=   self.get_feature_extractor()
										A__				=   self.get_tokenizer()

										A__				=   WhisperProcessor(tokenizer=__SCREAMING_SNAKE_CASE	,      feature_extractor=__SCREAMING_SNAKE_CASE )

										A__				=   floats_list((3, 1_0_0_0) )

										A__				=   feature_extractor(__SCREAMING_SNAKE_CASE	,      return_tensors='''np''' )
										A__				=   processor(__SCREAMING_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 						A__  (					self :							Union[str, Any] ):
										A__				=   self.get_feature_extractor()
										A__				=   self.get_tokenizer()

										A__				=   WhisperProcessor(tokenizer=__SCREAMING_SNAKE_CASE	,      feature_extractor=__SCREAMING_SNAKE_CASE )

										A__				=   '''This is a test string'''

										A__				=   processor(text=__SCREAMING_SNAKE_CASE )

										A__				=   tokenizer(__SCREAMING_SNAKE_CASE )

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





					def 						A__  (					self :							str ):
										A__				=   self.get_feature_extractor()
										A__				=   self.get_tokenizer()

										A__				=   WhisperProcessor(tokenizer=__SCREAMING_SNAKE_CASE	,      feature_extractor=__SCREAMING_SNAKE_CASE )

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

										A__				=   processor.batch_decode(__SCREAMING_SNAKE_CASE )
										A__				=   tokenizer.batch_decode(__SCREAMING_SNAKE_CASE )

										self.assertListEqual(__SCREAMING_SNAKE_CASE	,      __SCREAMING_SNAKE_CASE )





					def 						A__  (					self :							Dict ):
										A__				=   self.get_feature_extractor()
										A__				=   self.get_tokenizer()

										A__				=   WhisperProcessor(tokenizer=__SCREAMING_SNAKE_CASE	,      feature_extractor=__SCREAMING_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 						A__  (					self :							Optional[Any] ):
										A__				=   self.get_feature_extractor()
										A__				=   self.get_tokenizer()

										A__				=   WhisperProcessor(tokenizer=__SCREAMING_SNAKE_CASE	,      feature_extractor=__SCREAMING_SNAKE_CASE )
										A__				=   processor.get_decoder_prompt_ids(task='''transcribe'''	,      no_timestamps=__SCREAMING_SNAKE_CASE )

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

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





					def 						A__  (					self :							Dict ):
										A__				=   WhisperProcessor(tokenizer=self.get_tokenizer()	,      feature_extractor=self.get_feature_extractor() )
										A__				=   processor.get_prompt_ids('''Mr. Quilter''' )
										A__				=   processor.tokenizer.decode(__SCREAMING_SNAKE_CASE )

										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(__SCREAMING_SNAKE_CASE	,      '''<|startofprev|> Mr. Quilter''' )





					def 						A__  (					self :							Tuple ):
										A__				=   WhisperProcessor(tokenizer=self.get_tokenizer()	,      feature_extractor=self.get_feature_extractor() )
										A__				=   processor.get_prompt_ids('''''' )
										A__				=   processor.tokenizer.decode(__SCREAMING_SNAKE_CASE )

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





					def 						A__  (					self :							Dict ):
										A__				=   WhisperProcessor(tokenizer=self.get_tokenizer()	,      feature_extractor=self.get_feature_extractor() )

										def _test_prompt_error_raised_helper(_lowerCamelCase :							Tuple	,      _lowerCamelCase :							Union[str, Any] ):
															with pytest.raises(__SCREAMING_SNAKE_CASE ) as excinfo:
																				processor.get_prompt_ids(__SCREAMING_SNAKE_CASE )
															A__				=   F'''Encountered text in the prompt corresponding to disallowed special token: {special_token}.'''
															self.assertEqual(__SCREAMING_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|>''' )








"""simple docstring"""







import argparse
import logging
import sys
from unittest.mock import patch

import run_glue_with_pabee

from transformers.testing_utils import TestCasePlus


logging.basicConfig(level=logging.DEBUG)

__snake_case	:				List[str]  					=			logging.getLogger()



def   a_	(			):
     A__				=   argparse.ArgumentParser()
     parser.add_argument('''-f'''      )
     A__				=   parser.parse_args()
     return args.f

class       UpperCamelCase  (					a				):





     """simple docstring"""






     def 						A__  (					self :							Optional[int] ):
          A__				=   logging.StreamHandler(sys.stdout )
          logger.addHandler(_lowerCamelCase )

          A__				=   self.get_auto_remove_tmp_dir()
          A__				=   F'''
            run_glue_with_pabee.py
            --model_type albert
            --model_name_or_path albert-base-v2
            --data_dir ./tests/fixtures/tests_samples/MRPC/
            --output_dir {tmp_dir}
            --overwrite_output_dir
            --task_name mrpc
            --do_train
            --do_eval
            --per_gpu_train_batch_size=2
            --per_gpu_eval_batch_size=1
            --learning_rate=2e-5
            --max_steps=50
            --warmup_steps=2
            --seed=42
            --max_seq_length=128
            '''.split()

          with patch.object(_lowerCamelCase	,      '''argv'''	,      _lowerCamelCase ):
               A__				=   run_glue_with_pabee.main()
               for value in result.values():
                    self.assertGreaterEqual(_lowerCamelCase	,      0.75 )



