






'''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    	=   logging.get_logger(__name__)




class a_    (    UpperCamelCase_							):
   _snake_case        =							["""input_features""", """attention_mask"""]







   def __init__(self					,   __a=8_0					,   __a=1_6_0_0_0					,   __a=0.0					,   __a=1_0					,   __a=2_5					,   __a="hamming_window"					,   __a=3_2_7_6_8.0					,   __a=0.97					,   __a=1.0					,   __a=True					,   __a=True					,   __a=False					,   **__a					,   )			->      Tuple:
      """simple docstring"""






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

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

      __snake_case				:      int     							= win_length * sampling_rate // 1_0_0_0
      __snake_case				:      int     							= hop_length * sampling_rate // 1_0_0_0

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







   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      int:
      """simple docstring"""






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

      __snake_case				:      Tuple     							= 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					,   )

      __snake_case				:      List[str]     							= 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     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a					,   __a)			->      List[Any]:
      """simple docstring"""






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

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

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

      return x







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a = None)			->      Any:
      """simple docstring"""






      __snake_case				:      Tuple     							= 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					,   __a					,   __a = False					,   __a = None					,   __a = False					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   **__a					,   )			->      List[Any]:
      """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.')

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

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

      # always return batch
      if not is_batched:
         __snake_case				:      Union[str, Any]     							= [raw_speech]

      # extract fbank features
      __snake_case				:      List[str]     							= [self._extract_mfsc_features(__lowerCAmelCase) for one_waveform in raw_speech]

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

      __snake_case				:      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
      __snake_case				:      Union[str, Any]     							= padded_inputs.get('input_features')
      if isinstance(input_features[0]					,   __lowerCAmelCase):
         __snake_case				:      Dict     							= [np.asarray(__lowerCAmelCase					,   dtype=np.floataa) for feature in input_features]

      __snake_case				:      Any     							= padded_inputs.get('attention_mask')
      if attention_mask is not None:
         __snake_case				:      Any     							= [np.asarray(__lowerCAmelCase					,   dtype=np.intaa) for array in attention_mask]

      if self.normalize_means or self.normalize_vars:
         __snake_case				:      List[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
         )
         __snake_case				:      List[str]     							= self.normalize(
             padded_inputs['input_features']					,   attention_mask=__lowerCAmelCase)

      if return_tensors is not None:
         __snake_case				:      Dict     							= padded_inputs.convert_to_tensors(__lowerCAmelCase)

      return padded_inputs







'''simple docstring'''






import unittest

from transformers import load_tool
from transformers.utils import is_torch_available


if is_torch_available():
      import torch

from transformers.testing_utils import require_torch

from .test_tools_common import ToolTesterMixin




@require_torch
class a_    (    unittest.TestCase		,      UpperCamelCase_							):







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[Any]:
      """simple docstring"""






      __snake_case				:      List[str]     							= load_tool('text-to-speech')
      self.tool.setup()







   def     SCREAMING_SNAKE_CASE__					(self)			->      Optional[Any]:
      """simple docstring"""






      torch.manual_seed(0)
      __snake_case				:      Dict     							= self.tool('hey')
      __snake_case				:      List[Any]     							= result.to_raw()
      self.assertTrue(
          torch.allclose(
              resulting_tensor[:3]					,   torch.tensor([-0.0_005_966_668_832_115_829, -0.0_003_657_640_190_795_064, -0.00_013_439_502_799_883_485])					,   ))







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[Any]:
      """simple docstring"""






      torch.manual_seed(0)
      __snake_case				:      Any     							= self.tool('hey')
      __snake_case				:      Any     							= result.to_raw()
      self.assertTrue(
          torch.allclose(
              resulting_tensor[:3]					,   torch.tensor([-0.0_005_966_668_832_115_829, -0.0_003_657_640_190_795_064, -0.00_013_439_502_799_883_485])					,   ))