
from __future__ import annotations

import unittest

from transformers import DistilBertConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
	import tensorflow as tf

	from transformers.models.distilbert.modeling_tf_distilbert import (
	    TF_DISTILBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
	    TFDistilBertForMaskedLM,
	    TFDistilBertForMultipleChoice,
	    TFDistilBertForQuestionAnswering,
	    TFDistilBertForSequenceClassification,
	    TFDistilBertForTokenClassification,
	    TFDistilBertModel,
	)







class 			A_  :




	'''simple docstring'''


	def __init__(			self						,       snake_case						,       ):
		lowercase	    =       parent
		lowercase	    =       13
		lowercase	    =       7
		lowercase	    =       True
		lowercase	    =       True
		lowercase	    =       False
		lowercase	    =       True
		lowercase	    =       99
		lowercase	    =       32
		lowercase	    =       2
		lowercase	    =       4
		lowercase	    =       37
		lowercase	    =       'gelu'
		lowercase	    =       0.1
		lowercase	    =       0.1
		lowercase	    =       512
		lowercase	    =       16
		lowercase	    =       2
		lowercase	    =       0.02
		lowercase	    =       3
		lowercase	    =       4
		lowercase	    =       None





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       ids_tensor([self.batch_size, self.seq_length]						,       self.vocab_size			)

		lowercase	    =       None
		if self.use_input_mask:
			lowercase	    =       random_attention_mask([self.batch_size, self.seq_length]			)

		lowercase	    =       None
		lowercase	    =       None
		lowercase	    =       None
		if self.use_labels:
			lowercase	    =       ids_tensor([self.batch_size]						,       self.type_sequence_label_size			)
			lowercase	    =       ids_tensor([self.batch_size, self.seq_length]						,       self.num_labels			)
			lowercase	    =       ids_tensor([self.batch_size]						,       self.num_choices			)

		lowercase	    =       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						,       )

		return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels





	def       SCREAMING_SNAKE_CASE__		(			self						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case			):
		lowercase	    =       TFDistilBertModel(config=snake_case			)
		lowercase	    =       {'input_ids': input_ids, 'attention_mask': input_mask}

		lowercase	    =       model(snake_case			)

		lowercase	    =       [input_ids, input_mask]

		lowercase	    =       model(snake_case			)

		self.parent.assertEqual(result.last_hidden_state.shape						,       (self.batch_size, self.seq_length, self.hidden_size)			)





	def       SCREAMING_SNAKE_CASE__		(			self						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case			):
		lowercase	    =       TFDistilBertForMaskedLM(config=snake_case			)
		lowercase	    =       {'input_ids': input_ids, 'attention_mask': input_mask}
		lowercase	    =       model(snake_case			)
		self.parent.assertEqual(result.logits.shape						,       (self.batch_size, self.seq_length, self.vocab_size)			)





	def       SCREAMING_SNAKE_CASE__		(			self						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case			):
		lowercase	    =       TFDistilBertForQuestionAnswering(config=snake_case			)
		lowercase	    =       {
		    'input_ids': input_ids,
		    'attention_mask': input_mask,
		}
		lowercase	    =       model(snake_case			)
		self.parent.assertEqual(result.start_logits.shape						,       (self.batch_size, self.seq_length)			)
		self.parent.assertEqual(result.end_logits.shape						,       (self.batch_size, self.seq_length)			)





	def       SCREAMING_SNAKE_CASE__		(			self						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case			):
		lowercase	    =       self.num_labels
		lowercase	    =       TFDistilBertForSequenceClassification(snake_case			)
		lowercase	    =       {'input_ids': input_ids, 'attention_mask': input_mask}
		lowercase	    =       model(snake_case			)
		self.parent.assertEqual(result.logits.shape						,       (self.batch_size, self.num_labels)			)





	def       SCREAMING_SNAKE_CASE__		(			self						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case			):
		lowercase	    =       self.num_choices
		lowercase	    =       TFDistilBertForMultipleChoice(snake_case			)
		lowercase	    =       tf.tile(tf.expand_dims(snake_case						,       1			)						,       (1, self.num_choices, 1)			)
		lowercase	    =       tf.tile(tf.expand_dims(snake_case						,       1			)						,       (1, self.num_choices, 1)			)
		lowercase	    =       {
		    'input_ids': multiple_choice_inputs_ids,
		    'attention_mask': multiple_choice_input_mask,
		}
		lowercase	    =       model(snake_case			)
		self.parent.assertEqual(result.logits.shape						,       (self.batch_size, self.num_choices)			)





	def       SCREAMING_SNAKE_CASE__		(			self						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case			):
		lowercase	    =       self.num_labels
		lowercase	    =       TFDistilBertForTokenClassification(snake_case			)
		lowercase	    =       {'input_ids': input_ids, 'attention_mask': input_mask}
		lowercase	    =       model(snake_case			)
		self.parent.assertEqual(result.logits.shape						,       (self.batch_size, self.seq_length, self.num_labels)			)





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       self.prepare_config_and_inputs()
		((lowercase)    ,   (lowercase)    ,   (lowercase)    ,   (lowercase)    ,   (lowercase)    ,   (lowercase))	    =       config_and_inputs
		lowercase	    =       {'input_ids': input_ids, 'attention_mask': input_mask}
		return config, inputs_dict







