



"""simple docstring"""






import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


lowerCAmelCase__              =							logging.get_logger(__name__)






class 			__snake_case			(      _lowercase):
  snake_case__							:  Tuple         =		["input_features", "is_longer"]




  def __init__(					self	:		Optional[int]		,   __lowerCAmelCase	:		Tuple=6_4		,   __lowerCAmelCase	:		Union[str, Any]=4_8_0_0_0		,   __lowerCAmelCase	:		int=4_8_0		,   __lowerCAmelCase	:		Union[str, Any]=1_0		,   __lowerCAmelCase	:		int=1_0_2_4		,   __lowerCAmelCase	:		Optional[Any]=0.0		,   __lowerCAmelCase	:		int=False		,   __lowerCAmelCase	:		float = 0		,   __lowerCAmelCase	:		float = 1_4_0_0_0		,   __lowerCAmelCase	:		int = None		,   __lowerCAmelCase	:		str = "fusion"		,   __lowerCAmelCase	:		str = "repeatpad"		,   **__lowerCAmelCase	:		List[str]		,   ):






         """simple docstring"""






         super().__init__(
             feature_size=__lowerCAmelCase		,   sampling_rate=__lowerCAmelCase		,   padding_value=__lowerCAmelCase		,   return_attention_mask=__lowerCAmelCase		,   **__lowerCAmelCase		,   )
         _lowerCamelCase		:							str            =					top_db
         _lowerCamelCase		:							Optional[int]            =					truncation
         _lowerCamelCase		:							Tuple            =					padding
         _lowerCamelCase		:							Any            =					fft_window_size
         _lowerCamelCase		:							Tuple            =					(fft_window_size >> 1) + 1
         _lowerCamelCase		:							int            =					hop_length
         _lowerCamelCase		:							List[Any]            =					max_length_s
         _lowerCamelCase		:							Optional[Any]            =					max_length_s * sampling_rate
         _lowerCamelCase		:							List[str]            =					sampling_rate
         _lowerCamelCase		:							Tuple            =					frequency_min
         _lowerCamelCase		:							List[Any]            =					frequency_max
         _lowerCamelCase		:							str            =					mel_filter_bank(
             num_frequency_bins=self.nb_frequency_bins		,   num_mel_filters=__lowerCAmelCase		,   min_frequency=__lowerCAmelCase		,   max_frequency=__lowerCAmelCase		,   sampling_rate=__lowerCAmelCase		,   norm=__lowerCAmelCase		,   mel_scale='''htk'''		,   )
         _lowerCamelCase		:							Dict            =					mel_filter_bank(
             num_frequency_bins=self.nb_frequency_bins		,   num_mel_filters=__lowerCAmelCase		,   min_frequency=__lowerCAmelCase		,   max_frequency=__lowerCAmelCase		,   sampling_rate=__lowerCAmelCase		,   norm='''slaney'''		,   mel_scale='''slaney'''		,   )




  def 		SCREAMING_SNAKE_CASE							(					self	:		List[Any]						):






         """simple docstring"""






         _lowerCamelCase		:							Union[str, Any]            =					copy.deepcopy(self.__dict__						)
         _lowerCamelCase		:							Optional[Any]            =					self.__class__.__name__
         if "mel_filters" in output:
                del output["mel_filters"]
         if "mel_filters_slaney" in output:
                del output["mel_filters_slaney"]
         return output




  def 		SCREAMING_SNAKE_CASE							(					self	:		Any		,   __lowerCAmelCase	:		np.array		,   __lowerCAmelCase	:		Optional[np.array] = None						):






         """simple docstring"""






         _lowerCamelCase		:							Tuple            =					spectrogram(
             __lowerCAmelCase		,   window_function(self.fft_window_size		,   '''hann'''						)		,   frame_length=self.fft_window_size		,   hop_length=self.hop_length		,   power=2.0		,   mel_filters=__lowerCAmelCase		,   log_mel='''dB'''		,   )
         return log_mel_spectrogram.T




  def 		SCREAMING_SNAKE_CASE							(					self	:		Union[str, Any]		,   __lowerCAmelCase	:		Any		,   __lowerCAmelCase	:		Any		,   __lowerCAmelCase	:		Optional[int]						):






         """simple docstring"""






         _lowerCamelCase		:							int            =					np.array_split(list(range(0		,   total_frames - chunk_frames + 1						)						)		,   3						)
         if len(ranges[1]						) == 0:
                # if the audio is too short, we just use the first chunk
                _lowerCamelCase		:							Dict            =					[0]
         if len(ranges[2]						) == 0:
                # if the audio is too short, we just use the first chunk
                _lowerCamelCase		:							Dict            =					[0]
         # randomly choose index for each part
         _lowerCamelCase		:							Dict            =					np.random.choice(ranges[0]						)
         _lowerCamelCase		:							List[str]            =					np.random.choice(ranges[1]						)
         _lowerCamelCase		:							int            =					np.random.choice(ranges[2]						)

         _lowerCamelCase		:							Tuple            =					mel[idx_front : idx_front + chunk_frames, :]
         _lowerCamelCase		:							int            =					mel[idx_middle : idx_middle + chunk_frames, :]
         _lowerCamelCase		:							Dict            =					mel[idx_back : idx_back + chunk_frames, :]

         _lowerCamelCase		:							str            =					torch.tensor(mel[None, None, :]						)
         _lowerCamelCase		:							List[Any]            =					torch.nn.functional.interpolate(
             __lowerCAmelCase		,   size=[chunk_frames, 6_4]		,   mode='''bilinear'''		,   align_corners=__lowerCAmelCase						)
         _lowerCamelCase		:							Dict            =					mel_shrink[0][0].numpy()
         _lowerCamelCase		:							Optional[Any]            =					np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]		,   axis=0						)
         return mel_fusion




  def 		SCREAMING_SNAKE_CASE							(					self	:		Optional[int]		,   __lowerCAmelCase	:		np.array		,   __lowerCAmelCase	:		List[str]		,   __lowerCAmelCase	:		Union[str, Any]		,   __lowerCAmelCase	:		Optional[Any]						):






         """simple docstring"""






         if waveform.shape[0] > max_length:
                if truncation == "rand_trunc":
                       _lowerCamelCase		:							Any            =					True
                       # random crop to max_length (for compatibility) -> this should be handled by self.pad
                       _lowerCamelCase		:							int            =					len(__lowerCAmelCase						) - max_length
                       _lowerCamelCase		:							str            =					np.random.randint(0		,   overflow + 1						)
                       _lowerCamelCase		:							List[str]            =					waveform[idx : idx + max_length]
                       _lowerCamelCase		:							List[Any]            =					self._np_extract_fbank_features(__lowerCAmelCase		,   self.mel_filters_slaney						)[None, :]
                elif truncation == "fusion":
                       _lowerCamelCase		:							Any            =					self._np_extract_fbank_features(__lowerCAmelCase		,   self.mel_filters						)
                       _lowerCamelCase		:							Tuple            =					max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
                       _lowerCamelCase		:							int            =					mel.shape[0]
                       if chunk_frames == total_frames:
                              # there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
                              # In this case, we just use the whole audio.
                              _lowerCamelCase		:							str            =					np.stack([mel, mel, mel, mel]		,   axis=0						)
                              _lowerCamelCase		:							Any            =					False
                       else:
                              _lowerCamelCase		:							Any            =					self._random_mel_fusion(__lowerCAmelCase		,   __lowerCAmelCase		,   __lowerCAmelCase						)
                              _lowerCamelCase		:							Optional[int]            =					True
                else:
                       raise NotImplementedError(f'''data_truncating {truncation} not implemented'''						)

         else:
                _lowerCamelCase		:							Optional[Any]            =					False
                # only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
                if waveform.shape[0] < max_length:
                       if padding == "repeat":
                              _lowerCamelCase		:							Tuple            =					int(max_length / len(__lowerCAmelCase						)						)
                              _lowerCamelCase		:							Dict            =					np.stack(np.tile(__lowerCAmelCase		,   n_repeat + 1						)						)[:max_length]
                       if padding == "repeatpad":
                              _lowerCamelCase		:							Optional[Any]            =					int(max_length / len(__lowerCAmelCase						)						)
                              _lowerCamelCase		:							int            =					np.stack(np.tile(__lowerCAmelCase		,   __lowerCAmelCase						)						)
                       _lowerCamelCase		:							Tuple            =					np.pad(__lowerCAmelCase		,   (0, max_length - waveform.shape[0])		,   mode='''constant'''		,   constant_values=0						)

                if truncation == "fusion":
                       _lowerCamelCase		:							Tuple            =					self._np_extract_fbank_features(__lowerCAmelCase		,   self.mel_filters						)
                       _lowerCamelCase		:							Tuple            =					np.stack([input_mel, input_mel, input_mel, input_mel]		,   axis=0						)
                else:
                       _lowerCamelCase		:							List[Any]            =					self._np_extract_fbank_features(__lowerCAmelCase		,   self.mel_filters_slaney						)[None, :]

         return input_mel, longer




  def __call__(					self	:		Optional[Any]		,   __lowerCAmelCase	:		Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]		,   __lowerCAmelCase	:		str = None		,   __lowerCAmelCase	:		Optional[str] = None		,   __lowerCAmelCase	:		Optional[int] = None		,   __lowerCAmelCase	:		Optional[int] = None		,   __lowerCAmelCase	:		Optional[Union[str, TensorType]] = None		,   **__lowerCAmelCase	:		str		,   ):






         """simple docstring"""






         _lowerCamelCase		:							List[str]            =					truncation if truncation is not None else self.truncation
         _lowerCamelCase		:							Optional[Any]            =					padding if padding else self.padding

         if sampling_rate is not None:
                if sampling_rate != self.sampling_rate:
                       raise ValueError(
                           f'''The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a'''
                           f''' sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input'''
                           f''' was sampled with {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		:							List[str]            =					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}'''						)
         _lowerCamelCase		:							str            =					is_batched_numpy or (
             isinstance(__lowerCAmelCase		,   (list, tuple)						) and (isinstance(raw_speech[0]		,   (np.ndarray, tuple, list)						))
         )

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

         # always return batch
         if not is_batched:
                _lowerCamelCase		:							Dict            =					[np.asarray(__lowerCAmelCase						)]

         # convert to mel spectrogram, truncate and pad if needed.
         _lowerCamelCase		:							Dict            =					[
             self._get_input_mel(__lowerCAmelCase		,   max_length if max_length else self.nb_max_samples		,   __lowerCAmelCase		,   __lowerCAmelCase						)
             for waveform in raw_speech
         ]

         _lowerCamelCase		:							Any            =					[]
         _lowerCamelCase		:							Optional[int]            =					[]
         for mel, longer in padded_inputs:
                input_mel.append(__lowerCAmelCase						)
                is_longer.append(__lowerCAmelCase						)

         if truncation == "fusion" and sum(__lowerCAmelCase						) == 0:
                # if no audio is longer than 10s, then randomly select one audio to be longer
                _lowerCamelCase		:							List[Any]            =					np.random.randint(0		,   len(__lowerCAmelCase						)						)
                _lowerCamelCase		:							Union[str, Any]            =					True

         if isinstance(input_mel[0]		,   __lowerCAmelCase						):
                _lowerCamelCase		:							Optional[int]            =					[np.asarray(__lowerCAmelCase		,   dtype=np.floataa						) for feature in input_mel]

         # is_longer is a list of bool
         _lowerCamelCase		:							Optional[Any]            =					[[longer] for longer in is_longer]

         _lowerCamelCase		:							Tuple            =					{'''input_features''': input_mel, '''is_longer''': is_longer}
         _lowerCamelCase		:							Union[str, Any]            =					BatchFeature(__lowerCAmelCase						)

         if return_tensors is not None:
                _lowerCamelCase		:							List[str]            =					input_features.convert_to_tensors(__lowerCAmelCase						)

         return input_features






