





"""simple docstring"""




import warnings
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


a						: str     	=			logging.get_logger(__name__)


class        __UpperCamelCase				(						a__						):
							lowerCamelCase   :					List[Any]      					=["""input_values""", """attention_mask"""]






							def __init__(   self      ,					lowerCAmelCase__ = 1      ,					lowerCAmelCase__ = 1_6000      ,					lowerCAmelCase__ = 0.0      ,					lowerCAmelCase__ = False      ,					lowerCAmelCase__ = 80      ,					lowerCAmelCase__ = 16      ,					lowerCAmelCase__ = 64      ,					lowerCAmelCase__ = "hann_window"      ,					lowerCAmelCase__ = 1.0      ,					lowerCAmelCase__ = 80      ,					lowerCAmelCase__ = 7600      ,					lowerCAmelCase__ = 1E-10      ,					lowerCAmelCase__ = 2      ,					lowerCAmelCase__ = True      ,					**lowerCAmelCase__      ,					)	-> Tuple:
								super().__init__(feature_size=lowerCAmelCase__      ,					sampling_rate=lowerCAmelCase__      ,					padding_value=lowerCAmelCase__      ,					**lowerCAmelCase__      )
								a      :		Tuple	=				do_normalize
								a      :		Optional[int]	=				return_attention_mask

								a      :		Optional[Any]	=				num_mel_bins
								a      :		Optional[int]	=				hop_length
								a      :		str	=				win_length
								a      :		Optional[int]	=				win_function
								a      :		str	=				frame_signal_scale
								a      :		int	=				fmin
								a      :		Any	=				fmax
								a      :		Dict	=				mel_floor
								a      :		Tuple	=				reduction_factor

								a      :		List[Any]	=				win_length * sampling_rate // 1000
								a      :		Dict	=				hop_length * sampling_rate // 1000
								a      :		Optional[int]	=				optimal_fft_length(self.sample_size      )
								a      :		List[str]	=				(self.n_fft // 2) + 1

								a      :		str	=				window_function(window_length=self.sample_size      ,					name=self.win_function      ,					periodic=lowerCAmelCase__      )

								a      :		Optional[int]	=				mel_filter_bank(
								    num_frequency_bins=self.n_freqs      ,					num_mel_filters=self.num_mel_bins      ,					min_frequency=self.fmin      ,					max_frequency=self.fmax      ,					sampling_rate=self.sampling_rate      ,					norm="slaney"      ,					mel_scale="slaney"      ,					)

								if frame_signal_scale != 1.0:
									warnings.warn(
									    "The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"      ,					lowerCAmelCase__      ,					)
								if reduction_factor != 2.0:
									warnings.warn(
									    "The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"      ,					lowerCAmelCase__      ,					)






							@staticmethod
							# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
							def     __a   (   lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__ = 0.0      )	-> List[np.ndarray]:
								if attention_mask is not None:
									a      :		int	=				np.array(lowerCAmelCase__      ,					np.intaa      )
									a      :		Any	=				[]

									for vector, length in zip(lowerCAmelCase__      ,					attention_mask.sum(-1      )      ):
										a      :		str	=				(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7      )
										if length < normed_slice.shape[0]:
											a      :		Optional[int]	=				padding_value

										normed_input_values.append(lowerCAmelCase__      )
								else:
									a      :		Dict	=				[(x - x.mean()) / np.sqrt(x.var() + 1E-7      ) for x in input_values]

								return normed_input_values






							def     __a   (   self      ,					lowerCAmelCase__      ,					)	-> np.ndarray:
								a      :		Optional[int]	=				spectrogram(
								    lowerCAmelCase__      ,					window=self.window      ,					frame_length=self.sample_size      ,					hop_length=self.sample_stride      ,					fft_length=self.n_fft      ,					mel_filters=self.mel_filters      ,					mel_floor=self.mel_floor      ,					log_mel="log10"      ,					)
								return log_mel_spec.T






							def __call__(   self      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = False      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = False      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					**lowerCAmelCase__      ,					)	-> BatchFeature:
								if audio is None and audio_target is None:
									raise ValueError("You must provide either `audio` or `audio_target` values."      )

								if sampling_rate is not None:
									if sampling_rate != self.sampling_rate:
										raise ValueError(
										    f"""The model corresponding to this feature extractor: {self} was trained using a sampling rate of"""
										    f""" {self.sampling_rate}. Please make sure that the provided audio input was sampled with"""
										    f""" {self.sampling_rate} and not {sampling_rate}."""      )
								else:
									logger.warning(
									    "It is strongly recommended to pass the ``sampling_rate`` argument to this function. "
									    "Failing to do so can result in silent errors that might be hard to debug."      )

								if audio is not None:
									a      :		Tuple	=				self._process_audio(
									    lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					**lowerCAmelCase__      ,					)
								else:
									a      :		Any	=				None

								if audio_target is not None:
									a      :		Tuple	=				self._process_audio(
									    lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					**lowerCAmelCase__      ,					)

									if inputs is None:
										return inputs_target
									else:
										a      :		Dict	=				inputs_target["input_values"]
										a      :		List[Any]	=				inputs_target.get("attention_mask"      )
										if decoder_attention_mask is not None:
											a      :		int	=				decoder_attention_mask

								return inputs






							def     __a   (   self      ,					lowerCAmelCase__      ,					lowerCAmelCase__ = False      ,					lowerCAmelCase__ = False      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = False      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					**lowerCAmelCase__      ,					)	-> BatchFeature:
								a      :		Union[str, Any]	=				isinstance(lowerCAmelCase__      ,					np.ndarray      ) and len(speech.shape      ) > 1
								if is_batched_numpy and len(speech.shape      ) > 2:
									raise ValueError(f"""Only mono-channel audio is supported for input to {self}"""      )
								a      :		Optional[int]	=				is_batched_numpy or (
								    isinstance(lowerCAmelCase__      ,					(list, tuple)      ) and (isinstance(speech[0]      ,					(np.ndarray, tuple, list)      ))
								)

								if is_batched:
									a      :		List[str]	=				[np.asarray(lowerCAmelCase__      ,					dtype=np.floataa      ) for speech in speech]
								elif not is_batched and not isinstance(lowerCAmelCase__      ,					np.ndarray      ):
									a      :		str	=				np.asarray(lowerCAmelCase__      ,					dtype=np.floataa      )
								elif isinstance(lowerCAmelCase__      ,					np.ndarray      ) and speech.dtype is np.dtype(np.floataa      ):
									a      :		int	=				speech.astype(np.floataa      )

								# always return batch
								if not is_batched:
									a      :		Optional[Any]	=				[speech]

								# needed to make pad() work on spectrogram inputs
								a      :		Union[str, Any]	=				self.feature_size

								# convert into correct format for padding
								if is_target:
									a      :		Union[str, Any]	=				[self._extract_mel_features(lowerCAmelCase__      ) for waveform in speech]
									a      :		str	=				BatchFeature({"input_values": features}      )
									a      :		Dict	=				self.num_mel_bins
								else:
									a      :		List[Any]	=				BatchFeature({"input_values": speech}      )

								a      :		Any	=				self.pad(
								    lowerCAmelCase__      ,					padding=lowerCAmelCase__      ,					max_length=lowerCAmelCase__      ,					truncation=lowerCAmelCase__      ,					pad_to_multiple_of=lowerCAmelCase__      ,					return_attention_mask=lowerCAmelCase__      ,					**lowerCAmelCase__      ,					)

								a      :		int	=				feature_size_hack

								# convert input values to correct format
								a      :		Optional[int]	=				padded_inputs["input_values"]
								if not isinstance(input_values[0]      ,					np.ndarray      ):
									a      :		Dict	=				[np.asarray(lowerCAmelCase__      ,					dtype=np.floataa      ) for array in input_values]
								elif (
								    not isinstance(lowerCAmelCase__      ,					np.ndarray      )
								    and isinstance(input_values[0]      ,					np.ndarray      )
								    and input_values[0].dtype is np.dtype(np.floataa      )
								):
									a      :		Tuple	=				[array.astype(np.floataa      ) for array in input_values]
								elif isinstance(lowerCAmelCase__      ,					np.ndarray      ) and input_values.dtype is np.dtype(np.floataa      ):
									a      :		Optional[Any]	=				input_values.astype(np.floataa      )

								# convert attention_mask to correct format
								a      :		Tuple	=				padded_inputs.get("attention_mask"      )
								if attention_mask is not None:
									a      :		str	=				[np.asarray(lowerCAmelCase__      ,					dtype=np.intaa      ) for array in attention_mask]

								# zero-mean and unit-variance normalization
								if not is_target and self.do_normalize:
									a      :		Union[str, Any]	=				(
									    attention_mask
									    if self._get_padding_strategies(lowerCAmelCase__      ,					max_length=lowerCAmelCase__      ) is not PaddingStrategy.DO_NOT_PAD
									    else None
									)
									a      :		Optional[Any]	=				self.zero_mean_unit_var_norm(
									    padded_inputs["input_values"]      ,					attention_mask=lowerCAmelCase__      ,					padding_value=self.padding_value      )

								if return_tensors is not None:
									a      :		Optional[Any]	=				padded_inputs.convert_to_tensors(lowerCAmelCase__      )

								return padded_inputs






							def     __a   (   self      )	-> Dict[str, Any]:
								a      :		Dict	=				super().to_dict()

								# Don't serialize these as they are derived from the other properties.
								a      :		str	=				["window", "mel_filters", "sample_size", "sample_stride", "n_fft", "n_freqs"]
								for name in names:
									if name in output:
										del output[name]

								return output







'''simple docstring'''


import json
import os
import tempfile

import datasets
from utils import generate_example_dataset, get_duration


_lowerCamelCase	:       List[str]      	=	5_0000
_lowerCamelCase	:       Optional[int]      	=	5000

_lowerCamelCase	,_lowerCamelCase	:       int      	=	os.path.split(__file__)
_lowerCamelCase	:       str      	=	os.path.join(RESULTS_BASEPATH, "results", RESULTS_FILENAME.replace(".py", ".json"))







@get_duration
def 				__lowerCamelCase				(   A__    ,						A__    )       ->	Any:






   """simple docstring"""
   for i in range(A__    ):
      UpperCamelCase					=  dataset[i]







@get_duration
def 				__lowerCamelCase				(   A__    ,						A__    ,						A__    )       ->	int:






   """simple docstring"""
   for i in range(0    ,						len(A__    )    ,						A__    ):
      UpperCamelCase					=  dataset[i : i + batch_size]







@get_duration
def 				__lowerCamelCase				(   A__    ,						A__    ,						A__    )       ->	List[Any]:






   """simple docstring"""
   with dataset.formatted_as(type=A__    ):
      for i in range(A__    ):
         UpperCamelCase					=  dataset[i]







@get_duration
def 				__lowerCamelCase				(   A__    ,						A__    ,						A__    ,						A__    )       ->	int:






   """simple docstring"""
   with dataset.formatted_as(type=A__    ):
      for i in range(0    ,						A__    ,						A__    ):
         UpperCamelCase					=  dataset[i : i + batch_size]







def 				__lowerCamelCase				(   )       ->	List[str]:






   """simple docstring"""
   UpperCamelCase					=  {'num examples': SPEED_TEST_N_EXAMPLES}
   UpperCamelCase					=  [
       (read, {'length': SMALL_TEST}),
       (read, {'length': SPEED_TEST_N_EXAMPLES}),
       (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 10}),
       (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 100}),
       (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 1_000}),
       (read_formatted, {'type': 'numpy', 'length': SMALL_TEST}),
       (read_formatted, {'type': 'pandas', 'length': SMALL_TEST}),
       (read_formatted, {'type': 'torch', 'length': SMALL_TEST}),
       (read_formatted, {'type': 'tensorflow', 'length': SMALL_TEST}),
       (read_formatted_batch, {'type': 'numpy', 'length': SMALL_TEST, 'batch_size': 10}),
       (read_formatted_batch, {'type': 'numpy', 'length': SMALL_TEST, 'batch_size': 1_000}),
   ]

   UpperCamelCase					=  [
       (read, {'length': SMALL_TEST}),
       (read, {'length': SPEED_TEST_N_EXAMPLES}),
       (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 10}),
       (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 100}),
       (read_batch, {'length': SPEED_TEST_N_EXAMPLES, 'batch_size': 1_000}),
       (read_formatted, {'type': 'numpy', 'length': SMALL_TEST}),
       (read_formatted_batch, {'type': 'numpy', 'length': SMALL_TEST, 'batch_size': 10}),
       (read_formatted_batch, {'type': 'numpy', 'length': SMALL_TEST, 'batch_size': 1_000}),
   ]
   with tempfile.TemporaryDirectory() as tmp_dir:
      print('generating dataset'    )
      UpperCamelCase					=  datasets.Features(
          {'list': datasets.Sequence(datasets.Value('float32'    )    ), 'numbers': datasets.Value('float32'    )}    )
      UpperCamelCase					=  generate_example_dataset(
          os.path.join(A__    ,						'dataset.arrow'    )    ,						A__    ,						num_examples=A__    ,						seq_shapes={'list': (100,)}    ,						)
      print('first set of iterations'    )
      for func, kwargs in functions:
         print(func.__name__    ,						str(A__    )    )
         UpperCamelCase					=  func(A__    ,						**A__    )

      print('shuffling dataset'    )
      UpperCamelCase					=  dataset.shuffle()
      print('Second set of iterations (after shuffling'    )
      for func, kwargs in functions_shuffled:
         print('shuffled '    ,						func.__name__    ,						str(A__    )    )
         UpperCamelCase					=  func(
             A__    ,						**A__    )

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


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

