
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


UpperCAmelCase          =							logging.get_logger(__name__)



class  lowercase				(   lowercase__		):
			lowercase			     =    ['''input_values''', '''attention_mask''']






			def __init__(self     :  Tuple			,SCREAMING_SNAKE_CASE_     :  Dict=1			,SCREAMING_SNAKE_CASE_     :  str=16_000			,SCREAMING_SNAKE_CASE_     :  Dict=128			,SCREAMING_SNAKE_CASE_     :  List[Any]=1_024			,SCREAMING_SNAKE_CASE_     :  str=0.0			,SCREAMING_SNAKE_CASE_     :  Union[str, Any]=True			,SCREAMING_SNAKE_CASE_     :  Tuple=-4.2_67_73_93			,SCREAMING_SNAKE_CASE_     :  Union[str, Any]=4.5_68_99_74			,SCREAMING_SNAKE_CASE_     :  str=False			,**SCREAMING_SNAKE_CASE_     :  List[str]			,)					->  Union[str, Any]:





				"""simple docstring"""


				super().__init__(feature_size=SCREAMING_SNAKE_CASE_			,sampling_rate=SCREAMING_SNAKE_CASE_			,padding_value=SCREAMING_SNAKE_CASE_			,**SCREAMING_SNAKE_CASE_ )
				lowerCAmelCase				       =     num_mel_bins
				lowerCAmelCase				       =     max_length
				lowerCAmelCase				       =     do_normalize
				lowerCAmelCase				       =     mean
				lowerCAmelCase				       =     std
				lowerCAmelCase				       =     return_attention_mask






			def 				UpperCAmelCase       (self     :  List[str]			,SCREAMING_SNAKE_CASE_     :  np.ndarray			,SCREAMING_SNAKE_CASE_     :  int			,)					->  np.ndarray:





				"""simple docstring"""


				lowerCAmelCase				       =     torch.from_numpy(SCREAMING_SNAKE_CASE_ ).unsqueeze(0 )
				lowerCAmelCase				       =     ta_kaldi.fbank(
				    SCREAMING_SNAKE_CASE_			,htk_compat=SCREAMING_SNAKE_CASE_			,sample_frequency=self.sampling_rate			,use_energy=SCREAMING_SNAKE_CASE_			,window_type='''hanning'''			,num_mel_bins=self.num_mel_bins			,dither=0.0			,frame_shift=10			,)

				lowerCAmelCase				       =     fbank.shape[0]
				lowerCAmelCase				       =     max_length - n_frames

				# pad or truncate, depending on difference
				if difference > 0:
					lowerCAmelCase				       =     torch.nn.ZeroPadad((0, 0, 0, difference) )
					lowerCAmelCase				       =     pad_module(SCREAMING_SNAKE_CASE_ )
				elif difference < 0:
					lowerCAmelCase				       =     fbank[0:max_length, :]

				lowerCAmelCase				       =     fbank.numpy()

				return fbank






			def 				UpperCAmelCase       (self     :  Union[str, Any]			,SCREAMING_SNAKE_CASE_     :  np.ndarray )					->  np.ndarray:





				"""simple docstring"""


				return (input_values - (self.mean)) / (self.std * 2)






			def __call__(self     :  List[Any]			,SCREAMING_SNAKE_CASE_     :  Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]			,SCREAMING_SNAKE_CASE_     :  Optional[int] = None			,SCREAMING_SNAKE_CASE_     :  Optional[Union[str, TensorType]] = None			,**SCREAMING_SNAKE_CASE_     :  int			,)					->  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				       =     isinstance(SCREAMING_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}""" )
				lowerCAmelCase				       =     is_batched_numpy or (
				    isinstance(SCREAMING_SNAKE_CASE_			,(list, tuple) ) and (isinstance(raw_speech[0]			,(np.ndarray, tuple, list) ))
				)

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

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

				# extract fbank features and pad/truncate to max_length
				lowerCAmelCase				       =     [self._extract_fbank_features(SCREAMING_SNAKE_CASE_			,max_length=self.max_length ) for waveform in raw_speech]

				# convert into BatchFeature
				lowerCAmelCase				       =     BatchFeature({'''input_values''': features} )

				# make sure list is in array format
				lowerCAmelCase				       =     padded_inputs.get('''input_values''' )
				if isinstance(input_values[0]			,SCREAMING_SNAKE_CASE_ ):
					lowerCAmelCase				       =     [np.asarray(SCREAMING_SNAKE_CASE_			,dtype=np.floataa ) for feature in input_values]

				# normalization
				if self.do_normalize:
					lowerCAmelCase				       =     [self.normalize(SCREAMING_SNAKE_CASE_ ) for feature in input_values]

				if return_tensors is not None:
					lowerCAmelCase				       =     padded_inputs.convert_to_tensors(SCREAMING_SNAKE_CASE_ )

				return padded_inputs




"""simple docstring"""






import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin



class  lowercase				(   lowercase__		):
			lowercase			     =    '''AutoFeatureExtractor'''
			lowercase			     =    '''Speech2Text2Tokenizer'''






			def __init__(self     :  Tuple			,SCREAMING_SNAKE_CASE_     :  Union[str, Any]			,SCREAMING_SNAKE_CASE_     :  Optional[int] )					->  int:





				"""simple docstring"""


				super().__init__(SCREAMING_SNAKE_CASE_			,SCREAMING_SNAKE_CASE_ )
				lowerCAmelCase				       =     self.feature_extractor
				lowerCAmelCase				       =     False






			def __call__(self     :  Tuple			,*SCREAMING_SNAKE_CASE_     :  Optional[Any]			,**SCREAMING_SNAKE_CASE_     :  Dict )					->  Any:





				"""simple docstring"""


				if self._in_target_context_manager:
					return self.current_processor(*SCREAMING_SNAKE_CASE_			,**SCREAMING_SNAKE_CASE_ )

				if "raw_speech" in kwargs:
					warnings.warn('''Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.''' )
					lowerCAmelCase				       =     kwargs.pop('''raw_speech''' )
				else:
					lowerCAmelCase				       =     kwargs.pop('''audio'''			,SCREAMING_SNAKE_CASE_ )
				lowerCAmelCase				       =     kwargs.pop('''sampling_rate'''			,SCREAMING_SNAKE_CASE_ )
				lowerCAmelCase				       =     kwargs.pop('''text'''			,SCREAMING_SNAKE_CASE_ )
				if len(SCREAMING_SNAKE_CASE_ ) > 0:
					lowerCAmelCase				       =     args[0]
					lowerCAmelCase				       =     args[1:]

				if audio is None and text is None:
					raise ValueError('''You need to specify either an `audio` or `text` input to process.''' )

				if audio is not None:
					lowerCAmelCase				       =     self.feature_extractor(SCREAMING_SNAKE_CASE_			,*SCREAMING_SNAKE_CASE_			,sampling_rate=SCREAMING_SNAKE_CASE_			,**SCREAMING_SNAKE_CASE_ )
				if text is not None:
					lowerCAmelCase				       =     self.tokenizer(SCREAMING_SNAKE_CASE_			,**SCREAMING_SNAKE_CASE_ )

				if text is None:
					return inputs
				elif audio is None:
					return encodings
				else:
					lowerCAmelCase				       =     encodings['''input_ids''']
					return inputs






			def 				UpperCAmelCase       (self     :  Optional[int]			,*SCREAMING_SNAKE_CASE_     :  Any			,**SCREAMING_SNAKE_CASE_     :  List[Any] )					->  str:





				"""simple docstring"""


				return self.tokenizer.batch_decode(*SCREAMING_SNAKE_CASE_			,**SCREAMING_SNAKE_CASE_ )






			def 				UpperCAmelCase       (self     :  Optional[int]			,*SCREAMING_SNAKE_CASE_     :  int			,**SCREAMING_SNAKE_CASE_     :  Any )					->  Dict:





				"""simple docstring"""


				return self.tokenizer.decode(*SCREAMING_SNAKE_CASE_			,**SCREAMING_SNAKE_CASE_ )






			@contextmanager
			def 				UpperCAmelCase       (self     :  Optional[int] )					->  Union[str, Any]:





				"""simple docstring"""


				warnings.warn(
				    '''`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '''
				    '''labels by using the argument `text` of the regular `__call__` method (either in the same call as '''
				    '''your audio inputs, or in a separate call.''' )
				lowerCAmelCase				       =     True
				lowerCAmelCase				       =     self.tokenizer
				yield
				lowerCAmelCase				       =     self.feature_extractor
				lowerCAmelCase				       =     False


