



'''simple docstring'''


import itertools
import random
import unittest

import numpy as np

from transformers import BatchFeature, SpeechTaFeatureExtractor
from transformers.testing_utils import require_torch
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
						import torch


lowercase_      :				Optional[int]   					=    random.Random()







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      int ,  lowercase_    :      int=1.0 ,  lowercase_    :      List[Any]=None ,  lowercase_    :      Optional[Any]=None						):
						if rng is None:
												lowercase        =							global_rng

						lowercase        =							[]
						for batch_idx in range(shape[0]						):
												values.append([]						)
												for _ in range(shape[1]						):
																		values[-1].append(rng.random() * scale						)

						return values




@require_torch
class      __UpperCamelCase       (unittest.TestCase			):







	def __init__(    self ,				_lowerCAmelCase ,				_lowerCAmelCase=7 ,				_lowerCAmelCase=400 ,				_lowerCAmelCase=2000 ,				_lowerCAmelCase=1 ,				_lowerCAmelCase=0.0 ,				_lowerCAmelCase=1_6000 ,				_lowerCAmelCase=True ,				_lowerCAmelCase=80 ,				_lowerCAmelCase=16 ,				_lowerCAmelCase=64 ,				_lowerCAmelCase="hann_window" ,				_lowerCAmelCase=80 ,				_lowerCAmelCase=7600 ,				_lowerCAmelCase=1E-10 ,				_lowerCAmelCase=True ,				)    ->    Tuple:
							'''simple docstring'''






							lowercase        =							parent
							lowercase        =							batch_size
							lowercase        =							min_seq_length
							lowercase        =							max_seq_length
							lowercase        =							(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
							lowercase        =							feature_size
							lowercase        =							padding_value
							lowercase        =							sampling_rate
							lowercase        =							do_normalize
							lowercase        =							num_mel_bins
							lowercase        =							hop_length
							lowercase        =							win_length
							lowercase        =							win_function
							lowercase        =							fmin
							lowercase        =							fmax
							lowercase        =							mel_floor
							lowercase        =							return_attention_mask







	def   _a   (    self						)    ->    List[Any]:
							'''simple docstring'''






							return {
							    "feature_size": self.feature_size,
							    "padding_value": self.padding_value,
							    "sampling_rate": self.sampling_rate,
							    "do_normalize": self.do_normalize,
							    "num_mel_bins": self.num_mel_bins,
							    "hop_length": self.hop_length,
							    "win_length": self.win_length,
							    "win_function": self.win_function,
							    "fmin": self.fmin,
							    "fmax": self.fmax,
							    "mel_floor": self.mel_floor,
							    "return_attention_mask": self.return_attention_mask,
							}







	def   _a   (    self ,				_lowerCAmelCase=False ,				_lowerCAmelCase=False						)    ->    Union[str, Any]:
							'''simple docstring'''






							def _flatten(_lowerCAmelCase						):
													return list(itertools.chain(*_lowerCAmelCase						)						)

							if equal_length:
													lowercase        =							floats_list((self.batch_size, self.max_seq_length)						)
							else:
													# make sure that inputs increase in size
													lowercase        =							[
													    _flatten(floats_list((x, self.feature_size)						)						)
													    for x in range(self.min_seq_length ,				self.max_seq_length ,				self.seq_length_diff						)
													]

							if numpify:
													lowercase        =							[np.asarray(_lowerCAmelCase						) for x in speech_inputs]

							return speech_inputs







	def   _a   (    self ,				_lowerCAmelCase=False ,				_lowerCAmelCase=False						)    ->    Any:
							'''simple docstring'''






							if equal_length:
													lowercase        =							[floats_list((self.max_seq_length, self.num_mel_bins)						) for _ in range(self.batch_size						)]
							else:
													# make sure that inputs increase in size
													lowercase        =							[
													    floats_list((x, self.num_mel_bins)						)
													    for x in range(self.min_seq_length ,				self.max_seq_length ,				self.seq_length_diff						)
													]

							if numpify:
													lowercase        =							[np.asarray(_lowerCAmelCase						) for x in speech_inputs]

							return speech_inputs




@require_torch
class      __UpperCamelCase       (_UpperCAmelCase			,   unittest.TestCase			):
	__A			  =					SpeechTaFeatureExtractor







	def   _a   (    self						)    ->    Tuple:
							'''simple docstring'''






							lowercase        =							SpeechTaFeatureExtractionTester(self						)







	def   _a   (    self ,				_lowerCAmelCase						)    ->    Optional[Any]:
							'''simple docstring'''






							self.assertTrue(np.all(np.mean(_lowerCAmelCase ,				axis=0						) < 1E-3						)						)
							self.assertTrue(np.all(np.abs(np.var(_lowerCAmelCase ,				axis=0						) - 1						) < 1E-3						)						)







	def   _a   (    self						)    ->    Tuple:
							'''simple docstring'''






							lowercase        =							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
							# create three inputs of length 800, 1000, and 1200
							lowercase        =							[floats_list((1, x)						)[0] for x in range(800 ,				1400 ,				200						)]
							lowercase        =							[np.asarray(_lowerCAmelCase						) for speech_input in speech_inputs]

							# Test not batched input
							lowercase        =							feat_extract(speech_inputs[0] ,				return_tensors="""np"""						).input_values
							lowercase        =							feat_extract(np_speech_inputs[0] ,				return_tensors="""np"""						).input_values
							self.assertTrue(np.allclose(_lowerCAmelCase ,				_lowerCAmelCase ,				atol=1E-3						)						)

							# Test batched
							lowercase        =							feat_extract(_lowerCAmelCase ,				return_tensors="""np"""						).input_values
							lowercase        =							feat_extract(_lowerCAmelCase ,				return_tensors="""np"""						).input_values
							for enc_seq_a, enc_seq_a in zip(_lowerCAmelCase ,				_lowerCAmelCase						):
													self.assertTrue(np.allclose(_lowerCAmelCase ,				_lowerCAmelCase ,				atol=1E-3						)						)







	def   _a   (    self						)    ->    List[str]:
							'''simple docstring'''






							lowercase        =							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
							lowercase        =							[floats_list((1, x)						)[0] for x in range(800 ,				1400 ,				200						)]

							lowercase        =							["""longest""", """max_length""", """do_not_pad"""]
							lowercase        =							[None, 1600, None]
							for max_length, padding in zip(_lowerCAmelCase ,				_lowerCAmelCase						):
													lowercase        =							feat_extract(_lowerCAmelCase ,				padding=_lowerCAmelCase ,				max_length=_lowerCAmelCase ,				return_tensors="""np"""						)
													lowercase        =							processed.input_values

													self._check_zero_mean_unit_variance(input_values[0][:800]						)
													self.assertTrue(input_values[0][800:].sum() < 1E-6						)
													self._check_zero_mean_unit_variance(input_values[1][:1000]						)
													self.assertTrue(input_values[0][1000:].sum() < 1E-6						)
													self._check_zero_mean_unit_variance(input_values[2][:1200]						)







	def   _a   (    self						)    ->    List[str]:
							'''simple docstring'''






							lowercase        =							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
							lowercase        =							range(800 ,				1400 ,				200						)
							lowercase        =							[floats_list((1, x)						)[0] for x in lengths]

							lowercase        =							["""longest""", """max_length""", """do_not_pad"""]
							lowercase        =							[None, 1600, None]

							for max_length, padding in zip(_lowerCAmelCase ,				_lowerCAmelCase						):
													lowercase        =							feat_extract(_lowerCAmelCase ,				max_length=_lowerCAmelCase ,				padding=_lowerCAmelCase						)
													lowercase        =							processed.input_values

													self._check_zero_mean_unit_variance(input_values[0][:800]						)
													self._check_zero_mean_unit_variance(input_values[1][:1000]						)
													self._check_zero_mean_unit_variance(input_values[2][:1200]						)







	def   _a   (    self						)    ->    Any:
							'''simple docstring'''






							lowercase        =							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
							lowercase        =							[floats_list((1, x)						)[0] for x in range(800 ,				1400 ,				200						)]
							lowercase        =							feat_extract(
							    _lowerCAmelCase ,				truncation=_lowerCAmelCase ,				max_length=1000 ,				padding="""max_length""" ,				return_tensors="""np"""						)
							lowercase        =							processed.input_values

							self._check_zero_mean_unit_variance(input_values[0, :800]						)
							self._check_zero_mean_unit_variance(input_values[1]						)
							self._check_zero_mean_unit_variance(input_values[2]						)







	def   _a   (    self						)    ->    Optional[int]:
							'''simple docstring'''






							lowercase        =							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
							lowercase        =							[floats_list((1, x)						)[0] for x in range(800 ,				1400 ,				200						)]
							lowercase        =							feat_extract(
							    _lowerCAmelCase ,				truncation=_lowerCAmelCase ,				max_length=1000 ,				padding="""longest""" ,				return_tensors="""np"""						)
							lowercase        =							processed.input_values

							self._check_zero_mean_unit_variance(input_values[0, :800]						)
							self._check_zero_mean_unit_variance(input_values[1, :1000]						)
							self._check_zero_mean_unit_variance(input_values[2]						)

							# make sure that if max_length < longest -> then pad to max_length
							self.assertTrue(input_values.shape == (3, 1000)						)

							lowercase        =							[floats_list((1, x)						)[0] for x in range(800 ,				1400 ,				200						)]
							lowercase        =							feat_extract(
							    _lowerCAmelCase ,				truncation=_lowerCAmelCase ,				max_length=2000 ,				padding="""longest""" ,				return_tensors="""np"""						)
							lowercase        =							processed.input_values

							self._check_zero_mean_unit_variance(input_values[0, :800]						)
							self._check_zero_mean_unit_variance(input_values[1, :1000]						)
							self._check_zero_mean_unit_variance(input_values[2]						)

							# make sure that if max_length > longest -> then pad to longest
							self.assertTrue(input_values.shape == (3, 1200)						)







	def   _a   (    self						)    ->    Any:
							'''simple docstring'''






							lowercase        =							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
							lowercase        =							np.random.rand(100						).astype(np.floataa						)
							lowercase        =							np_speech_inputs.tolist()

							for inputs in [py_speech_inputs, np_speech_inputs]:
													lowercase        =							feature_extractor.pad([{"""input_values""": inputs}] ,				return_tensors="""np"""						)
													self.assertTrue(np_processed.input_values.dtype == np.floataa						)
													lowercase        =							feature_extractor.pad([{"""input_values""": inputs}] ,				return_tensors="""pt"""						)
													self.assertTrue(pt_processed.input_values.dtype == torch.floataa						)







	def   _a   (    self						)    ->    List[Any]:
							'''simple docstring'''






							lowercase        =							self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
							# create three inputs of length 800, 1000, and 1200
							lowercase        =							[floats_list((1, x)						)[0] for x in range(800 ,				1400 ,				200						)]
							lowercase        =							[np.asarray(_lowerCAmelCase						) for speech_input in speech_inputs]

							# Test feature size
							lowercase        =							feature_extractor(audio_target=_lowerCAmelCase ,				padding=_lowerCAmelCase ,				return_tensors="""np"""						).input_values
							self.assertTrue(input_values.ndim == 3						)
							self.assertTrue(input_values.shape[-1] == feature_extractor.num_mel_bins						)

							# Test not batched input
							lowercase        =							feature_extractor(speech_inputs[0] ,				return_tensors="""np"""						).input_values
							lowercase        =							feature_extractor(np_speech_inputs[0] ,				return_tensors="""np"""						).input_values
							self.assertTrue(np.allclose(_lowerCAmelCase ,				_lowerCAmelCase ,				atol=1E-3						)						)

							# Test batched
							lowercase        =							feature_extractor(_lowerCAmelCase ,				return_tensors="""np"""						).input_values
							lowercase        =							feature_extractor(_lowerCAmelCase ,				return_tensors="""np"""						).input_values
							for enc_seq_a, enc_seq_a in zip(_lowerCAmelCase ,				_lowerCAmelCase						):
													self.assertTrue(np.allclose(_lowerCAmelCase ,				_lowerCAmelCase ,				atol=1E-3						)						)

							# Test 2-D numpy arrays are batched.
							lowercase        =							[floats_list((1, x)						)[0] for x in (800, 800, 800)]
							lowercase        =							np.asarray(_lowerCAmelCase						)
							lowercase        =							feature_extractor(_lowerCAmelCase ,				return_tensors="""np"""						).input_values
							lowercase        =							feature_extractor(_lowerCAmelCase ,				return_tensors="""np"""						).input_values
							for enc_seq_a, enc_seq_a in zip(_lowerCAmelCase ,				_lowerCAmelCase						):
													self.assertTrue(np.allclose(_lowerCAmelCase ,				_lowerCAmelCase ,				atol=1E-3						)						)







	def   _a   (    self						)    ->    List[str]:
							'''simple docstring'''






							lowercase        =							self.feat_extract_tester.prepare_inputs_for_target()
							lowercase        =							self.feature_extraction_class(**self.feat_extract_dict						)
							lowercase        =							feat_extract.model_input_names[0]

							lowercase        =							BatchFeature({input_name: speech_inputs}						)

							self.assertTrue(all(len(_lowerCAmelCase						) == len(_lowerCAmelCase						) for x, y in zip(_lowerCAmelCase ,				processed_features[input_name]						)						)						)

							lowercase        =							self.feat_extract_tester.prepare_inputs_for_target(equal_length=_lowerCAmelCase						)
							lowercase        =							BatchFeature({input_name: speech_inputs} ,				tensor_type="""np"""						)

							lowercase        =							processed_features[input_name]

							if len(batch_features_input.shape						) < 3:
													lowercase        =							batch_features_input[:, :, None]

							self.assertTrue(
							    batch_features_input.shape
							    == (self.feat_extract_tester.batch_size, len(speech_inputs[0]						), self.feat_extract_tester.num_mel_bins)						)







	@require_torch
	def   _a   (    self						)    ->    Any:
							'''simple docstring'''






							lowercase        =							self.feat_extract_tester.prepare_inputs_for_target(equal_length=_lowerCAmelCase						)
							lowercase        =							self.feature_extraction_class(**self.feat_extract_dict						)
							lowercase        =							feat_extract.model_input_names[0]

							lowercase        =							BatchFeature({input_name: speech_inputs} ,				tensor_type="""pt"""						)

							lowercase        =							processed_features[input_name]

							if len(batch_features_input.shape						) < 3:
													lowercase        =							batch_features_input[:, :, None]

							self.assertTrue(
							    batch_features_input.shape
							    == (self.feat_extract_tester.batch_size, len(speech_inputs[0]						), self.feat_extract_tester.num_mel_bins)						)







	@require_torch
	def   _a   (    self						)    ->    str:
							'''simple docstring'''






							lowercase        =							self.feature_extraction_class(**self.feat_extract_dict						)
							lowercase        =							self.feat_extract_tester.prepare_inputs_for_target()
							lowercase        =							feat_extract.model_input_names[0]

							lowercase        =							BatchFeature({input_name: speech_inputs}						)

							lowercase        =							feat_extract.num_mel_bins  # hack!

							lowercase        =							feat_extract.pad(_lowerCAmelCase ,				padding="""longest""" ,				return_tensors="""np"""						)[input_name]
							lowercase        =							feat_extract.pad(_lowerCAmelCase ,				padding="""longest""" ,				return_tensors="""pt"""						)[input_name]

							self.assertTrue(abs(input_np.astype(np.floataa						).sum() - input_pt.numpy().astype(np.floataa						).sum()						) < 1E-2						)







	def   _a   (    self						)    ->    int:
							'''simple docstring'''






							lowercase        =							self.feat_extract_dict
							lowercase        =							True
							lowercase        =							self.feature_extraction_class(**_lowerCAmelCase						)
							lowercase        =							self.feat_extract_tester.prepare_inputs_for_target()
							lowercase        =							[len(_lowerCAmelCase						) for x in speech_inputs]
							lowercase        =							feat_extract.model_input_names[0]

							lowercase        =							BatchFeature({input_name: speech_inputs}						)

							lowercase        =							feat_extract.num_mel_bins  # hack!

							lowercase        =							feat_extract.pad(_lowerCAmelCase ,				padding="""longest""" ,				return_tensors="""np"""						)
							self.assertIn("""attention_mask""" ,				_lowerCAmelCase						)
							self.assertListEqual(list(processed.attention_mask.shape						) ,				list(processed[input_name].shape[:2]						)						)
							self.assertListEqual(processed.attention_mask.sum(-1						).tolist() ,				_lowerCAmelCase						)







	def   _a   (    self						)    ->    Dict:
							'''simple docstring'''






							lowercase        =							self.feat_extract_dict
							lowercase        =							True
							lowercase        =							self.feature_extraction_class(**_lowerCAmelCase						)
							lowercase        =							self.feat_extract_tester.prepare_inputs_for_target()
							lowercase        =							[len(_lowerCAmelCase						) for x in speech_inputs]
							lowercase        =							feat_extract.model_input_names[0]

							lowercase        =							BatchFeature({input_name: speech_inputs}						)
							lowercase        =							min(_lowerCAmelCase						)

							lowercase        =							feat_extract.num_mel_bins  # hack!

							lowercase        =							feat_extract.pad(
							    _lowerCAmelCase ,				padding="""max_length""" ,				max_length=_lowerCAmelCase ,				truncation=_lowerCAmelCase ,				return_tensors="""np"""						)
							self.assertIn("""attention_mask""" ,				_lowerCAmelCase						)
							self.assertListEqual(
							    list(processed_pad.attention_mask.shape						) ,				[processed_pad[input_name].shape[0], max_length]						)
							self.assertListEqual(
							    processed_pad.attention_mask[:, :max_length].sum(-1						).tolist() ,				[max_length for x in speech_inputs]						)







	def   _a   (    self ,				_lowerCAmelCase						)    ->    Optional[Any]:
							'''simple docstring'''






							from datasets import load_dataset

							lowercase        =							load_dataset("""hf-internal-testing/librispeech_asr_dummy""" ,				"""clean""" ,				split="""validation"""						)
							# automatic decoding with librispeech
							lowercase        =							ds.sort("""id"""						).select(range(_lowerCAmelCase						)						)[:num_samples]["""audio"""]

							return [x["array"] for x in speech_samples]







	def   _a   (    self						)    ->    Dict:
							'''simple docstring'''






							lowercase        =							torch.tensor(
							    [2.38_04E-03, 2.07_52E-03, 1.98_36E-03, 2.10_57E-03, 1.61_74E-03,
							     3.05_18E-04, 9.15_53E-05, 3.35_69E-04, 9.76_56E-04, 1.83_11E-03,
							     2.01_42E-03, 2.10_57E-03, 1.73_95E-03, 4.57_76E-04, -3.96_73E-04,
							     4.57_76E-04, 1.00_71E-03, 9.15_53E-05, 4.88_28E-04, 1.15_97E-03,
							     7.32_42E-04, 9.46_04E-04, 1.80_05E-03, 1.83_11E-03, 8.85_01E-04,
							     4.27_25E-04, 4.88_28E-04, 7.32_42E-04, 1.09_86E-03, 2.10_57E-03]						)
							# fmt: on

							lowercase        =							self._load_datasamples(1						)
							lowercase        =							SpeechTaFeatureExtractor()
							lowercase        =							feature_extractor(_lowerCAmelCase ,				return_tensors="""pt"""						).input_values
							self.assertEquals(input_values.shape ,				(1, 9_3680)						)
							self.assertTrue(torch.allclose(input_values[0, :30] ,				_lowerCAmelCase ,				atol=1E-6						)						)







	def   _a   (    self						)    ->    Optional[int]:
							'''simple docstring'''






							lowercase        =							torch.tensor(
							    [-2.6870, -3.0104, -3.1356, -3.5352, -3.0044, -3.0353, -3.4719, -3.6777,
							     -3.1520, -2.9435, -2.6553, -2.8795, -2.9944, -2.5921, -3.0279, -3.0386,
							     -3.0864, -3.1291, -3.2353, -2.7444, -2.6831, -2.7287, -3.1761, -3.1571,
							     -3.2726, -3.0582, -3.1007, -3.4533, -3.4695, -3.0998]						)
							# fmt: on

							lowercase        =							self._load_datasamples(1						)
							lowercase        =							SpeechTaFeatureExtractor()
							lowercase        =							feature_extractor(audio_target=_lowerCAmelCase ,				return_tensors="""pt"""						).input_values
							self.assertEquals(input_values.shape ,				(1, 366, 80)						)
							self.assertTrue(torch.allclose(input_values[0, 0, :30] ,				_lowerCAmelCase ,				atol=1E-4						)						)







'''simple docstring'''


import json
import os
import tempfile

import transformers

import datasets
from utils import generate_example_dataset, get_duration


lowercase_      :				int   					=    50_0000

lowercase_					,       lowercase_      :				Union[str, Any]   					=    os.path.split(__file__)
lowercase_      :				Optional[Any]   					=    os.path.join(RESULTS_BASEPATH, '''results''', RESULTS_FILENAME.replace('''.py''', '''.json'''))







@get_duration
def      SCREAMING_SNAKE_CASE				(				lowercase_    :      datasets.Dataset ,  **lowercase_    :      Dict						):
						lowercase        =							dataset.map(**lowercase_						)







@get_duration
def      SCREAMING_SNAKE_CASE				(				lowercase_    :      datasets.Dataset ,  **lowercase_    :      Optional[int]						):
						lowercase        =							dataset.filter(**lowercase_						)







def      SCREAMING_SNAKE_CASE				(				):
						lowercase        =							{"""num examples""": SPEED_TEST_N_EXAMPLES}
						with tempfile.TemporaryDirectory() as tmp_dir:
												lowercase        =							datasets.Features({"""text""": datasets.Value("""string"""						), """numbers""": datasets.Value("""float32"""						)}						)
												lowercase        =							generate_example_dataset(
												    os.path.join(lowercase_ ,  """dataset.arrow"""						) ,  lowercase_ ,  num_examples=lowercase_						)

												lowercase        =							transformers.AutoTokenizer.from_pretrained("""bert-base-cased""" ,  use_fast=lowercase_						)

												def tokenize(lowercase_    :      Dict						):
																		return tokenizer(examples["""text"""]						)

												lowercase        =							map(lowercase_						)

												lowercase        =							map(lowercase_ ,  batched=lowercase_						)

												lowercase        =							map(lowercase_ ,  function=lambda lowercase_						: None ,  batched=lowercase_						)

												with dataset.formatted_as(type="""numpy"""						):
																		lowercase        =							map(lowercase_ ,  function=lambda lowercase_						: None ,  batched=lowercase_						)

												with dataset.formatted_as(type="""pandas"""						):
																		lowercase        =							map(lowercase_ ,  function=lambda lowercase_						: None ,  batched=lowercase_						)

												with dataset.formatted_as(type="""torch""" ,  columns="""numbers"""						):
																		lowercase        =							map(lowercase_ ,  function=lambda lowercase_						: None ,  batched=lowercase_						)

												with dataset.formatted_as(type="""tensorflow""" ,  columns="""numbers"""						):
																		lowercase        =							map(lowercase_ ,  function=lambda lowercase_						: None ,  batched=lowercase_						)

												lowercase        =							map(lowercase_ ,  function=lowercase_ ,  batched=lowercase_						)

												lowercase        =							filter(lowercase_						)

												# Activate later when tokenizer support batched inputs
												# with dataset.formatted_as(type='numpy'):
												#     times[func.__name__ + " fast-tokenizer batched numpy"] = func(dataset, function=tokenize, batched=True)

						with open(lowercase_ ,  """wb"""						) as f:
												f.write(json.dumps(lowercase_						).encode("""utf-8"""						)						)


if __name__ == "__main__":  # useful to run the profiler
						benchmark_map_filter()


