



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


snake_case   					=				logging.get_logger(__name__)



class     UpperCAmelCase    (	__SCREAMING_SNAKE_CASE  ):
   A__		:	Any           =  ['''input_features''', '''attention_mask''']



   def __init__(       self    :	List[Any]							,       __lowerCamelCase    :	str=8_0							,       __lowerCamelCase    :	str=1_6_0_0_0							,       __lowerCamelCase    :	List[str]=8_0							,       __lowerCamelCase    :	Dict=0.0							,       __lowerCamelCase    :	int=True							,       __lowerCamelCase    :	Tuple=True							,       __lowerCamelCase    :	Tuple=True							,       **__lowerCamelCase    :	List[str]							,       ):



     """simple docstring"""
     super().__init__(feature_size=__lowerCamelCase							,       sampling_rate=__lowerCamelCase							,       padding_value=__lowerCamelCase							,       **__lowerCamelCase				)
     _snake_case   	=      num_mel_bins
     _snake_case   	=      do_ceptral_normalize
     _snake_case   	=      normalize_means
     _snake_case   	=      normalize_vars
     _snake_case   	=      True



   def    __UpperCAmelCase    (       self    :	Optional[int]							,       __lowerCamelCase    :	np.ndarray							,       ):



     """simple docstring"""
     _snake_case   	=      waveform * (2**1_5)  # Kaldi compliance: 16-bit signed integers
     _snake_case   	=      torch.from_numpy(__lowerCamelCase				).unsqueeze(0				)
     _snake_case   	=      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"""
     # make sure we normalize float32 arrays
     if normalize_means:
       _snake_case   	=      x[:input_length].mean(axis=0				)
       _snake_case   	=      np.subtract(__lowerCamelCase							,       __lowerCamelCase				)
     if normalize_vars:
       _snake_case   	=      x[:input_length].std(axis=0				)
       _snake_case   	=      np.divide(__lowerCamelCase							,       __lowerCamelCase				)

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

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

     return x



   def    __UpperCAmelCase    (       self    :	Tuple							,       __lowerCamelCase    :	List[np.ndarray]							,       __lowerCamelCase    :	Optional[np.ndarray] = None				):



     """simple docstring"""
     _snake_case   	=      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    :	int							,       __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.'''				)

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

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

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

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

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

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

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

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

     if return_tensors is not None:
       _snake_case   	=      padded_inputs.convert_to_tensors(__lowerCamelCase				)

     return padded_inputs













def   __lowercase       (			snake_case			):







				"""simple docstring"""






				return "".join([hex(snake_case			)[2:].zfill(2			).upper() for byte in list(snake_case			)]			)

def   __lowercase       (			snake_case			):







				"""simple docstring"""






				if (len(snake_case			) % 2) != 0:
								raise ValueError(
								    '''Base16 encoded data is invalid:
Data does not have an even number of hex digits.'''			)
				# Check the character set - the standard base16 alphabet
				# is uppercase according to RFC3548 section 6
				if not set(snake_case			) <= set('''0123456789ABCDEF'''			):
								raise ValueError(
								    '''Base16 encoded data is invalid:
Data is not uppercase hex or it contains invalid characters.'''			)
				# For every two hexadecimal digits (= a byte), turn it into an integer.
				# Then, string the result together into bytes, and return it.
				return bytes(int(data[i] + data[i + 1],	1_6			) for i in range(0,	len(snake_case			),	2			)			)


if __name__ == "__main__":
					import doctest

					doctest.testmod()






