


'''simple docstring'''


from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


__magic_name__			=      logging.get_logger(__name__)



class 	__lowerCAmelCase		(							__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							["""input_features""", """attention_mask"""]






  def __init__(							self :       Dict	,_a :       List[Any]=80	,_a :       Tuple=16000	,_a :       List[str]=80	,_a :       int=0.0	,_a :       str=True	,_a :       List[Any]=True	,_a :       Any=True	,**_a :       int	,):

        '''simple docstring'''




        super().__init__(feature_size=_a	,sampling_rate=_a	,padding_value=_a	,**_a )
        A_						:  Optional[int]          =				num_mel_bins
        A_						:  Union[str, Any]          =				do_ceptral_normalize
        A_						:  int          =				normalize_means
        A_						:  Tuple          =				normalize_vars
        A_						:  Tuple          =				True






  def        _a							(							self :       List[str]	,_a :       np.ndarray	,):

        '''simple docstring'''




        A_						:  str          =				waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
        A_						:  Union[str, Any]          =				torch.from_numpy(_a ).unsqueeze(0 )
        A_						:  Optional[int]          =				ta_kaldi.fbank(_a	,num_mel_bins=self.num_mel_bins	,sample_frequency=self.sampling_rate )
        return features.numpy()






  @staticmethod
  def        _a							(							_a :       np.ndarray	,_a :       int	,_a :       Optional[bool] = True	,_a :       Optional[bool] = True	,_a :       float = 0.0	,):

        '''simple docstring'''




        if normalize_means:
              A_						:  int          =				x[:input_length].mean(axis=0 )
              A_						:  Dict          =				np.subtract(_a	,_a )
        if normalize_vars:
              A_						:  Dict          =				x[:input_length].std(axis=0 )
              A_						:  List[Any]          =				np.divide(_a	,_a )

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

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

        return x






  def        _a							(							self :       Dict	,_a :       List[np.ndarray]	,_a :       Optional[np.ndarray] = None ):

        '''simple docstring'''




        A_						:  Dict          =				attention_mask.sum(-1 ) if attention_mask is not None else [x.shape[0] for x in input_features]
        return [
            self.utterance_cmvn(_a	,_a	,self.normalize_means	,self.normalize_vars	,self.padding_value )
            for x, n in zip(_a	,_a )
        ]






  def __call__(							self :       Any	,_a :       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]	,_a :       Union[bool, str, PaddingStrategy] = False	,_a :       Optional[int] = None	,_a :       bool = False	,_a :       Optional[int] = None	,_a :       Optional[Union[str, TensorType]] = None	,_a :       Optional[int] = None	,_a :       Optional[bool] = None	,**_a :       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.""" )

        A_						:  Dict          =				isinstance(_a	,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_						:  Optional[int]          =				is_batched_numpy or (
            isinstance(_a	,(list, tuple) ) and (isinstance(raw_speech[0]	,(np.ndarray, tuple, list) ))
        )

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

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

        # extract fbank features
        A_						:  Tuple          =				[self._extract_fbank_features(_a ) for waveform in raw_speech]

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

        A_						:  Tuple          =				self.pad(
            _a	,padding=_a	,max_length=_a	,truncation=_a	,pad_to_multiple_of=_a	,return_attention_mask=_a	,**_a	,)

        # make sure list is in array format
        A_						:  Optional[int]          =				padded_inputs.get("""input_features""" )
        if isinstance(input_features[0]	,_a ):
              A_						:  int          =				[np.asarray(_a	,dtype=np.floataa ) for feature in input_features]

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

        # Utterance-level cepstral mean and variance normalization
        if self.do_ceptral_normalize:
              A_						:  Tuple          =				(
                  np.array(_a	,dtype=np.intaa )
                  if self._get_padding_strategies(_a	,max_length=_a ) is not PaddingStrategy.DO_NOT_PAD
                  else None
              )
              A_						:  int          =				self.normalize(
                  padded_inputs["""input_features"""]	,attention_mask=_a )

        if return_tensors is not None:
              A_						:  int          =				padded_inputs.convert_to_tensors(_a )

        return padded_inputs




'''simple docstring'''


from ..utils import DummyObject, requires_backends



class 	__lowerCAmelCase		(							metaclass=__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							["""torch""", """torchsde"""]






  def __init__(							self :       Any	,*_a :       Union[str, Any]	,**_a :       Optional[int] ):

        '''simple docstring'''




        requires_backends(self	,["""torch""", """torchsde"""] )






  @classmethod
  def        _a							(							cls :       Optional[int]	,*_a :       List[Any]	,**_a :       Any ):

        '''simple docstring'''




        requires_backends(cls	,["""torch""", """torchsde"""] )






  @classmethod
  def        _a							(							cls :       List[Any]	,*_a :       Tuple	,**_a :       Union[str, Any] ):

        '''simple docstring'''




        requires_backends(cls	,["""torch""", """torchsde"""] )
