





import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...models.auto.modeling_auto import MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
from ...utils import logging
from ..auto import CONFIG_MAPPING


A   : Optional[int]			=				logging.get_logger(__name__)

A   : int			=				{
    '''salesforce/blip2-opt-2.7b''': '''https://huggingface.co/salesforce/blip2-opt-2.7b/resolve/main/config.json''',
}


class   A      (SCREAMING_SNAKE_CASE   ):
			'''simple docstring'''




			__lowerCamelCase     :	Optional[Any]          =   '''blip_2_vision_model'''






			def __init__(      self       :   Tuple							,     __lowerCAmelCase       :   List[Any]=14_08							,     __lowerCAmelCase       :   List[Any]=61_44							,     __lowerCAmelCase       :   Optional[Any]=39							,     __lowerCAmelCase       :   Union[str, Any]=16							,     __lowerCAmelCase       :   Dict=2_24							,     __lowerCAmelCase       :   Any=14							,     __lowerCAmelCase       :   str="gelu"							,     __lowerCAmelCase       :   Optional[int]=0.0_0_0_0_1							,     __lowerCAmelCase       :   Optional[int]=0.0							,     __lowerCAmelCase       :   int=1e-10							,     __lowerCAmelCase       :   Tuple=True							,     **__lowerCAmelCase       :   Any							,     )     ->     Dict:







					"""simple docstring"""






					super().__init__(**__lowerCAmelCase       )

					A__  				=       hidden_size
					A__  				=       intermediate_size
					A__  				=       num_hidden_layers
					A__  				=       num_attention_heads
					A__  				=       patch_size
					A__  				=       image_size
					A__  				=       initializer_range
					A__  				=       attention_dropout
					A__  				=       layer_norm_eps
					A__  				=       hidden_act
					A__  				=       qkv_bias







			@classmethod
			def        a_	(      cls       :   Tuple							,     __lowerCAmelCase       :   Union[str, os.PathLike]							,     **__lowerCAmelCase       :   List[str]       )     ->     "PretrainedConfig":







					"""simple docstring"""






					cls._set_token_in_kwargs(__lowerCAmelCase       )

					A__				,      A__  				=       cls.get_config_dict(__lowerCAmelCase							,     **__lowerCAmelCase       )

					# get the vision config dict if we are loading from Blip2Config
					if config_dict.get("""model_type"""       ) == "blip-2":
							A__  				=       config_dict["""vision_config"""]

					if "model_type" in config_dict and hasattr(cls							,     """model_type"""       ) and config_dict["model_type"] != cls.model_type:
							logger.warning(
							    f'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
							    f'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'       )

					return cls.from_dict(__lowerCAmelCase							,     **__lowerCAmelCase       )


