
"""simple docstring"""


from collections import namedtuple

_lowercase							      = namedtuple('''from_to''', '''from_ to''')

_lowercase							      = {
    '''cubicmeter''': from_to(1, 1),
    '''litre''': from_to(0.001, 10_00),
    '''kilolitre''': from_to(1, 1),
    '''gallon''': from_to(0.00_454, 264.172),
    '''cubicyard''': from_to(0.76_455, 1.30_795),
    '''cubicfoot''': from_to(0.028, 35.3_147),
    '''cup''': from_to(0.000_236_588, 4_226.75),
}






def 					_snake_case (						snake_case__      :   float		,					snake_case__      :   str		,					snake_case__      :   str   ):
	if from_type not in METRIC_CONVERSION:
		raise ValueError(
		    F'Invalid \'from_type\' value: {from_type!r}  Supported values are:\n'
		    + ', '.join(snake_case__   )   )
	if to_type not in METRIC_CONVERSION:
		raise ValueError(
		    F'Invalid \'to_type\' value: {to_type!r}.  Supported values are:\n'
		    + ', '.join(snake_case__   )   )
	return value * METRIC_CONVERSION[from_type].from_ * METRIC_CONVERSION[to_type].to


if __name__ == "__main__":
		import doctest

		doctest.testmod()







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


lowercase_				     =     logging.get_logger(__name__)


class 					snake_case		(							_lowerCAmelCase						):


   '''simple docstring'''

   A_						:	int						 = ["input_features", "attention_mask"]


   def __init__(	self :     Optional[Any],       _lowerCamelCase :     Union[str, Any]=80,       _lowerCamelCase :     int=1_60_00,       _lowerCamelCase :     Any=80,       _lowerCamelCase :     List[str]=0.0,       _lowerCamelCase :     int=True,       _lowerCamelCase :     Optional[Any]=True,       _lowerCamelCase :     Optional[int]=True,       **_lowerCamelCase :     List[str],       ):

     '''simple docstring'''
     super().__init__(feature_size=_lowerCamelCase,       sampling_rate=_lowerCamelCase,       padding_value=_lowerCamelCase,       **_lowerCamelCase  )
     __A									= num_mel_bins
     __A									= do_ceptral_normalize
     __A									= normalize_means
     __A									= normalize_vars
     __A									= True


   def 				_SCREAMING_SNAKE_CASE				(	self :     Dict,       _lowerCamelCase :     np.ndarray,       ):

     '''simple docstring'''
     __A									= waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
     __A									= torch.from_numpy(_lowerCamelCase  ).unsqueeze(0  )
     __A									= ta_kaldi.fbank(_lowerCamelCase,       num_mel_bins=self.num_mel_bins,       sample_frequency=self.sampling_rate  )
     return features.numpy()


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

     '''simple docstring'''
     # make sure we normalize float32 arrays
     if normalize_means:
       __A									= x[:input_length].mean(axis=0  )
       __A									= np.subtract(_lowerCamelCase,       _lowerCamelCase  )
     if normalize_vars:
       __A									= x[:input_length].std(axis=0  )
       __A									= np.divide(_lowerCamelCase,       _lowerCamelCase  )

     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 :     str,       _lowerCamelCase :     List[np.ndarray],       _lowerCamelCase :     Optional[np.ndarray] = None  ):

     '''simple docstring'''
     __A									= 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 :     Optional[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 :     Optional[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									= 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}'  )
     __A									= is_batched_numpy or (
         isinstance(_lowerCamelCase,       (list, tuple)  ) and (isinstance(raw_speech[0],       (np.ndarray, tuple, list)  ))
     )

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

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

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

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

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

     if return_tensors is not None:
       __A									= padded_inputs.convert_to_tensors(_lowerCamelCase  )

     return padded_inputs




