

"""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 TensorType, logging


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






class     _UpperCAmelCase			(		snake_case__):
   __a :       Optional[Any]									=							["""input_values""", """attention_mask"""]
   def __init__(		self      ,		_A=1      ,		_A=1_60_00      ,		_A=1_28      ,		_A=10_24      ,		_A=0.0      ,		_A=True      ,		_A=-4.2677393      ,		_A=4.5689974      ,		_A=False      ,		**_A      ,		)    ->    str:






    '''simple docstring'''







    super().__init__(feature_size=__snake_case      ,		sampling_rate=__snake_case      ,		padding_value=__snake_case      ,		**__snake_case     )
    _UpperCAmelCase    :							Optional[int]									=					num_mel_bins
    _UpperCAmelCase    :							Dict									=					max_length
    _UpperCAmelCase    :							int									=					do_normalize
    _UpperCAmelCase    :							int									=					mean
    _UpperCAmelCase    :							Tuple									=					std
    _UpperCAmelCase    :							List[Any]									=					return_attention_mask
   def 							__snake_case (		self      ,		_A      ,		_A      ,		)    ->    List[str]:






    '''simple docstring'''







    _UpperCAmelCase    :							List[Any]									=					torch.from_numpy(__snake_case     ).unsqueeze(0     )
    _UpperCAmelCase    :							Dict									=					ta_kaldi.fbank(
        __snake_case      ,		htk_compat=__snake_case      ,		sample_frequency=self.sampling_rate      ,		use_energy=__snake_case      ,		window_type="""hanning"""      ,		num_mel_bins=self.num_mel_bins      ,		dither=0.0      ,		frame_shift=10      ,		)

    _UpperCAmelCase    :							List[Any]									=					fbank.shape[0]
    _UpperCAmelCase    :							Optional[Any]									=					max_length - n_frames

    # pad or truncate, depending on difference
    if difference > 0:
     _UpperCAmelCase    :							str									=					torch.nn.ZeroPadad((0, 0, 0, difference)     )
     _UpperCAmelCase    :							List[Any]									=					pad_module(__snake_case     )
    elif difference < 0:
     _UpperCAmelCase    :							int									=					fbank[0:max_length, :]

    _UpperCAmelCase    :							Optional[int]									=					fbank.numpy()

    return fbank
   def 							__snake_case (		self      ,		_A     )    ->    Tuple:






    '''simple docstring'''







    return (input_values - (self.mean)) / (self.std * 2)
   def __call__(		self      ,		_A      ,		_A = None      ,		_A = None      ,		**_A      ,		)    ->    Union[str, 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."""     )

    _UpperCAmelCase    :							Optional[Any]									=					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}'''     )
    _UpperCAmelCase    :							str									=					is_batched_numpy or (
        isinstance(__snake_case      ,		(list, tuple)     ) and (isinstance(raw_speech[0]      ,		(np.ndarray, tuple, list)     ))
    )

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

    # always return batch
    if not is_batched:
     _UpperCAmelCase    :							Optional[int]									=					[raw_speech]

    # extract fbank features and pad/truncate to max_length
    _UpperCAmelCase    :							List[Any]									=					[self._extract_fbank_features(__snake_case      ,		max_length=self.max_length     ) for waveform in raw_speech]

    # convert into BatchFeature
    _UpperCAmelCase    :							Optional[Any]									=					BatchFeature({"""input_values""": features}     )

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

    # normalization
    if self.do_normalize:
     _UpperCAmelCase    :							str									=					[self.normalize(__snake_case     ) for feature in input_values]

    if return_tensors is not None:
     _UpperCAmelCase    :							str									=					padded_inputs.convert_to_tensors(__snake_case     )

    return padded_inputs








"""simple docstring"""






import safs






class     _UpperCAmelCase			(		safs.SaFileSystem):
			pass





