


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   _lowerCamelCase (					UpperCamelCase			):




       """simple docstring"""
       snake_case      =    ["input_features", "attention_mask"]



       def __init__(      self				,       _SCREAMING_SNAKE_CASE=80				,       _SCREAMING_SNAKE_CASE=1_6000				,       _SCREAMING_SNAKE_CASE=0.0				,       _SCREAMING_SNAKE_CASE=10				,       _SCREAMING_SNAKE_CASE=25				,       _SCREAMING_SNAKE_CASE="hamming_window"				,       _SCREAMING_SNAKE_CASE=3_2_7_6_8.0				,       _SCREAMING_SNAKE_CASE=0.9_7				,       _SCREAMING_SNAKE_CASE=1.0				,       _SCREAMING_SNAKE_CASE=True				,       _SCREAMING_SNAKE_CASE=True				,       _SCREAMING_SNAKE_CASE=False				,       **_SCREAMING_SNAKE_CASE				,       )->Any:




         '''simple docstring'''





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

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

         A_      :				List[str]		     =					win_length * sampling_rate // 1000
         A_      :				Optional[int]		     =					hop_length * sampling_rate // 1000

         A_      :				Union[str, Any]		     =					optimal_fft_length(self.sample_size		)
         A_      :				Dict		     =					(self.n_fft // 2) + 1



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE		)->np.ndarray:




         '''simple docstring'''





         if self.win_function == "hamming_window":
           A_      :				int		     =					window_function(window_length=self.sample_size				,       name=self.win_function				,       periodic=_SCREAMING_SNAKE_CASE		)
         else:
           A_      :				str		     =					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_      :				Tuple		     =					spectrogram(
             one_waveform * self.frame_signal_scale				,       window=_SCREAMING_SNAKE_CASE				,       frame_length=self.sample_size				,       hop_length=self.sample_stride				,       fft_length=self.n_fft				,       center=_SCREAMING_SNAKE_CASE				,       preemphasis=self.preemphasis_coeff				,       mel_filters=_SCREAMING_SNAKE_CASE				,       mel_floor=self.mel_floor				,       log_mel='''log'''				,       )
         return msfc_features.T



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)->Dict:




         '''simple docstring'''





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

         if input_length < x.shape[0]:
           A_      :				Tuple		     =					padding_value

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

         return x



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE = None		)->List[np.ndarray]:




         '''simple docstring'''





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







       def __call__(      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE = False				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = False				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       **_SCREAMING_SNAKE_CASE				,       )->BatchFeature:




         '''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_      :				Any		     =					isinstance(_SCREAMING_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_      :				int		     =					is_batched_numpy or (
             isinstance(_SCREAMING_SNAKE_CASE				,       (list, tuple)		) and (isinstance(raw_speech[0]				,       (np.ndarray, tuple, list)		))
         )

         if is_batched:
           A_      :				List[str]		     =					[np.asarray(_SCREAMING_SNAKE_CASE				,       dtype=np.floataa		) for speech in raw_speech]
         elif not is_batched and not isinstance(_SCREAMING_SNAKE_CASE				,       np.ndarray		):
           A_      :				Dict		     =					np.asarray(_SCREAMING_SNAKE_CASE				,       dtype=np.floataa		)
         elif isinstance(_SCREAMING_SNAKE_CASE				,       np.ndarray		) and raw_speech.dtype is np.dtype(np.floataa		):
           A_      :				List[Any]		     =					raw_speech.astype(np.floataa		)

         # always return batch
         if not is_batched:
           A_      :				Dict		     =					[raw_speech]

         # extract fbank features
         A_      :				Tuple		     =					[self._extract_mfsc_features(_SCREAMING_SNAKE_CASE		) for one_waveform in raw_speech]

         # convert into correct format for padding
         A_      :				Union[str, Any]		     =					BatchFeature({'''input_features''': features}		)

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

         A_      :				Dict		     =					padded_inputs.get('''attention_mask'''		)
         if attention_mask is not None:
           A_      :				Union[str, Any]		     =					[np.asarray(_SCREAMING_SNAKE_CASE				,       dtype=np.intaa		) for array in attention_mask]

         if self.normalize_means or self.normalize_vars:
           A_      :				List[str]		     =					(
               np.array(_SCREAMING_SNAKE_CASE				,       dtype=np.intaa		)
               if self._get_padding_strategies(_SCREAMING_SNAKE_CASE				,       max_length=_SCREAMING_SNAKE_CASE		) is not PaddingStrategy.DO_NOT_PAD
               and padding
               else None
           )
           A_      :				List[Any]		     =					self.normalize(
               padded_inputs['''input_features''']				,       attention_mask=_SCREAMING_SNAKE_CASE		)

         if return_tensors is not None:
           A_      :				Optional[int]		     =					padded_inputs.convert_to_tensors(_SCREAMING_SNAKE_CASE		)

         return padded_inputs








import inspect
import warnings
from typing import Any, Dict, Optional, Union

from packaging import version


def   _SCREAMING_SNAKE_CASE	( *SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE = None ,			SCREAMING_SNAKE_CASE=True ,			SCREAMING_SNAKE_CASE=2       ):
  from .. import __version__

  A_      :				Union[str, Any]		     =					take_from
  A_      :				Optional[Any]		     =					()
  if not isinstance(args[0] ,			SCREAMING_SNAKE_CASE       ):
    A_      :				List[str]		     =					(args,)

  for attribute, version_name, message in args:
    if version.parse(version.parse(SCREAMING_SNAKE_CASE       ).base_version       ) >= version.parse(SCREAMING_SNAKE_CASE       ):
      raise ValueError(
          f'''The deprecation tuple {(attribute, version_name, message)} should be removed since diffusers\''''
          f''' version {__version__} is >= {version_name}'''       )

    A_      :				List[Any]		     =					None
    if isinstance(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ) and attribute in deprecated_kwargs:
      values += (deprecated_kwargs.pop(SCREAMING_SNAKE_CASE       ),)
      A_      :				Optional[Any]		     =					f'''The `{attribute}` argument is deprecated and will be removed in version {version_name}.'''
    elif hasattr(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ):
      values += (getattr(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ),)
      A_      :				int		     =					f'''The `{attribute}` attribute is deprecated and will be removed in version {version_name}.'''
    elif deprecated_kwargs is None:
      A_      :				List[Any]		     =					f'''`{attribute}` is deprecated and will be removed in version {version_name}.'''

    if warning is not None:
      A_      :				Union[str, Any]		     =					warning + ''' ''' if standard_warn else ''''''
      warnings.warn(warning + message ,			SCREAMING_SNAKE_CASE ,			stacklevel=SCREAMING_SNAKE_CASE       )

  if isinstance(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ) and len(SCREAMING_SNAKE_CASE       ) > 0:
    A_      :				Dict		     =					inspect.getouterframes(inspect.currentframe()       )[1]
    A_      :				Optional[int]		     =					call_frame.filename
    A_      :				Optional[int]		     =					call_frame.lineno
    A_      :				str		     =					call_frame.function
    A_     ,       A_      :				List[str]		     =					next(iter(deprecated_kwargs.items()       )       )
    raise TypeError(f'''{function} in {filename} line {line_number-1} got an unexpected keyword argument `{key}`'''       )

  if len(SCREAMING_SNAKE_CASE       ) == 0:
    return
  elif len(SCREAMING_SNAKE_CASE       ) == 1:
    return values[0]
  return values




