






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

__lowerCamelCase      :    List[Any]   		=   random.Random()







def   SCREAMING_SNAKE_CASE       (				snake_case_      :							Tuple  ,						snake_case_      :							Union[str, Any]=1.0  ,						snake_case_      :							List[str]=None  ,						snake_case_      :							str=None     ):
	if rng is None:
		snake_case__				:   str     =						global_rng

	snake_case__				:   str     =						[]
	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 							SCREAMING_SNAKE_CASE__		(       unittest.TestCase		):





					"""simple docstring"""






					def __init__(      self  :						Tuple							,	__A  :						Union[str, Any]							,	__A  :						List[Any]=7							,	__A  :						int=4_0_0							,	__A  :						Union[str, Any]=2_0_0_0							,	__A  :						Union[str, Any]=1_0							,	__A  :						Optional[Any]=1_6_0							,	__A  :						int=8							,	__A  :						Dict=0.0							,	__A  :						Optional[int]=4_0_0_0							,	__A  :						Optional[int]=False							,	__A  :						Union[str, Any]=True							,	):
						snake_case__				:   List[str]     =						parent
						snake_case__				:   Optional[Any]     =						batch_size
						snake_case__				:   List[Any]     =						min_seq_length
						snake_case__				:   Union[str, Any]     =						max_seq_length
						snake_case__				:   Tuple     =						(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
						snake_case__				:   Optional[Any]     =						padding_value
						snake_case__				:   List[str]     =						sampling_rate
						snake_case__				:   Optional[int]     =						return_attention_mask
						snake_case__				:   int     =						do_normalize
						snake_case__				:   Tuple     =						feature_size
						snake_case__				:   int     =						chunk_length
						snake_case__				:   Any     =						hop_length



					def  _lowercase  (      self  :						int ):
						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  _lowercase  (      self  :						List[str]							,	__A  :						Any=False							,	__A  :						Dict=False ):
						def _flatten(__A  :						Any ):
							return list(itertools.chain(*__A ) )

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



@require_torch
@require_torchaudio
class 							SCREAMING_SNAKE_CASE__		(       UpperCamelCase_				,     unittest.TestCase		):





					"""simple docstring"""






					a_			=	WhisperFeatureExtractor if is_speech_available() else None



					def  _lowercase  (      self  :						Union[str, Any] ):
						snake_case__				:   Optional[Any]     =						WhisperFeatureExtractionTester(self )



					def  _lowercase  (      self  :						Tuple ):
						snake_case__				:   int     =						self.feature_extraction_class(**self.feat_extract_dict )

						with tempfile.TemporaryDirectory() as tmpdirname:
							snake_case__				:   List[str]     =						feat_extract_first.save_pretrained(__A )[0]
							check_json_file_has_correct_format(__A )
							snake_case__				:   List[str]     =						self.feature_extraction_class.from_pretrained(__A )

						snake_case__				:   Tuple     =						feat_extract_first.to_dict()
						snake_case__				:   List[Any]     =						feat_extract_second.to_dict()
						snake_case__				:   int     =						feat_extract_first.mel_filters
						snake_case__				:   Any     =						feat_extract_second.mel_filters
						self.assertTrue(np.allclose(__A							,	__A ) )
						self.assertEqual(__A							,	__A )



					def  _lowercase  (      self  :						Optional[Any] ):
						snake_case__				:   List[str]     =						self.feature_extraction_class(**self.feat_extract_dict )

						with tempfile.TemporaryDirectory() as tmpdirname:
							snake_case__				:   Any     =						os.path.join(__A							,	"feat_extract.json" )
							feat_extract_first.to_json_file(__A )
							snake_case__				:   List[str]     =						self.feature_extraction_class.from_json_file(__A )

						snake_case__				:   Union[str, Any]     =						feat_extract_first.to_dict()
						snake_case__				:   List[Any]     =						feat_extract_second.to_dict()
						snake_case__				:   Any     =						feat_extract_first.mel_filters
						snake_case__				:   Tuple     =						feat_extract_second.mel_filters
						self.assertTrue(np.allclose(__A							,	__A ) )
						self.assertEqual(__A							,	__A )



					def  _lowercase  (      self  :						Any ):
						# Tests that all call wrap to encode_plus and batch_encode_plus
						snake_case__				:   Tuple     =						self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
						# create three inputs of length 800, 1000, and 1200
						snake_case__				:   int     =						[floats_list((1, x) )[0] for x in range(8_0_0							,	1_4_0_0							,	2_0_0 )]
						snake_case__				:   Any     =						[np.asarray(__A ) for speech_input in speech_inputs]

						# Test feature size
						snake_case__				:   List[str]     =						feature_extractor(__A							,	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
						snake_case__				:   Any     =						feature_extractor(speech_inputs[0]							,	return_tensors="np" ).input_features
						snake_case__				:   int     =						feature_extractor(np_speech_inputs[0]							,	return_tensors="np" ).input_features
						self.assertTrue(np.allclose(__A							,	__A							,	atol=1e-3 ) )

						# Test batched
						snake_case__				:   List[str]     =						feature_extractor(__A							,	return_tensors="np" ).input_features
						snake_case__				:   Tuple     =						feature_extractor(__A							,	return_tensors="np" ).input_features
						for enc_seq_a, enc_seq_a in zip(__A							,	__A ):
							self.assertTrue(np.allclose(__A							,	__A							,	atol=1e-3 ) )

						# Test 2-D numpy arrays are batched.
						snake_case__				:   Optional[Any]     =						[floats_list((1, x) )[0] for x in (8_0_0, 8_0_0, 8_0_0)]
						snake_case__				:   List[Any]     =						np.asarray(__A )
						snake_case__				:   Tuple     =						feature_extractor(__A							,	return_tensors="np" ).input_features
						snake_case__				:   Union[str, Any]     =						feature_extractor(__A							,	return_tensors="np" ).input_features
						for enc_seq_a, enc_seq_a in zip(__A							,	__A ):
							self.assertTrue(np.allclose(__A							,	__A							,	atol=1e-3 ) )

						# Test truncation required
						snake_case__				:   Optional[Any]     =						[floats_list((1, x) )[0] for x in range(2_0_0							,	(feature_extractor.n_samples + 5_0_0)							,	2_0_0 )]
						snake_case__				:   Optional[Any]     =						[np.asarray(__A ) for speech_input in speech_inputs]

						snake_case__				:   Union[str, Any]     =						[x[: feature_extractor.n_samples] for x in speech_inputs]
						snake_case__				:   List[Any]     =						[np.asarray(__A ) for speech_input in speech_inputs_truncated]

						snake_case__				:   Union[str, Any]     =						feature_extractor(__A							,	return_tensors="np" ).input_features
						snake_case__				:   Tuple     =						feature_extractor(__A							,	return_tensors="np" ).input_features
						for enc_seq_a, enc_seq_a in zip(__A							,	__A ):
							self.assertTrue(np.allclose(__A							,	__A							,	atol=1e-3 ) )



					def  _lowercase  (      self  :						Optional[Any] ):
						import torch

						snake_case__				:   int     =						self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
						snake_case__				:   List[Any]     =						np.random.rand(1_0_0							,	3_2 ).astype(np.floataa )
						snake_case__				:   Union[str, Any]     =						np_speech_inputs.tolist()

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



					def  _lowercase  (      self  :						List[str]							,	__A  :						Dict ):
						snake_case__				:   Any     =						load_dataset("hf-internal-testing/librispeech_asr_dummy"							,	"clean"							,	split="validation" )
						# automatic decoding with librispeech
						snake_case__				:   Optional[int]     =						ds.sort("id" ).select(range(__A ) )[:num_samples]["audio"]

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



					def  _lowercase  (      self  :						Union[str, Any] ):
						# fmt: off
						snake_case__				:   List[str]     =						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

						snake_case__				:   Tuple     =						self._load_datasamples(1 )
						snake_case__				:   Any     =						WhisperFeatureExtractor()
						snake_case__				:   List[str]     =						feature_extractor(__A							,	return_tensors="pt" ).input_features
						self.assertEqual(input_features.shape							,	(1, 8_0, 3_0_0_0) )
						self.assertTrue(torch.allclose(input_features[0, 0, :3_0]							,	__A							,	atol=1e-4 ) )






					def  _lowercase  (      self  :						Tuple ):
						snake_case__				:   Tuple     =						self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict() )
						snake_case__				:   Dict     =						self._load_datasamples(1 )[0]
						snake_case__				:   Dict     =						((audio - audio.min()) / (audio.max() - audio.min())) * 6_5_5_3_5  # Rescale to [0, 65535] to show issue
						snake_case__				:   List[str]     =						feat_extract.zero_mean_unit_var_norm([audio]							,	attention_mask=__A )[0]

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











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


__lowerCamelCase      :    Union[str, Any]   		=   logging.get_logger(__name__)

__lowerCamelCase      :    Dict   		=   {
    """Salesforce/instruct-blip-flan-t5""": """https://huggingface.co/Salesforce/instruct-blip-flan-t5/resolve/main/config.json""",
}



class 							SCREAMING_SNAKE_CASE__		(       UpperCamelCase_		):





					"""simple docstring"""






					a_			=	"instructblip_vision_model"



					def __init__(      self  :						List[Any]							,	__A  :						Dict=1_4_0_8							,	__A  :						Tuple=6_1_4_4							,	__A  :						str=3_9							,	__A  :						int=1_6							,	__A  :						str=2_2_4							,	__A  :						Any=1_4							,	__A  :						Dict="gelu"							,	__A  :						List[Any]=1e-6							,	__A  :						Any=0.0							,	__A  :						List[Any]=1e-1_0							,	__A  :						Union[str, Any]=True							,	**__A  :						Tuple							,	):
						super().__init__(**__A )

						snake_case__				:   List[str]     =						hidden_size
						snake_case__				:   Optional[int]     =						intermediate_size
						snake_case__				:   List[str]     =						num_hidden_layers
						snake_case__				:   List[Any]     =						num_attention_heads
						snake_case__				:   str     =						patch_size
						snake_case__				:   int     =						image_size
						snake_case__				:   int     =						initializer_range
						snake_case__				:   Optional[int]     =						attention_dropout
						snake_case__				:   str     =						layer_norm_eps
						snake_case__				:   Optional[Any]     =						hidden_act
						snake_case__				:   Tuple     =						qkv_bias






					@classmethod
					def  _lowercase  (      cls  :						List[str]							,	__A  :						Union[str, os.PathLike]							,	**__A  :						Optional[Any] ):
						cls._set_token_in_kwargs(__A )

						snake_case__,		snake_case__				:   str     =						cls.get_config_dict(__A							,	**__A )

						# get the vision config dict if we are loading from InstructBlipConfig
						if config_dict.get("model_type" ) == "instructblip":
							snake_case__				:   Union[str, Any]     =						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 							SCREAMING_SNAKE_CASE__		(       UpperCamelCase_		):





					"""simple docstring"""






					a_			=	"instructblip_qformer"



					def __init__(      self  :						Any							,	__A  :						Union[str, Any]=3_0_5_2_2							,	__A  :						Union[str, Any]=7_6_8							,	__A  :						Optional[int]=1_2							,	__A  :						Dict=1_2							,	__A  :						Dict=3_0_7_2							,	__A  :						List[str]="gelu"							,	__A  :						Union[str, Any]=0.1							,	__A  :						Tuple=0.1							,	__A  :						Any=5_1_2							,	__A  :						Optional[int]=0.0_2							,	__A  :						List[str]=1e-1_2							,	__A  :						Any=0							,	__A  :						Optional[Any]="absolute"							,	__A  :						str=2							,	__A  :						Any=1_4_0_8							,	**__A  :						List[str]							,	):
						super().__init__(pad_token_id=__A							,	**__A )

						snake_case__				:   Dict     =						vocab_size
						snake_case__				:   Optional[int]     =						hidden_size
						snake_case__				:   Optional[Any]     =						num_hidden_layers
						snake_case__				:   str     =						num_attention_heads
						snake_case__				:   int     =						hidden_act
						snake_case__				:   Optional[Any]     =						intermediate_size
						snake_case__				:   Union[str, Any]     =						hidden_dropout_prob
						snake_case__				:   List[Any]     =						attention_probs_dropout_prob
						snake_case__				:   List[Any]     =						max_position_embeddings
						snake_case__				:   int     =						initializer_range
						snake_case__				:   Dict     =						layer_norm_eps
						snake_case__				:   str     =						position_embedding_type
						snake_case__				:   Dict     =						cross_attention_frequency
						snake_case__				:   List[str]     =						encoder_hidden_size






					@classmethod
					def  _lowercase  (      cls  :						List[Any]							,	__A  :						Union[str, os.PathLike]							,	**__A  :						Optional[int] ):
						cls._set_token_in_kwargs(__A )

						snake_case__,		snake_case__				:   Tuple     =						cls.get_config_dict(__A							,	**__A )

						# get the qformer config dict if we are loading from InstructBlipConfig
						if config_dict.get("model_type" ) == "instructblip":
							snake_case__				:   List[Any]     =						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 							SCREAMING_SNAKE_CASE__		(       UpperCamelCase_		):





					"""simple docstring"""






					a_			=	"instructblip"
					a_			=	True



					def __init__(      self  :						List[str]							,	__A  :						Optional[Any]=None							,	__A  :						Tuple=None							,	__A  :						Optional[int]=None							,	__A  :						Optional[Any]=3_2							,	**__A  :						Optional[int] ):
						super().__init__(**__A )

						if vision_config is None:
							snake_case__				:   Any     =						{}
							logger.info("vision_config is None. initializing the InstructBlipVisionConfig with default values." )

						if qformer_config is None:
							snake_case__				:   Optional[Any]     =						{}
							logger.info("qformer_config is None. Initializing the InstructBlipQFormerConfig with default values." )

						if text_config is None:
							snake_case__				:   Optional[int]     =						{}
							logger.info("text_config is None. Initializing the text config with default values (`OPTConfig`)." )

						snake_case__				:   List[Any]     =						InstructBlipVisionConfig(**__A )
						snake_case__				:   Union[str, Any]     =						InstructBlipQFormerConfig(**__A )
						snake_case__				:   Dict     =						text_config["model_type"] if "model_type" in text_config else "opt"
						snake_case__				:   List[Any]     =						CONFIG_MAPPING[text_model_type](**__A )

						snake_case__				:   Union[str, Any]     =						self.text_config.tie_word_embeddings
						snake_case__				:   Tuple     =						self.text_config.is_encoder_decoder

						snake_case__				:   str     =						num_query_tokens
						snake_case__				:   Dict     =						self.vision_config.hidden_size
						snake_case__				:   List[Any]     =						self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
						snake_case__				:   int     =						1.0
						snake_case__				:   Optional[int]     =						0.0_2



					@classmethod
					def  _lowercase  (      cls  :						List[str]							,	__A  :						InstructBlipVisionConfig							,	__A  :						InstructBlipQFormerConfig							,	__A  :						PretrainedConfig							,	**__A  :						int							,	):
						return cls(
						    vision_config=vision_config.to_dict()							,	qformer_config=qformer_config.to_dict()							,	text_config=text_config.to_dict()							,	**__A							,	)






					def  _lowercase  (      self  :						Optional[int] ):
						snake_case__				:   Any     =						copy.deepcopy(self.__dict__ )
						snake_case__				:   Optional[Any]     =						self.vision_config.to_dict()
						snake_case__				:   List[str]     =						self.qformer_config.to_dict()
						snake_case__				:   List[Any]     =						self.text_config.to_dict()
						snake_case__				:   List[Any]     =						self.__class__.model_type
						return output



