



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


UpperCAmelCase						:		List[Any]			= logging.get_logger(__name__)


class    SCREAMING_SNAKE_CASE__ (							__UpperCAmelCase   ):
    lowercase__      =   ["input_features", "attention_mask"]

    def __init__(				self			:				Union[str, Any] ,  lowerCAmelCase_			:				str=8_0 ,  lowerCAmelCase_			:				List[str]=1_6_0_0_0 ,  lowerCAmelCase_			:				List[Any]=8_0 ,  lowerCAmelCase_			:				str=0.0 ,  lowerCAmelCase_			:				Union[str, Any]=True ,  lowerCAmelCase_			:				Union[str, Any]=True ,  lowerCAmelCase_			:				Optional[Any]=True ,  **lowerCAmelCase_			:				Optional[Any] ,  ):







      """simple docstring"""







      super().__init__(feature_size=lowerCAmelCase_ ,  sampling_rate=lowerCAmelCase_ ,  padding_value=lowerCAmelCase_ ,  **lowerCAmelCase_)
      lowercase_								= num_mel_bins
      lowercase_								= do_ceptral_normalize
      lowercase_								= normalize_means
      lowercase_								= normalize_vars
      lowercase_								= True

    def 					_UpperCAmelCase	(				self			:				str ,  lowerCAmelCase_			:				np.ndarray ,  ):







      """simple docstring"""







      lowercase_								= waveform * (2**1_5)  # Kaldi compliance: 16-bit signed integers
      lowercase_								= torch.from_numpy(lowerCAmelCase_).unsqueeze(0)
      lowercase_								= ta_kaldi.fbank(lowerCAmelCase_ ,  num_mel_bins=self.num_mel_bins ,  sample_frequency=self.sampling_rate)
      return features.numpy()

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







      """simple docstring"""







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

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

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

      return x

    def 					_UpperCAmelCase	(				self			:				Optional[Any] ,  lowerCAmelCase_			:				List[np.ndarray] ,  lowerCAmelCase_			:				Optional[np.ndarray] = None):







      """simple docstring"""







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




    def __call__(				self			:				List[Any] ,  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[Union[str, TensorType]] = None ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				Optional[bool] = None ,  **lowerCAmelCase_			:				Dict ,  ):







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

      lowercase_								= 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}''')
      lowercase_								= is_batched_numpy or (
          isinstance(lowerCAmelCase_ ,  (list, tuple)) and (isinstance(raw_speech[0] ,  (np.ndarray, tuple, list)))
      )

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

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

      # extract fbank features
      lowercase_								= [self._extract_fbank_features(lowerCAmelCase_) for waveform in raw_speech]

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

      lowercase_								= 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
      lowercase_								= padded_inputs.get("""input_features""")
      if isinstance(input_features[0] ,  lowerCAmelCase_):
        lowercase_								= [np.asarray(lowerCAmelCase_ ,  dtype=np.floataa) for feature in input_features]

      lowercase_								= padded_inputs.get("""attention_mask""")
      if attention_mask is not None:
        lowercase_								= [np.asarray(lowerCAmelCase_ ,  dtype=np.intaa) for array in attention_mask]

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

      if return_tensors is not None:
        lowercase_								= padded_inputs.convert_to_tensors(lowerCAmelCase_)

      return padded_inputs






"""simple docstring"""
def 			_SCREAMING_SNAKE_CASE						(__lowerCAmelCase ,  __lowerCAmelCase							) ->						int:







  '''simple docstring'''


  while a != 0:
    lowercase_					,    lowercase_								= b % a, a
  return b


def 			_SCREAMING_SNAKE_CASE						(__lowerCAmelCase ,  __lowerCAmelCase							) ->						int:







  '''simple docstring'''


  if gcd(__lowerCAmelCase ,  __lowerCAmelCase							) != 1:
    lowercase_								= F'''mod inverse of {a!r} and {m!r} does not exist'''
    raise ValueError(__lowerCAmelCase							)
  lowercase_					,    lowercase_					,    lowercase_								= 1, 0, a
  lowercase_					,    lowercase_					,    lowercase_								= 0, 1, m
  while va != 0:
    lowercase_								= ua // va
    lowercase_					,    lowercase_					,    lowercase_					,    lowercase_					,    lowercase_					,    lowercase_								= (ua - q * va), (ua - q * va), (ua - q * va), va, va, va
  return ua % m

