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


A__   :		Union[str, Any] 		=logging.get_logger(__name__)







class  __A				( _SCREAMING_SNAKE_CASE			):
 lowerCamelCase				       =['''input_features''', '''attention_mask''']
 def __init__(   self			:    Optional[Any]				,   lowerCamelCase			:    Any=80				,   lowerCamelCase			:    Dict=1_60_00				,   lowerCamelCase			:    Optional[int]=0.0				,   lowerCamelCase			:    Tuple=10				,   lowerCamelCase			:    Union[str, Any]=25				,   lowerCamelCase			:    int="hamming_window"				,   lowerCamelCase			:    Dict=3_27_68.0				,   lowerCamelCase			:    Any=0.97				,   lowerCamelCase			:    List[str]=1.0				,   lowerCamelCase			:    str=True				,   lowerCamelCase			:    Any=True				,   lowerCamelCase			:    Tuple=False				,   **lowerCamelCase			:    Optional[int]				,   ):



   """simple docstring"""





   super().__init__(feature_size=lowerCamelCase				,   sampling_rate=lowerCamelCase				,   padding_value=lowerCamelCase				,   **lowerCamelCase       )

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

   __A		:     Tuple												=		win_length * sampling_rate // 10_00
   __A		:     Any												=		hop_length * sampling_rate // 10_00

   __A		:     Dict												=		optimal_fft_length(self.sample_size       )
   __A		:     str												=		(self.n_fft // 2) + 1
 def 		lowercase_(   self			:    List[Any]				,   lowerCamelCase			:    np.array       ):



   """simple docstring"""





   if self.win_function == "hamming_window":
     __A		:     Optional[int]												=		window_function(window_length=self.sample_size				,   name=self.win_function				,   periodic=lowerCamelCase       )
   else:
     __A		:     int												=		window_function(window_length=self.sample_size				,   name=self.win_function       )

   __A		:     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				,   )

   __A		:     Union[str, Any]												=		spectrogram(
       one_waveform * self.frame_signal_scale				,   window=lowerCamelCase				,   frame_length=self.sample_size				,   hop_length=self.sample_stride				,   fft_length=self.n_fft				,   center=lowerCamelCase				,   preemphasis=self.preemphasis_coeff				,   mel_filters=lowerCamelCase				,   mel_floor=self.mel_floor				,   log_mel="""log"""				,   )
   return msfc_features.T
 def 		lowercase_(   self			:    Optional[Any]				,   lowerCamelCase			:    int				,   lowerCamelCase			:    Tuple				,   lowerCamelCase			:    Optional[int]       ):



   """simple docstring"""





   if self.normalize_means:
     __A		:     List[str]												=		x[:input_length].mean(axis=0       )
     __A		:     Union[str, Any]												=		np.subtract(lowerCamelCase				,   lowerCamelCase       )
   if self.normalize_vars:
     __A		:     List[Any]												=		x[:input_length].std(axis=0       )
     __A		:     int												=		np.divide(lowerCamelCase				,   lowerCamelCase       )

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

   # make sure array is in float32
   __A		:     Dict												=		x.astype(np.floataa       )

   return x
 def 		lowercase_(   self			:    List[str]				,   lowerCamelCase			:    List[np.ndarray]				,   lowerCamelCase			:    Optional[np.ndarray] = None       ):



   """simple docstring"""





   __A		:     str												=		attention_mask.sum(-1       ) if attention_mask is not None else [x.shape[0] for x in input_features]
   return [self._normalize_one(lowerCamelCase				,   lowerCamelCase				,   self.padding_value       ) for x, n in zip(lowerCamelCase				,   lowerCamelCase       )]






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



   """simple docstring"""





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

   __A		:     Union[str, 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}"       )
   __A		:     Union[str, Any]												=		is_batched_numpy or (
       isinstance(lowerCamelCase				,   (list, tuple)       ) and (isinstance(raw_speech[0]				,   (np.ndarray, tuple, list)       ))
   )

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

   # always return batch
   if not is_batched:
     __A		:     int												=		[raw_speech]

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

   # convert into correct format for padding
   __A		:     Tuple												=		BatchFeature({"""input_features""": features}       )

   __A		:     Tuple												=		self.pad(
       lowerCamelCase				,   padding=lowerCamelCase				,   max_length=lowerCamelCase				,   truncation=lowerCamelCase				,   pad_to_multiple_of=lowerCamelCase				,   return_attention_mask=lowerCamelCase				,   **lowerCamelCase				,   )
   # make sure list is in array format
   __A		:     Any												=		padded_inputs.get("""input_features"""       )
   if isinstance(input_features[0]				,   lowerCamelCase       ):
     __A		:     List[Any]												=		[np.asarray(lowerCamelCase				,   dtype=np.floataa       ) for feature in input_features]

   __A		:     List[str]												=		padded_inputs.get("""attention_mask"""       )
   if attention_mask is not None:
     __A		:     int												=		[np.asarray(lowerCamelCase				,   dtype=np.intaa       ) for array in attention_mask]

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

   if return_tensors is not None:
     __A		:     Optional[Any]												=		padded_inputs.convert_to_tensors(lowerCamelCase       )

   return padded_inputs





'''simple docstring'''
from packaging import version

from .import_utils import is_accelerate_available


if is_accelerate_available():
     import accelerate







def        A_     (						__SCREAMING_SNAKE_CASE						:  Dict )				->       Optional[int]:





  """simple docstring"""

  if not is_accelerate_available():
    return method
  __A		:     int												=		version.parse(accelerate.__version__ ).base_version
  if version.parse(__SCREAMING_SNAKE_CASE ) < version.parse("""0.17.0""" ):
    return method

  def wrapper(self						:  str			,     *__SCREAMING_SNAKE_CASE						:  Tuple			,     **__SCREAMING_SNAKE_CASE						:  Tuple ):
    if hasattr(self			,     """_hf_hook""" ) and hasattr(self._hf_hook			,     """pre_forward""" ):
      self._hf_hook.pre_forward(self )
    return method(self			,     *__SCREAMING_SNAKE_CASE			,     **__SCREAMING_SNAKE_CASE )

  return wrapper



