code: 
  
  
  
  
  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_03_58
  lowerCAmelCase__   	=				5_03_62
  
  
  
  
  
  
  
  @require_torch
  @require_torchaudio
  @require_sentencepiece
  class   lowerCAmelCase__ (       unittest.TestCase):
  
  
  
  
       '''simple docstring'''
  
  
       def 				_lowerCamelCase				(		self)    ->       Dict:
              _A   : List[str]       					=				"openai/whisper-small.en"
              _A   : Tuple       					=				tempfile.mkdtemp()
  
  
       def 				_lowerCamelCase				(		self				,		**__lowerCamelCase)    ->       str:
              return WhisperTokenizer.from_pretrained(self.checkpoint				,		**__lowerCamelCase)
  
  
       def 				_lowerCamelCase				(		self				,		**__lowerCamelCase)    ->       Union[str, Any]:
              return WhisperFeatureExtractor.from_pretrained(self.checkpoint				,		**__lowerCamelCase)
  
  
       def 				_lowerCamelCase				(		self)    ->       List[Any]:
              shutil.rmtree(self.tmpdirname)
  
  
       def 				_lowerCamelCase				(		self)    ->       str:
              _A   : str       					=				self.get_tokenizer()
              _A   : Dict       					=				self.get_feature_extractor()
  
              _A   : Any       					=				WhisperProcessor(tokenizer=__lowerCamelCase				,		feature_extractor=__lowerCamelCase)
  
              processor.save_pretrained(self.tmpdirname)
              _A   : List[Any]       					=				WhisperProcessor.from_pretrained(self.tmpdirname)
  
              self.assertEqual(processor.tokenizer.get_vocab()				,		tokenizer.get_vocab())
              self.assertIsInstance(processor.tokenizer				,		__lowerCamelCase)
  
              self.assertEqual(processor.feature_extractor.to_json_string()				,		feature_extractor.to_json_string())
              self.assertIsInstance(processor.feature_extractor				,		__lowerCamelCase)
  
  
       def 				_lowerCamelCase				(		self)    ->       Union[str, Any]:
              _A   : Tuple       					=				WhisperProcessor(tokenizer=self.get_tokenizer()				,		feature_extractor=self.get_feature_extractor())
              processor.save_pretrained(self.tmpdirname)
  
              _A   : Union[str, Any]       					=				self.get_tokenizer(bos_token="(BOS)"				,		eos_token="(EOS)")
              _A   : List[str]       					=				self.get_feature_extractor(do_normalize=__lowerCamelCase				,		padding_value=1.0)
  
              _A   : Union[str, Any]       					=				WhisperProcessor.from_pretrained(
                  self.tmpdirname				,		bos_token="(BOS)"				,		eos_token="(EOS)"				,		do_normalize=__lowerCamelCase				,		padding_value=1.0)
  
              self.assertEqual(processor.tokenizer.get_vocab()				,		tokenizer_add_kwargs.get_vocab())
              self.assertIsInstance(processor.tokenizer				,		__lowerCamelCase)
  
              self.assertEqual(processor.feature_extractor.to_json_string()				,		feature_extractor_add_kwargs.to_json_string())
              self.assertIsInstance(processor.feature_extractor				,		__lowerCamelCase)
  
  
       def 				_lowerCamelCase				(		self)    ->       List[str]:
              _A   : List[str]       					=				self.get_feature_extractor()
              _A   : Optional[Any]       					=				self.get_tokenizer()
  
              _A   : int       					=				WhisperProcessor(tokenizer=__lowerCamelCase				,		feature_extractor=__lowerCamelCase)
  
              _A   : Any       					=				floats_list((3, 1_0_0_0))
  
              _A   : Dict       					=				feature_extractor(__lowerCamelCase				,		return_tensors="np")
              _A   : Dict       					=				processor(__lowerCamelCase				,		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 				_lowerCamelCase				(		self)    ->       List[str]:
              _A   : List[str]       					=				self.get_feature_extractor()
              _A   : str       					=				self.get_tokenizer()
  
              _A   : Dict       					=				WhisperProcessor(tokenizer=__lowerCamelCase				,		feature_extractor=__lowerCamelCase)
  
              _A   : Any       					=				"This is a test string"
  
              _A   : str       					=				processor(text=__lowerCamelCase)
  
              _A   : Union[str, Any]       					=				tokenizer(__lowerCamelCase)
  
              for key in encoded_tok.keys():
                     self.assertListEqual(encoded_tok[key]				,		encoded_processor[key])
  
  
       def 				_lowerCamelCase				(		self)    ->       Union[str, Any]:
              _A   : List[str]       					=				self.get_feature_extractor()
              _A   : List[Any]       					=				self.get_tokenizer()
  
              _A   : Optional[Any]       					=				WhisperProcessor(tokenizer=__lowerCamelCase				,		feature_extractor=__lowerCamelCase)
  
              _A   : int       					=				[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]
  
              _A   : Optional[int]       					=				processor.batch_decode(__lowerCamelCase)
              _A   : Optional[int]       					=				tokenizer.batch_decode(__lowerCamelCase)
  
              self.assertListEqual(__lowerCamelCase				,		__lowerCamelCase)
  
  
       def 				_lowerCamelCase				(		self)    ->       List[Any]:
              _A   : Optional[Any]       					=				self.get_feature_extractor()
              _A   : Union[str, Any]       					=				self.get_tokenizer()
  
              _A   : Dict       					=				WhisperProcessor(tokenizer=__lowerCamelCase				,		feature_extractor=__lowerCamelCase)
  
              self.assertListEqual(
                  processor.model_input_names				,		feature_extractor.model_input_names				,		msg="`processor` and `feature_extractor` model input names do not match"				,		)
  
  
       def 				_lowerCamelCase				(		self)    ->       str:
              _A   : Dict       					=				self.get_feature_extractor()
              _A   : Any       					=				self.get_tokenizer()
  
              _A   : Dict       					=				WhisperProcessor(tokenizer=__lowerCamelCase				,		feature_extractor=__lowerCamelCase)
              _A   : Tuple       					=				processor.get_decoder_prompt_ids(task="transcribe"				,		no_timestamps=__lowerCamelCase)
  
              self.assertIsInstance(__lowerCamelCase				,		__lowerCamelCase)
              for ids in forced_decoder_ids:
                     self.assertIsInstance(__lowerCamelCase				,		(list, tuple))
  
              _A   : int       					=				[TRANSCRIBE, NOTIMESTAMPS]
              self.assertListEqual([ids[-1] for ids in forced_decoder_ids]				,		__lowerCamelCase)
  
  
       def 				_lowerCamelCase				(		self)    ->       List[str]:
              _A   : List[str]       					=				WhisperProcessor(tokenizer=self.get_tokenizer()				,		feature_extractor=self.get_feature_extractor())
              _A   : Any       					=				processor.get_prompt_ids("Mr. Quilter")
              _A   : int       					=				processor.tokenizer.decode(__lowerCamelCase)
  
              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(__lowerCamelCase				,		"<|startofprev|> Mr. Quilter")
  
  
       def 				_lowerCamelCase				(		self)    ->       Any:
              _A   : Optional[Any]       					=				WhisperProcessor(tokenizer=self.get_tokenizer()				,		feature_extractor=self.get_feature_extractor())
              _A   : Optional[Any]       					=				processor.get_prompt_ids("")
              _A   : Tuple       					=				processor.tokenizer.decode(__lowerCamelCase)
  
              self.assertListEqual(prompt_ids.tolist()				,		[5_0_3_6_0, 2_2_0])
              self.assertEqual(__lowerCamelCase				,		"<|startofprev|> ")
  
  
       def 				_lowerCamelCase				(		self)    ->       int:
              _A   : Optional[Any]       					=				WhisperProcessor(tokenizer=self.get_tokenizer()				,		feature_extractor=self.get_feature_extractor())
  
              def _test_prompt_error_raised_helper(__lowerCamelCase				,		__lowerCamelCase):
                     with pytest.raises(__lowerCamelCase) as excinfo:
                            processor.get_prompt_ids(__lowerCamelCase)
                     _A   : Any       					=				F"Encountered text in the prompt corresponding to disallowed special token: {special_token}."
                     self.assertEqual(__lowerCamelCase				,		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|>")
  
style_context: 
  
  
  import importlib
  import inspect
  from functools import wraps
  from typing import TYPE_CHECKING, Optional, Union
  
  from .download.streaming_download_manager import (
      xbasename,
      xdirname,
      xet_parse,
      xexists,
      xgetsize,
      xglob,
      xgzip_open,
      xisdir,
      xisfile,
      xjoin,
      xlistdir,
      xnumpy_load,
      xopen,
      xpandas_read_csv,
      xpandas_read_excel,
      xPath,
      xrelpath,
      xsio_loadmat,
      xsplit,
      xsplitext,
      xwalk,
      xxml_dom_minidom_parse,
  )
  from .utils.logging import get_logger
  from .utils.patching import patch_submodule
  from .utils.py_utils import get_imports
  
  
  lowerCAmelCase__   	=				get_logger(__name__)
  
  
  if TYPE_CHECKING:
      from .builder import DatasetBuilder
  def 							_UpperCAmelCase       (UpperCamelCase__  :						Optional[Any]       ,							UpperCamelCase__  :						Optional[Union[str, bool]] = None		):
         _A   : Dict       					=				importlib.import_module(UpperCamelCase__		)
  
         # TODO(QL): always update the module to add subsequent new authentication
         if hasattr(UpperCamelCase__       ,							"_patched_for_streaming"		) and module._patched_for_streaming:
                return
  
         def wrap_auth(UpperCamelCase__  :						Optional[int]		):
                @wraps(UpperCamelCase__		)
                def wrapper(*UpperCamelCase__  :						str       ,							**UpperCamelCase__  :						Optional[int]		):
                       return function(*UpperCamelCase__       ,							use_auth_token=UpperCamelCase__       ,							**UpperCamelCase__		)
  
                _A   : List[Any]       					=				"wrap_auth"
                return wrapper
  
         # open files in a streaming fashion
         patch_submodule(UpperCamelCase__       ,							"open"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.listdir"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.walk"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"glob.glob"       ,							wrap_auth(UpperCamelCase__		)		).start()
         # allow to navigate in remote zip files
         patch_submodule(UpperCamelCase__       ,							"os.path.join"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.dirname"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.basename"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.relpath"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.split"       ,							UpperCamelCase__		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.splitext"       ,							UpperCamelCase__		).start()
         # allow checks on paths
         patch_submodule(UpperCamelCase__       ,							"os.path.exists"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.isdir"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.isfile"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"os.path.getsize"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"pathlib.Path"       ,							UpperCamelCase__		).start()
         # file readers
         patch_submodule(UpperCamelCase__       ,							"gzip.open"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"numpy.load"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"pandas.read_csv"       ,							wrap_auth(UpperCamelCase__		)       ,							attrs=["__version__"]		).start()
         patch_submodule(UpperCamelCase__       ,							"pandas.read_excel"       ,							wrap_auth(UpperCamelCase__		)       ,							attrs=["__version__"]		).start()
         patch_submodule(UpperCamelCase__       ,							"scipy.io.loadmat"       ,							wrap_auth(UpperCamelCase__		)       ,							attrs=["__version__"]		).start()
         patch_submodule(UpperCamelCase__       ,							"xml.etree.ElementTree.parse"       ,							wrap_auth(UpperCamelCase__		)		).start()
         patch_submodule(UpperCamelCase__       ,							"xml.dom.minidom.parse"       ,							wrap_auth(UpperCamelCase__		)		).start()
         _A   : Any       					=				True
  
  
  
  
  
  
  def 							_UpperCAmelCase       (UpperCamelCase__  :						"DatasetBuilder"		):
         extend_module_for_streaming(builder.__module__       ,							use_auth_token=builder.use_auth_token		)
         # if needed, we also have to extend additional internal imports (like wmt14 -> wmt_utils)
         if not builder.__module__.startswith("datasets."		):  # check that it's not a packaged builder like csv
                for imports in get_imports(inspect.getfile(builder.__class__		)		):
                       if imports[0] == "internal":
                              _A   : Optional[Any]       					=				imports[1]
                              _A   : int       					=				".".join(builder.__module__.split("."		)[:-1] + [internal_import_name]		)
                              extend_module_for_streaming(UpperCamelCase__       ,							use_auth_token=builder.use_auth_token		)
  
      # builders can inherit from other builders that might use streaming functionality
      # (for example, ImageFolder and AudioFolder inherit from FolderBuilder which implements examples generation)
      # but these parents builders are not patched automatically as they are not instantiated, so we patch them here
         from .builder import DatasetBuilder
  
         _A   : Tuple       					=				[
             cls.__module__
             for cls in type(UpperCamelCase__		).__mro__[1:]  # make sure it's not the same module we've already patched
             if issubclass(cls       ,							UpperCamelCase__		) and cls.__module__ != DatasetBuilder.__module__
         ]  # check it's not a standard builder from datasets.builder
         for module in parent_builder_modules:
                extend_module_for_streaming(UpperCamelCase__       ,							use_auth_token=builder.use_auth_token		)
  