


'''simple docstring'''



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      :Dict			    =       logging.get_logger(__name__)

__UpperCAmelCase      :Optional[int]			    =       {
    "Salesforce/instruct-blip-flan-t5": "https://huggingface.co/Salesforce/instruct-blip-flan-t5/resolve/main/config.json",
}



class      a    (		_a	):


     """simple docstring"""






     SCREAMING_SNAKE_CASE	:			Tuple							       =       "instructblip_vision_model"

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

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






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

          __UpperCAmelCase							,       __UpperCAmelCase     :			Any  							=      cls.get_config_dict(snake_case							,			**snake_case      )

          # get the vision config dict if we are loading from InstructBlipConfig
          if config_dict.get('''model_type'''      ) == "instructblip":
               __UpperCAmelCase     :			int  							=      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(snake_case							,			**snake_case      )



class      a    (		_a	):


     """simple docstring"""






     SCREAMING_SNAKE_CASE	:			Optional[int]							       =       "instructblip_qformer"

     def __init__(    self					:		str							,			snake_case					:		List[str]=3_0522							,			snake_case					:		Optional[int]=768							,			snake_case					:		Tuple=12							,			snake_case					:		str=12							,			snake_case					:		Union[str, Any]=3072							,			snake_case					:		Any="gelu"							,			snake_case					:		Dict=0.1							,			snake_case					:		Dict=0.1							,			snake_case					:		List[str]=512							,			snake_case					:		Optional[int]=0.02							,			snake_case					:		Union[str, Any]=1E-12							,			snake_case					:		Tuple=0							,			snake_case					:		Union[str, Any]="absolute"							,			snake_case					:		Tuple=2							,			snake_case					:		List[Any]=1408							,			**snake_case					:		Optional[int]							,			)						->							Dict:
          super().__init__(pad_token_id=snake_case							,			**snake_case      )

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






     @classmethod
     def 			lowerCamelCase__			(    cls					:		Optional[int]							,			snake_case					:		Union[str, os.PathLike]							,			**snake_case					:		Dict      )						->							"PretrainedConfig":
          cls._set_token_in_kwargs(snake_case      )

          __UpperCAmelCase							,       __UpperCAmelCase     :			int  							=      cls.get_config_dict(snake_case							,			**snake_case      )

          # get the qformer config dict if we are loading from InstructBlipConfig
          if config_dict.get('''model_type'''      ) == "instructblip":
               __UpperCAmelCase     :			Tuple  							=      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(snake_case							,			**snake_case      )



class      a    (		_a	):


     """simple docstring"""






     SCREAMING_SNAKE_CASE	:			Optional[Any]							       =       "instructblip"
     SCREAMING_SNAKE_CASE	:			List[str]							       =       True

     def __init__(    self					:		List[Any]							,			snake_case					:		Optional[Any]=None							,			snake_case					:		Tuple=None							,			snake_case					:		Optional[Any]=None							,			snake_case					:		Tuple=32							,			**snake_case					:		Any      )						->							Any:
          super().__init__(**snake_case      )

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

          if qformer_config is None:
               __UpperCAmelCase     :			Tuple  							=      {}
               logger.info('''qformer_config is None. Initializing the InstructBlipQFormerConfig with default values.'''      )

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

          __UpperCAmelCase     :			Optional[Any]  							=      InstructBlipVisionConfig(**snake_case      )
          __UpperCAmelCase     :			Any  							=      InstructBlipQFormerConfig(**snake_case      )
          __UpperCAmelCase     :			Optional[int]  							=      text_config['''model_type'''] if '''model_type''' in text_config else '''opt'''
          __UpperCAmelCase     :			str  							=      CONFIG_MAPPING[text_model_type](**snake_case      )

          __UpperCAmelCase     :			Optional[Any]  							=      self.text_config.tie_word_embeddings
          __UpperCAmelCase     :			Optional[Any]  							=      self.text_config.is_encoder_decoder

          __UpperCAmelCase     :			List[str]  							=      num_query_tokens
          __UpperCAmelCase     :			Tuple  							=      self.vision_config.hidden_size
          __UpperCAmelCase     :			List[Any]  							=      self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
          __UpperCAmelCase     :			Optional[Any]  							=      1.0
          __UpperCAmelCase     :			str  							=      0.02

     @classmethod
     def 			lowerCamelCase__			(    cls					:		Union[str, Any]							,			snake_case					:		InstructBlipVisionConfig							,			snake_case					:		InstructBlipQFormerConfig							,			snake_case					:		PretrainedConfig							,			**snake_case					:		Optional[Any]							,			)						->							str:
          return cls(
              vision_config=vision_config.to_dict()							,			qformer_config=qformer_config.to_dict()							,			text_config=text_config.to_dict()							,			**snake_case							,			)






     def 			lowerCamelCase__			(    self					:		Union[str, Any]      )						->							Dict:
          __UpperCAmelCase     :			Tuple  							=      copy.deepcopy(self.__dict__      )
          __UpperCAmelCase     :			Any  							=      self.vision_config.to_dict()
          __UpperCAmelCase     :			str  							=      self.qformer_config.to_dict()
          __UpperCAmelCase     :			List[str]  							=      self.text_config.to_dict()
          __UpperCAmelCase     :			Dict  							=      self.__class__.model_type
          return output



