





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


lowerCAmelCase_    =    logging.get_logger(__name__)



class  __lowerCAmelCase	(      lowerCamelCase_					):
			lowerCamelCase_					:					List[Any]	      =				['''input_features''', '''attention_mask''']






			def __init__(self ,							__magic_name__=80 ,							__magic_name__=1_6000 ,							__magic_name__=80 ,							__magic_name__=0.0 ,							__magic_name__=True ,							__magic_name__=True ,							__magic_name__=True ,							**__magic_name__ ,							)					->       Union[str, Any]:


							'''simple docstring'''



							super().__init__(feature_size=__snake_case ,							sampling_rate=__snake_case ,							padding_value=__snake_case ,							**__snake_case   )
							snake_case_						:   Optional[Any]           =  num_mel_bins
							snake_case_						:   Dict           =  do_ceptral_normalize
							snake_case_						:   List[Any]           =  normalize_means
							snake_case_						:   int           =  normalize_vars
							snake_case_						:   List[Any]           =  True






			def       lowerCamelCase  (self ,							__magic_name__ ,							)					->       np.ndarray:


							'''simple docstring'''



							snake_case_						:   Tuple           =  waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
							snake_case_						:   Tuple           =  torch.from_numpy(__snake_case   ).unsqueeze(0   )
							snake_case_						:   Union[str, Any]           =  ta_kaldi.fbank(__snake_case ,							num_mel_bins=self.num_mel_bins ,							sample_frequency=self.sampling_rate   )
							return features.numpy()






			@staticmethod
			def       lowerCamelCase  (__magic_name__ ,							__magic_name__ ,							__magic_name__ = True ,							__magic_name__ = True ,							__magic_name__ = 0.0 ,							)					->       np.ndarray:


							'''simple docstring'''



							if normalize_means:
											snake_case_						:   str           =  x[:input_length].mean(axis=0   )
											snake_case_						:   Dict           =  np.subtract(__snake_case ,							__snake_case   )
							if normalize_vars:
											snake_case_						:   str           =  x[:input_length].std(axis=0   )
											snake_case_						:   Union[str, Any]           =  np.divide(__snake_case ,							__snake_case   )

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

							# make sure array is in float32
							snake_case_						:   str           =  x.astype(np.floataa   )

							return x






			def       lowerCamelCase  (self ,							__magic_name__ ,							__magic_name__ = None   )					->       List[np.ndarray]:


							'''simple docstring'''



							snake_case_						:   Optional[Any]           =  attention_mask.sum(-1   ) if attention_mask is not None else [x.shape[0] for x in input_features]
							return [
							    self.utterance_cmvn(__snake_case ,							__snake_case ,							self.normalize_means ,							self.normalize_vars ,							self.padding_value   )
							    for x, n in zip(__snake_case ,							__snake_case   )
							]






			def __call__(self ,							__magic_name__ ,							__magic_name__ = False ,							__magic_name__ = None ,							__magic_name__ = False ,							__magic_name__ = None ,							__magic_name__ = None ,							__magic_name__ = None ,							__magic_name__ = None ,							**__magic_name__ ,							)					->       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.'''   )

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

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

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

							# extract fbank features
							snake_case_						:   List[str]           =  [self._extract_fbank_features(__snake_case   ) for waveform in raw_speech]

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

							snake_case_						:   List[Any]           =  self.pad(
							    __snake_case ,							padding=__snake_case ,							max_length=__snake_case ,							truncation=__snake_case ,							pad_to_multiple_of=__snake_case ,							return_attention_mask=__snake_case ,							**__snake_case ,							)

							# make sure list is in array format
							snake_case_						:   int           =  padded_inputs.get('''input_features'''   )
							if isinstance(input_features[0] ,							__snake_case   ):
											snake_case_						:   List[str]           =  [np.asarray(__snake_case ,							dtype=np.floataa   ) for feature in input_features]

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

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

							if return_tensors is not None:
											snake_case_						:   Tuple           =  padded_inputs.convert_to_tensors(__snake_case   )

							return padded_inputs







import unittest

import numpy as np

from transformers import AlbertConfig, is_flax_available
from transformers.testing_utils import require_flax, slow

from ...test_modeling_flax_common import FlaxModelTesterMixin, ids_tensor, random_attention_mask


if is_flax_available():
	import jax.numpy as jnp

	from transformers.models.albert.modeling_flax_albert import (
	    FlaxAlbertForMaskedLM,
	    FlaxAlbertForMultipleChoice,
	    FlaxAlbertForPreTraining,
	    FlaxAlbertForQuestionAnswering,
	    FlaxAlbertForSequenceClassification,
	    FlaxAlbertForTokenClassification,
	    FlaxAlbertModel,
	)



class  __lowerCAmelCase	(      unittest.TestCase					):






			def __init__(self ,							__magic_name__ ,							__magic_name__=13 ,							__magic_name__=7 ,							__magic_name__=True ,							__magic_name__=True ,							__magic_name__=True ,							__magic_name__=True ,							__magic_name__=99 ,							__magic_name__=32 ,							__magic_name__=5 ,							__magic_name__=4 ,							__magic_name__=37 ,							__magic_name__="gelu" ,							__magic_name__=0.1 ,							__magic_name__=0.1 ,							__magic_name__=512 ,							__magic_name__=16 ,							__magic_name__=2 ,							__magic_name__=0.02 ,							__magic_name__=4 ,							)					->       Tuple:


							'''simple docstring'''



							snake_case_						:   Union[str, Any]           =  parent
							snake_case_						:   Optional[Any]           =  batch_size
							snake_case_						:   List[Any]           =  seq_length
							snake_case_						:   Tuple           =  is_training
							snake_case_						:   List[str]           =  use_attention_mask
							snake_case_						:   Any           =  use_token_type_ids
							snake_case_						:   Dict           =  use_labels
							snake_case_						:   Optional[Any]           =  vocab_size
							snake_case_						:   Dict           =  hidden_size
							snake_case_						:   List[Any]           =  num_hidden_layers
							snake_case_						:   Union[str, Any]           =  num_attention_heads
							snake_case_						:   Any           =  intermediate_size
							snake_case_						:   Optional[int]           =  hidden_act
							snake_case_						:   Optional[int]           =  hidden_dropout_prob
							snake_case_						:   Optional[Any]           =  attention_probs_dropout_prob
							snake_case_						:   Optional[int]           =  max_position_embeddings
							snake_case_						:   Optional[int]           =  type_vocab_size
							snake_case_						:   List[Any]           =  type_sequence_label_size
							snake_case_						:   Dict           =  initializer_range
							snake_case_						:   Dict           =  num_choices






			def       lowerCamelCase  (self   )					->       Tuple:


							'''simple docstring'''



							snake_case_						:   int           =  ids_tensor([self.batch_size, self.seq_length] ,							self.vocab_size   )

							snake_case_						:   Any           =  None
							if self.use_attention_mask:
											snake_case_						:   List[Any]           =  random_attention_mask([self.batch_size, self.seq_length]   )

							snake_case_						:   List[Any]           =  None
							if self.use_token_type_ids:
											snake_case_						:   Dict           =  ids_tensor([self.batch_size, self.seq_length] ,							self.type_vocab_size   )

							snake_case_						:   List[Any]           =  AlbertConfig(
							    vocab_size=self.vocab_size ,							hidden_size=self.hidden_size ,							num_hidden_layers=self.num_hidden_layers ,							num_attention_heads=self.num_attention_heads ,							intermediate_size=self.intermediate_size ,							hidden_act=self.hidden_act ,							hidden_dropout_prob=self.hidden_dropout_prob ,							attention_probs_dropout_prob=self.attention_probs_dropout_prob ,							max_position_embeddings=self.max_position_embeddings ,							type_vocab_size=self.type_vocab_size ,							is_decoder=__magic_name__ ,							initializer_range=self.initializer_range ,							)

							return config, input_ids, token_type_ids, attention_mask






			def       lowerCamelCase  (self   )					->       List[str]:


							'''simple docstring'''



							snake_case_						:   Tuple           =  self.prepare_config_and_inputs()
							snake_case_      ,							snake_case_      ,							snake_case_      ,							snake_case_						:   Optional[int]           =  config_and_inputs
							snake_case_						:   int           =  {'''input_ids''': input_ids, '''token_type_ids''': token_type_ids, '''attention_mask''': attention_mask}
							return config, inputs_dict




@require_flax
class  __lowerCAmelCase	(      _a,					unittest.TestCase					):
			lowerCamelCase_					:					Optional[int]	      =				(
			    (
			        FlaxAlbertModel,
			        FlaxAlbertForPreTraining,
			        FlaxAlbertForMaskedLM,
			        FlaxAlbertForMultipleChoice,
			        FlaxAlbertForQuestionAnswering,
			        FlaxAlbertForSequenceClassification,
			        FlaxAlbertForTokenClassification,
			        FlaxAlbertForQuestionAnswering,
			    )
			    if is_flax_available()
			    else ()
			)






			def       lowerCamelCase  (self   )					->       List[str]:


							'''simple docstring'''



							snake_case_						:   Optional[Any]           =  FlaxAlbertModelTester(self   )






			@slow
			def       lowerCamelCase  (self   )					->       Tuple:


							'''simple docstring'''



							for model_class_name in self.all_model_classes:
											snake_case_						:   Dict           =  model_class_name.from_pretrained('''albert-base-v2'''   )
											snake_case_						:   Union[str, Any]           =  model(np.ones((1, 1)   )   )
											self.assertIsNotNone(__magic_name__   )




@require_flax
class  __lowerCAmelCase	(      unittest.TestCase					):






			@slow
			def       lowerCamelCase  (self   )					->       Optional[Any]:


							'''simple docstring'''



							snake_case_						:   Optional[Any]           =  FlaxAlbertModel.from_pretrained('''albert-base-v2'''   )
							snake_case_						:   Optional[int]           =  np.array([[0, 345, 232, 328, 740, 140, 1695, 69, 6078, 1588, 2]]   )
							snake_case_						:   Dict           =  np.array([[0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]   )
							snake_case_						:   Union[str, Any]           =  model(__magic_name__ ,							attention_mask=__magic_name__   )[0]
							snake_case_						:   Tuple           =  (1, 11, 768)
							self.assertEqual(output.shape ,							__magic_name__   )
							snake_case_						:   str           =  np.array(
							    [[[-0.6_513, 1.5_035, -0.2_766], [-0.6_515, 1.5_046, -0.2_780], [-0.6_512, 1.5_049, -0.2_784]]]   )

							self.assertTrue(jnp.allclose(output[:, 1:4, 1:4] ,							__magic_name__ ,							atol=1e-4   )   )
