





"""simple docstring"""





from typing import 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 ....file_utils import PaddingStrategy, TensorType
from ....utils import logging


_UpperCamelCase					      =   logging.get_logger(__name__)
class 		SCREAMING_SNAKE_CASE_				(     snake_case__ ):
				"""simple docstring"""
				__snake_case :							Optional[Any]									=		["""input_features""", """attention_mask"""]




				def __init__(  self	:Any    ,     __lowercase	:Tuple=80    ,     __lowercase	:int=1_6000    ,     __lowercase	:Dict=0.0    ,     __lowercase	:Optional[Any]=10    ,     __lowercase	:Tuple=25    ,     __lowercase	:Dict="hamming_window"    ,     __lowercase	:Dict=32768.0    ,     __lowercase	:List[Any]=0.97    ,     __lowercase	:int=1.0    ,     __lowercase	:List[str]=True    ,     __lowercase	:List[Any]=True    ,     __lowercase	:Tuple=False    ,     **__lowercase	:Dict    ,     ):
											super().__init__(feature_size=__lowercase    ,     sampling_rate=__lowercase    ,     padding_value=__lowercase    ,     **__lowercase      )

											__lowerCamelCase   :   List[Any]       =feature_size
											__lowerCamelCase   :   List[str]       =sampling_rate
											__lowerCamelCase   :   Dict       =padding_value
											__lowerCamelCase   :   Tuple       =hop_length
											__lowerCamelCase   :   Any       =win_length
											__lowerCamelCase   :   Union[str, Any]       =frame_signal_scale
											__lowerCamelCase   :   Union[str, Any]       =preemphasis_coeff
											__lowerCamelCase   :   Union[str, Any]       =mel_floor
											__lowerCamelCase   :   Optional[Any]       =normalize_means
											__lowerCamelCase   :   Tuple       =normalize_vars
											__lowerCamelCase   :   Dict       =win_function
											__lowerCamelCase   :   int       =return_attention_mask

											__lowerCamelCase   :   Optional[Any]       =win_length * sampling_rate // 1000
											__lowerCamelCase   :   int       =hop_length * sampling_rate // 1000

											__lowerCamelCase   :   Optional[int]       =optimal_fft_length(self.sample_size      )
											__lowerCamelCase   :   Optional[Any]       =(self.n_fft // 2) + 1




				def        __lowercase	(  self	:Any    ,     __lowercase	:np.array      ):
											if self.win_function == "hamming_window":
																		__lowerCamelCase   :   Any       =window_function(window_length=self.sample_size    ,     name=self.win_function    ,     periodic=__lowercase      )
											else:
																		__lowerCamelCase   :   List[Any]       =window_function(window_length=self.sample_size    ,     name=self.win_function      )

											__lowerCamelCase   :   Dict       =mel_filter_bank(
											    num_frequency_bins=self.n_freqs    ,     num_mel_filters=self.feature_size    ,     min_frequency=0.0    ,     max_frequency=self.sampling_rate / 2.0    ,     sampling_rate=self.sampling_rate    ,     )

											__lowerCamelCase   :   Union[str, Any]       =spectrogram(
											    one_waveform * self.frame_signal_scale    ,     window=__lowercase    ,     frame_length=self.sample_size    ,     hop_length=self.sample_stride    ,     fft_length=self.n_fft    ,     center=__lowercase    ,     preemphasis=self.preemphasis_coeff    ,     mel_filters=__lowercase    ,     mel_floor=self.mel_floor    ,     log_mel='''log'''    ,     )
											return msfc_features.T




				def        __lowercase	(  self	:str    ,     __lowercase	:Dict    ,     __lowercase	:Any    ,     __lowercase	:Optional[int]      ):
											# make sure we normalize float32 arrays
											if self.normalize_means:
																		__lowerCamelCase   :   List[str]       =x[:input_length].mean(axis=0      )
																		__lowerCamelCase   :   Optional[int]       =np.subtract(__lowercase    ,     __lowercase      )
											if self.normalize_vars:
																		__lowerCamelCase   :   Optional[Any]       =x[:input_length].std(axis=0      )
																		__lowerCamelCase   :   Any       =np.divide(__lowercase    ,     __lowercase      )

											if input_length < x.shape[0]:
																		__lowerCamelCase   :   int       =padding_value

											# make sure array is in float32
											__lowerCamelCase   :   List[Any]       =x.astype(np.floataa      )

											return x




				def        __lowercase	(  self	:Optional[int]    ,     __lowercase	:List[np.ndarray]    ,     __lowercase	:Optional[np.ndarray] = None      ):
											__lowerCamelCase   :   Optional[Any]       =attention_mask.sum(-1      ) if attention_mask is not None else [x.shape[0] for x in input_features]
											return [self._normalize_one(__lowercase    ,     __lowercase    ,     self.padding_value      ) for x, n in zip(__lowercase    ,     __lowercase      )]




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

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

											__lowerCamelCase   :   Dict       =isinstance(__lowercase    ,     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}'      )
											__lowerCamelCase   :   Optional[Any]       =is_batched_numpy or (
											    isinstance(__lowercase    ,     (list, tuple)      ) and (isinstance(raw_speech[0]    ,     (np.ndarray, tuple, list)      ))
											)

											if is_batched:
																		__lowerCamelCase   :   Dict       =[np.asarray(__lowercase    ,     dtype=np.floataa      ) for speech in raw_speech]
											elif not is_batched and not isinstance(__lowercase    ,     np.ndarray      ):
																		__lowerCamelCase   :   str       =np.asarray(__lowercase    ,     dtype=np.floataa      )
											elif isinstance(__lowercase    ,     np.ndarray      ) and raw_speech.dtype is np.dtype(np.floataa      ):
																		__lowerCamelCase   :   Tuple       =raw_speech.astype(np.floataa      )

											# always return batch
											if not is_batched:
																		__lowerCamelCase   :   Optional[Any]       =[raw_speech]

											# extract fbank features
											__lowerCamelCase   :   Optional[int]       =[self._extract_mfsc_features(__lowercase      ) for one_waveform in raw_speech]

											# convert into correct format for padding
											__lowerCamelCase   :   List[Any]       =BatchFeature({'''input_features''': features}      )

											__lowerCamelCase   :   Dict       =self.pad(
											    __lowercase    ,     padding=__lowercase    ,     max_length=__lowercase    ,     truncation=__lowercase    ,     pad_to_multiple_of=__lowercase    ,     return_attention_mask=__lowercase    ,     **__lowercase    ,     )
											# make sure list is in array format
											__lowerCamelCase   :   Optional[int]       =padded_inputs.get('''input_features'''      )
											if isinstance(input_features[0]    ,     __lowercase      ):
																		__lowerCamelCase   :   str       =[np.asarray(__lowercase    ,     dtype=np.floataa      ) for feature in input_features]

											__lowerCamelCase   :   List[str]       =padded_inputs.get('''attention_mask'''      )
											if attention_mask is not None:
																		__lowerCamelCase   :   Dict       =[np.asarray(__lowercase    ,     dtype=np.intaa      ) for array in attention_mask]

											if self.normalize_means or self.normalize_vars:
																		__lowerCamelCase   :   Dict       =(
																		    np.array(__lowercase    ,     dtype=np.intaa      )
																		    if self._get_padding_strategies(__lowercase    ,     max_length=__lowercase      ) is not PaddingStrategy.DO_NOT_PAD
																		    and padding
																		    else None
																		)
																		__lowerCamelCase   :   Optional[int]       =self.normalize(
																		    padded_inputs['''input_features''']    ,     attention_mask=__lowercase      )

											if return_tensors is not None:
																		__lowerCamelCase   :   Any       =padded_inputs.convert_to_tensors(__lowercase      )

											return padded_inputs











"""simple docstring"""





from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, require_tokenizers, slow


if is_tf_available():
							import numpy as np
							import tensorflow as tf

							from transformers import TFXLMRobertaModel
@require_tf
@require_sentencepiece
@require_tokenizers
class 		SCREAMING_SNAKE_CASE_				(     unittest.TestCase ):
				"""simple docstring"""




				@slow
				def        __lowercase	(  self	:Union[str, Any]      ):
											__lowerCamelCase   :   List[Any]       =TFXLMRobertaModel.from_pretrained('''jplu/tf-xlm-roberta-base'''      )

											__lowerCamelCase   :   Dict       ={
											    '''input_ids''': tf.convert_to_tensor([[0, 2646, 1_0269, 83, 9_9942, 2]]    ,     dtype=tf.intaa      ),  # "My dog is cute"
											    '''attention_mask''': tf.convert_to_tensor([[1, 1, 1, 1, 1, 1]]    ,     dtype=tf.intaa      ),
											}

											__lowerCamelCase   :   Optional[int]       =model(__lowercase      )['''last_hidden_state''']
											__lowerCamelCase   :   str       =tf.TensorShape((1, 6, 768)      )
											self.assertEqual(output.shape    ,     __lowercase      )
											# compare the actual values for a slice.
											__lowerCamelCase   :   str       =tf.convert_to_tensor(
											    [
											        [
											            [0.0681762, 0.10894451, 0.06772504],
											            [-0.06423668, 0.02366615, 0.04329344],
											            [-0.06057295, 0.09974135, -0.00070584],
											        ]
											    ]    ,     dtype=tf.floataa    ,     )

											self.assertTrue(np.allclose(output[:, :3, :3].numpy()    ,     expected_slice.numpy()    ,     atol=1e-4      )      )




