


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


__snake_case   							=						logging.get_logger(__name__)

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



class 	__lowerCamelCase			(_a						):
       _lowercase											=     """blip_2_vision_model"""







       def __init__(			self: Optional[int],A_: str=1408,A_: Union[str, Any]=6144,A_: Tuple=39,A_: Optional[int]=16,A_: List[str]=224,A_: Optional[Any]=14,A_: Union[str, Any]="gelu",A_: List[Any]=0.0_0_0_0_1,A_: List[Any]=0.0,A_: List[Any]=1E-10,A_: Optional[int]=True,**A_: Union[str, Any],):






            '''simple docstring'''




            super().__init__(**A_					)

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







       @classmethod
       def 		snake_case_	(			cls: Tuple,A_: Union[str, os.PathLike],**A_: Tuple					):






            '''simple docstring'''




            cls._set_token_in_kwargs(A_					)

            __UpperCamelCase,    __UpperCamelCase =			cls.get_config_dict(A_,**A_					)

            # get the vision config dict if we are loading from Blip2Config
            if config_dict.get('model_type'					) == "blip-2":
                 __UpperCamelCase =			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(A_,**A_					)



class 	__lowerCamelCase			(_a						):
       _lowercase											=     """blip_2_qformer"""







       def __init__(			self: Optional[Any],A_: Optional[int]=3_0522,A_: List[Any]=768,A_: Any=12,A_: Dict=12,A_: int=3072,A_: Dict="gelu",A_: Union[str, Any]=0.1,A_: Any=0.1,A_: Any=512,A_: Dict=0.0_2,A_: str=1E-12,A_: List[Any]=0,A_: Optional[int]="absolute",A_: List[str]=2,A_: Optional[Any]=1408,**A_: Optional[Any],):






            '''simple docstring'''




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

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







       @classmethod
       def 		snake_case_	(			cls: int,A_: Union[str, os.PathLike],**A_: int					):






            '''simple docstring'''




            cls._set_token_in_kwargs(A_					)

            __UpperCamelCase,    __UpperCamelCase =			cls.get_config_dict(A_,**A_					)

            # get the qformer config dict if we are loading from Blip2Config
            if config_dict.get('model_type'					) == "blip-2":
                 __UpperCamelCase =			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(A_,**A_					)



class 	__lowerCamelCase			(_a						):
       _lowercase											=     """blip-2"""
       _lowercase											=     True







       def __init__(			self: Union[str, Any],A_: Optional[int]=None,A_: List[str]=None,A_: Optional[Any]=None,A_: List[str]=32,**A_: Optional[Any]					):






            '''simple docstring'''




            super().__init__(**A_					)

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

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

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

            __UpperCamelCase =			BlipaVisionConfig(**A_					)
            __UpperCamelCase =			BlipaQFormerConfig(**A_					)
            __UpperCamelCase =			text_config['model_type'] if 'model_type' in text_config else 'opt'
            __UpperCamelCase =			CONFIG_MAPPING[text_model_type](**A_					)

            __UpperCamelCase =			self.text_config.tie_word_embeddings
            __UpperCamelCase =			self.text_config.is_encoder_decoder

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







       @classmethod
       def 		snake_case_	(			cls: List[Any],A_: BlipaVisionConfig,A_: BlipaQFormerConfig,A_: PretrainedConfig,**A_: Optional[int],):






            '''simple docstring'''




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







       def 		snake_case_	(			self: Optional[int]					):






            '''simple docstring'''




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






import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_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_torch_available():
				import torch

if is_datasets_available():
				from datasets import load_dataset

lowerCAmelCase       		=    random.Random()




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_=1.0      ,       lowercase_=None      ,       lowercase_=None      )		->     Union[str, Any]:




						'''simple docstring'''







						if rng is None:
												__UpperCAmelCase			:							str	    =	global_rng

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

						return values




