

'''simple docstring'''

import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


A_							    =				logging.get_logger(__name__)

A_							    =				{
    "Salesforce/blip-vqa-base": "https://huggingface.co/Salesforce/blip-vqa-base/resolve/main/config.json",
    "Salesforce/blip-vqa-capfit-large": (
        "https://huggingface.co/Salesforce/blip-vqa-base-capfit/resolve/main/config.json"
    ),
    "Salesforce/blip-image-captioning-base": (
        "https://huggingface.co/Salesforce/blip-image-captioning-base/resolve/main/config.json"
    ),
    "Salesforce/blip-image-captioning-large": (
        "https://huggingface.co/Salesforce/blip-image-captioning-large/resolve/main/config.json"
    ),
    "Salesforce/blip-itm-base-coco": "https://huggingface.co/Salesforce/blip-itm-base-coco/resolve/main/config.json",
    "Salesforce/blip-itm-large-coco": "https://huggingface.co/Salesforce/blip-itm-large-coco/resolve/main/config.json",
    "Salesforce/blip-itm-base-flikr": "https://huggingface.co/Salesforce/blip-itm-base-flikr/resolve/main/config.json",
    "Salesforce/blip-itm-large-flikr": (
        "https://huggingface.co/Salesforce/blip-itm-large-flikr/resolve/main/config.json"
    ),
}