'''simple docstring'''



import gc
import random
import unittest

import torch

from diffusers import (
    IFImgaImgPipeline,
    IFImgaImgSuperResolutionPipeline,
    IFInpaintingPipeline,
    IFInpaintingSuperResolutionPipeline,
    IFPipeline,
    IFSuperResolutionPipeline,
)
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import floats_tensor, load_numpy, require_torch_gpu, skip_mps, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference
from . import IFPipelineTesterMixin



@skip_mps
class      a    (		_a     ,  _a     ,  unittest.TestCase	):


     """simple docstring"""






     SCREAMING_SNAKE_CASE	:			Dict							       =       IFPipeline
     SCREAMING_SNAKE_CASE	:			Optional[int]							       =       TEXT_TO_IMAGE_PARAMS - {"width", "height", "latents"}
     SCREAMING_SNAKE_CASE	:			Optional[int]							       =       TEXT_TO_IMAGE_BATCH_PARAMS
     SCREAMING_SNAKE_CASE	:			str							       =       PipelineTesterMixin.required_optional_params - {"latents"}

     def 			lowerCamelCase__			(    self					:		Tuple      )						->							Optional[Any]:
          return self._get_dummy_components()

     def 			lowerCamelCase__			(    self					:		Optional[Any]							,			snake_case					:		Optional[int]							,			snake_case					:		List[Any]=0      )						->							Optional[Any]:
          if str(snake_case      ).startswith('''mps'''      ):
               __UpperCAmelCase     :			Optional[Any]  							=      torch.manual_seed(snake_case      )
          else:
               __UpperCAmelCase     :			Dict  							=      torch.Generator(device=snake_case      ).manual_seed(snake_case      )

          __UpperCAmelCase     :			str  							=      {
              '''prompt''': '''A painting of a squirrel eating a burger''',
              '''generator''': generator,
              '''num_inference_steps''': 2,
              '''output_type''': '''numpy''',
          }

          return inputs

     def 			lowerCamelCase__			(    self					:		List[str]      )						->							Optional[int]:
          self._test_save_load_optional_components()

     @unittest.skipIf(torch_device != '''cuda'''							,			reason='''float16 requires CUDA'''      )
     def 			lowerCamelCase__			(    self					:		Dict      )						->							Union[str, Any]:
          # Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
          super().test_save_load_floataa(expected_max_diff=1E-1      )

     def 			lowerCamelCase__			(    self					:		List[str]      )						->							Tuple:
          self._test_attention_slicing_forward_pass(expected_max_diff=1E-2      )

     def 			lowerCamelCase__			(    self					:		Any      )						->							List[str]:
          self._test_save_load_local()

     def 			lowerCamelCase__			(    self					:		Union[str, Any]      )						->							str:
          self._test_inference_batch_single_identical(
              expected_max_diff=1E-2							,			)






     @unittest.skipIf(
         torch_device != '''cuda''' or not is_xformers_available()							,			reason='''XFormers attention is only available with CUDA and `xformers` installed'''							,			)
     def 			lowerCamelCase__			(    self					:		str      )						->							Optional[Any]:
          self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3      )



