


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


snake_case							:       str     			=      logging.get_logger(__name__)







class        snake_case_   (lowerCamelCase_      ):
					UpperCAmelCase__					:    List[Any]          =			['''input_features''', '''is_longer''']
					def __init__(      self   :Union[str, Any]			,__snake_case   :Optional[Any]=64			,__snake_case   :Dict=4_80_00			,__snake_case   :List[Any]=4_80			,__snake_case   :str=10			,__snake_case   :int=10_24			,__snake_case   :List[Any]=0.0			,__snake_case   :Optional[Any]=False			,__snake_case   :float = 0			,__snake_case   :float = 1_40_00			,__snake_case   :int = None			,__snake_case   :str = "fusion"			,__snake_case   :str = "repeatpad"			,**__snake_case   :List[str]			,)					->    Any:
												super().__init__(
												    feature_size=__snake_case			,sampling_rate=__snake_case			,padding_value=__snake_case			,return_attention_mask=__snake_case			,**__snake_case			,)
												a__             =     top_db
												a__             =     truncation
												a__             =     padding
												a__             =     fft_window_size
												a__             =     (fft_window_size >> 1) + 1
												a__             =     hop_length
												a__             =     max_length_s
												a__             =     max_length_s * sampling_rate
												a__             =     sampling_rate
												a__             =     frequency_min
												a__             =     frequency_max
												a__             =     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'			,)
												a__             =     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  lowerCamelCase__(      self   :List[Any]      )					->    Dict[str, Any]:
												a__             =     copy.deepcopy(self.__dict__      )
												a__             =     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  lowerCamelCase__(      self   :Any			,__snake_case   :np.array			,__snake_case   :Optional[np.array] = None      )					->    np.ndarray:
												a__             =     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  lowerCamelCase__(      self   :List[Any]			,__snake_case   :Optional[int]			,__snake_case   :Optional[int]			,__snake_case   :Any      )					->    Dict:
												a__             =     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
																			a__             =     [0]
												if len(ranges[2]      ) == 0:
																			# if the audio is too short, we just use the first chunk
																			a__             =     [0]
												# randomly choose index for each part
												a__             =     np.random.choice(ranges[0]      )
												a__             =     np.random.choice(ranges[1]      )
												a__             =     np.random.choice(ranges[2]      )

												a__             =     mel[idx_front : idx_front + chunk_frames, :]
												a__             =     mel[idx_middle : idx_middle + chunk_frames, :]
												a__             =     mel[idx_back : idx_back + chunk_frames, :]

												a__             =     torch.tensor(mel[None, None, :]      )
												a__             =     torch.nn.functional.interpolate(
												    __snake_case			,size=[chunk_frames, 64]			,mode='bilinear'			,align_corners=__snake_case      )
												a__             =     mel_shrink[0][0].numpy()
												a__             =     np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]			,axis=0      )
												return mel_fusion
					def  lowerCamelCase__(      self   :List[Any]			,__snake_case   :np.array			,__snake_case   :List[str]			,__snake_case   :Optional[int]			,__snake_case   :int      )					->    np.array:
												if waveform.shape[0] > max_length:
																			if truncation == "rand_trunc":
																										a__             =     True
																										# random crop to max_length (for compatibility) -> this should be handled by self.pad
																										a__             =     len(__snake_case      ) - max_length
																										a__             =     np.random.randint(0			,overflow + 1      )
																										a__             =     waveform[idx : idx + max_length]
																										a__             =     self._np_extract_fbank_features(__snake_case			,self.mel_filters_slaney      )[None, :]
																			elif truncation == "fusion":
																										a__             =     self._np_extract_fbank_features(__snake_case			,self.mel_filters      )
																										a__             =     max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
																										a__             =     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.
																																	a__             =     np.stack([mel, mel, mel, mel]			,axis=0      )
																																	a__             =     False
																										else:
																																	a__             =     self._random_mel_fusion(__snake_case			,__snake_case			,__snake_case      )
																																	a__             =     True
																			else:
																										raise NotImplementedError(F'data_truncating {truncation} not implemented'      )

												else:
																			a__             =     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":
																																	a__             =     int(max_length / len(__snake_case      )      )
																																	a__             =     np.stack(np.tile(__snake_case			,n_repeat + 1      )      )[:max_length]
																										if padding == "repeatpad":
																																	a__             =     int(max_length / len(__snake_case      )      )
																																	a__             =     np.stack(np.tile(__snake_case			,__snake_case      )      )
																										a__             =     np.pad(__snake_case			,(0, max_length - waveform.shape[0])			,mode='constant'			,constant_values=0      )

																			if truncation == "fusion":
																										a__             =     self._np_extract_fbank_features(__snake_case			,self.mel_filters      )
																										a__             =     np.stack([input_mel, input_mel, input_mel, input_mel]			,axis=0      )
																			else:
																										a__             =     self._np_extract_fbank_features(__snake_case			,self.mel_filters_slaney      )[None, :]

												return input_mel, longer



					def __call__(      self   :Dict			,__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   :Optional[Any]			,)					->    BatchFeature:
												a__             =     truncation if truncation is not None else self.truncation
												a__             =     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.'      )

												a__             =     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}'      )
												a__             =     is_batched_numpy or (
												    isinstance(__snake_case			,(list, tuple)      ) and (isinstance(raw_speech[0]			,(np.ndarray, tuple, list)      ))
												)

												if is_batched:
																			a__             =     [np.asarray(__snake_case			,dtype=np.floataa      ) for speech in raw_speech]
												elif not is_batched and not isinstance(__snake_case			,np.ndarray      ):
																			a__             =     np.asarray(__snake_case			,dtype=np.floataa      )
												elif isinstance(__snake_case			,np.ndarray      ) and raw_speech.dtype is np.dtype(np.floataa      ):
																			a__             =     raw_speech.astype(np.floataa      )

												# always return batch
												if not is_batched:
																			a__             =     [np.asarray(__snake_case      )]

												# convert to mel spectrogram, truncate and pad if needed.
												a__             =     [
												    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
												]

												a__             =     []
												a__             =     []
												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
																			a__             =     np.random.randint(0			,len(__snake_case      )      )
																			a__             =     True

												if isinstance(input_mel[0]			,__snake_case      ):
																			a__             =     [np.asarray(__snake_case			,dtype=np.floataa      ) for feature in input_mel]

												# is_longer is a list of bool
												a__             =     [[longer] for longer in is_longer]

												a__             =     {'input_features': input_mel, 'is_longer': is_longer}
												a__             =     BatchFeature(__snake_case      )

												if return_tensors is not None:
																			a__             =     input_features.convert_to_tensors(__snake_case      )

												return input_features








import sys

from .dependency_versions_table import deps
from .utils.versions import require_version, require_version_core


# define which module versions we always want to check at run time
# (usually the ones defined in `install_requires` in setup.py)
#
# order specific notes:
# - tqdm must be checked before tokenizers

snake_case							:       List[str]     			=      '''python tqdm regex requests packaging filelock numpy tokenizers'''.split()
if sys.version_info < (3, 7):
						pkgs_to_check_at_runtime.append('''dataclasses''')
if sys.version_info < (3, 8):
						pkgs_to_check_at_runtime.append('''importlib_metadata''')

for pkg in pkgs_to_check_at_runtime:
						if pkg in deps:
												if pkg == "tokenizers":
																		# must be loaded here, or else tqdm check may fail
																		from .utils import is_tokenizers_available

																		if not is_tokenizers_available():
																								continue  # not required, check version only if installed

												require_version_core(deps[pkg])
						else:
												raise ValueError(f"""can't find {pkg} in {deps.keys()}, check dependency_versions_table.py""")






def 							__lowercase   ( __lowerCAmelCase							:				Optional[Any]					,    __lowerCAmelCase							:				Tuple=None   ):
							require_version(deps[pkg]					,    __lowerCAmelCase   )