'''simple docstring'''
import unittest

import numpy as np

from transformers import DistilBertConfig, 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.distilbert.modeling_flax_distilbert import (
					    FlaxDistilBertForMaskedLM,
					    FlaxDistilBertForMultipleChoice,
					    FlaxDistilBertForQuestionAnswering,
					    FlaxDistilBertForSequenceClassification,
					    FlaxDistilBertForTokenClassification,
					    FlaxDistilBertModel,
					)




class        _UpperCamelCase	(    unittest.TestCase):


				'''simple docstring'''


				def __init__(     self							,      a_							,      a_=1_3							,      a_=7							,      a_=True							,      a_=True							,      a_=True							,      a_=True							,      a_=9_9							,      a_=3_2							,      a_=5							,      a_=4							,      a_=3_7							,      a_="gelu"							,      a_=0.1							,      a_=0.1							,      a_=5_1_2							,      a_=1_6							,      a_=2							,      a_=0.02							,      a_=4							,      )    ->  Tuple:
					lowercase						:		Optional[Any]   					=		parent
					lowercase						:		int   					=		batch_size
					lowercase						:		int   					=		seq_length
					lowercase						:		List[str]   					=		is_training
					lowercase						:		str   					=		use_attention_mask
					lowercase						:		List[str]   					=		use_token_type_ids
					lowercase						:		Optional[Any]   					=		use_labels
					lowercase						:		Dict   					=		vocab_size
					lowercase						:		Union[str, Any]   					=		hidden_size
					lowercase						:		int   					=		num_hidden_layers
					lowercase						:		Optional[Any]   					=		num_attention_heads
					lowercase						:		Optional[Any]   					=		intermediate_size
					lowercase						:		int   					=		hidden_act
					lowercase						:		Dict   					=		hidden_dropout_prob
					lowercase						:		Optional[Any]   					=		attention_probs_dropout_prob
					lowercase						:		Dict   					=		max_position_embeddings
					lowercase						:		Tuple   					=		type_vocab_size
					lowercase						:		int   					=		type_sequence_label_size
					lowercase						:		Optional[Any]   					=		initializer_range
					lowercase						:		Union[str, Any]   					=		num_choices

				def 	a__		(     self							)    ->  Optional[int]:
					lowercase						:		List[Any]   					=		ids_tensor([self.batch_size, self.seq_length]							,      self.vocab_size							)

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

					lowercase						:		Optional[Any]   					=		DistilBertConfig(
					    vocab_size=self.vocab_size							,      dim=self.hidden_size							,      n_layers=self.num_hidden_layers							,      n_heads=self.num_attention_heads							,      hidden_dim=self.intermediate_size							,      hidden_act=self.hidden_act							,      dropout=self.hidden_dropout_prob							,      attention_dropout=self.attention_probs_dropout_prob							,      max_position_embeddings=self.max_position_embeddings							,      initializer_range=self.initializer_range							,      tie_weights_=a_							,      )

					return config, input_ids, attention_mask


				def 	a__		(     self							)    ->  List[str]:
					lowercase						:		Dict   					=		self.prepare_config_and_inputs()
					lowercase					,		lowercase					,		lowercase						:		Dict   					=		config_and_inputs
					lowercase						:		Any   					=		{"input_ids": input_ids, "attention_mask": attention_mask}
					return config, inputs_dict




