


"""simple docstring"""




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						=      logging.get_logger(__name__)
class 					lowerCAmelCase	(					lowerCamelCase_    ):




		'''simple docstring'''







		SCREAMING_SNAKE_CASE_						:     str								=     ["""input_values""", """attention_mask"""]







		def __init__(	self						,       lowerCAmelCase__ = 1						,       lowerCAmelCase__ = 16_000						,       lowerCAmelCase__ = 0.0						,       lowerCAmelCase__ = False						,       lowerCAmelCase__ = 80						,       lowerCAmelCase__ = 16						,       lowerCAmelCase__ = 64						,       lowerCAmelCase__ = "hann_window"						,       lowerCAmelCase__ = 1.0						,       lowerCAmelCase__ = 80						,       lowerCAmelCase__ = 7_600						,       lowerCAmelCase__ = 1e-10						,       lowerCAmelCase__ = 2						,       lowerCAmelCase__ = True						,       **lowerCAmelCase__						,       )     ->							Dict:
								super().__init__(feature_size=lowerCAmelCase__						,       sampling_rate=lowerCAmelCase__						,       padding_value=lowerCAmelCase__						,       **lowerCAmelCase__				)
								SCREAMING_SNAKE_CASE													=					do_normalize
								SCREAMING_SNAKE_CASE													=					return_attention_mask

								SCREAMING_SNAKE_CASE													=					num_mel_bins
								SCREAMING_SNAKE_CASE													=					hop_length
								SCREAMING_SNAKE_CASE													=					win_length
								SCREAMING_SNAKE_CASE													=					win_function
								SCREAMING_SNAKE_CASE													=					frame_signal_scale
								SCREAMING_SNAKE_CASE													=					fmin
								SCREAMING_SNAKE_CASE													=					fmax
								SCREAMING_SNAKE_CASE													=					mel_floor
								SCREAMING_SNAKE_CASE													=					reduction_factor

								SCREAMING_SNAKE_CASE													=					win_length * sampling_rate // 1_000
								SCREAMING_SNAKE_CASE													=					hop_length * sampling_rate // 1_000
								SCREAMING_SNAKE_CASE													=					optimal_fft_length(self.sample_size				)
								SCREAMING_SNAKE_CASE													=					(self.n_fft // 2) + 1

								SCREAMING_SNAKE_CASE													=					window_function(window_length=self.sample_size						,       name=self.win_function						,       periodic=lowerCAmelCase__				)

								SCREAMING_SNAKE_CASE													=					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'						,       lowerCAmelCase__						,       )
								if reduction_factor != 2.0:
														warnings.warn(
														    'The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers'						,       lowerCAmelCase__						,       )







		@staticmethod
		# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
		def 							__A						(	lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__ = 0.0				)     ->							List[np.ndarray]:
								if attention_mask is not None:
														SCREAMING_SNAKE_CASE													=					np.array(lowerCAmelCase__						,       np.intaa				)
														SCREAMING_SNAKE_CASE													=					[]

														for vector, length in zip(lowerCAmelCase__						,       attention_mask.sum(-1				)				):
																				SCREAMING_SNAKE_CASE													=					(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7				)
																				if length < normed_slice.shape[0]:
																										SCREAMING_SNAKE_CASE													=					padding_value

																				normed_input_values.append(lowerCAmelCase__				)
								else:
														SCREAMING_SNAKE_CASE													=					[(x - x.mean()) / np.sqrt(x.var() + 1e-7				) for x in input_values]

								return normed_input_values







		def 							__A						(	self						,       lowerCAmelCase__						,       )     ->							np.ndarray:
								SCREAMING_SNAKE_CASE													=					spectrogram(
								    lowerCAmelCase__						,       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						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = False						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = False						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       **lowerCAmelCase__						,       )     ->							BatchFeature:
								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:
														SCREAMING_SNAKE_CASE													=					self._process_audio(
														    lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       **lowerCAmelCase__						,       )
								else:
														SCREAMING_SNAKE_CASE													=					None

								if audio_target is not None:
														SCREAMING_SNAKE_CASE													=					self._process_audio(
														    lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__						,       **lowerCAmelCase__						,       )

														if inputs is None:
																				return inputs_target
														else:
																				SCREAMING_SNAKE_CASE													=					inputs_target['input_values']
																				SCREAMING_SNAKE_CASE													=					inputs_target.get('attention_mask'				)
																				if decoder_attention_mask is not None:
																										SCREAMING_SNAKE_CASE													=					decoder_attention_mask

								return inputs







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__ = False						,       lowerCAmelCase__ = False						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = False						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       lowerCAmelCase__ = None						,       **lowerCAmelCase__						,       )     ->							BatchFeature:
								SCREAMING_SNAKE_CASE													=					isinstance(lowerCAmelCase__						,       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}'				)
								SCREAMING_SNAKE_CASE													=					is_batched_numpy or (
								    isinstance(lowerCAmelCase__						,       (list, tuple)				) and (isinstance(speech[0]						,       (np.ndarray, tuple, list)				))
								)

								if is_batched:
														SCREAMING_SNAKE_CASE													=					[np.asarray(lowerCAmelCase__						,       dtype=np.floataa				) for speech in speech]
								elif not is_batched and not isinstance(lowerCAmelCase__						,       np.ndarray				):
														SCREAMING_SNAKE_CASE													=					np.asarray(lowerCAmelCase__						,       dtype=np.floataa				)
								elif isinstance(lowerCAmelCase__						,       np.ndarray				) and speech.dtype is np.dtype(np.floataa				):
														SCREAMING_SNAKE_CASE													=					speech.astype(np.floataa				)

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

								# needed to make pad() work on spectrogram inputs
								SCREAMING_SNAKE_CASE													=					self.feature_size

								# convert into correct format for padding
								if is_target:
														SCREAMING_SNAKE_CASE													=					[self._extract_mel_features(lowerCAmelCase__				) for waveform in speech]
														SCREAMING_SNAKE_CASE													=					BatchFeature({'input_values': features}				)
														SCREAMING_SNAKE_CASE													=					self.num_mel_bins
								else:
														SCREAMING_SNAKE_CASE													=					BatchFeature({'input_values': speech}				)

								SCREAMING_SNAKE_CASE													=					self.pad(
								    lowerCAmelCase__						,       padding=lowerCAmelCase__						,       max_length=lowerCAmelCase__						,       truncation=lowerCAmelCase__						,       pad_to_multiple_of=lowerCAmelCase__						,       return_attention_mask=lowerCAmelCase__						,       **lowerCAmelCase__						,       )

								SCREAMING_SNAKE_CASE													=					feature_size_hack

								# convert input values to correct format
								SCREAMING_SNAKE_CASE													=					padded_inputs['input_values']
								if not isinstance(input_values[0]						,       np.ndarray				):
														SCREAMING_SNAKE_CASE													=					[np.asarray(lowerCAmelCase__						,       dtype=np.floataa				) for array in input_values]
								elif (
								    not isinstance(lowerCAmelCase__						,       np.ndarray				)
								    and isinstance(input_values[0]						,       np.ndarray				)
								    and input_values[0].dtype is np.dtype(np.floataa				)
								):
														SCREAMING_SNAKE_CASE													=					[array.astype(np.floataa				) for array in input_values]
								elif isinstance(lowerCAmelCase__						,       np.ndarray				) and input_values.dtype is np.dtype(np.floataa				):
														SCREAMING_SNAKE_CASE													=					input_values.astype(np.floataa				)

								# convert attention_mask to correct format
								SCREAMING_SNAKE_CASE													=					padded_inputs.get('attention_mask'				)
								if attention_mask is not None:
														SCREAMING_SNAKE_CASE													=					[np.asarray(lowerCAmelCase__						,       dtype=np.intaa				) for array in attention_mask]

								# zero-mean and unit-variance normalization
								if not is_target and self.do_normalize:
														SCREAMING_SNAKE_CASE													=					(
														    attention_mask
														    if self._get_padding_strategies(lowerCAmelCase__						,       max_length=lowerCAmelCase__				) is not PaddingStrategy.DO_NOT_PAD
														    else None
														)
														SCREAMING_SNAKE_CASE													=					self.zero_mean_unit_var_norm(
														    padded_inputs['input_values']						,       attention_mask=lowerCAmelCase__						,       padding_value=self.padding_value				)

								if return_tensors is not None:
														SCREAMING_SNAKE_CASE													=					padded_inputs.convert_to_tensors(lowerCAmelCase__				)

								return padded_inputs







		def 							__A						(	self				)     ->							Dict[str, Any]:
								SCREAMING_SNAKE_CASE													=					super().to_dict()

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

								return output