class 				_snake_case			(							_UpperCamelCase    ):
   _A							:  Optional[int]      =       "blip_text_model"
   def __init__(			self      :   Tuple  ,SCREAMING_SNAKE_CASE__      :   List[str]=30_524  ,SCREAMING_SNAKE_CASE__      :   Any=768  ,SCREAMING_SNAKE_CASE__      :   int=768  ,SCREAMING_SNAKE_CASE__      :   List[str]=3_072  ,SCREAMING_SNAKE_CASE__      :   Dict=768  ,SCREAMING_SNAKE_CASE__      :   int=12  ,SCREAMING_SNAKE_CASE__      :   Optional[Any]=8  ,SCREAMING_SNAKE_CASE__      :   Optional[Any]=512  ,SCREAMING_SNAKE_CASE__      :   Any="gelu"  ,SCREAMING_SNAKE_CASE__      :   Optional[int]=1e-12  ,SCREAMING_SNAKE_CASE__      :   Dict=0.0  ,SCREAMING_SNAKE_CASE__      :   List[Any]=0.0  ,SCREAMING_SNAKE_CASE__      :   Tuple=0.02  ,SCREAMING_SNAKE_CASE__      :   Optional[Any]=30_522  ,SCREAMING_SNAKE_CASE__      :   Any=2  ,SCREAMING_SNAKE_CASE__      :   List[Any]=0  ,SCREAMING_SNAKE_CASE__      :   Tuple=102  ,SCREAMING_SNAKE_CASE__      :   str=True  ,SCREAMING_SNAKE_CASE__      :   Union[str, Any]=True  ,**SCREAMING_SNAKE_CASE__      :   List[Any]  ,):
     super().__init__(
         pad_token_id=a_  ,bos_token_id=a_  ,eos_token_id=a_  ,sep_token_id=a_  ,**a_  ,)

     SCREAMING_SNAKE_CASE:List[Any]       =       vocab_size
     SCREAMING_SNAKE_CASE:Optional[Any]       =       hidden_size
     SCREAMING_SNAKE_CASE:Optional[int]       =       encoder_hidden_size
     SCREAMING_SNAKE_CASE:Dict       =       intermediate_size
     SCREAMING_SNAKE_CASE:List[Any]       =       projection_dim
     SCREAMING_SNAKE_CASE:str       =       hidden_dropout_prob
     SCREAMING_SNAKE_CASE:Optional[int]       =       num_hidden_layers
     SCREAMING_SNAKE_CASE:List[str]       =       num_attention_heads
     SCREAMING_SNAKE_CASE:Dict       =       max_position_embeddings
     SCREAMING_SNAKE_CASE:Dict       =       layer_norm_eps
     SCREAMING_SNAKE_CASE:int       =       hidden_act
     SCREAMING_SNAKE_CASE:Dict       =       initializer_range
     SCREAMING_SNAKE_CASE:Any       =       attention_probs_dropout_prob
     SCREAMING_SNAKE_CASE:Tuple       =       is_decoder
     SCREAMING_SNAKE_CASE:Optional[int]       =       use_cache







   @classmethod
   def      __UpperCamelCase    (			cls      :   Dict  ,SCREAMING_SNAKE_CASE__      :   Union[str, os.PathLike]  ,**SCREAMING_SNAKE_CASE__      :   Optional[int]				):
     cls._set_token_in_kwargs(a_				)

     SCREAMING_SNAKE_CASE:Optional[int]       =       cls.get_config_dict(a_  ,**a_				)

     # get the text config dict if we are loading from BlipConfig
     if config_dict.get("model_type"				) == "blip":
       SCREAMING_SNAKE_CASE:Tuple       =       config_dict["text_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 				_snake_case			(							_UpperCamelCase    ):
   _A							:  int      =       "blip_vision_model"
   def __init__(			self      :   int  ,SCREAMING_SNAKE_CASE__      :   List[str]=768  ,SCREAMING_SNAKE_CASE__      :   Any=3_072  ,SCREAMING_SNAKE_CASE__      :   Union[str, Any]=512  ,SCREAMING_SNAKE_CASE__      :   Optional[int]=12  ,SCREAMING_SNAKE_CASE__      :   Union[str, Any]=12  ,SCREAMING_SNAKE_CASE__      :   Union[str, Any]=384  ,SCREAMING_SNAKE_CASE__      :   Dict=16  ,SCREAMING_SNAKE_CASE__      :   Tuple="gelu"  ,SCREAMING_SNAKE_CASE__      :   Any=1e-5  ,SCREAMING_SNAKE_CASE__      :   Optional[int]=0.0  ,SCREAMING_SNAKE_CASE__      :   Any=1e-10  ,**SCREAMING_SNAKE_CASE__      :   Optional[int]  ,):
     super().__init__(**a_				)

     SCREAMING_SNAKE_CASE:str       =       hidden_size
     SCREAMING_SNAKE_CASE:List[str]       =       intermediate_size
     SCREAMING_SNAKE_CASE:Optional[int]       =       projection_dim
     SCREAMING_SNAKE_CASE:Dict       =       num_hidden_layers
     SCREAMING_SNAKE_CASE:Dict       =       num_attention_heads
     SCREAMING_SNAKE_CASE:Optional[int]       =       patch_size
     SCREAMING_SNAKE_CASE:Optional[int]       =       image_size
     SCREAMING_SNAKE_CASE:int       =       initializer_range
     SCREAMING_SNAKE_CASE:Union[str, Any]       =       attention_dropout
     SCREAMING_SNAKE_CASE:List[str]       =       layer_norm_eps
     SCREAMING_SNAKE_CASE:Tuple       =       hidden_act







   @classmethod
   def      __UpperCamelCase    (			cls      :   str  ,SCREAMING_SNAKE_CASE__      :   Union[str, os.PathLike]  ,**SCREAMING_SNAKE_CASE__      :   Optional[Any]				):
     cls._set_token_in_kwargs(a_				)

     SCREAMING_SNAKE_CASE:Tuple       =       cls.get_config_dict(a_  ,**a_				)

     # get the vision config dict if we are loading from BlipConfig
     if config_dict.get("model_type"				) == "blip":
       SCREAMING_SNAKE_CASE:List[str]       =       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 				_snake_case			(							_UpperCamelCase    ):
   _A							:  Tuple      =       "blip"
   _A							:  List[str]      =       True
   def __init__(			self      :   List[Any]  ,SCREAMING_SNAKE_CASE__      :   Optional[Any]=None  ,SCREAMING_SNAKE_CASE__      :   List[str]=None  ,SCREAMING_SNAKE_CASE__      :   Dict=512  ,SCREAMING_SNAKE_CASE__      :   Tuple=2.6_592  ,SCREAMING_SNAKE_CASE__      :   List[Any]=256  ,**SCREAMING_SNAKE_CASE__      :   Tuple  ,):
     super().__init__(**a_				)

     if text_config is None:
       SCREAMING_SNAKE_CASE:List[str]       =       {}
       logger.info("`text_config` is `None`. Initializing the `BlipTextConfig` with default values."				)

     if vision_config is None:
       SCREAMING_SNAKE_CASE:Tuple       =       {}
       logger.info("`vision_config` is `None`. Initializing the `BlipVisionConfig` with default values."				)

     SCREAMING_SNAKE_CASE:int       =       BlipTextConfig(**a_				)
     SCREAMING_SNAKE_CASE:str       =       BlipVisionConfig(**a_				)

     SCREAMING_SNAKE_CASE:Optional[int]       =       self.vision_config.hidden_size

     SCREAMING_SNAKE_CASE:Any       =       projection_dim
     SCREAMING_SNAKE_CASE:Optional[Any]       =       logit_scale_init_value
     SCREAMING_SNAKE_CASE:Dict       =       1.0
     SCREAMING_SNAKE_CASE:Tuple       =       0.02
     SCREAMING_SNAKE_CASE:Any       =       image_text_hidden_size
   @classmethod
   def      __UpperCamelCase    (			cls      :   Union[str, Any]  ,SCREAMING_SNAKE_CASE__      :   BlipTextConfig  ,SCREAMING_SNAKE_CASE__      :   BlipVisionConfig  ,**SCREAMING_SNAKE_CASE__      :   Any				):
     return cls(text_config=text_config.to_dict()  ,vision_config=vision_config.to_dict()  ,**a_				)







   def      __UpperCamelCase    (			self      :   Optional[Any]				):
     SCREAMING_SNAKE_CASE:Optional[int]       =       copy.deepcopy(self.__dict__				)
     SCREAMING_SNAKE_CASE:Optional[Any]       =       self.text_config.to_dict()
     SCREAMING_SNAKE_CASE:Dict       =       self.vision_config.to_dict()
     SCREAMING_SNAKE_CASE:Any       =       self.__class__.model_type
     return output













"""simple docstring"""






import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np



class 					__UpperCAmelCase							:
				def       UpperCAmelCase      (		self    :			List[str]    )	->	List[str]:



							'''simple docstring'''



							torch.manual_seed(0    )
							a__  :	List[str]		 =			TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

							torch.manual_seed(0    )
							a__  :	int		 =			AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

							torch.manual_seed(0    )
							a__  :	Any		 =			UNetaDConditionModel(
							    sample_size=32	, layers_per_block=1	, block_out_channels=[32, 64]	, down_block_types=[
							        "ResnetDownsampleBlock2D",
							        "SimpleCrossAttnDownBlock2D",
							    ]	, mid_block_type="UNetMidBlock2DSimpleCrossAttn"	, up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]	, in_channels=3	, out_channels=6	, cross_attention_dim=32	, encoder_hid_dim=32	, attention_head_dim=8	, addition_embed_type="text"	, addition_embed_type_num_heads=2	, cross_attention_norm="group_norm"	, resnet_time_scale_shift="scale_shift"	, act_fn="gelu"	, )
							unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

							torch.manual_seed(0    )
							a__  :	str		 =			DDPMScheduler(
							    num_train_timesteps=10_00	, beta_schedule="squaredcos_cap_v2"	, beta_start=0.0001	, beta_end=0.02	, thresholding=a_	, dynamic_thresholding_ratio=0.95	, sample_max_value=1.0	, prediction_type="epsilon"	, variance_type="learned_range"	, )

							torch.manual_seed(0    )
							a__  :	Optional[int]		 =			IFWatermarker()

							return {
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "unet": unet,
							    "scheduler": scheduler,
							    "watermarker": watermarker,
							    "safety_checker": None,
							    "feature_extractor": None,
							}
				def       UpperCAmelCase      (		self    :			Optional[Any]    )	->	Union[str, Any]:



							'''simple docstring'''



							torch.manual_seed(0    )
							a__  :	List[Any]		 =			TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

							torch.manual_seed(0    )
							a__  :	List[str]		 =			AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

							torch.manual_seed(0    )
							a__  :	Dict		 =			UNetaDConditionModel(
							    sample_size=32	, layers_per_block=[1, 2]	, block_out_channels=[32, 64]	, down_block_types=[
							        "ResnetDownsampleBlock2D",
							        "SimpleCrossAttnDownBlock2D",
							    ]	, mid_block_type="UNetMidBlock2DSimpleCrossAttn"	, up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]	, in_channels=6	, out_channels=6	, cross_attention_dim=32	, encoder_hid_dim=32	, attention_head_dim=8	, addition_embed_type="text"	, addition_embed_type_num_heads=2	, cross_attention_norm="group_norm"	, resnet_time_scale_shift="scale_shift"	, act_fn="gelu"	, class_embed_type="timestep"	, mid_block_scale_factor=1.414	, time_embedding_act_fn="gelu"	, time_embedding_dim=32	, )
							unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

							torch.manual_seed(0    )
							a__  :	Optional[int]		 =			DDPMScheduler(
							    num_train_timesteps=10_00	, beta_schedule="squaredcos_cap_v2"	, beta_start=0.0001	, beta_end=0.02	, thresholding=a_	, dynamic_thresholding_ratio=0.95	, sample_max_value=1.0	, prediction_type="epsilon"	, variance_type="learned_range"	, )

							torch.manual_seed(0    )
							a__  :	Tuple		 =			DDPMScheduler(
							    num_train_timesteps=10_00	, beta_schedule="squaredcos_cap_v2"	, beta_start=0.0001	, beta_end=0.02	, )

							torch.manual_seed(0    )
							a__  :	Optional[Any]		 =			IFWatermarker()

							return {
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "unet": unet,
							    "scheduler": scheduler,
							    "image_noising_scheduler": image_noising_scheduler,
							    "watermarker": watermarker,
							    "safety_checker": None,
							    "feature_extractor": None,
							}
				def       UpperCAmelCase      (		self    :			List[Any]    )	->	Tuple:



							'''simple docstring'''



							a__  :	Dict		 =			self.get_dummy_components()
							a__  :	Any		 =			self.pipeline_class(**a_    )
							pipe.to(a_    )
							pipe.set_progress_bar_config(disable=a_    )

							a__  :	Any		 =			self.get_dummy_inputs(a_    )

							a__  :	Optional[int]		 =			inputs["prompt"]
							a__  :	List[Any]		 =			inputs["generator"]
							a__  :	Optional[int]		 =			inputs["num_inference_steps"]
							a__  :	Any		 =			inputs["output_type"]

							if "image" in inputs:
										a__  :	Any		 =			inputs["image"]
							else:
										a__  :	Dict		 =			None

							if "mask_image" in inputs:
										a__  :	Optional[int]		 =			inputs["mask_image"]
							else:
										a__  :	Any		 =			None

							if "original_image" in inputs:
										a__  :	List[Any]		 =			inputs["original_image"]
							else:
										a__  :	str		 =			None

							a__ ,	a__  :	Optional[int]		 =			pipe.encode_prompt(a_    )

							# inputs with prompt converted to embeddings
							a__  :	Union[str, Any]		 =			{
							    "prompt_embeds": prompt_embeds,
							    "negative_prompt_embeds": negative_prompt_embeds,
							    "generator": generator,
							    "num_inference_steps": num_inference_steps,
							    "output_type": output_type,
							}

							if image is not None:
										a__  :	Dict		 =			image

							if mask_image is not None:
										a__  :	Any		 =			mask_image

							if original_image is not None:
										a__  :	Optional[int]		 =			original_image

							# set all optional components to None
							for optional_component in pipe._optional_components:
										setattr(a_	, a_	, a_    )

							a__  :	int		 =			pipe(**a_    )[0]

							with tempfile.TemporaryDirectory() as tmpdir:
										pipe.save_pretrained(a_    )
										a__  :	List[str]		 =			self.pipeline_class.from_pretrained(a_    )
										pipe_loaded.to(a_    )
										pipe_loaded.set_progress_bar_config(disable=a_    )

							pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

							for optional_component in pipe._optional_components:
										self.assertTrue(
										    getattr(a_	, a_    ) is None	, F"`{optional_component}` did not stay set to None after loading."	, )

							a__  :	Union[str, Any]		 =			self.get_dummy_inputs(a_    )

							a__  :	str		 =			inputs["generator"]
							a__  :	Dict		 =			inputs["num_inference_steps"]
							a__  :	Optional[int]		 =			inputs["output_type"]

							# inputs with prompt converted to embeddings
							a__  :	List[Any]		 =			{
							    "prompt_embeds": prompt_embeds,
							    "negative_prompt_embeds": negative_prompt_embeds,
							    "generator": generator,
							    "num_inference_steps": num_inference_steps,
							    "output_type": output_type,
							}

							if image is not None:
										a__  :	Dict		 =			image

							if mask_image is not None:
										a__  :	Any		 =			mask_image

							if original_image is not None:
										a__  :	Dict		 =			original_image

							a__  :	Optional[Any]		 =			pipe_loaded(**a_    )[0]

							a__  :	int		 =			np.abs(to_np(a_    ) - to_np(a_    )    ).max()
							self.assertLess(a_	, 1E-4    )






				def       UpperCAmelCase      (		self    :			int    )	->	Any:



							'''simple docstring'''



							a__  :	Dict		 =			self.get_dummy_components()
							a__  :	Dict		 =			self.pipeline_class(**a_    )
							pipe.to(a_    )
							pipe.set_progress_bar_config(disable=a_    )

							a__  :	List[str]		 =			self.get_dummy_inputs(a_    )
							a__  :	Dict		 =			pipe(**a_    )[0]

							with tempfile.TemporaryDirectory() as tmpdir:
										pipe.save_pretrained(a_    )
										a__  :	str		 =			self.pipeline_class.from_pretrained(a_    )
										pipe_loaded.to(a_    )
										pipe_loaded.set_progress_bar_config(disable=a_    )

							pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

							a__  :	Optional[int]		 =			self.get_dummy_inputs(a_    )
							a__  :	Optional[int]		 =			pipe_loaded(**a_    )[0]

							a__  :	List[Any]		 =			np.abs(to_np(a_    ) - to_np(a_    )    ).max()
							self.assertLess(a_	, 1E-4    )