@require_flax
class        _UpperCamelCase	(    SCREAMING_SNAKE_CASE						,			unittest.TestCase):


				'''simple docstring'''


				_snake_case						       =				(
				    (
				        FlaxDistilBertModel,
				        FlaxDistilBertForMaskedLM,
				        FlaxDistilBertForMultipleChoice,
				        FlaxDistilBertForQuestionAnswering,
				        FlaxDistilBertForSequenceClassification,
				        FlaxDistilBertForTokenClassification,
				        FlaxDistilBertForQuestionAnswering,
				    )
				    if is_flax_available()
				    else ()
				)

				def 	a__		(     self							)    ->  Dict:
					lowercase						:		Optional[int]   					=		FlaxDistilBertModelTester(self							)


				@slow
				def 	a__		(     self							)    ->  Union[str, Any]:
					for model_class_name in self.all_model_classes:
						lowercase						:		int   					=		model_class_name.from_pretrained("distilbert-base-uncased"							)
						lowercase						:		Tuple   					=		model(np.ones((1, 1)							)							)
						self.assertIsNotNone(a_							)




@require_flax
class        _UpperCamelCase	(    unittest.TestCase):


				'''simple docstring'''


				@slow
				def 	a__		(     self							)    ->  Union[str, Any]:
					lowercase						:		str   					=		FlaxDistilBertModel.from_pretrained("distilbert-base-uncased"							)
					lowercase						:		int   					=		np.array([[0, 3_4_5, 2_3_2, 3_2_8, 7_4_0, 1_4_0, 1_6_9_5, 6_9, 6_0_7_8, 1_5_8_8, 2]]							)
					lowercase						:		int   					=		np.array([[0, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]							)
					lowercase						:		List[str]   					=		model(a_							,      attention_mask=a_							)[0]
					lowercase						:		int   					=		(1, 1_1, 7_6_8)
					self.assertEqual(output.shape							,      a_							)
					lowercase						:		List[str]   					=		np.array([[[-0.16_39, 0.32_99, 0.16_48], [-0.17_46, 0.32_89, 0.17_10], [-0.18_84, 0.33_57, 0.18_10]]]							)

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


