





'''simple docstring'''






import json
import os
import shutil
import tempfile
import unittest

from transformers import is_speech_available, is_torch_available
from transformers.models.speechta import SpeechTaTokenizer
from transformers.testing_utils import get_tests_dir, require_torch
from transformers.utils import FEATURE_EXTRACTOR_NAME


if is_speech_available() and is_torch_available():
	from transformers import SpeechTaFeatureExtractor, SpeechTaProcessor

	from .test_feature_extraction_speechta import floats_list


__UpperCamelCase      :   List[Any]								=     get_tests_dir('''fixtures/test_sentencepiece_bpe_char.model''')



@require_torch
class  lowerCamelCase__				(						unittest.TestCase						):


	"""simple docstring"""





	def 				_lowerCamelCase				(					self			)       ->				Any:
					_A :				List[str]				=  tempfile.mkdtemp()

					_A :				Optional[Any]				=  SpeechTaTokenizer(UpperCAmelCase__			)
					tokenizer.save_pretrained(self.tmpdirname			)

					_A :				Optional[int]				=  {
					    '''feature_size''': 1,
					    '''padding_value''': 0.0,
					    '''sampling_rate''': 1_6_0_0_0,
					    '''do_normalize''': False,
					    '''num_mel_bins''': 8_0,
					    '''hop_length''': 1_6,
					    '''win_length''': 6_4,
					    '''win_function''': '''hann_window''',
					    '''fmin''': 8_0,
					    '''fmax''': 7_6_0_0,
					    '''mel_floor''': 1e-10,
					    '''reduction_factor''': 2,
					    '''return_attention_mask''': True,
					}

					_A :				Optional[Any]				=  os.path.join(self.tmpdirname   ,     UpperCAmelCase__			)
					with open(self.feature_extraction_file   ,     '''w'''   ,     encoding='''utf-8'''			) as fp:
									fp.write(json.dumps(UpperCAmelCase__			) + '''\n'''			)




	def 				_lowerCamelCase				(					self   ,     **UpperCAmelCase__			)       ->				List[str]:
					return SpeechTaTokenizer.from_pretrained(self.tmpdirname   ,     **UpperCAmelCase__			)




	def 				_lowerCamelCase				(					self   ,     **UpperCAmelCase__			)       ->				str:
					return SpeechTaFeatureExtractor.from_pretrained(self.tmpdirname   ,     **UpperCAmelCase__			)




	def 				_lowerCamelCase				(					self			)       ->				Optional[int]:
					shutil.rmtree(self.tmpdirname			)




	def 				_lowerCamelCase				(					self			)       ->				str:
					_A :				Tuple				=  self.get_tokenizer()
					_A :				Union[str, Any]				=  self.get_feature_extractor()

					_A :				Union[str, Any]				=  SpeechTaProcessor(tokenizer=UpperCAmelCase__   ,     feature_extractor=UpperCAmelCase__			)

					processor.save_pretrained(self.tmpdirname			)
					_A :				List[str]				=  SpeechTaProcessor.from_pretrained(self.tmpdirname			)

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

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




	def 				_lowerCamelCase				(					self			)       ->				Any:
					_A :				int				=  SpeechTaProcessor(tokenizer=self.get_tokenizer()   ,     feature_extractor=self.get_feature_extractor()			)
					processor.save_pretrained(self.tmpdirname			)

					_A :				int				=  self.get_tokenizer(bos_token='''(BOS)'''   ,     eos_token='''(EOS)'''			)
					_A :				int				=  self.get_feature_extractor(do_normalize=UpperCAmelCase__   ,     padding_value=1.0			)

					_A :				str				=  SpeechTaProcessor.from_pretrained(
					    self.tmpdirname   ,     bos_token='''(BOS)'''   ,     eos_token='''(EOS)'''   ,     do_normalize=UpperCAmelCase__   ,     padding_value=1.0			)

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

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




	def 				_lowerCamelCase				(					self			)       ->				Union[str, Any]:
					_A :				Optional[int]				=  self.get_feature_extractor()
					_A :				str				=  self.get_tokenizer()

					_A :				str				=  SpeechTaProcessor(tokenizer=UpperCAmelCase__   ,     feature_extractor=UpperCAmelCase__			)

					_A :				Tuple				=  floats_list((3, 1_0_0_0)			)

					_A :				int				=  feature_extractor(audio=UpperCAmelCase__   ,     return_tensors='''np'''			)
					_A :				str				=  processor(audio=UpperCAmelCase__   ,     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 :				str				=  self.get_feature_extractor()
					_A :				Optional[int]				=  self.get_tokenizer()

					_A :				Optional[Any]				=  SpeechTaProcessor(tokenizer=UpperCAmelCase__   ,     feature_extractor=UpperCAmelCase__			)

					_A :				List[str]				=  floats_list((3, 1_0_0_0)			)

					_A :				Dict				=  feature_extractor(audio_target=UpperCAmelCase__   ,     return_tensors='''np'''			)
					_A :				List[str]				=  processor(audio_target=UpperCAmelCase__   ,     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			)       ->				int:
					_A :				Dict				=  self.get_feature_extractor()
					_A :				Optional[Any]				=  self.get_tokenizer()

					_A :				Optional[int]				=  SpeechTaProcessor(tokenizer=UpperCAmelCase__   ,     feature_extractor=UpperCAmelCase__			)

					_A :				int				=  '''This is a test string'''

					_A :				Union[str, Any]				=  processor(text=UpperCAmelCase__			)
					_A :				List[Any]				=  tokenizer(UpperCAmelCase__			)

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




	def 				_lowerCamelCase				(					self			)       ->				Dict:
					_A :				Union[str, Any]				=  self.get_feature_extractor()
					_A :				str				=  self.get_tokenizer()

					_A :				Any				=  SpeechTaProcessor(tokenizer=UpperCAmelCase__   ,     feature_extractor=UpperCAmelCase__			)

					_A :				Optional[int]				=  '''This is a test string'''

					_A :				str				=  processor(text_target=UpperCAmelCase__			)
					_A :				List[str]				=  tokenizer(UpperCAmelCase__			)

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




	def 				_lowerCamelCase				(					self			)       ->				Tuple:
					_A :				List[str]				=  self.get_feature_extractor()
					_A :				Dict				=  self.get_tokenizer()

					_A :				Tuple				=  SpeechTaProcessor(tokenizer=UpperCAmelCase__   ,     feature_extractor=UpperCAmelCase__			)

					_A :				Dict				=  [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

					_A :				str				=  processor.batch_decode(UpperCAmelCase__			)
					_A :				Dict				=  tokenizer.batch_decode(UpperCAmelCase__			)

					self.assertListEqual(UpperCAmelCase__   ,     UpperCAmelCase__			)




	def 				_lowerCamelCase				(					self			)       ->				str:
					_A :				int				=  self.get_feature_extractor()
					_A :				Tuple				=  self.get_tokenizer()

					_A :				Any				=  SpeechTaProcessor(tokenizer=UpperCAmelCase__   ,     feature_extractor=UpperCAmelCase__			)

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







'''simple docstring'''






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


__UpperCamelCase      :   List[Any]								=     get_logger(__name__)


if TYPE_CHECKING:
	from .builder import DatasetBuilder





def      lowercase  (  lowerCAmelCase :						List[str]   , lowerCAmelCase :						Optional[Union[str, bool]] = None):



				"""simple docstring"""


				_A :				Union[str, Any]				=  importlib.import_module(lowerCAmelCase)

				# TODO(QL): always update the module to add subsequent new authentication
				if hasattr(lowerCAmelCase   , '''_patched_for_streaming''') and module._patched_for_streaming:
								return

				def wrap_auth(lowerCAmelCase :						Union[str, Any]):
								@wraps(lowerCAmelCase)
								def wrapper(*lowerCAmelCase :						Union[str, Any]   , **lowerCAmelCase :						int):
												return function(*lowerCAmelCase   , use_auth_token=lowerCAmelCase   , **lowerCAmelCase)

								_A :				Optional[Any]				=  '''wrap_auth'''
								return wrapper

				# open files in a streaming fashion
				patch_submodule(lowerCAmelCase   , '''open'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.listdir'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.walk'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''glob.glob'''   , wrap_auth(lowerCAmelCase)).start()
				# allow to navigate in remote zip files
				patch_submodule(lowerCAmelCase   , '''os.path.join'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.dirname'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.basename'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.relpath'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.split'''   , lowerCAmelCase).start()
				patch_submodule(lowerCAmelCase   , '''os.path.splitext'''   , lowerCAmelCase).start()
				# allow checks on paths
				patch_submodule(lowerCAmelCase   , '''os.path.exists'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.path.isdir'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.path.isfile'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''os.path.getsize'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''pathlib.Path'''   , lowerCAmelCase).start()
				# file readers
				patch_submodule(lowerCAmelCase   , '''gzip.open'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''numpy.load'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''pandas.read_csv'''   , wrap_auth(lowerCAmelCase)   , attrs=['''__version__''']).start()
				patch_submodule(lowerCAmelCase   , '''pandas.read_excel'''   , wrap_auth(lowerCAmelCase)   , attrs=['''__version__''']).start()
				patch_submodule(lowerCAmelCase   , '''scipy.io.loadmat'''   , wrap_auth(lowerCAmelCase)   , attrs=['''__version__''']).start()
				patch_submodule(lowerCAmelCase   , '''xml.etree.ElementTree.parse'''   , wrap_auth(lowerCAmelCase)).start()
				patch_submodule(lowerCAmelCase   , '''xml.dom.minidom.parse'''   , wrap_auth(lowerCAmelCase)).start()
				_A :				Optional[int]				=  True





def      lowercase  (  lowerCAmelCase :						"DatasetBuilder"):



				"""simple docstring"""


				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[int]				=  imports[1]
																_A :				str				=  '''.'''.join(builder.__module__.split('''.''')[:-1] + [internal_import_name])
																extend_module_for_streaming(lowerCAmelCase   , 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(lowerCAmelCase).__mro__[1:]  # make sure it's not the same module we've already patched
				    if issubclass(cls   , lowerCAmelCase) 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(lowerCAmelCase   , use_auth_token=builder.use_auth_token)
