





from math import ceil
from typing import List, Optional, Union

import numpy as np

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


__UpperCamelCase:	Optional[int] = logging.get_logger(__name__)


class       __lowerCAmelCase					(       _UpperCamelCase      ):


	'''simple docstring'''





	_A													=				["audio_values", "audio_mask"]

	def __init__(						self:						Any,					lowerCamelCase_:						int=2048,					lowerCamelCase_:						str=1,					lowerCamelCase_:						str=[16, 16],					lowerCamelCase_:						Union[str, Any]=128,					lowerCamelCase_:						Tuple=44100,					lowerCamelCase_:						Union[str, Any]=86,					lowerCamelCase_:						Optional[Any]=2048,					lowerCamelCase_:						Dict=0.0,					**lowerCamelCase_:						str,					):
			super().__init__(
			    feature_size=lowerCamelCase_,					sampling_rate=lowerCamelCase_,					padding_value=lowerCamelCase_,					**lowerCamelCase_,					)

			lowercase__					:      List[str]      =		spectrogram_length
			lowercase__					:      Union[str, Any]      =		num_channels
			lowercase__					:      int      =		patch_size
			lowercase__					:      Union[str, Any]      =		feature_size // self.patch_size[1]
			lowercase__					:      Optional[int]      =		n_fft
			lowercase__					:      int      =		sampling_rate // hop_length_to_sampling_rate
			lowercase__					:      str      =		sampling_rate
			lowercase__					:      Union[str, Any]      =		padding_value
			lowercase__					:      Any      =		mel_filter_bank(
			    num_frequency_bins=1 + n_fft // 2,					num_mel_filters=lowerCamelCase_,					min_frequency=0.0,					max_frequency=2_2_0_5_0.0,					sampling_rate=lowerCamelCase_,					norm='slaney',					mel_scale='slaney',					).T

	def    snake_case__(						self:						Tuple,					lowerCamelCase_:						np.array			):
			lowercase__					:      Optional[Any]      =		spectrogram(
			    lowerCamelCase_,					window_function(self.n_fft,					'hann'			),					frame_length=self.n_fft,					hop_length=self.hop_length,					power=2.0,					mel_filters=self.mel_filters.T,					log_mel='dB',					db_range=8_0.0,					)
			lowercase__					:      Optional[Any]      =		log_spec[:, :-1]
			lowercase__					:      int      =		log_spec - 2_0.0
			lowercase__					:      Dict      =		np.clip(log_spec / 4_0.0,					-2.0,					0.0			) + 1.0
			return log_spec





	def __call__(						self:						Dict,					lowerCamelCase_:						Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]],					lowerCamelCase_:						Optional[Union[str, TensorType]] = None,					lowerCamelCase_:						Optional[bool] = True,					lowerCamelCase_:						Optional[int] = None,					lowerCamelCase_:						bool = False,					lowerCamelCase_:						bool = False,					**lowerCamelCase_:						str,					):

			if sampling_rate is not None:
					if sampling_rate != self.sampling_rate:
							raise ValueError(
							    'This feature extractor is set to support sampling rate'
							    F""" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"""
							    F""" 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__					:      Any      =		isinstance(lowerCamelCase_,					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__					:      Optional[Any]      =		is_batched_numpy or (
			    isinstance(lowerCamelCase_,					(list, tuple)			) and (isinstance(raw_speech[0],					(np.ndarray, tuple, list)			))
			)
			if is_batched:
					lowercase__					:      int      =		[np.asarray([speech],					dtype=np.floataa			).T for speech in raw_speech]
			elif not is_batched and not isinstance(lowerCamelCase_,					np.ndarray			):
					lowercase__					:      Optional[int]      =		np.asarray(lowerCamelCase_,					dtype=np.floataa			)
			elif isinstance(lowerCamelCase_,					np.ndarray			) and raw_speech.dtype is np.dtype(np.floataa			):
					lowercase__					:      str      =		raw_speech.astype(np.floataa			)
			# always return batch
			if not is_batched:
					lowercase__					:      Tuple      =		[np.asarray([raw_speech]			).T]

			# Convert audio signals to log mel spectrograms, truncate by time axis
			lowercase__					:      List[str]      =		[
			    self._np_extract_fbank_features(waveform.squeeze()			).T[: self.spectrogram_length] for waveform in raw_speech
			]
			if isinstance(audio_features[0],					lowerCamelCase_			):
					lowercase__					:      str      =		[np.asarray(lowerCamelCase_,					dtype=np.floataa			) for feature in audio_features]

			# Create audio attention mask
			lowercase__					:      Dict      =		max(
			    [ceil(feature.shape[0] / self.patch_size[0]			) * self.freq_len for feature in audio_features]			)  # The maximum number of audio patches in a batch
			if return_attention_mask:
					lowercase__					:      int      =		[
					    (ceil(feature.shape[0] / self.patch_size[0]			) * self.freq_len) * [1]
					    + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]			) * self.freq_len) * [0]
					    for feature in audio_features
					]
					lowercase__					:      Any      =		np.array(lowerCamelCase_			).astype(np.floataa			)

			# convert into correct format for padding
			lowercase__					:      Any      =		max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
			lowercase__					:      Tuple      =		np.ones([len(lowerCamelCase_			), 1, max_time_len, self.feature_size]			).astype(np.floataa			)
			lowercase__					:      Dict      =		padded_audio_features * self.padding_value
			for i in range(len(lowerCamelCase_			)			):
					lowercase__					:      Union[str, Any]      =		audio_features[i]
					lowercase__					:      str      =		feature

			# return as BatchFeature
			if return_attention_mask:
					lowercase__					:      Dict      =		{'audio_values': padded_audio_features, 'audio_mask': audio_mask}
			else:
					lowercase__					:      int      =		{'audio_values': padded_audio_features}

			lowercase__					:      List[Any]      =		BatchFeature(data=lowerCamelCase_,					tensor_type=lowerCamelCase_			)
			return encoded_inputs









from math import ceil
from typing import List, Optional, Union

import numpy as np

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


__UpperCamelCase:	Optional[int] = logging.get_logger(__name__)


class       __lowerCAmelCase					(       _UpperCamelCase      ):


	'''simple docstring'''





	_A													=				["audio_values", "audio_mask"]

	def __init__(						self:						Any,					lowerCamelCase_:						int=2048,					lowerCamelCase_:						str=1,					lowerCamelCase_:						str=[16, 16],					lowerCamelCase_:						Union[str, Any]=128,					lowerCamelCase_:						Tuple=44100,					lowerCamelCase_:						Union[str, Any]=86,					lowerCamelCase_:						Optional[Any]=2048,					lowerCamelCase_:						Dict=0.0,					**lowerCamelCase_:						str,					):
			super().__init__(
			    feature_size=lowerCamelCase_,					sampling_rate=lowerCamelCase_,					padding_value=lowerCamelCase_,					**lowerCamelCase_,					)

			lowercase__					:      List[str]      =		spectrogram_length
			lowercase__					:      Union[str, Any]      =		num_channels
			lowercase__					:      int      =		patch_size
			lowercase__					:      Union[str, Any]      =		feature_size // self.patch_size[1]
			lowercase__					:      Optional[int]      =		n_fft
			lowercase__					:      int      =		sampling_rate // hop_length_to_sampling_rate
			lowercase__					:      str      =		sampling_rate
			lowercase__					:      Union[str, Any]      =		padding_value
			lowercase__					:      Any      =		mel_filter_bank(
			    num_frequency_bins=1 + n_fft // 2,					num_mel_filters=lowerCamelCase_,					min_frequency=0.0,					max_frequency=2_2_0_5_0.0,					sampling_rate=lowerCamelCase_,					norm='slaney',					mel_scale='slaney',					).T

	def    snake_case__(						self:						Tuple,					lowerCamelCase_:						np.array			):
			lowercase__					:      Optional[Any]      =		spectrogram(
			    lowerCamelCase_,					window_function(self.n_fft,					'hann'			),					frame_length=self.n_fft,					hop_length=self.hop_length,					power=2.0,					mel_filters=self.mel_filters.T,					log_mel='dB',					db_range=8_0.0,					)
			lowercase__					:      Optional[Any]      =		log_spec[:, :-1]
			lowercase__					:      int      =		log_spec - 2_0.0
			lowercase__					:      Dict      =		np.clip(log_spec / 4_0.0,					-2.0,					0.0			) + 1.0
			return log_spec





	def __call__(						self:						Dict,					lowerCamelCase_:						Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]],					lowerCamelCase_:						Optional[Union[str, TensorType]] = None,					lowerCamelCase_:						Optional[bool] = True,					lowerCamelCase_:						Optional[int] = None,					lowerCamelCase_:						bool = False,					lowerCamelCase_:						bool = False,					**lowerCamelCase_:						str,					):

			if sampling_rate is not None:
					if sampling_rate != self.sampling_rate:
							raise ValueError(
							    'This feature extractor is set to support sampling rate'
							    F""" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"""
							    F""" 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__					:      Any      =		isinstance(lowerCamelCase_,					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__					:      Optional[Any]      =		is_batched_numpy or (
			    isinstance(lowerCamelCase_,					(list, tuple)			) and (isinstance(raw_speech[0],					(np.ndarray, tuple, list)			))
			)
			if is_batched:
					lowercase__					:      int      =		[np.asarray([speech],					dtype=np.floataa			).T for speech in raw_speech]
			elif not is_batched and not isinstance(lowerCamelCase_,					np.ndarray			):
					lowercase__					:      Optional[int]      =		np.asarray(lowerCamelCase_,					dtype=np.floataa			)
			elif isinstance(lowerCamelCase_,					np.ndarray			) and raw_speech.dtype is np.dtype(np.floataa			):
					lowercase__					:      str      =		raw_speech.astype(np.floataa			)
			# always return batch
			if not is_batched:
					lowercase__					:      Tuple      =		[np.asarray([raw_speech]			).T]

			# Convert audio signals to log mel spectrograms, truncate by time axis
			lowercase__					:      List[str]      =		[
			    self._np_extract_fbank_features(waveform.squeeze()			).T[: self.spectrogram_length] for waveform in raw_speech
			]
			if isinstance(audio_features[0],					lowerCamelCase_			):
					lowercase__					:      str      =		[np.asarray(lowerCamelCase_,					dtype=np.floataa			) for feature in audio_features]

			# Create audio attention mask
			lowercase__					:      Dict      =		max(
			    [ceil(feature.shape[0] / self.patch_size[0]			) * self.freq_len for feature in audio_features]			)  # The maximum number of audio patches in a batch
			if return_attention_mask:
					lowercase__					:      int      =		[
					    (ceil(feature.shape[0] / self.patch_size[0]			) * self.freq_len) * [1]
					    + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]			) * self.freq_len) * [0]
					    for feature in audio_features
					]
					lowercase__					:      Any      =		np.array(lowerCamelCase_			).astype(np.floataa			)

			# convert into correct format for padding
			lowercase__					:      Any      =		max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
			lowercase__					:      Tuple      =		np.ones([len(lowerCamelCase_			), 1, max_time_len, self.feature_size]			).astype(np.floataa			)
			lowercase__					:      Dict      =		padded_audio_features * self.padding_value
			for i in range(len(lowerCamelCase_			)			):
					lowercase__					:      Union[str, Any]      =		audio_features[i]
					lowercase__					:      str      =		feature

			# return as BatchFeature
			if return_attention_mask:
					lowercase__					:      Dict      =		{'audio_values': padded_audio_features, 'audio_mask': audio_mask}
			else:
					lowercase__					:      int      =		{'audio_values': padded_audio_features}

			lowercase__					:      List[Any]      =		BatchFeature(data=lowerCamelCase_,					tensor_type=lowerCamelCase_			)
			return encoded_inputs