class   A      (SCREAMING_SNAKE_CASE   ):
			'''simple docstring'''




			__lowerCamelCase     :	List[str]          =   '''blip_2_qformer'''






			def __init__(      self       :   int							,     __lowerCAmelCase       :   List[Any]=3_05_22							,     __lowerCAmelCase       :   List[str]=7_68							,     __lowerCAmelCase       :   str=12							,     __lowerCAmelCase       :   Optional[Any]=12							,     __lowerCAmelCase       :   Union[str, Any]=30_72							,     __lowerCAmelCase       :   str="gelu"							,     __lowerCAmelCase       :   List[str]=0.1							,     __lowerCAmelCase       :   List[Any]=0.1							,     __lowerCAmelCase       :   Optional[Any]=5_12							,     __lowerCAmelCase       :   List[Any]=0.0_2							,     __lowerCAmelCase       :   Any=1e-12							,     __lowerCAmelCase       :   Union[str, Any]=0							,     __lowerCAmelCase       :   int="absolute"							,     __lowerCAmelCase       :   Tuple=2							,     __lowerCAmelCase       :   List[Any]=14_08							,     **__lowerCAmelCase       :   List[Any]							,     )     ->     Any:







					"""simple docstring"""






					super().__init__(pad_token_id=__lowerCAmelCase							,     **__lowerCAmelCase       )

					A__  				=       vocab_size
					A__  				=       hidden_size
					A__  				=       num_hidden_layers
					A__  				=       num_attention_heads
					A__  				=       hidden_act
					A__  				=       intermediate_size
					A__  				=       hidden_dropout_prob
					A__  				=       attention_probs_dropout_prob
					A__  				=       max_position_embeddings
					A__  				=       initializer_range
					A__  				=       layer_norm_eps
					A__  				=       position_embedding_type
					A__  				=       cross_attention_frequency
					A__  				=       encoder_hidden_size







			@classmethod
			def        a_	(      cls       :   Any							,     __lowerCAmelCase       :   Union[str, os.PathLike]							,     **__lowerCAmelCase       :   int       )     ->     "PretrainedConfig":







					"""simple docstring"""






					cls._set_token_in_kwargs(__lowerCAmelCase       )

					A__				,      A__  				=       cls.get_config_dict(__lowerCAmelCase							,     **__lowerCAmelCase       )

					# get the qformer config dict if we are loading from Blip2Config
					if config_dict.get("""model_type"""       ) == "blip-2":
							A__  				=       config_dict["""qformer_config"""]

					if "model_type" in config_dict and hasattr(cls							,     """model_type"""       ) and config_dict["model_type"] != cls.model_type:
							logger.warning(
							    f'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
							    f'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'       )

					return cls.from_dict(__lowerCAmelCase							,     **__lowerCAmelCase       )


class   A      (SCREAMING_SNAKE_CASE   ):
			'''simple docstring'''




			__lowerCamelCase     :	List[str]          =   '''blip-2'''
			__lowerCamelCase     :	List[str]          =   True






			def __init__(      self       :   Dict							,     __lowerCAmelCase       :   str=None							,     __lowerCAmelCase       :   List[str]=None							,     __lowerCAmelCase       :   List[Any]=None							,     __lowerCAmelCase       :   Any=32							,     **__lowerCAmelCase       :   int       )     ->     List[str]:







					"""simple docstring"""






					super().__init__(**__lowerCAmelCase       )

					if vision_config is None:
							A__  				=       {}
							logger.info("""vision_config is None. initializing the Blip2VisionConfig with default values."""       )

					if qformer_config is None:
							A__  				=       {}
							logger.info("""qformer_config is None. Initializing the Blip2QFormerConfig with default values."""       )

					if text_config is None:
							A__  				=       {}
							logger.info("""text_config is None. Initializing the text config with default values (`OPTConfig`)."""       )

					A__  				=       BlipaVisionConfig(**__lowerCAmelCase       )
					A__  				=       BlipaQFormerConfig(**__lowerCAmelCase       )
					A__  				=       text_config["""model_type"""] if """model_type""" in text_config else """opt"""
					A__  				=       CONFIG_MAPPING[text_model_type](**__lowerCAmelCase       )

					A__  				=       self.text_config.tie_word_embeddings
					A__  				=       self.text_config.is_encoder_decoder

					A__  				=       num_query_tokens
					A__  				=       self.vision_config.hidden_size
					A__  				=       self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
					A__  				=       1.0
					A__  				=       0.0_2






			@classmethod
			def        a_	(      cls       :   Any							,     __lowerCAmelCase       :   BlipaVisionConfig							,     __lowerCAmelCase       :   BlipaQFormerConfig							,     __lowerCAmelCase       :   PretrainedConfig							,     **__lowerCAmelCase       :   Optional[Any]							,     )     ->     int:







					"""simple docstring"""






					return cls(
					    vision_config=vision_config.to_dict()							,     qformer_config=qformer_config.to_dict()							,     text_config=text_config.to_dict()							,     **__lowerCAmelCase							,     )







			def        a_	(      self       :   Optional[int]       )     ->     Tuple:







					"""simple docstring"""






					A__  				=       copy.deepcopy(self.__dict__       )
					A__  				=       self.vision_config.to_dict()
					A__  				=       self.qformer_config.to_dict()
					A__  				=       self.text_config.to_dict()
					A__  				=       self.__class__.model_type
					return output










