




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

import numpy as np

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


UpperCAmelCase_		:				str									=     logging.get_logger(__name__)




class 		__A      (			UpperCamelCase__			):
						UpperCamelCase     						=					["""input_values""", """attention_mask"""]

						def __init__(		self   :str    , __snake_case   :int = 1    , __snake_case   :int = 1_60_00    , __snake_case   :float = 0.0    , __snake_case   :bool = False    , __snake_case   :int = 80    , __snake_case   :int = 16    , __snake_case   :int = 64    , __snake_case   :str = "hann_window"    , __snake_case   :float = 1.0    , __snake_case   :float = 80    , __snake_case   :float = 76_00    , __snake_case   :float = 1E-10    , __snake_case   :int = 2    , __snake_case   :bool = True    , **__snake_case   :List[Any]    , ):
										'''simple docstring'''

										super().__init__(feature_size=__snake_case    , sampling_rate=__snake_case    , padding_value=__snake_case    , **__snake_case )
										__magic_name__	:	Optional[int]								=do_normalize
										__magic_name__	:	Any								=return_attention_mask

										__magic_name__	:	List[Any]								=num_mel_bins
										__magic_name__	:	str								=hop_length
										__magic_name__	:	str								=win_length
										__magic_name__	:	List[Any]								=win_function
										__magic_name__	:	Optional[int]								=frame_signal_scale
										__magic_name__	:	List[str]								=fmin
										__magic_name__	:	int								=fmax
										__magic_name__	:	Optional[int]								=mel_floor
										__magic_name__	:	Any								=reduction_factor

										__magic_name__	:	int								=win_length * sampling_rate // 10_00
										__magic_name__	:	List[str]								=hop_length * sampling_rate // 10_00
										__magic_name__	:	Tuple								=optimal_fft_length(self.sample_size )
										__magic_name__	:	List[Any]								=(self.n_fft // 2) + 1

										__magic_name__	:	int								=window_function(window_length=self.sample_size    , name=self.win_function    , periodic=__snake_case )

										__magic_name__	:	Union[str, Any]								=mel_filter_bank(
										    num_frequency_bins=self.n_freqs    , num_mel_filters=self.num_mel_bins    , min_frequency=self.fmin    , max_frequency=self.fmax    , sampling_rate=self.sampling_rate    , norm="""slaney"""    , mel_scale="""slaney"""    , )

										if frame_signal_scale != 1.0:
														warnings.warn(
														    """The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"""    , __snake_case    , )
										if reduction_factor != 2.0:
														warnings.warn(
														    """The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"""    , __snake_case    , )

						@staticmethod
						# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
						def     A__						(		__snake_case   :List[np.ndarray]    , __snake_case   :List[np.ndarray]    , __snake_case   :float = 0.0 ):
										'''simple docstring'''

										if attention_mask is not None:
														__magic_name__	:	Tuple								=np.array(__snake_case    , np.intaa )
														__magic_name__	:	Optional[int]								=[]

														for vector, length in zip(__snake_case    , attention_mask.sum(-1 ) ):
																		__magic_name__	:	Union[str, Any]								=(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7 )
																		if length < normed_slice.shape[0]:
																						__magic_name__	:	str								=padding_value

																		normed_input_values.append(__snake_case )
										else:
														__magic_name__	:	Any								=[(x - x.mean()) / np.sqrt(x.var() + 1E-7 ) for x in input_values]

										return normed_input_values

						def     A__						(		self   :Dict    , __snake_case   :np.ndarray    , ):
										'''simple docstring'''

										__magic_name__	:	Union[str, Any]								=spectrogram(
										    __snake_case    , window=self.window    , frame_length=self.sample_size    , hop_length=self.sample_stride    , fft_length=self.n_fft    , mel_filters=self.mel_filters    , mel_floor=self.mel_floor    , log_mel="""log10"""    , )
										return log_mel_spec.T

						def __call__(		self   :int    , __snake_case   :Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None    , __snake_case   :Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None    , __snake_case   :Union[bool, str, PaddingStrategy] = False    , __snake_case   :Optional[int] = None    , __snake_case   :bool = False    , __snake_case   :Optional[int] = None    , __snake_case   :Optional[bool] = None    , __snake_case   :Optional[Union[str, TensorType]] = None    , __snake_case   :Optional[int] = None    , **__snake_case   :Any    , ):
										'''simple docstring'''

										if audio is None and audio_target is None:
														raise ValueError("""You must provide either `audio` or `audio_target` values.""" )

										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 audio 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.""" )

										if audio is not None:
														__magic_name__	:	Union[str, Any]								=self._process_audio(
														    __snake_case    , __snake_case    , __snake_case    , __snake_case    , __snake_case    , __snake_case    , __snake_case    , __snake_case    , **__snake_case    , )
										else:
														__magic_name__	:	List[str]								=None

										if audio_target is not None:
														__magic_name__	:	Any								=self._process_audio(
														    __snake_case    , __snake_case    , __snake_case    , __snake_case    , __snake_case    , __snake_case    , __snake_case    , __snake_case    , **__snake_case    , )

														if inputs is None:
																		return inputs_target
														else:
																		__magic_name__	:	List[str]								=inputs_target["""input_values"""]
																		__magic_name__	:	List[str]								=inputs_target.get("""attention_mask""" )
																		if decoder_attention_mask is not None:
																						__magic_name__	:	str								=decoder_attention_mask

										return inputs

						def     A__						(		self   :List[str]    , __snake_case   :Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]    , __snake_case   :bool = False    , __snake_case   :Union[bool, str, PaddingStrategy] = False    , __snake_case   :Optional[int] = None    , __snake_case   :bool = False    , __snake_case   :Optional[int] = None    , __snake_case   :Optional[bool] = None    , __snake_case   :Optional[Union[str, TensorType]] = None    , **__snake_case   :Union[str, Any]    , ):
										'''simple docstring'''

										__magic_name__	:	Optional[Any]								=isinstance(__snake_case    , np.ndarray ) and len(speech.shape ) > 1
										if is_batched_numpy and len(speech.shape ) > 2:
														raise ValueError(f"Only mono-channel audio is supported for input to {self}" )
										__magic_name__	:	Optional[int]								=is_batched_numpy or (
										    isinstance(__snake_case    , (list, tuple) ) and (isinstance(speech[0]    , (np.ndarray, tuple, list) ))
										)

										if is_batched:
														__magic_name__	:	Any								=[np.asarray(__snake_case    , dtype=np.floataa ) for speech in speech]
										elif not is_batched and not isinstance(__snake_case    , np.ndarray ):
														__magic_name__	:	Dict								=np.asarray(__snake_case    , dtype=np.floataa )
										elif isinstance(__snake_case    , np.ndarray ) and speech.dtype is np.dtype(np.floataa ):
														__magic_name__	:	List[str]								=speech.astype(np.floataa )

										# always return batch
										if not is_batched:
														__magic_name__	:	int								=[speech]

										# needed to make pad() work on spectrogram inputs
										__magic_name__	:	List[str]								=self.feature_size

										# convert into correct format for padding
										if is_target:
														__magic_name__	:	Optional[int]								=[self._extract_mel_features(__snake_case ) for waveform in speech]
														__magic_name__	:	Optional[int]								=BatchFeature({"""input_values""": features} )
														__magic_name__	:	List[str]								=self.num_mel_bins
										else:
														__magic_name__	:	int								=BatchFeature({"""input_values""": speech} )

										__magic_name__	:	List[str]								=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    , )

										__magic_name__	:	Optional[int]								=feature_size_hack

										# convert input values to correct format
										__magic_name__	:	Tuple								=padded_inputs["""input_values"""]
										if not isinstance(input_values[0]    , np.ndarray ):
														__magic_name__	:	Optional[Any]								=[np.asarray(__snake_case    , dtype=np.floataa ) for array in input_values]
										elif (
										    not isinstance(__snake_case    , np.ndarray )
										    and isinstance(input_values[0]    , np.ndarray )
										    and input_values[0].dtype is np.dtype(np.floataa )
										):
														__magic_name__	:	Union[str, Any]								=[array.astype(np.floataa ) for array in input_values]
										elif isinstance(__snake_case    , np.ndarray ) and input_values.dtype is np.dtype(np.floataa ):
														__magic_name__	:	Any								=input_values.astype(np.floataa )

										# convert attention_mask to correct format
										__magic_name__	:	Optional[Any]								=padded_inputs.get("""attention_mask""" )
										if attention_mask is not None:
														__magic_name__	:	str								=[np.asarray(__snake_case    , dtype=np.intaa ) for array in attention_mask]

										# zero-mean and unit-variance normalization
										if not is_target and self.do_normalize:
														__magic_name__	:	Optional[Any]								=(
														    attention_mask
														    if self._get_padding_strategies(__snake_case    , max_length=__snake_case ) is not PaddingStrategy.DO_NOT_PAD
														    else None
														)
														__magic_name__	:	Dict								=self.zero_mean_unit_var_norm(
														    padded_inputs["""input_values"""]    , attention_mask=__snake_case    , padding_value=self.padding_value )

										if return_tensors is not None:
														__magic_name__	:	Any								=padded_inputs.convert_to_tensors(__snake_case )

										return padded_inputs


						def     A__						(		self   :Union[str, Any] ):
										'''simple docstring'''

										__magic_name__	:	Dict								=super().to_dict()

										# Don't serialize these as they are derived from the other properties.
										__magic_name__	:	Optional[Any]								=["""window""", """mel_filters""", """sample_size""", """sample_stride""", """n_fft""", """n_freqs"""]
										for name in names:
														if name in output:
																		del output[name]

										return output










from __future__ import annotations

import unittest

from transformers import 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 numpy
							import tensorflow as tf

							from transformers import (
							    TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
							    TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
							    TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST,
							    BertConfig,
							    DPRConfig,
							    TFDPRContextEncoder,
							    TFDPRQuestionEncoder,
							    TFDPRReader,
							)




class 		__A      :

						def __init__(		self   :Optional[int]    , __snake_case   :Optional[Any]    , __snake_case   :List[Any]=13    , __snake_case   :str=7    , __snake_case   :Tuple=True    , __snake_case   :Dict=True    , __snake_case   :Optional[Any]=True    , __snake_case   :str=True    , __snake_case   :int=99    , __snake_case   :int=32    , __snake_case   :Dict=2    , __snake_case   :Optional[Any]=4    , __snake_case   :Dict=37    , __snake_case   :Optional[int]="gelu"    , __snake_case   :Tuple=0.1    , __snake_case   :Tuple=0.1    , __snake_case   :int=5_12    , __snake_case   :int=16    , __snake_case   :int=2    , __snake_case   :Optional[int]=0.02    , __snake_case   :Union[str, Any]=3    , __snake_case   :Any=4    , __snake_case   :str=None    , __snake_case   :str=0    , ):
										'''simple docstring'''

										__magic_name__	:	Optional[int]								=parent
										__magic_name__	:	int								=batch_size
										__magic_name__	:	Any								=seq_length
										__magic_name__	:	List[str]								=is_training
										__magic_name__	:	Any								=use_input_mask
										__magic_name__	:	Union[str, Any]								=use_token_type_ids
										__magic_name__	:	Union[str, Any]								=use_labels
										__magic_name__	:	Optional[Any]								=vocab_size
										__magic_name__	:	Optional[Any]								=hidden_size
										__magic_name__	:	Optional[Any]								=num_hidden_layers
										__magic_name__	:	Optional[Any]								=num_attention_heads
										__magic_name__	:	Tuple								=intermediate_size
										__magic_name__	:	Tuple								=hidden_act
										__magic_name__	:	Tuple								=hidden_dropout_prob
										__magic_name__	:	Any								=attention_probs_dropout_prob
										__magic_name__	:	Union[str, Any]								=max_position_embeddings
										__magic_name__	:	int								=type_vocab_size
										__magic_name__	:	Optional[Any]								=type_sequence_label_size
										__magic_name__	:	Union[str, Any]								=initializer_range
										__magic_name__	:	Optional[Any]								=num_labels
										__magic_name__	:	Any								=num_choices
										__magic_name__	:	Optional[Any]								=scope
										__magic_name__	:	str								=projection_dim

						def     A__						(		self   :Dict ):
										'''simple docstring'''

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

										__magic_name__	:	Union[str, Any]								=None
										if self.use_input_mask:
														# follow test_modeling_tf_ctrl.py
														__magic_name__	:	List[Any]								=random_attention_mask([self.batch_size, self.seq_length] )

										__magic_name__	:	Optional[int]								=None
										if self.use_token_type_ids:
														__magic_name__	:	str								=ids_tensor([self.batch_size, self.seq_length]    , self.type_vocab_size )

										__magic_name__	:	List[str]								=None
										__magic_name__	:	List[Any]								=None
										__magic_name__	:	List[str]								=None
										if self.use_labels:
														__magic_name__	:	str								=ids_tensor([self.batch_size]    , self.type_sequence_label_size )
														__magic_name__	:	Optional[int]								=ids_tensor([self.batch_size, self.seq_length]    , self.num_labels )
														__magic_name__	:	Union[str, Any]								=ids_tensor([self.batch_size]    , self.num_choices )

										__magic_name__	:	Optional[int]								=BertConfig(
										    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=__snake_case    , initializer_range=self.initializer_range    , )
										__magic_name__	:	Dict								=DPRConfig(projection_dim=self.projection_dim    , **config.to_dict() )

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

						def     A__						(		self   :List[Any]    , __snake_case   :List[Any]    , __snake_case   :List[str]    , __snake_case   :Any    , __snake_case   :Any    , __snake_case   :Optional[Any]    , __snake_case   :Optional[Any]    , __snake_case   :List[str] ):
										'''simple docstring'''

										__magic_name__	:	List[Any]								=TFDPRContextEncoder(config=__snake_case )
										__magic_name__	:	Union[str, Any]								=model(__snake_case    , attention_mask=__snake_case    , token_type_ids=__snake_case )
										__magic_name__	:	str								=model(__snake_case    , token_type_ids=__snake_case )
										__magic_name__	:	Optional[int]								=model(__snake_case )
										self.parent.assertEqual(result.pooler_output.shape    , (self.batch_size, self.projection_dim or self.hidden_size) )

						def     A__						(		self   :int    , __snake_case   :str    , __snake_case   :Tuple    , __snake_case   :Optional[Any]    , __snake_case   :Optional[Any]    , __snake_case   :Tuple    , __snake_case   :List[Any]    , __snake_case   :int ):
										'''simple docstring'''

										__magic_name__	:	int								=TFDPRQuestionEncoder(config=__snake_case )
										__magic_name__	:	Union[str, Any]								=model(__snake_case    , attention_mask=__snake_case    , token_type_ids=__snake_case )
										__magic_name__	:	List[str]								=model(__snake_case    , token_type_ids=__snake_case )
										__magic_name__	:	Union[str, Any]								=model(__snake_case )
										self.parent.assertEqual(result.pooler_output.shape    , (self.batch_size, self.projection_dim or self.hidden_size) )

						def     A__						(		self   :List[Any]    , __snake_case   :List[str]    , __snake_case   :Optional[int]    , __snake_case   :str    , __snake_case   :Any    , __snake_case   :str    , __snake_case   :Optional[Any]    , __snake_case   :Dict ):
										'''simple docstring'''

										__magic_name__	:	Optional[int]								=TFDPRReader(config=__snake_case )
										__magic_name__	:	List[str]								=model(__snake_case    , attention_mask=__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) )
										self.parent.assertEqual(result.relevance_logits.shape    , (self.batch_size,) )


						def     A__						(		self   :Optional[Any] ):
										'''simple docstring'''

										__magic_name__	:	str								=self.prepare_config_and_inputs()
										(
										    (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, (
										    __magic_name__
										)				, 
										)	:	Union[str, Any]								=config_and_inputs
										__magic_name__	:	List[str]								={"""input_ids""": input_ids}
										return config, inputs_dict




@require_tf
class 		__A      (			UpperCamelCase__					,						UpperCamelCase__					,						unittest.TestCase			):
						UpperCamelCase     						=					(
						    (
						        TFDPRContextEncoder,
						        TFDPRQuestionEncoder,
						        TFDPRReader,
						    )
						    if is_tf_available()
						    else ()
						)
						UpperCamelCase     						=					{"""feature-extraction""": TFDPRQuestionEncoder} if is_tf_available() else {}

						UpperCamelCase     						=					False
						UpperCamelCase     						=					False
						UpperCamelCase     						=					False
						UpperCamelCase     						=					False
						UpperCamelCase     						=					False

						def     A__						(		self   :Union[str, Any] ):
										'''simple docstring'''

										__magic_name__	:	Union[str, Any]								=TFDPRModelTester(self )
										__magic_name__	:	Optional[Any]								=ConfigTester(self    , config_class=__snake_case    , hidden_size=37 )

						def     A__						(		self   :Dict ):
										'''simple docstring'''

										self.config_tester.run_common_tests()

						def     A__						(		self   :Tuple ):
										'''simple docstring'''

										__magic_name__	:	str								=self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_dpr_context_encoder(*__snake_case )

						def     A__						(		self   :List[Any] ):
										'''simple docstring'''

										__magic_name__	:	int								=self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_dpr_question_encoder(*__snake_case )

						def     A__						(		self   :Any ):
										'''simple docstring'''

										__magic_name__	:	Optional[int]								=self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_dpr_reader(*__snake_case )


						@slow
						def     A__						(		self   :Tuple ):
										'''simple docstring'''

										for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														__magic_name__	:	Dict								=TFDPRContextEncoder.from_pretrained(__snake_case )
														self.assertIsNotNone(__snake_case )

										for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														__magic_name__	:	Dict								=TFDPRContextEncoder.from_pretrained(__snake_case )
														self.assertIsNotNone(__snake_case )

										for model_name in TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														__magic_name__	:	Union[str, Any]								=TFDPRQuestionEncoder.from_pretrained(__snake_case )
														self.assertIsNotNone(__snake_case )

										for model_name in TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														__magic_name__	:	List[str]								=TFDPRReader.from_pretrained(__snake_case )
														self.assertIsNotNone(__snake_case )




@require_tf
class 		__A      (			unittest.TestCase			):

						@slow
						def     A__						(		self   :Optional[Any] ):
										'''simple docstring'''

										__magic_name__	:	int								=TFDPRQuestionEncoder.from_pretrained("""facebook/dpr-question_encoder-single-nq-base""" )

										__magic_name__	:	int								=tf.constant(
										    [[1_01, 75_92, 10_10, 20_03, 20_26, 38_99, 1_01_40, 10_29, 1_02]] )  # [CLS] hello, is my dog cute? [SEP]
										__magic_name__	:	Optional[Any]								=model(__snake_case )[0]  # embedding shape = (1, 768)
										# compare the actual values for a slice.
										__magic_name__	:	Tuple								=tf.constant(
										    [
										        [
										            0.03236253,
										            0.12753335,
										            0.16818509,
										            0.00279786,
										            0.3896933,
										            0.24264945,
										            0.2178971,
										            -0.02335227,
										            -0.08481959,
										            -0.14324117,
										        ]
										    ] )
										self.assertTrue(numpy.allclose(output[:, :10].numpy()    , expected_slice.numpy()    , atol=1E-4 ) )