@slow
@require_torch_gpu
class      a    (		unittest.TestCase	):


     """simple docstring"""






     def 			lowerCamelCase__			(    self					:		Dict      )						->							Dict:
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()

     def 			lowerCamelCase__			(    self					:		Any      )						->							Tuple:
          # if

          __UpperCAmelCase     :			Union[str, Any]  							=      IFPipeline.from_pretrained('''DeepFloyd/IF-I-XL-v1.0'''							,			variant='''fp16'''							,			torch_dtype=torch.floataa      )

          __UpperCAmelCase     :			List[str]  							=      IFSuperResolutionPipeline.from_pretrained(
              '''DeepFloyd/IF-II-L-v1.0'''							,			variant='''fp16'''							,			torch_dtype=torch.floataa							,			text_encoder=snake_case							,			tokenizer=snake_case      )

          # pre compute text embeddings and remove T5 to save memory

          pipe_a.text_encoder.to('''cuda'''      )

          __UpperCAmelCase							,       __UpperCAmelCase     :			Tuple  							=      pipe_a.encode_prompt('''anime turtle'''							,			device='''cuda'''      )

          del pipe_a.tokenizer
          del pipe_a.text_encoder
          gc.collect()

          __UpperCAmelCase     :			Any  							=      None
          __UpperCAmelCase     :			Optional[Any]  							=      None

          pipe_a.enable_model_cpu_offload()
          pipe_a.enable_model_cpu_offload()

          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )
          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )

          self._test_if(snake_case							,			snake_case							,			snake_case							,			snake_case      )

          pipe_a.remove_all_hooks()
          pipe_a.remove_all_hooks()

          # img2img

          __UpperCAmelCase     :			Union[str, Any]  							=      IFImgaImgPipeline(**pipe_a.components      )
          __UpperCAmelCase     :			int  							=      IFImgaImgSuperResolutionPipeline(**pipe_a.components      )

          pipe_a.enable_model_cpu_offload()
          pipe_a.enable_model_cpu_offload()

          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )
          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )

          self._test_if_imgaimg(snake_case							,			snake_case							,			snake_case							,			snake_case      )

          pipe_a.remove_all_hooks()
          pipe_a.remove_all_hooks()

          # inpainting

          __UpperCAmelCase     :			List[str]  							=      IFInpaintingPipeline(**pipe_a.components      )
          __UpperCAmelCase     :			List[str]  							=      IFInpaintingSuperResolutionPipeline(**pipe_a.components      )

          pipe_a.enable_model_cpu_offload()
          pipe_a.enable_model_cpu_offload()

          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )
          pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()      )

          self._test_if_inpainting(snake_case							,			snake_case							,			snake_case							,			snake_case      )

     def 			lowerCamelCase__			(    self					:		List[str]							,			snake_case					:		Any							,			snake_case					:		Union[str, Any]							,			snake_case					:		Optional[int]							,			snake_case					:		str      )						->							Optional[int]:
          # pipeline 1

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			Dict  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )
          __UpperCAmelCase     :			List[str]  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			num_inference_steps=2							,			generator=snake_case							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			List[Any]  							=      output.images[0]

          assert image.shape == (64, 64, 3)

          __UpperCAmelCase     :			List[str]  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 13 * 10**9

          __UpperCAmelCase     :			Optional[Any]  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )

          # pipeline 2

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			int  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )

          __UpperCAmelCase     :			Optional[int]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )

          __UpperCAmelCase     :			List[Any]  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			generator=snake_case							,			num_inference_steps=2							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			Optional[Any]  							=      output.images[0]

          assert image.shape == (256, 256, 3)

          __UpperCAmelCase     :			List[str]  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 4 * 10**9

          __UpperCAmelCase     :			Tuple  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_superresolution_stage_II.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )

     def 			lowerCamelCase__			(    self					:		Optional[int]							,			snake_case					:		Optional[int]							,			snake_case					:		Union[str, Any]							,			snake_case					:		str							,			snake_case					:		Dict      )						->							str:
          # pipeline 1

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			Optional[Any]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )

          __UpperCAmelCase     :			Dict  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )

          __UpperCAmelCase     :			Dict  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			num_inference_steps=2							,			generator=snake_case							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			int  							=      output.images[0]

          assert image.shape == (64, 64, 3)

          __UpperCAmelCase     :			Any  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 10 * 10**9

          __UpperCAmelCase     :			Optional[Any]  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )

          # pipeline 2

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			Optional[int]  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )

          __UpperCAmelCase     :			Optional[Any]  							=      floats_tensor((1, 3, 256, 256)							,			rng=random.Random(0      )      ).to(snake_case      )
          __UpperCAmelCase     :			List[str]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )

          __UpperCAmelCase     :			List[str]  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			original_image=snake_case							,			generator=snake_case							,			num_inference_steps=2							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			int  							=      output.images[0]

          assert image.shape == (256, 256, 3)

          __UpperCAmelCase     :			int  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 4 * 10**9

          __UpperCAmelCase     :			Any  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img_superresolution_stage_II.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )






     def 			lowerCamelCase__			(    self					:		str							,			snake_case					:		Union[str, Any]							,			snake_case					:		Optional[int]							,			snake_case					:		Optional[Any]							,			snake_case					:		Optional[Any]      )						->							Optional[int]:
          # pipeline 1

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			List[Any]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )
          __UpperCAmelCase     :			List[str]  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(1      )      ).to(snake_case      )

          __UpperCAmelCase     :			Optional[Any]  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )
          __UpperCAmelCase     :			Dict  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			mask_image=snake_case							,			num_inference_steps=2							,			generator=snake_case							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			Union[str, Any]  							=      output.images[0]

          assert image.shape == (64, 64, 3)

          __UpperCAmelCase     :			Tuple  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 10 * 10**9

          __UpperCAmelCase     :			Tuple  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )

          # pipeline 2

          _start_torch_memory_measurement()

          __UpperCAmelCase     :			int  							=      torch.Generator(device='''cpu'''      ).manual_seed(0      )

          __UpperCAmelCase     :			Tuple  							=      floats_tensor((1, 3, 64, 64)							,			rng=random.Random(0      )      ).to(snake_case      )
          __UpperCAmelCase     :			Any  							=      floats_tensor((1, 3, 256, 256)							,			rng=random.Random(0      )      ).to(snake_case      )
          __UpperCAmelCase     :			Tuple  							=      floats_tensor((1, 3, 256, 256)							,			rng=random.Random(1      )      ).to(snake_case      )

          __UpperCAmelCase     :			Union[str, Any]  							=      pipe_a(
              prompt_embeds=snake_case							,			negative_prompt_embeds=snake_case							,			image=snake_case							,			mask_image=snake_case							,			original_image=snake_case							,			generator=snake_case							,			num_inference_steps=2							,			output_type='''np'''							,			)

          __UpperCAmelCase     :			List[Any]  							=      output.images[0]

          assert image.shape == (256, 256, 3)

          __UpperCAmelCase     :			Any  							=      torch.cuda.max_memory_allocated()
          assert mem_bytes < 4 * 10**9

          __UpperCAmelCase     :			str  							=      load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting_superresolution_stage_II.npy'''      )
          assert_mean_pixel_difference(snake_case							,			snake_case      )




def      _a     (						):





     '''simple docstring'''






     torch.cuda.empty_cache()
     torch.cuda.reset_max_memory_allocated()
     torch.cuda.reset_peak_memory_stats()