





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


__A									=				logging.get_logger(__name__)



class    _SCREAMING_SNAKE_CASE						(	__SCREAMING_SNAKE_CASE    ):







 '''simple docstring'''
 lowercase_		=			["input_features", "attention_mask"]

 def __init__(self				:   List[Any]       , UpperCAmelCase_				:   Dict=80       , UpperCAmelCase_				:   Optional[Any]=16_000       , UpperCAmelCase_				:   str=80       , UpperCAmelCase_				:   List[Any]=0.0       , UpperCAmelCase_				:   Optional[Any]=True       , UpperCAmelCase_				:   Optional[int]=True       , UpperCAmelCase_				:   Optional[int]=True       , **UpperCAmelCase_				:   List[str]       , )       ->Optional[Any]:



   '''simple docstring'''



   super().__init__(feature_size=UpperCAmelCase_       , sampling_rate=UpperCAmelCase_       , padding_value=UpperCAmelCase_       , **UpperCAmelCase_)
   lowerCamelCase__:      Dict			  =num_mel_bins
   lowerCamelCase__:      int			  =do_ceptral_normalize
   lowerCamelCase__:      int			  =normalize_means
   lowerCamelCase__:      Optional[Any]			  =normalize_vars
   lowerCamelCase__:      Optional[int]			  =True

 def 							SCREAMING_SNAKE_CASE_					(self				:   Optional[Any]       , UpperCAmelCase_				:   np.ndarray       , )       ->np.ndarray:



   '''simple docstring'''



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

 @staticmethod
 def 							SCREAMING_SNAKE_CASE_					(UpperCAmelCase_				:   np.ndarray       , UpperCAmelCase_				:   int       , UpperCAmelCase_				:   Optional[bool] = True       , UpperCAmelCase_				:   Optional[bool] = True       , UpperCAmelCase_				:   float = 0.0       , )       ->np.ndarray:



   '''simple docstring'''



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

   if input_length < x.shape[0]:
     lowerCamelCase__:      List[Any]			  =padding_value

   # make sure array is in float32
   lowerCamelCase__:      List[str]			  =x.astype(np.floataa)

   return x

 def 							SCREAMING_SNAKE_CASE_					(self				:   Optional[Any]       , UpperCAmelCase_				:   List[np.ndarray]       , UpperCAmelCase_				:   Optional[np.ndarray] = None)       ->List[np.ndarray]:



   '''simple docstring'''



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







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

   lowerCamelCase__:      Optional[Any]			  =isinstance(UpperCAmelCase_       , 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}""")
   lowerCamelCase__:      int			  =is_batched_numpy or (
       isinstance(UpperCAmelCase_       , (list, tuple)) and (isinstance(raw_speech[0]       , (np.ndarray, tuple, list)))
   )

   if is_batched:
     lowerCamelCase__:      str			  =[np.asarray(UpperCAmelCase_       , dtype=np.floataa) for speech in raw_speech]
   elif not is_batched and not isinstance(UpperCAmelCase_       , np.ndarray):
     lowerCamelCase__:      str			  =np.asarray(UpperCAmelCase_       , dtype=np.floataa)
   elif isinstance(UpperCAmelCase_       , np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
     lowerCamelCase__:      Union[str, Any]			  =raw_speech.astype(np.floataa)

   # always return batch
   if not is_batched:
     lowerCamelCase__:      List[str]			  =[raw_speech]

   # extract fbank features
   lowerCamelCase__:      str			  =[self._extract_fbank_features(UpperCAmelCase_) for waveform in raw_speech]

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

   lowerCamelCase__:      Union[str, Any]			  =self.pad(
       UpperCAmelCase_       , padding=UpperCAmelCase_       , max_length=UpperCAmelCase_       , truncation=UpperCAmelCase_       , pad_to_multiple_of=UpperCAmelCase_       , return_attention_mask=UpperCAmelCase_       , **UpperCAmelCase_       , )

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

   lowerCamelCase__:      Union[str, Any]			  =padded_inputs.get("attention_mask")
   if attention_mask is not None:
     lowerCamelCase__:      Optional[int]			  =[np.asarray(UpperCAmelCase_       , dtype=np.intaa) for array in attention_mask]

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

   if return_tensors is not None:
     lowerCamelCase__:      str			  =padded_inputs.convert_to_tensors(UpperCAmelCase_)

   return padded_inputs








import os


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_doctest_list.py
__A									=				"."


if __name__ == "__main__":
     __A									=				os.path.join(REPO_PATH, "utils/documentation_tests.txt")
     __A									=				[]
     __A									=				[]
     with open(doctest_file_path) as fp:
          for line in fp:
               __A									=				line.strip()
               __A									=				os.path.join(REPO_PATH, line)
               if not (os.path.isfile(path) or os.path.isdir(path)):
                    non_existent_paths.append(line)
               all_paths.append(path)
     if len(non_existent_paths) > 0:
          __A									=				"\n".join(non_existent_paths)
          raise ValueError(f'`utils/documentation_tests.txt` contains non-existent paths:\n{non_existent_paths}')
     if all_paths != sorted(all_paths):
          raise ValueError("Files in `utils/documentation_tests.txt` are not in alphabetical order.")

