


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


lowerCamelCase_       :       List[str]      						=  logging.get_logger(__name__)






class 				__A    (	_UpperCamelCase      ):






 """simple docstring"""




 __lowerCAmelCase			 =  ['audio_values', 'audio_mask']


 def __init__(		self			,   __A=2048			,   __A=1			,   __A=[16, 16]			,   __A=128			,   __A=4_4100			,   __A=86			,   __A=2048			,   __A=0.0			,   **__A			,   ) ->					Union[str, Any]:
     super().__init__(
         feature_size=_UpperCAmelCase			,   sampling_rate=_UpperCAmelCase			,   padding_value=_UpperCAmelCase			,   **_UpperCAmelCase			,   )

     a					=spectrogram_length
     a					=num_channels
     a					=patch_size
     a					=feature_size // self.patch_size[1]
     a					=n_fft
     a					=sampling_rate // hop_length_to_sampling_rate
     a					=sampling_rate
     a					=padding_value
     a					=mel_filter_bank(
         num_frequency_bins=1 + n_fft // 2			,   num_mel_filters=_UpperCAmelCase			,   min_frequency=0.0			,   max_frequency=2_2050.0			,   sampling_rate=_UpperCAmelCase			,   norm='''slaney'''			,   mel_scale='''slaney'''			,   ).T


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A						) ->					Optional[int]:
     a					=spectrogram(
         _UpperCAmelCase			,   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=80.0			,   )
     a					=log_spec[:, :-1]
     a					=log_spec - 20.0
     a					=np.clip(log_spec / 40.0			,   -2.0			,   0.0						) + 1.0
     return log_spec





 def __call__(		self			,   __A			,   __A = None			,   __A = True			,   __A = None			,   __A = False			,   __A = False			,   **__A			,   ) ->					List[Any]:

     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.'''						)

     a					=isinstance(_UpperCAmelCase			,   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(_UpperCAmelCase			,   (list, tuple)						) and (isinstance(raw_speech[0]			,   (np.ndarray, tuple, list)						))
     )
     if is_batched:
         a					=[np.asarray([speech]			,   dtype=np.floataa						).T for speech in raw_speech]
     elif not is_batched and not isinstance(_UpperCAmelCase			,   np.ndarray						):
         a					=np.asarray(_UpperCAmelCase			,   dtype=np.floataa						)
     elif isinstance(_UpperCAmelCase			,   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([raw_speech]						).T]

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

     # Create audio attention mask
     a					=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:
         a					=[
             (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
         ]
         a					=np.array(_UpperCAmelCase						).astype(np.floataa						)

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

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

     a					=BatchFeature(data=_UpperCAmelCase			,   tensor_type=_UpperCAmelCase						)
     return encoded_inputs



"""simple docstring"""







from ...processing_utils import ProcessorMixin






class 				__A    (	_SCREAMING_SNAKE_CASE      ):






 """simple docstring"""




 __lowerCAmelCase			 =  "WhisperFeatureExtractor"
 __lowerCAmelCase			 =  "WhisperTokenizer"


 def __init__(		self			,   __A			,   __A						) ->					Dict:
     super().__init__(__A			,   __A						)
     a					=self.feature_extractor
     a					=False


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A=None			,   __A=None			,   __A=True						) ->					int:
     return self.tokenizer.get_decoder_prompt_ids(task=__A			,   language=__A			,   no_timestamps=__A						)


 def __call__(		self			,   *__A			,   **__A						) ->					Tuple:
     # For backward compatibility
     if self._in_target_context_manager:
         return self.current_processor(*__A			,   **__A						)

     a					=kwargs.pop('''audio'''			,   __A						)
     a					=kwargs.pop('''sampling_rate'''			,   __A						)
     a					=kwargs.pop('''text'''			,   __A						)
     if len(__A						) > 0:
         a					=args[0]
         a					=args[1:]

     if audio is None and text is None:
         raise ValueError('''You need to specify either an `audio` or `text` input to process.'''						)

     if audio is not None:
         a					=self.feature_extractor(__A			,   *__A			,   sampling_rate=__A			,   **__A						)
     if text is not None:
         a					=self.tokenizer(__A			,   **__A						)

     if text is None:
         return inputs

     elif audio is None:
         return encodings
     else:
         a					=encodings['''input_ids''']
         return inputs


 def 				SCREAMING_SNAKE_CASE							(		self			,   *__A			,   **__A						) ->					Optional[Any]:
     return self.tokenizer.batch_decode(*__A			,   **__A						)


 def 				SCREAMING_SNAKE_CASE							(		self			,   *__A			,   **__A						) ->					Union[str, Any]:
     return self.tokenizer.decode(*__A			,   **__A						)





 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A="np"						) ->					Optional[Any]:
     return self.tokenizer.get_prompt_ids(__A			,   return_tensors=__A						)