


"""simple docstring"""




import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / 'utils'))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


UpperCAmelCase_				:			Any   =	get_tests_dir('fixtures')
class 			SCREAMING_SNAKE_CASE__  (    unittest.TestCase      ):
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[Any]					)		->      Tuple:
             # A mock response for an HTTP head request to emulate server down
             a_ :		Tuple							=  mock.Mock()
             a_ :		Optional[int]							=  5_0_0
             a_ :		Union[str, Any]							=  {}
             a_ :		int							=  HTTPError
             a_ :		List[Any]							=  {}

             # Download this model to make sure it's in the cache.
             a_ :		Tuple							=  WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'					)
             # Under the mock environment we get a 500 error when trying to reach the model.
             with mock.patch('requests.Session.request'					,							return_value=SCREAMING_SNAKE_CASE__					) as mock_head:
                    a_ :		str							=  WavaVecaFeatureExtractor.from_pretrained('hf-internal-testing/tiny-random-wav2vec2'					)
                    # This check we did call the fake head request
                    mock_head.assert_called()

      def 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      Union[str, Any]:
             # This test is for deprecated behavior and can be removed in v5
             a_ :		Tuple							=  WavaVecaFeatureExtractor.from_pretrained(
                 'https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json'					)
@is_staging_test
class 			SCREAMING_SNAKE_CASE__  (    unittest.TestCase      ):
      @classmethod
      def 	SCREAMING_SNAKE_CASE							(	cls    :	Optional[Any]					)		->      Any:
             a_ :		int							=  TOKEN
             HfFolder.save_token(SCREAMING_SNAKE_CASE__					)
      @classmethod
      def 	SCREAMING_SNAKE_CASE							(	cls    :	int					)		->      Optional[int]:
             try:
                    delete_repo(token=cls._token					,							repo_id='test-feature-extractor'					)
             except HTTPError:
                    pass

             try:
                    delete_repo(token=cls._token					,							repo_id='valid_org/test-feature-extractor-org'					)
             except HTTPError:
                    pass

             try:
                    delete_repo(token=cls._token					,							repo_id='test-dynamic-feature-extractor'					)
             except HTTPError:
                    pass
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      List[Any]:
             a_ :		Any							=  WavaVecaFeatureExtractor.from_pretrained(SCREAMING_SNAKE_CASE__					)
             feature_extractor.push_to_hub('test-feature-extractor'					,							use_auth_token=self._token					)

             a_ :		Tuple							=  WavaVecaFeatureExtractor.from_pretrained(F"""{USER}/test-feature-extractor"""					)
             for k, v in feature_extractor.__dict__.items():
                    self.assertEqual(SCREAMING_SNAKE_CASE__					,							getattr(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)					)

             # Reset repo
             delete_repo(token=self._token					,							repo_id='test-feature-extractor'					)

             # Push to hub via save_pretrained
             with tempfile.TemporaryDirectory() as tmp_dir:
                    feature_extractor.save_pretrained(
                        SCREAMING_SNAKE_CASE__					,							repo_id='test-feature-extractor'					,							push_to_hub=SCREAMING_SNAKE_CASE__					,							use_auth_token=self._token					)

             a_ :		Tuple							=  WavaVecaFeatureExtractor.from_pretrained(F"""{USER}/test-feature-extractor"""					)
             for k, v in feature_extractor.__dict__.items():
                    self.assertEqual(SCREAMING_SNAKE_CASE__					,							getattr(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					)		->      Optional[Any]:
             a_ :		Union[str, Any]							=  WavaVecaFeatureExtractor.from_pretrained(SCREAMING_SNAKE_CASE__					)
             feature_extractor.push_to_hub('valid_org/test-feature-extractor'					,							use_auth_token=self._token					)

             a_ :		Union[str, Any]							=  WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor'					)
             for k, v in feature_extractor.__dict__.items():
                    self.assertEqual(SCREAMING_SNAKE_CASE__					,							getattr(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)					)

             # Reset repo
             delete_repo(token=self._token					,							repo_id='valid_org/test-feature-extractor'					)

             # Push to hub via save_pretrained
             with tempfile.TemporaryDirectory() as tmp_dir:
                    feature_extractor.save_pretrained(
                        SCREAMING_SNAKE_CASE__					,							repo_id='valid_org/test-feature-extractor-org'					,							push_to_hub=SCREAMING_SNAKE_CASE__					,							use_auth_token=self._token					)

             a_ :		Union[str, Any]							=  WavaVecaFeatureExtractor.from_pretrained('valid_org/test-feature-extractor-org'					)
             for k, v in feature_extractor.__dict__.items():
                    self.assertEqual(SCREAMING_SNAKE_CASE__					,							getattr(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)					)

      def 	SCREAMING_SNAKE_CASE							(	self    :	Any					)		->      List[str]:
             CustomFeatureExtractor.register_for_auto_class()
             a_ :		List[Any]							=  CustomFeatureExtractor.from_pretrained(SCREAMING_SNAKE_CASE__					)

             feature_extractor.push_to_hub('test-dynamic-feature-extractor'					,							use_auth_token=self._token					)

             # This has added the proper auto_map field to the config
             self.assertDictEqual(
                 feature_extractor.auto_map					,							{'AutoFeatureExtractor': 'custom_feature_extraction.CustomFeatureExtractor'}					,							)

             a_ :		Tuple							=  AutoFeatureExtractor.from_pretrained(
                 F"""{USER}/test-dynamic-feature-extractor"""					,							trust_remote_code=SCREAMING_SNAKE_CASE__					)
             # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
             self.assertEqual(new_feature_extractor.__class__.__name__					,							'CustomFeatureExtractor'					)






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


UpperCAmelCase_				:			str   =	logging.get_logger(__name__)

UpperCAmelCase_				:			str   =	{
    'Salesforce/instruct-blip-flan-t5': 'https://huggingface.co/Salesforce/instruct-blip-flan-t5/resolve/main/config.json',
}
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      snake_case__    :					int					       =						'''instructblip_vision_model'''
      def __init__(	self    :	Optional[int]					,							SCREAMING_SNAKE_CASE__    :	List[Any]=1_4_0_8					,							SCREAMING_SNAKE_CASE__    :	int=6_1_4_4					,							SCREAMING_SNAKE_CASE__    :	Dict=3_9					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=1_6					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=2_2_4					,							SCREAMING_SNAKE_CASE__    :	Optional[int]=1_4					,							SCREAMING_SNAKE_CASE__    :	List[Any]="gelu"					,							SCREAMING_SNAKE_CASE__    :	List[str]=1E-6					,							SCREAMING_SNAKE_CASE__    :	Tuple=0.0					,							SCREAMING_SNAKE_CASE__    :	List[str]=1E-10					,							SCREAMING_SNAKE_CASE__    :	str=True					,							**SCREAMING_SNAKE_CASE__    :	Dict					,							)		->      int:
             super().__init__(**SCREAMING_SNAKE_CASE__					)

             a_ :		List[str]							=  hidden_size
             a_ :		Any							=  intermediate_size
             a_ :		str							=  num_hidden_layers
             a_ :		Dict							=  num_attention_heads
             a_ :		str							=  patch_size
             a_ :		Any							=  image_size
             a_ :		Dict							=  initializer_range
             a_ :		List[Any]							=  attention_dropout
             a_ :		Union[str, Any]							=  layer_norm_eps
             a_ :		Optional[Any]							=  hidden_act
             a_ :		Optional[Any]							=  qkv_bias

      @classmethod
      def 	SCREAMING_SNAKE_CASE							(	cls    :	Any					,							SCREAMING_SNAKE_CASE__    :	Union[str, os.PathLike]					,							**SCREAMING_SNAKE_CASE__    :	List[str]					)		->      "PretrainedConfig":
             cls._set_token_in_kwargs(SCREAMING_SNAKE_CASE__					)

             a_							,		a_ :		Dict							=  cls.get_config_dict(SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)

             # get the vision config dict if we are loading from InstructBlipConfig
             if config_dict.get('model_type'					) == "instructblip":
                    a_ :		List[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(SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      snake_case__    :					List[Any]					       =						'''instructblip_qformer'''
      def __init__(	self    :	Optional[int]					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=3_0_5_2_2					,							SCREAMING_SNAKE_CASE__    :	str=7_6_8					,							SCREAMING_SNAKE_CASE__    :	List[str]=1_2					,							SCREAMING_SNAKE_CASE__    :	Any=1_2					,							SCREAMING_SNAKE_CASE__    :	Optional[int]=3_0_7_2					,							SCREAMING_SNAKE_CASE__    :	List[str]="gelu"					,							SCREAMING_SNAKE_CASE__    :	Dict=0.1					,							SCREAMING_SNAKE_CASE__    :	int=0.1					,							SCREAMING_SNAKE_CASE__    :	Optional[int]=5_1_2					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]=0.02					,							SCREAMING_SNAKE_CASE__    :	int=1E-12					,							SCREAMING_SNAKE_CASE__    :	Dict=0					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]="absolute"					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=2					,							SCREAMING_SNAKE_CASE__    :	str=1_4_0_8					,							**SCREAMING_SNAKE_CASE__    :	Dict					,							)		->      Optional[Any]:
             super().__init__(pad_token_id=SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)

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

      @classmethod
      def 	SCREAMING_SNAKE_CASE							(	cls    :	str					,							SCREAMING_SNAKE_CASE__    :	Union[str, os.PathLike]					,							**SCREAMING_SNAKE_CASE__    :	Union[str, Any]					)		->      "PretrainedConfig":
             cls._set_token_in_kwargs(SCREAMING_SNAKE_CASE__					)

             a_							,		a_ :		Union[str, Any]							=  cls.get_config_dict(SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)

             # get the qformer config dict if we are loading from InstructBlipConfig
             if config_dict.get('model_type'					) == "instructblip":
                    a_ :		Dict							=  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(SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      snake_case__    :					Union[str, Any]					       =						'''instructblip'''
      snake_case__    :					List[str]					       =						True
      def __init__(	self    :	Tuple					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]=None					,							SCREAMING_SNAKE_CASE__    :	Dict=None					,							SCREAMING_SNAKE_CASE__    :	List[Any]=None					,							SCREAMING_SNAKE_CASE__    :	int=3_2					,							**SCREAMING_SNAKE_CASE__    :	Union[str, Any]					)		->      Tuple:
             super().__init__(**SCREAMING_SNAKE_CASE__					)

             if vision_config is None:
                    a_ :		Dict							=  {}
                    logger.info('vision_config is None. initializing the InstructBlipVisionConfig with default values.'					)

             if qformer_config is None:
                    a_ :		List[str]							=  {}
                    logger.info('qformer_config is None. Initializing the InstructBlipQFormerConfig with default values.'					)

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

             a_ :		int							=  InstructBlipVisionConfig(**SCREAMING_SNAKE_CASE__					)
             a_ :		int							=  InstructBlipQFormerConfig(**SCREAMING_SNAKE_CASE__					)
             a_ :		Optional[Any]							=  text_config['model_type'] if 'model_type' in text_config else 'opt'
             a_ :		List[Any]							=  CONFIG_MAPPING[text_model_type](**SCREAMING_SNAKE_CASE__					)

             a_ :		Tuple							=  self.text_config.tie_word_embeddings
             a_ :		List[Any]							=  self.text_config.is_encoder_decoder

             a_ :		Optional[Any]							=  num_query_tokens
             a_ :		int							=  self.vision_config.hidden_size
             a_ :		List[str]							=  self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
             a_ :		Optional[int]							=  1.0
             a_ :		Any							=  0.02
      @classmethod
      def 	SCREAMING_SNAKE_CASE							(	cls    :	List[Any]					,							SCREAMING_SNAKE_CASE__    :	InstructBlipVisionConfig					,							SCREAMING_SNAKE_CASE__    :	InstructBlipQFormerConfig					,							SCREAMING_SNAKE_CASE__    :	PretrainedConfig					,							**SCREAMING_SNAKE_CASE__    :	List[str]					,							)		->      Any:
             return cls(
                 vision_config=vision_config.to_dict()					,							qformer_config=qformer_config.to_dict()					,							text_config=text_config.to_dict()					,							**SCREAMING_SNAKE_CASE__					,							)

      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[int]					)		->      Union[str, Any]:
             a_ :		List[Any]							=  copy.deepcopy(self.__dict__					)
             a_ :		Optional[int]							=  self.vision_config.to_dict()
             a_ :		Optional[Any]							=  self.qformer_config.to_dict()
             a_ :		Union[str, Any]							=  self.text_config.to_dict()
             a_ :		Optional[Any]							=  self.__class__.model_type
             return output


