

'''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			:     Union[str, Any]										=      logging.get_logger(__name__)



class lowerCamelCase			(  lowercase_						):



			'''simple docstring'''






			__snake_case        =	['input_features', 'attention_mask']


			def __init__(				self	:						List[Any]		,      lowerCAmelCase_	:						Union[str, Any]=80		,      lowerCAmelCase_	:						int=1_60_00		,      lowerCAmelCase_	:						Union[str, Any]=80		,      lowerCAmelCase_	:						Tuple=0.0		,      lowerCAmelCase_	:						List[Any]=True		,      lowerCAmelCase_	:						Optional[Any]=True		,      lowerCAmelCase_	:						Dict=True		,      **lowerCAmelCase_	:						Tuple		,      )      ->		Optional[int]:





									'''simple docstring'''
									super().__init__(feature_size=lowerCAmelCase_		,      sampling_rate=lowerCAmelCase_		,      padding_value=lowerCAmelCase_		,      **lowerCAmelCase_			)
									A__					:       Dict     =num_mel_bins
									A__					:       Optional[int]     =do_ceptral_normalize
									A__					:       Union[str, Any]     =normalize_means
									A__					:       Union[str, Any]     =normalize_vars
									A__					:       Any     =True


			def lowercase__						(				self	:						List[str]		,      lowerCAmelCase_	:						np.ndarray		,      )      ->		np.ndarray:





									'''simple docstring'''
									A__					:       int     =waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
									A__					:       Optional[int]     =torch.from_numpy(lowerCAmelCase_			).unsqueeze(0			)
									A__					:       List[str]     =ta_kaldi.fbank(lowerCAmelCase_		,      num_mel_bins=self.num_mel_bins		,      sample_frequency=self.sampling_rate			)
									return features.numpy()


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





									'''simple docstring'''
									# make sure we normalize float32 arrays
									if normalize_means:
															A__					:       List[Any]     =x[:input_length].mean(axis=0			)
															A__					:       int     =np.subtract(lowerCAmelCase_		,      lowerCAmelCase_			)
									if normalize_vars:
															A__					:       Union[str, Any]     =x[:input_length].std(axis=0			)
															A__					:       str     =np.divide(lowerCAmelCase_		,      lowerCAmelCase_			)

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

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

									return x


			def lowercase__						(				self	:						Union[str, Any]		,      lowerCAmelCase_	:						List[np.ndarray]		,      lowerCAmelCase_	:						Optional[np.ndarray] = None			)      ->		List[np.ndarray]:





									'''simple docstring'''
									A__					:       Union[str, Any]     =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	:						str		,      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_	:						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."""			)

									A__					:       Dict     =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__					:       Union[str, Any]     =is_batched_numpy or (
									    isinstance(lowerCAmelCase_		,      (list, tuple)			) and (isinstance(raw_speech[0]		,      (np.ndarray, tuple, list)			))
									)

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

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

									# extract fbank features
									A__					:       int     =[self._extract_fbank_features(lowerCAmelCase_			) for waveform in raw_speech]

									# convert into correct format for padding
									A__					:       List[str]     =BatchFeature({"""input_features""": features}			)

									A__					:       int     =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__					:       Any     =padded_inputs.get("""input_features"""			)
									if isinstance(input_features[0]		,      lowerCAmelCase_			):
															A__					:       List[str]     =[np.asarray(lowerCAmelCase_		,      dtype=np.floataa			) for feature in input_features]

									A__					:       Dict     =padded_inputs.get("""attention_mask"""			)
									if attention_mask is not None:
															A__					:       List[str]     =[np.asarray(lowerCAmelCase_		,      dtype=np.intaa			) for array in attention_mask]

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

									if return_tensors is not None:
															A__					:       Union[str, Any]     =padded_inputs.convert_to_tensors(lowerCAmelCase_			)

									return padded_inputs




'''simple docstring'''



from typing import Dict

from .base import GenericTensor, Pipeline



class lowerCamelCase			(  lowercase_						):



			'''simple docstring'''






			def lowercase__						(				self	:						List[str]		,      lowerCAmelCase_	:						Tuple=None		,      lowerCAmelCase_	:						Dict=None		,      lowerCAmelCase_	:						Any=None		,      **lowerCAmelCase_	:						Union[str, Any]			)      ->		Optional[Any]:





									'''simple docstring'''
									if tokenize_kwargs is None:
															A__					:       List[Any]     ={}

									if truncation is not None:
															if "truncation" in tokenize_kwargs:
																					raise ValueError(
																					    """truncation parameter defined twice (given as keyword argument as well as in tokenize_kwargs)"""			)
															A__					:       str     =truncation

									A__					:       Optional[int]     =tokenize_kwargs

									A__					:       List[Any]     ={}
									if return_tensors is not None:
															A__					:       Any     =return_tensors

									return preprocess_params, {}, postprocess_params


			def lowercase__						(				self	:						int		,      lowerCAmelCase_	:						Union[str, Any]		,      **lowerCAmelCase_	:						Dict			)      ->		Dict[str, GenericTensor]:





									'''simple docstring'''
									A__					:       List[str]     =self.framework
									A__					:       Union[str, Any]     =self.tokenizer(lowerCAmelCase_		,      return_tensors=lowerCAmelCase_		,      **lowerCAmelCase_			)
									return model_inputs


			def lowercase__						(				self	:						Optional[int]		,      lowerCAmelCase_	:						Dict			)      ->		Optional[Any]:





									'''simple docstring'''
									A__					:       Union[str, Any]     =self.model(**lowerCAmelCase_			)
									return model_outputs


			def lowercase__						(				self	:						Tuple		,      lowerCAmelCase_	:						Tuple		,      lowerCAmelCase_	:						Any=False			)      ->		List[Any]:





									'''simple docstring'''
									# [0] is the first available tensor, logits or last_hidden_state.
									if return_tensors:
															return model_outputs[0]
									if self.framework == "pt":
															return model_outputs[0].tolist()
									elif self.framework == "tf":
															return model_outputs[0].numpy().tolist()



			def __call__(				self	:						int		,      *lowerCAmelCase_	:						Any		,      **lowerCAmelCase_	:						Union[str, Any]			)      ->		List[str]:





									'''simple docstring'''
									return super().__call__(*lowerCAmelCase_		,      **lowerCAmelCase_			)