import itertools
import os
import random
import tempfile
import unittest

import numpy as np
from datasets import load_dataset

from transformers import is_speech_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
					from transformers import WhisperFeatureExtractor

if is_torch_available():
					import torch

A   : Dict			=				random.Random()







def        __lowerCamelCase (							__a       :Dict   ,					__a       :str=1.0   ,					__a       :List[Any]=None   ,					__a       :List[str]=None   )		->    Optional[int]:






		"""simple docstring"""

		if rng is None:
				A__  				=       global_rng

		A__  				=       []
		for batch_idx in range(shape[0]   ):
				values.append([]   )
				for _ in range(shape[1]   ):
						values[-1].append(rng.random() * scale   )

		return values


@require_torch
@require_torchaudio
class   A      (unittest.TestCase   ):
			'''simple docstring'''




			def __init__(      self       :   Dict							,     __lowerCAmelCase       :   Any							,     __lowerCAmelCase       :   str=7							,     __lowerCAmelCase       :   List[Any]=4_00							,     __lowerCAmelCase       :   Optional[Any]=20_00							,     __lowerCAmelCase       :   Dict=10							,     __lowerCAmelCase       :   Union[str, Any]=1_60							,     __lowerCAmelCase       :   List[Any]=8							,     __lowerCAmelCase       :   List[Any]=0.0							,     __lowerCAmelCase       :   int=40_00							,     __lowerCAmelCase       :   Any=False							,     __lowerCAmelCase       :   List[str]=True							,     )     ->     Optional[int]:







					"""simple docstring"""






					A__  				=       parent
					A__  				=       batch_size
					A__  				=       min_seq_length
					A__  				=       max_seq_length
					A__  				=       (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
					A__  				=       padding_value
					A__  				=       sampling_rate
					A__  				=       return_attention_mask
					A__  				=       do_normalize
					A__  				=       feature_size
					A__  				=       chunk_length
					A__  				=       hop_length






			def        a_	(      self       :   Optional[Any]       )     ->     Optional[int]:







					"""simple docstring"""






					return {
					    "feature_size": self.feature_size,
					    "hop_length": self.hop_length,
					    "chunk_length": self.chunk_length,
					    "padding_value": self.padding_value,
					    "sampling_rate": self.sampling_rate,
					    "return_attention_mask": self.return_attention_mask,
					    "do_normalize": self.do_normalize,
					}







			def        a_	(      self       :   List[str]							,     __lowerCAmelCase       :   List[Any]=False							,     __lowerCAmelCase       :   int=False       )     ->     str:







					"""simple docstring"""






					def _flatten(__lowerCAmelCase       :   Optional[int]       ):
							return list(itertools.chain(*__lowerCAmelCase       )       )

					if equal_length:
							A__  				=       [floats_list((self.max_seq_length, self.feature_size)       ) for _ in range(self.batch_size       )]
					else:
							# make sure that inputs increase in size
							A__  				=       [
							    floats_list((x, self.feature_size)       )
							    for x in range(self.min_seq_length							,     self.max_seq_length							,     self.seq_length_diff       )
							]
					if numpify:
							A__  				=       [np.asarray(__lowerCAmelCase       ) for x in speech_inputs]
					return speech_inputs


@require_torch
@require_torchaudio
class   A      (SCREAMING_SNAKE_CASE       ,					unittest.TestCase   ):
			'''simple docstring'''




			__lowerCamelCase     :	int          =   WhisperFeatureExtractor if is_speech_available() else None






			def        a_	(      self       :   Any       )     ->     str:







					"""simple docstring"""






					A__  				=       WhisperFeatureExtractionTester(self       )






			def        a_	(      self       :   Dict       )     ->     Tuple:







					"""simple docstring"""






					A__  				=       self.feature_extraction_class(**self.feat_extract_dict       )

					with tempfile.TemporaryDirectory() as tmpdirname:
							A__  				=       feat_extract_first.save_pretrained(__lowerCAmelCase       )[0]
							check_json_file_has_correct_format(__lowerCAmelCase       )
							A__  				=       self.feature_extraction_class.from_pretrained(__lowerCAmelCase       )

					A__  				=       feat_extract_first.to_dict()
					A__  				=       feat_extract_second.to_dict()
					A__  				=       feat_extract_first.mel_filters
					A__  				=       feat_extract_second.mel_filters
					self.assertTrue(np.allclose(__lowerCAmelCase							,     __lowerCAmelCase       )       )
					self.assertEqual(__lowerCAmelCase							,     __lowerCAmelCase       )






			def        a_	(      self       :   Dict       )     ->     str:







					"""simple docstring"""






					A__  				=       self.feature_extraction_class(**self.feat_extract_dict       )

					with tempfile.TemporaryDirectory() as tmpdirname:
							A__  				=       os.path.join(__lowerCAmelCase							,     """feat_extract.json"""       )
							feat_extract_first.to_json_file(__lowerCAmelCase       )
							A__  				=       self.feature_extraction_class.from_json_file(__lowerCAmelCase       )

					A__  				=       feat_extract_first.to_dict()
					A__  				=       feat_extract_second.to_dict()
					A__  				=       feat_extract_first.mel_filters
					A__  				=       feat_extract_second.mel_filters
					self.assertTrue(np.allclose(__lowerCAmelCase							,     __lowerCAmelCase       )       )
					self.assertEqual(__lowerCAmelCase							,     __lowerCAmelCase       )






			def        a_	(      self       :   Union[str, Any]       )     ->     List[Any]:







					"""simple docstring"""






					A__  				=       self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()       )
					# create three inputs of length 800, 1000, and 1200
					A__  				=       [floats_list((1, x)       )[0] for x in range(8_00							,     14_00							,     2_00       )]
					A__  				=       [np.asarray(__lowerCAmelCase       ) for speech_input in speech_inputs]

					# Test feature size
					A__  				=       feature_extractor(__lowerCAmelCase							,     padding="""max_length"""							,     return_tensors="""np"""       ).input_features
					self.assertTrue(input_features.ndim == 3       )
					self.assertTrue(input_features.shape[-1] == feature_extractor.nb_max_frames       )
					self.assertTrue(input_features.shape[-2] == feature_extractor.feature_size       )

					# Test not batched input
					A__  				=       feature_extractor(speech_inputs[0]							,     return_tensors="""np"""       ).input_features
					A__  				=       feature_extractor(np_speech_inputs[0]							,     return_tensors="""np"""       ).input_features
					self.assertTrue(np.allclose(__lowerCAmelCase							,     __lowerCAmelCase							,     atol=1e-3       )       )

					# Test batched
					A__  				=       feature_extractor(__lowerCAmelCase							,     return_tensors="""np"""       ).input_features
					A__  				=       feature_extractor(__lowerCAmelCase							,     return_tensors="""np"""       ).input_features
					for enc_seq_a, enc_seq_a in zip(__lowerCAmelCase							,     __lowerCAmelCase       ):
							self.assertTrue(np.allclose(__lowerCAmelCase							,     __lowerCAmelCase							,     atol=1e-3       )       )

					# Test 2-D numpy arrays are batched.
					A__  				=       [floats_list((1, x)       )[0] for x in (8_00, 8_00, 8_00)]
					A__  				=       np.asarray(__lowerCAmelCase       )
					A__  				=       feature_extractor(__lowerCAmelCase							,     return_tensors="""np"""       ).input_features
					A__  				=       feature_extractor(__lowerCAmelCase							,     return_tensors="""np"""       ).input_features
					for enc_seq_a, enc_seq_a in zip(__lowerCAmelCase							,     __lowerCAmelCase       ):
							self.assertTrue(np.allclose(__lowerCAmelCase							,     __lowerCAmelCase							,     atol=1e-3       )       )

					# Test truncation required
					A__  				=       [floats_list((1, x)       )[0] for x in range(2_00							,     (feature_extractor.n_samples + 5_00)							,     2_00       )]
					A__  				=       [np.asarray(__lowerCAmelCase       ) for speech_input in speech_inputs]

					A__  				=       [x[: feature_extractor.n_samples] for x in speech_inputs]
					A__  				=       [np.asarray(__lowerCAmelCase       ) for speech_input in speech_inputs_truncated]

					A__  				=       feature_extractor(__lowerCAmelCase							,     return_tensors="""np"""       ).input_features
					A__  				=       feature_extractor(__lowerCAmelCase							,     return_tensors="""np"""       ).input_features
					for enc_seq_a, enc_seq_a in zip(__lowerCAmelCase							,     __lowerCAmelCase       ):
							self.assertTrue(np.allclose(__lowerCAmelCase							,     __lowerCAmelCase							,     atol=1e-3       )       )






			def        a_	(      self       :   Dict       )     ->     Union[str, Any]:







					"""simple docstring"""






					import torch

					A__  				=       self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()       )
					A__  				=       np.random.rand(1_00							,     32       ).astype(np.floataa       )
					A__  				=       np_speech_inputs.tolist()

					for inputs in [py_speech_inputs, np_speech_inputs]:
							A__  				=       feature_extractor.pad([{"""input_features""": inputs}]							,     return_tensors="""np"""       )
							self.assertTrue(np_processed.input_features.dtype == np.floataa       )
							A__  				=       feature_extractor.pad([{"""input_features""": inputs}]							,     return_tensors="""pt"""       )
							self.assertTrue(pt_processed.input_features.dtype == torch.floataa       )






			def        a_	(      self       :   List[Any]							,     __lowerCAmelCase       :   Any       )     ->     Dict:







					"""simple docstring"""






					A__  				=       load_dataset("""hf-internal-testing/librispeech_asr_dummy"""							,     """clean"""							,     split="""validation"""       )
					# automatic decoding with librispeech
					A__  				=       ds.sort("""id"""       ).select(range(__lowerCAmelCase       )       )[:num_samples]["""audio"""]

					return [x["array"] for x in speech_samples]






			def        a_	(      self       :   Dict       )     ->     Any:







					"""simple docstring"""






					A__  				=       torch.tensor(
					    [
					        0.1_1_9_3, -0.0_9_4_6, -0.1_0_9_8, -0.0_1_9_6, 0.0_2_2_5, -0.0_6_9_0, -0.1_7_3_6, 0.0_9_5_1,
					        0.0_9_7_1, -0.0_8_1_7, -0.0_7_0_2, 0.0_1_6_2, 0.0_2_6_0, 0.0_0_1_7, -0.0_1_9_2, -0.1_6_7_8,
					        0.0_7_0_9, -0.1_8_6_7, -0.0_6_5_5, -0.0_2_7_4, -0.0_2_3_4, -0.1_8_8_4, -0.0_5_1_6, -0.0_5_5_4,
					        -0.0_2_7_4, -0.1_4_2_5, -0.1_4_2_3, 0.0_8_3_7, 0.0_3_7_7, -0.0_8_5_4
					    ]       )
					# fmt: on

					A__  				=       self._load_datasamples(1       )
					A__  				=       WhisperFeatureExtractor()
					A__  				=       feature_extractor(__lowerCAmelCase							,     return_tensors="""pt"""       ).input_features
					self.assertEqual(input_features.shape							,     (1, 80, 30_00)       )
					self.assertTrue(torch.allclose(input_features[0, 0, :30]							,     __lowerCAmelCase							,     atol=1e-4       )       )







			def        a_	(      self       :   Dict       )     ->     Tuple:







					"""simple docstring"""






					A__  				=       self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()       )
					A__  				=       self._load_datasamples(1       )[0]
					A__  				=       ((audio - audio.min()) / (audio.max() - audio.min())) * 6_55_35  # Rescale to [0, 65535] to show issue
					A__  				=       feat_extract.zero_mean_unit_var_norm([audio]							,     attention_mask=__lowerCAmelCase       )[0]

					self.assertTrue(np.all(np.mean(__lowerCAmelCase       ) < 1e-3       )       )
					self.assertTrue(np.all(np.abs(np.var(__lowerCAmelCase       ) - 1       ) < 1e-3       )       )