"""simple docstring"""




from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_torch_available,
    is_vision_available,
)


__UpperCamelCase						=      {
    '''configuration_blip''': [
        '''BLIP_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''BlipConfig''',
        '''BlipTextConfig''',
        '''BlipVisionConfig''',
    ],
    '''processing_blip''': ['''BlipProcessor'''],
}

try:
	if not is_vision_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__UpperCamelCase						=      ['''BlipImageProcessor''']


try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__UpperCamelCase						=      [
	    '''BLIP_PRETRAINED_MODEL_ARCHIVE_LIST''',
	    '''BlipModel''',
	    '''BlipPreTrainedModel''',
	    '''BlipForConditionalGeneration''',
	    '''BlipForQuestionAnswering''',
	    '''BlipVisionModel''',
	    '''BlipTextModel''',
	    '''BlipForImageTextRetrieval''',
	]

try:
	if not is_tf_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__UpperCamelCase						=      [
	    '''TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST''',
	    '''TFBlipModel''',
	    '''TFBlipPreTrainedModel''',
	    '''TFBlipForConditionalGeneration''',
	    '''TFBlipForQuestionAnswering''',
	    '''TFBlipVisionModel''',
	    '''TFBlipTextModel''',
	    '''TFBlipForImageTextRetrieval''',
	]

if TYPE_CHECKING:
	from .configuration_blip import BLIP_PRETRAINED_CONFIG_ARCHIVE_MAP, BlipConfig, BlipTextConfig, BlipVisionConfig
	from .processing_blip import BlipProcessor

	try:
		if not is_vision_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .image_processing_blip import BlipImageProcessor

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_blip import (
		    BLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
		    BlipForConditionalGeneration,
		    BlipForImageTextRetrieval,
		    BlipForQuestionAnswering,
		    BlipModel,
		    BlipPreTrainedModel,
		    BlipTextModel,
		    BlipVisionModel,
		)

	try:
		if not is_tf_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_tf_blip import (
		    TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
		    TFBlipForConditionalGeneration,
		    TFBlipForImageTextRetrieval,
		    TFBlipForQuestionAnswering,
		    TFBlipModel,
		    TFBlipPreTrainedModel,
		    TFBlipTextModel,
		    TFBlipVisionModel,
		)

else:
	import sys

	__UpperCamelCase						=      _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)


