




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


_lowerCamelCase	:		Tuple							       =   logging.get_logger(__name__)


class 				UpperCamelCase_    ( UpperCAmelCase__     ):

    '''simple docstring'''






    UpperCAmelCase__							  =      ["""input_features""", """attention_mask"""]


    def __init__(			self      :	str					, UpperCAmelCase__      :	int=80					, UpperCAmelCase__      :	Any=16_000					, UpperCAmelCase__      :	Tuple=0.0					, UpperCAmelCase__      :	int=10					, UpperCAmelCase__      :	List[Any]=25					, UpperCAmelCase__      :	Optional[Any]="hamming_window"					, UpperCAmelCase__      :	List[Any]=32_768.0					, UpperCAmelCase__      :	Dict=0.97					, UpperCAmelCase__      :	Optional[Any]=1.0					, UpperCAmelCase__      :	Tuple=True					, UpperCAmelCase__      :	int=True					, UpperCAmelCase__      :	List[Any]=False					, **UpperCAmelCase__      :	str					, )						->Optional[int]:





           '''simple docstring'''

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

           A__       =    feature_size
           A__       =    sampling_rate
           A__       =    padding_value
           A__       =    hop_length
           A__       =    win_length
           A__       =    frame_signal_scale
           A__       =    preemphasis_coeff
           A__       =    mel_floor
           A__       =    normalize_means
           A__       =    normalize_vars
           A__       =    win_function
           A__       =    return_attention_mask

           A__       =    win_length * sampling_rate // 1_000
           A__       =    hop_length * sampling_rate // 1_000

           A__       =    optimal_fft_length(self.sample_size)
           A__       =    (self.n_fft // 2) + 1


    def        SCREAMING_SNAKE_CASE							(			self      :	Any					, UpperCAmelCase__      :	Optional[Any])						->Dict:





           '''simple docstring'''

           if self.win_function == "hamming_window":
                  A__       =    window_function(window_length=self.sample_size					, name=self.win_function					, periodic=snake_case_)
           else:
                  A__       =    window_function(window_length=self.sample_size					, name=self.win_function)

           A__       =    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__       =    spectrogram(
               one_waveform * self.frame_signal_scale					, window=snake_case_					, frame_length=self.sample_size					, hop_length=self.sample_stride					, fft_length=self.n_fft					, center=snake_case_					, preemphasis=self.preemphasis_coeff					, mel_filters=snake_case_					, mel_floor=self.mel_floor					, log_mel='''log'''					, )
           return msfc_features.T


    def        SCREAMING_SNAKE_CASE							(			self      :	str					, UpperCAmelCase__      :	List[str]					, UpperCAmelCase__      :	Optional[Any]					, UpperCAmelCase__      :	Tuple)						->Dict:





           '''simple docstring'''

           if self.normalize_means:
                  A__       =    x[:input_length].mean(axis=0)
                  A__       =    np.subtract(snake_case_					, snake_case_)
           if self.normalize_vars:
                  A__       =    x[:input_length].std(axis=0)
                  A__       =    np.divide(snake_case_					, snake_case_)

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

           # make sure array is in float32
           A__       =    x.astype(np.floataa)

           return x


    def        SCREAMING_SNAKE_CASE							(			self      :	Optional[int]					, UpperCAmelCase__      :	Union[str, Any]					, UpperCAmelCase__      :	Dict = None)						->int:





           '''simple docstring'''

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





    def __call__(			self      :	Optional[Any]					, UpperCAmelCase__      :	Any					, UpperCAmelCase__      :	List[Any] = False					, UpperCAmelCase__      :	Any = None					, UpperCAmelCase__      :	Optional[int] = False					, UpperCAmelCase__      :	Tuple = None					, UpperCAmelCase__      :	List[str] = None					, UpperCAmelCase__      :	Dict = None					, UpperCAmelCase__      :	Optional[int] = None					, **UpperCAmelCase__      :	List[Any]					, )						->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__       =    isinstance(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__       =    is_batched_numpy or (
               isinstance(snake_case_					, (list, tuple)) and (isinstance(raw_speech[0]					, (np.ndarray, tuple, list)))
           )

           if is_batched:
                  A__       =    [np.asarray(snake_case_					, dtype=np.floataa) for speech in raw_speech]
           elif not is_batched and not isinstance(snake_case_					, np.ndarray):
                  A__       =    np.asarray(snake_case_					, dtype=np.floataa)
           elif isinstance(snake_case_					, 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__       =    [raw_speech]

           # extract fbank features
           A__       =    [self._extract_mfsc_features(snake_case_) for one_waveform in raw_speech]

           # convert into correct format for padding
           A__       =    BatchFeature({'''input_features''': features})

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

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

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

           if return_tensors is not None:
                  A__       =    padded_inputs.convert_to_tensors(snake_case_)

           return padded_inputs









"""simple docstring"""




from transformers import BertTokenizerFast

from .custom_tokenization import CustomTokenizer

class  _UpperCAmelCase  (   UpperCAmelCase__		):



      '''simple docstring'''
      lowercase_      : List[str]	    =   CustomTokenizer
      pass