@require_tf
class 			A_  (     __lowerCamelCase ,      __lowerCamelCase ,      unittest.TestCase  ):




	'''simple docstring'''


	_UpperCamelCase	:	List[Any]									=			(
	    (
	        TFDistilBertModel,
	        TFDistilBertForMaskedLM,
	        TFDistilBertForQuestionAnswering,
	        TFDistilBertForSequenceClassification,
	        TFDistilBertForTokenClassification,
	        TFDistilBertForMultipleChoice,
	    )
	    if is_tf_available()
	    else None
	)
	_UpperCamelCase	:	int									=			(
	    {
	        """feature-extraction""": TFDistilBertModel,
	        """fill-mask""": TFDistilBertForMaskedLM,
	        """question-answering""": TFDistilBertForQuestionAnswering,
	        """text-classification""": TFDistilBertForSequenceClassification,
	        """token-classification""": TFDistilBertForTokenClassification,
	        """zero-shot""": TFDistilBertForSequenceClassification,
	    }
	    if is_tf_available()
	    else {}
	)
	_UpperCamelCase	:	List[str]									=			False
	_UpperCamelCase	:	Dict									=			False





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       TFDistilBertModelTester(self			)
		lowercase	    =       ConfigTester(self						,       config_class=snake_case						,       dim=37			)





	def       SCREAMING_SNAKE_CASE__		(			self			):
		self.config_tester.run_common_tests()





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_distilbert_model(*snake_case			)





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_distilbert_for_masked_lm(*snake_case			)





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_distilbert_for_question_answering(*snake_case			)





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_distilbert_for_sequence_classification(*snake_case			)





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_distilbert_for_multiple_choice(*snake_case			)





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_distilbert_for_token_classification(*snake_case			)





	@slow
	def       SCREAMING_SNAKE_CASE__		(			self			):
		for model_name in list(TF_DISTILBERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]			):
			lowercase	    =       TFDistilBertModel.from_pretrained(snake_case			)
			self.assertIsNotNone(snake_case			)







@require_tf
class 			A_  (     unittest.TestCase  ):




	'''simple docstring'''


	@slow
	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       TFDistilBertModel.from_pretrained('distilbert-base-uncased'			)
		lowercase	    =       tf.constant([[0, 1, 2, 3, 4, 5]]			)
		lowercase	    =       model(snake_case			)[0]

		lowercase	    =       [1, 6, 768]
		self.assertEqual(output.shape						,       snake_case			)

		lowercase	    =       tf.constant(
		    [
		        [
		            [0.19_261_885, -0.13_732_955, 0.4_119_799],
		            [0.22_150_156, -0.07_422_661, 0.39_037_204],
		            [0.22_756_018, -0.0_896_414, 0.3_701_467],
		        ]
		    ]			)
		tf.debugging.assert_near(output[:, :3, :3]						,       snake_case						,       atol=1E-4			)





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


lowercase				=logging.get_logger(__name__)






class    __magic_name__				(						lowerCAmelCase							):
       UpperCAmelCase		 =["input_features", "attention_mask"]






       def __init__(     self    ,				snake_case=8_0    ,				snake_case=1_6_0_0_0    ,				snake_case=8_0    ,				snake_case=0.0    ,				snake_case=True    ,				snake_case=True    ,				snake_case=True    ,				**snake_case    ,				)		->    str:




          '''simple docstring'''
          super().__init__(feature_size=snake_case    ,				sampling_rate=snake_case    ,				padding_value=snake_case    ,				**snake_case)
          _UpperCAmelCase  :  Optional[Any]				       =num_mel_bins
          _UpperCAmelCase  :  Optional[int]				       =do_ceptral_normalize
          _UpperCAmelCase  :  Optional[Any]				       =normalize_means
          _UpperCAmelCase  :  Optional[Any]				       =normalize_vars
          _UpperCAmelCase  :  Tuple				       =True






       def      lowerCAmelCase						(     self    ,				snake_case    ,				)		->    np.ndarray:




          '''simple docstring'''
          _UpperCAmelCase  :  List[Any]				       =waveform * (2**1_5)  # Kaldi compliance: 16-bit signed integers
          _UpperCAmelCase  :  Dict				       =torch.from_numpy(snake_case).unsqueeze(0)
          _UpperCAmelCase  :  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						(     snake_case    ,				snake_case    ,				snake_case = True    ,				snake_case = True    ,				snake_case = 0.0    ,				)		->    np.ndarray:




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

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

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

          return x






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




          '''simple docstring'''
          _UpperCAmelCase  :  str				       =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    ,				snake_case    ,				snake_case = False    ,				snake_case = None    ,				snake_case = False    ,				snake_case = None    ,				snake_case = None    ,				snake_case = None    ,				snake_case = None    ,				**snake_case    ,				)		->    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.')

          _UpperCAmelCase  :  Optional[int]				       =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}")
          _UpperCAmelCase  :  List[Any]				       =is_batched_numpy or (
              isinstance(snake_case    ,				(list, tuple)) and (isinstance(raw_speech[0]    ,				(np.ndarray, tuple, list)))
          )

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

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

          # extract fbank features
          _UpperCAmelCase  :  Optional[Any]				       =[self._extract_fbank_features(snake_case) for waveform in raw_speech]

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

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

          _UpperCAmelCase  :  int				       =padded_inputs.get('attention_mask')
          if attention_mask is not None:
             _UpperCAmelCase  :  Tuple				       =[np.asarray(snake_case    ,				dtype=np.intaa) for array in attention_mask]

          # Utterance-level cepstral mean and variance normalization
          if self.do_ceptral_normalize:
             _UpperCAmelCase  :  Optional[Any]				       =(
                 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
             )
             _UpperCAmelCase  :  Optional[int]				       =self.normalize(
                 padded_inputs['input_features']    ,				attention_mask=snake_case)

          if return_tensors is not None:
             _UpperCAmelCase  :  List[Any]				       =padded_inputs.convert_to_tensors(snake_case)

          return padded_inputs

