



"""simple docstring"""







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

import numpy as np
import torch

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


lowerCAmelCase_  		=  logging.get_logger(__name__)
class 			__A			(      A_       ):







		'''simple docstring'''





		lowerCAmelCase     :   List[Any]        =							["input_features", "is_longer"]







		def __init__(      self		:    Any     ,_snake_case		:    int=64     ,_snake_case		:    Optional[Any]=48_000     ,_snake_case		:    Dict=480     ,_snake_case		:    str=10     ,_snake_case		:    Tuple=1_024     ,_snake_case		:    Tuple=0.0     ,_snake_case		:    Optional[Any]=False     ,_snake_case		:    float = 0     ,_snake_case		:    float = 14_000     ,_snake_case		:    int = None     ,_snake_case		:    str = "fusion"     ,_snake_case		:    str = "repeatpad"     ,**_snake_case		:    List[Any]     ,)   ->						str:






				"""simple docstring"""



				super().__init__(
				    feature_size=snake_case__     ,sampling_rate=snake_case__     ,padding_value=snake_case__     ,return_attention_mask=snake_case__     ,**snake_case__     ,)
				lowercase__  :       Any												=     top_db
				lowercase__  :       Optional[int]												=     truncation
				lowercase__  :       int												=     padding
				lowercase__  :       Optional[Any]												=     fft_window_size
				lowercase__  :       str												=     (fft_window_size >> 1) + 1
				lowercase__  :       List[Any]												=     hop_length
				lowercase__  :       str												=     max_length_s
				lowercase__  :       Any												=     max_length_s * sampling_rate
				lowercase__  :       Dict												=     sampling_rate
				lowercase__  :       Optional[int]												=     frequency_min
				lowercase__  :       Union[str, Any]												=     frequency_max
				lowercase__  :       List[Any]												=     mel_filter_bank(
				    num_frequency_bins=self.nb_frequency_bins     ,num_mel_filters=snake_case__     ,min_frequency=snake_case__     ,max_frequency=snake_case__     ,sampling_rate=snake_case__     ,norm=snake_case__     ,mel_scale='''htk'''     ,)
				lowercase__  :       Optional[Any]												=     mel_filter_bank(
				    num_frequency_bins=self.nb_frequency_bins     ,num_mel_filters=snake_case__     ,min_frequency=snake_case__     ,max_frequency=snake_case__     ,sampling_rate=snake_case__     ,norm='''slaney'''     ,mel_scale='''slaney'''     ,)







		def    UpperCAmelCase       (      self		:    Dict		)   ->						Dict:






				"""simple docstring"""



				lowercase__  :       List[Any]												=     copy.deepcopy(self.__dict__		)
				lowercase__  :       int												=     self.__class__.__name__
				if "mel_filters" in output:
						del output["mel_filters"]
				if "mel_filters_slaney" in output:
						del output["mel_filters_slaney"]
				return output







		def    UpperCAmelCase       (      self		:    List[Any]     ,_snake_case		:    np.array     ,_snake_case		:    Optional[np.array] = None		)   ->						Tuple:






				"""simple docstring"""



				lowercase__  :       Tuple												=     spectrogram(
				    snake_case__     ,window_function(self.fft_window_size     ,'''hann'''		)     ,frame_length=self.fft_window_size     ,hop_length=self.hop_length     ,power=2.0     ,mel_filters=snake_case__     ,log_mel='''dB'''     ,)
				return log_mel_spectrogram.T







		def    UpperCAmelCase       (      self		:    Optional[Any]     ,_snake_case		:    Dict     ,_snake_case		:    int     ,_snake_case		:    str		)   ->						Optional[Any]:






				"""simple docstring"""



				lowercase__  :       Optional[Any]												=     np.array_split(list(range(0     ,total_frames - chunk_frames + 1		)		)     ,3		)
				if len(ranges[1]		) == 0:
						# if the audio is too short, we just use the first chunk
						lowercase__  :       Optional[Any]												=     [0]
				if len(ranges[2]		) == 0:
						# if the audio is too short, we just use the first chunk
						lowercase__  :       Optional[Any]												=     [0]
				# randomly choose index for each part
				lowercase__  :       List[str]												=     np.random.choice(ranges[0]		)
				lowercase__  :       int												=     np.random.choice(ranges[1]		)
				lowercase__  :       List[str]												=     np.random.choice(ranges[2]		)

				lowercase__  :       Tuple												=     mel[idx_front : idx_front + chunk_frames, :]
				lowercase__  :       Union[str, Any]												=     mel[idx_middle : idx_middle + chunk_frames, :]
				lowercase__  :       Optional[int]												=     mel[idx_back : idx_back + chunk_frames, :]

				lowercase__  :       Union[str, Any]												=     torch.tensor(mel[None, None, :]		)
				lowercase__  :       List[Any]												=     torch.nn.functional.interpolate(
				    snake_case__     ,size=[chunk_frames, 64]     ,mode='''bilinear'''     ,align_corners=snake_case__		)
				lowercase__  :       List[Any]												=     mel_shrink[0][0].numpy()
				lowercase__  :       Optional[Any]												=     np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]     ,axis=0		)
				return mel_fusion







		def    UpperCAmelCase       (      self		:    Dict     ,_snake_case		:    np.array     ,_snake_case		:    Union[str, Any]     ,_snake_case		:    Dict     ,_snake_case		:    List[str]		)   ->						Optional[int]:






				"""simple docstring"""



				if waveform.shape[0] > max_length:
						if truncation == "rand_trunc":
								lowercase__  :       str												=     True
								# random crop to max_length (for compatibility) -> this should be handled by self.pad
								lowercase__  :       List[Any]												=     len(snake_case__		) - max_length
								lowercase__  :       Dict												=     np.random.randint(0     ,overflow + 1		)
								lowercase__  :       str												=     waveform[idx : idx + max_length]
								lowercase__  :       Dict												=     self._np_extract_fbank_features(snake_case__     ,self.mel_filters_slaney		)[None, :]
						elif truncation == "fusion":
								lowercase__  :       Optional[Any]												=     self._np_extract_fbank_features(snake_case__     ,self.mel_filters		)
								lowercase__  :       Optional[Any]												=     max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
								lowercase__  :       Dict												=     mel.shape[0]
								if chunk_frames == total_frames:
										# there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
										# In this case, we just use the whole audio.
										lowercase__  :       int												=     np.stack([mel, mel, mel, mel]     ,axis=0		)
										lowercase__  :       Union[str, Any]												=     False
								else:
										lowercase__  :       Any												=     self._random_mel_fusion(snake_case__     ,snake_case__     ,snake_case__		)
										lowercase__  :       int												=     True
						else:
								raise NotImplementedError(f"""data_truncating {truncation} not implemented"""		)

				else:
						lowercase__  :       int												=     False
						# only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
						if waveform.shape[0] < max_length:
								if padding == "repeat":
										lowercase__  :       int												=     int(max_length / len(snake_case__		)		)
										lowercase__  :       Any												=     np.stack(np.tile(snake_case__     ,n_repeat + 1		)		)[:max_length]
								if padding == "repeatpad":
										lowercase__  :       Tuple												=     int(max_length / len(snake_case__		)		)
										lowercase__  :       int												=     np.stack(np.tile(snake_case__     ,snake_case__		)		)
								lowercase__  :       Dict												=     np.pad(snake_case__     ,(0, max_length - waveform.shape[0])     ,mode='''constant'''     ,constant_values=0		)

						if truncation == "fusion":
								lowercase__  :       Optional[int]												=     self._np_extract_fbank_features(snake_case__     ,self.mel_filters		)
								lowercase__  :       Any												=     np.stack([input_mel, input_mel, input_mel, input_mel]     ,axis=0		)
						else:
								lowercase__  :       Any												=     self._np_extract_fbank_features(snake_case__     ,self.mel_filters_slaney		)[None, :]

				return input_mel, longer







		def __call__(      self		:    Any     ,_snake_case		:    Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]     ,_snake_case		:    str = None     ,_snake_case		:    Optional[str] = None     ,_snake_case		:    Optional[int] = None     ,_snake_case		:    Optional[int] = None     ,_snake_case		:    Optional[Union[str, TensorType]] = None     ,**_snake_case		:    Any     ,)   ->						List[str]:






				"""simple docstring"""



				lowercase__  :       str												=     truncation if truncation is not None else self.truncation
				lowercase__  :       Tuple												=     padding if padding else self.padding

				if sampling_rate is not None:
						if sampling_rate != self.sampling_rate:
								raise ValueError(
								    f"""The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"""
								    f""" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"""
								    f""" was sampled with {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.'''		)

				lowercase__  :       Dict												=     isinstance(snake_case__     ,np.ndarray		) and len(raw_speech.shape		) > 1
				if is_batched_numpy and len(raw_speech.shape		) > 2:
						raise ValueError(f"""Only mono-channel audio is supported for input to {self}"""		)
				lowercase__  :       Union[str, Any]												=     is_batched_numpy or (
				    isinstance(snake_case__     ,(list, tuple)		) and (isinstance(raw_speech[0]     ,(np.ndarray, tuple, list)		))
				)

				if is_batched:
						lowercase__  :       Dict												=     [np.asarray(snake_case__     ,dtype=np.floataa		) for speech in raw_speech]
				elif not is_batched and not isinstance(snake_case__     ,np.ndarray		):
						lowercase__  :       Optional[Any]												=     np.asarray(snake_case__     ,dtype=np.floataa		)
				elif isinstance(snake_case__     ,np.ndarray		) and raw_speech.dtype is np.dtype(np.floataa		):
						lowercase__  :       Optional[int]												=     raw_speech.astype(np.floataa		)

				# always return batch
				if not is_batched:
						lowercase__  :       List[str]												=     [np.asarray(snake_case__		)]

				# convert to mel spectrogram, truncate and pad if needed.
				lowercase__  :       str												=     [
				    self._get_input_mel(snake_case__     ,max_length if max_length else self.nb_max_samples     ,snake_case__     ,snake_case__		)
				    for waveform in raw_speech
				]

				lowercase__  :       Optional[int]												=     []
				lowercase__  :       List[Any]												=     []
				for mel, longer in padded_inputs:
						input_mel.append(snake_case__		)
						is_longer.append(snake_case__		)

				if truncation == "fusion" and sum(snake_case__		) == 0:
						# if no audio is longer than 10s, then randomly select one audio to be longer
						lowercase__  :       Dict												=     np.random.randint(0     ,len(snake_case__		)		)
						lowercase__  :       List[str]												=     True

				if isinstance(input_mel[0]     ,snake_case__		):
						lowercase__  :       Union[str, Any]												=     [np.asarray(snake_case__     ,dtype=np.floataa		) for feature in input_mel]

				# is_longer is a list of bool
				lowercase__  :       int												=     [[longer] for longer in is_longer]

				lowercase__  :       Optional[int]												=     {"input_features": input_mel, "is_longer": is_longer}
				lowercase__  :       List[str]												=     BatchFeature(snake_case__		)

				if return_tensors is not None:
						lowercase__  :       str												=     input_features.convert_to_tensors(snake_case__		)

				return input_features


'''simple docstring'''





import unittest

from transformers.testing_utils import CaptureStdout
from transformers.tools.python_interpreter import evaluate







def snake_case_     (					lowercase__		):
  return x + 2






class 							UpperCAmelCase_							(    unittest.TestCase				):

 '''simple docstring'''







 def 							UpperCamelCase  (					self		:  Optional[int] ):



   '''simple docstring'''




   UpperCAmelCase__      :					List[str]				       =							"x = 3"
   UpperCAmelCase__      :					List[str]				       =							{}
   UpperCAmelCase__      :					Tuple				       =							evaluate(snake_case__			,				{}			,				state=snake_case__ )
   assert result == 3
   self.assertDictEqual(snake_case__			,				{"x": 3} )

   UpperCAmelCase__      :					Any				       =							"x = y"
   UpperCAmelCase__      :					List[str]				       =							{"y": 5}
   UpperCAmelCase__      :					Dict				       =							evaluate(snake_case__			,				{}			,				state=snake_case__ )
   # evaluate returns the value of the last assignment.
   assert result == 5
   self.assertDictEqual(snake_case__			,				{"x": 5, "y": 5} )







 def 							UpperCamelCase  (					self		:  Union[str, Any] ):



   '''simple docstring'''




   UpperCAmelCase__      :					int				       =							"y = add_two(x)"
   UpperCAmelCase__      :					Tuple				       =							{"x": 3}
   UpperCAmelCase__      :					Union[str, Any]				       =							evaluate(snake_case__			,				{"add_two": add_two}			,				state=snake_case__ )
   assert result == 5
   self.assertDictEqual(snake_case__			,				{"x": 3, "y": 5} )

   # Won't work without the tool
   with CaptureStdout() as out:
     UpperCAmelCase__      :					Any				       =							evaluate(snake_case__			,				{}			,				state=snake_case__ )
   assert result is None
   assert "tried to execute add_two" in out.out







 def 							UpperCamelCase  (					self		:  List[str] ):



   '''simple docstring'''




   UpperCAmelCase__      :					Tuple				       =							"x = 3"
   UpperCAmelCase__      :					List[Any]				       =							{}
   UpperCAmelCase__      :					Tuple				       =							evaluate(snake_case__			,				{}			,				state=snake_case__ )
   assert result == 3
   self.assertDictEqual(snake_case__			,				{"x": 3} )







 def 							UpperCamelCase  (					self		:  Dict ):



   '''simple docstring'''




   UpperCAmelCase__      :					int				       =							"test_dict = {'x': x, 'y': add_two(x)}"
   UpperCAmelCase__      :					List[Any]				       =							{"x": 3}
   UpperCAmelCase__      :					int				       =							evaluate(snake_case__			,				{"add_two": add_two}			,				state=snake_case__ )
   self.assertDictEqual(snake_case__			,				{"x": 3, "y": 5} )
   self.assertDictEqual(snake_case__			,				{"x": 3, "test_dict": {"x": 3, "y": 5}} )







 def 							UpperCamelCase  (					self		:  List[str] ):



   '''simple docstring'''




   UpperCAmelCase__      :					int				       =							"x = 3\ny = 5"
   UpperCAmelCase__      :					Dict				       =							{}
   UpperCAmelCase__      :					Union[str, Any]				       =							evaluate(snake_case__			,				{}			,				state=snake_case__ )
   # evaluate returns the value of the last assignment.
   assert result == 5
   self.assertDictEqual(snake_case__			,				{"x": 3, "y": 5} )







 def 							UpperCamelCase  (					self		:  Union[str, Any] ):



   '''simple docstring'''




   UpperCAmelCase__      :					Dict				       =							"text = f'This is x: {x}.'"
   UpperCAmelCase__      :					Dict				       =							{"x": 3}
   UpperCAmelCase__      :					Dict				       =							evaluate(snake_case__			,				{}			,				state=snake_case__ )
   # evaluate returns the value of the last assignment.
   assert result == "This is x: 3."
   self.assertDictEqual(snake_case__			,				{"x": 3, "text": "This is x: 3."} )







 def 							UpperCamelCase  (					self		:  List[Any] ):



   '''simple docstring'''




   UpperCAmelCase__      :					List[str]				       =							"if x <= 3:\n    y = 2\nelse:\n    y = 5"
   UpperCAmelCase__      :					Dict				       =							{"x": 3}
   UpperCAmelCase__      :					Dict				       =							evaluate(snake_case__			,				{}			,				state=snake_case__ )
   # evaluate returns the value of the last assignment.
   assert result == 2
   self.assertDictEqual(snake_case__			,				{"x": 3, "y": 2} )

   UpperCAmelCase__      :					Optional[Any]				       =							{"x": 8}
   UpperCAmelCase__      :					str				       =							evaluate(snake_case__			,				{}			,				state=snake_case__ )
   # evaluate returns the value of the last assignment.
   assert result == 5
   self.assertDictEqual(snake_case__			,				{"x": 8, "y": 5} )







 def 							UpperCamelCase  (					self		:  str ):



   '''simple docstring'''




   UpperCAmelCase__      :					Optional[int]				       =							"test_list = [x, add_two(x)]"
   UpperCAmelCase__      :					Any				       =							{"x": 3}
   UpperCAmelCase__      :					Any				       =							evaluate(snake_case__			,				{"add_two": add_two}			,				state=snake_case__ )
   self.assertListEqual(snake_case__			,				[3, 5] )
   self.assertDictEqual(snake_case__			,				{"x": 3, "test_list": [3, 5]} )







 def 							UpperCamelCase  (					self		:  str ):



   '''simple docstring'''




   UpperCAmelCase__      :					str				       =							"y = x"
   UpperCAmelCase__      :					str				       =							{"x": 3}
   UpperCAmelCase__      :					Any				       =							evaluate(snake_case__			,				{}			,				state=snake_case__ )
   assert result == 3
   self.assertDictEqual(snake_case__			,				{"x": 3, "y": 3} )







 def 							UpperCamelCase  (					self		:  int ):



   '''simple docstring'''




   UpperCAmelCase__      :					Optional[int]				       =							"test_list = [x, add_two(x)]\ntest_list[1]"
   UpperCAmelCase__      :					List[Any]				       =							{"x": 3}
   UpperCAmelCase__      :					List[Any]				       =							evaluate(snake_case__			,				{"add_two": add_two}			,				state=snake_case__ )
   assert result == 5
   self.assertDictEqual(snake_case__			,				{"x": 3, "test_list": [3, 5]} )

   UpperCAmelCase__      :					Union[str, Any]				       =							"test_dict = {'x': x, 'y': add_two(x)}\ntest_dict['y']"
   UpperCAmelCase__      :					str				       =							{"x": 3}
   UpperCAmelCase__      :					Any				       =							evaluate(snake_case__			,				{"add_two": add_two}			,				state=snake_case__ )
   assert result == 5
   self.assertDictEqual(snake_case__			,				{"x": 3, "test_dict": {"x": 3, "y": 5}} )







 def 							UpperCamelCase  (					self		:  Union[str, Any] ):



   '''simple docstring'''




   UpperCAmelCase__      :					List[Any]				       =							"x = 0\nfor i in range(3):\n    x = i"
   UpperCAmelCase__      :					List[Any]				       =							{}
   UpperCAmelCase__      :					Any				       =							evaluate(snake_case__			,				{"range": range}			,				state=snake_case__ )
   assert result == 2
   self.assertDictEqual(snake_case__			,				{"x": 2, "i": 2} )