class 							lowerCamelCase   (  unittest.TestCase ):


	def __init__(    self				,     lowercase__				,     lowercase__=7				,     lowercase__=4_0_0				,     lowercase__=2_0_0_0				,     lowercase__=2_0_4_8				,     lowercase__=1_2_8				,     lowercase__=1				,     lowercase__=5_1_2				,     lowercase__=3_0				,     lowercase__=4_4_1_0_0				,     ):
							__UpperCAmelCase			:							Optional[Any]	    =	parent
							__UpperCAmelCase			:							Any	    =	batch_size
							__UpperCAmelCase			:							int	    =	min_seq_length
							__UpperCAmelCase			:							List[str]	    =	max_seq_length
							__UpperCAmelCase			:							Dict	    =	(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
							__UpperCAmelCase			:							Any	    =	spectrogram_length
							__UpperCAmelCase			:							List[Any]	    =	feature_size
							__UpperCAmelCase			:							Union[str, Any]	    =	num_audio_channels
							__UpperCAmelCase			:							Optional[int]	    =	hop_length
							__UpperCAmelCase			:							Tuple	    =	chunk_length
							__UpperCAmelCase			:							Any	    =	sampling_rate


	def  A(    self):
							return {
							    "spectrogram_length": self.spectrogram_length,
							    "feature_size": self.feature_size,
							    "num_audio_channels": self.num_audio_channels,
							    "hop_length": self.hop_length,
							    "chunk_length": self.chunk_length,
							    "sampling_rate": self.sampling_rate,
							}


	def  A(    self				,     lowercase__=False				,     lowercase__=False):
							def _flatten(lowercase__):
													return list(itertools.chain(*lowercase__))

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






@require_torch
@require_torchaudio
class 							lowerCamelCase   (  _UpperCamelCase							,						unittest.TestCase ):
	_lowerCAmelCase     :		Optional[int]										=						TvltFeatureExtractor


	def  A(    self):
							__UpperCAmelCase			:							Dict	    =	TvltFeatureExtractionTester(self)


	def  A(    self):
							__UpperCAmelCase			:							Optional[int]	    =	self.feature_extraction_class(**self.feat_extract_dict)
							self.assertTrue(hasattr(lowercase__				,     '''spectrogram_length'''))
							self.assertTrue(hasattr(lowercase__				,     '''feature_size'''))
							self.assertTrue(hasattr(lowercase__				,     '''num_audio_channels'''))
							self.assertTrue(hasattr(lowercase__				,     '''hop_length'''))
							self.assertTrue(hasattr(lowercase__				,     '''chunk_length'''))
							self.assertTrue(hasattr(lowercase__				,     '''sampling_rate'''))


	def  A(    self):
							__UpperCAmelCase			:							str	    =	self.feature_extraction_class(**self.feat_extract_dict)

							with tempfile.TemporaryDirectory() as tmpdirname:
													__UpperCAmelCase			:							str	    =	feat_extract_first.save_pretrained(lowercase__)[0]
													check_json_file_has_correct_format(lowercase__)
													__UpperCAmelCase			:							Optional[int]	    =	self.feature_extraction_class.from_pretrained(lowercase__)

							__UpperCAmelCase			:							List[Any]	    =	feat_extract_first.to_dict()
							__UpperCAmelCase			:							Union[str, Any]	    =	feat_extract_second.to_dict()
							__UpperCAmelCase			:							Union[str, Any]	    =	dict_first.pop('''mel_filters''')
							__UpperCAmelCase			:							Union[str, Any]	    =	dict_second.pop('''mel_filters''')
							self.assertTrue(np.allclose(lowercase__				,     lowercase__))
							self.assertEqual(lowercase__				,     lowercase__)


	def  A(    self):
							__UpperCAmelCase			:							Optional[int]	    =	self.feature_extraction_class(**self.feat_extract_dict)

							with tempfile.TemporaryDirectory() as tmpdirname:
													__UpperCAmelCase			:							Union[str, Any]	    =	os.path.join(lowercase__				,     '''feat_extract.json''')
													feat_extract_first.to_json_file(lowercase__)
													__UpperCAmelCase			:							str	    =	self.feature_extraction_class.from_json_file(lowercase__)

							__UpperCAmelCase			:							Any	    =	feat_extract_first.to_dict()
							__UpperCAmelCase			:							Union[str, Any]	    =	feat_extract_second.to_dict()
							__UpperCAmelCase			:							Tuple	    =	dict_first.pop('''mel_filters''')
							__UpperCAmelCase			:							List[str]	    =	dict_second.pop('''mel_filters''')
							self.assertTrue(np.allclose(lowercase__				,     lowercase__))
							self.assertEqual(lowercase__				,     lowercase__)


	def  A(    self):
							# Initialize feature_extractor
							__UpperCAmelCase			:							Optional[int]	    =	self.feature_extraction_class(**self.feat_extract_dict)

							# create three inputs of length 800, 1000, and 1200
							__UpperCAmelCase			:							Optional[int]	    =	[floats_list((1, x))[0] for x in range(8_0_0				,     1_4_0_0				,     2_0_0)]
							__UpperCAmelCase			:							int	    =	[np.asarray(lowercase__) for speech_input in speech_inputs]

							# Test not batched input
							__UpperCAmelCase			:							Dict	    =	feature_extractor(np_speech_inputs[0]				,     return_tensors='''np'''				,     sampling_rate=4_4_1_0_0).audio_values

							self.assertTrue(encoded_audios.ndim == 4)
							self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size)
							self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length)
							self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels)

							# Test batched
							__UpperCAmelCase			:							List[str]	    =	feature_extractor(lowercase__				,     return_tensors='''np'''				,     sampling_rate=4_4_1_0_0).audio_values

							self.assertTrue(encoded_audios.ndim == 4)
							self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size)
							self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length)
							self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels)

							# Test audio masking
							__UpperCAmelCase			:							Tuple	    =	feature_extractor(
							    lowercase__				,     return_tensors='''np'''				,     sampling_rate=4_4_1_0_0				,     mask_audio=lowercase__).audio_values

							self.assertTrue(encoded_audios.ndim == 4)
							self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size)
							self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length)
							self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels)

							# Test 2-D numpy arrays are batched.
							__UpperCAmelCase			:							Any	    =	[floats_list((1, x))[0] for x in (8_0_0, 8_0_0, 8_0_0)]
							__UpperCAmelCase			:							Optional[Any]	    =	np.asarray(lowercase__)
							__UpperCAmelCase			:							Tuple	    =	feature_extractor(lowercase__				,     return_tensors='''np'''				,     sampling_rate=4_4_1_0_0).audio_values
							self.assertTrue(encoded_audios.ndim == 4)
							self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size)
							self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length)
							self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels)


	def  A(    self				,     lowercase__):
							__UpperCAmelCase			:							Optional[int]	    =	load_dataset('''hf-internal-testing/librispeech_asr_dummy'''				,     '''clean'''				,     split='''validation''')
							# automatic decoding with librispeech
							__UpperCAmelCase			:							int	    =	ds.sort('''id''').select(range(lowercase__))[:num_samples]['''audio''']

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


	def  A(    self):
							__UpperCAmelCase			:							Optional[Any]	    =	self._load_datasamples(1)
							__UpperCAmelCase			:							Tuple	    =	TvltFeatureExtractor()
							__UpperCAmelCase			:							Tuple	    =	feature_extractor(lowercase__				,     return_tensors='''pt''').audio_values

							self.assertEquals(audio_values.shape				,     (1, 1, 1_9_2, 1_2_8))

							__UpperCAmelCase			:							int	    =	torch.tensor([[-0.3_0_3_2, -0.2_7_0_8], [-0.4_4_3_4, -0.4_0_0_7]])
							self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]				,     lowercase__				,     atol=1e-4))

