
"""simple docstring"""





from typing import Dict, List, Optional, Tuple, Union

import torch

from ...models import AutoencoderKL, TransformeraDModel
from ...schedulers import KarrasDiffusionSchedulers
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput
class 						SCREAMING_SNAKE_CASE_ (						__a ):




		"""simple docstring"""





		def __init__(     self							,  lowerCAmelCase__							,  lowerCAmelCase__							,  lowerCAmelCase__							,  lowerCAmelCase__ = None							,  ):
				super().__init__()
				self.register_modules(transformer=lowerCAmelCase__							,  vae=lowerCAmelCase__							,  scheduler=lowerCAmelCase__)

				# create a imagenet -> id dictionary for easier use
				__SCREAMING_SNAKE_CASE            = {}
				if idalabel is not None:
						for key, value in idalabel.items():
								for label in value.split(""","""):
										__SCREAMING_SNAKE_CASE            = int(lowerCAmelCase__)
						__SCREAMING_SNAKE_CASE            = dict(sorted(self.labels.items()))




		def 	snake_case_     (     self							,  lowerCAmelCase__):

				if not isinstance(lowerCAmelCase__							,  lowerCAmelCase__):
						__SCREAMING_SNAKE_CASE            = list(lowerCAmelCase__)

				for l in label:
						if l not in self.labels:
								raise ValueError(
								    f"{l} does not exist. Please make sure to select one of the following labels: \n {self.labels}.")

				return [self.labels[l] for l in label]




		@torch.no_grad()
		def __call__(     self							,  lowerCAmelCase__							,  lowerCAmelCase__ = 4.0							,  lowerCAmelCase__ = None							,  lowerCAmelCase__ = 5_0							,  lowerCAmelCase__ = "pil"							,  lowerCAmelCase__ = True							,  ):
				__SCREAMING_SNAKE_CASE            = len(lowerCAmelCase__)
				__SCREAMING_SNAKE_CASE            = self.transformer.config.sample_size
				__SCREAMING_SNAKE_CASE            = self.transformer.config.in_channels

				__SCREAMING_SNAKE_CASE            = randn_tensor(
				    shape=(batch_size, latent_channels, latent_size, latent_size)							,  generator=lowerCAmelCase__							,  device=self.device							,  dtype=self.transformer.dtype							,  )
				__SCREAMING_SNAKE_CASE            = torch.cat([latents] * 2) if guidance_scale > 1 else latents

				__SCREAMING_SNAKE_CASE            = torch.tensor(lowerCAmelCase__							,  device=self.device).reshape(-1)
				__SCREAMING_SNAKE_CASE            = torch.tensor([1_0_0_0] * batch_size							,  device=self.device)
				__SCREAMING_SNAKE_CASE            = torch.cat([class_labels, class_null]							,  0) if guidance_scale > 1 else class_labels

				# set step values
				self.scheduler.set_timesteps(lowerCAmelCase__)

				for t in self.progress_bar(self.scheduler.timesteps):
						if guidance_scale > 1:
								__SCREAMING_SNAKE_CASE            = latent_model_input[: len(lowerCAmelCase__) // 2]
								__SCREAMING_SNAKE_CASE            = torch.cat([half, half]							,  dim=0)
						__SCREAMING_SNAKE_CASE            = self.scheduler.scale_model_input(lowerCAmelCase__							,  lowerCAmelCase__)

						__SCREAMING_SNAKE_CASE            = t
						if not torch.is_tensor(lowerCAmelCase__):
								# TODO: this requires sync between CPU and GPU. So try to pass timesteps as tensors if you can
								# This would be a good case for the `match` statement (Python 3.10+)
								__SCREAMING_SNAKE_CASE            = latent_model_input.device.type == """mps"""
								if isinstance(lowerCAmelCase__							,  lowerCAmelCase__):
										__SCREAMING_SNAKE_CASE            = torch.floataa if is_mps else torch.floataa
								else:
										__SCREAMING_SNAKE_CASE            = torch.intaa if is_mps else torch.intaa
								__SCREAMING_SNAKE_CASE            = torch.tensor([timesteps]							,  dtype=lowerCAmelCase__							,  device=latent_model_input.device)
						elif len(timesteps.shape) == 0:
								__SCREAMING_SNAKE_CASE            = timesteps[None].to(latent_model_input.device)
						# broadcast to batch dimension in a way that's compatible with ONNX/Core ML
						__SCREAMING_SNAKE_CASE            = timesteps.expand(latent_model_input.shape[0])
						# predict noise model_output
						__SCREAMING_SNAKE_CASE            = self.transformer(
						    lowerCAmelCase__							,  timestep=lowerCAmelCase__							,  class_labels=lowerCAmelCase__).sample

						# perform guidance
						if guidance_scale > 1:
								__SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE            = noise_pred[:, :latent_channels], noise_pred[:, latent_channels:]
								__SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE            = torch.split(lowerCAmelCase__							,  len(lowerCAmelCase__) // 2							,  dim=0)

								__SCREAMING_SNAKE_CASE            = uncond_eps + guidance_scale * (cond_eps - uncond_eps)
								__SCREAMING_SNAKE_CASE            = torch.cat([half_eps, half_eps]							,  dim=0)

								__SCREAMING_SNAKE_CASE            = torch.cat([eps, rest]							,  dim=1)

						# learned sigma
						if self.transformer.config.out_channels // 2 == latent_channels:
								__SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE            = torch.split(lowerCAmelCase__							,  lowerCAmelCase__							,  dim=1)
						else:
								__SCREAMING_SNAKE_CASE            = noise_pred

						# compute previous image: x_t -> x_t-1
						__SCREAMING_SNAKE_CASE            = self.scheduler.step(lowerCAmelCase__							,  lowerCAmelCase__							,  lowerCAmelCase__).prev_sample

				if guidance_scale > 1:
						__SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE            = latent_model_input.chunk(2							,  dim=0)
				else:
						__SCREAMING_SNAKE_CASE            = latent_model_input

				__SCREAMING_SNAKE_CASE            = 1 / self.vae.config.scaling_factor * latents
				__SCREAMING_SNAKE_CASE            = self.vae.decode(lowerCAmelCase__).sample

				__SCREAMING_SNAKE_CASE            = (samples / 2 + 0.5).clamp(0							,  1)

				# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
				__SCREAMING_SNAKE_CASE            = samples.cpu().permute(0							,  2							,  3							,  1).float().numpy()

				if output_type == "pil":
						__SCREAMING_SNAKE_CASE            = self.numpy_to_pil(lowerCAmelCase__)

				if not return_dict:
						return (samples,)

				return ImagePipelineOutput(images=lowerCAmelCase__)






"""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


__magic_name__       =			logging.get_logger(__name__)

__magic_name__       =			{
    "salesforce/blip2-opt-2.7b": "https://huggingface.co/salesforce/blip2-opt-2.7b/resolve/main/config.json",
}
class 						SCREAMING_SNAKE_CASE_ (						__a ):




		"""simple docstring"""





		__lowercase							:  Optional[Any]						   =		'''blip_2_vision_model'''




		def __init__(     self							,  lowerCAmelCase__=1_4_0_8							,  lowerCAmelCase__=6_1_4_4							,  lowerCAmelCase__=3_9							,  lowerCAmelCase__=1_6							,  lowerCAmelCase__=2_2_4							,  lowerCAmelCase__=1_4							,  lowerCAmelCase__="gelu"							,  lowerCAmelCase__=0.0_00_01							,  lowerCAmelCase__=0.0							,  lowerCAmelCase__=1E-10							,  lowerCAmelCase__=True							,  **lowerCAmelCase__							,  ):
				super().__init__(**lowerCAmelCase__)

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




		@classmethod
		def 	snake_case_     (     cls							,  lowerCAmelCase__							,  **lowerCAmelCase__):
				cls._set_token_in_kwargs(lowerCAmelCase__)

				__SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE            = cls.get_config_dict(lowerCAmelCase__							,  **lowerCAmelCase__)

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



class 						SCREAMING_SNAKE_CASE_ (						__a ):




		"""simple docstring"""





		__lowercase							:  Tuple						   =		'''blip_2_qformer'''




		def __init__(     self							,  lowerCAmelCase__=3_0_5_2_2							,  lowerCAmelCase__=7_6_8							,  lowerCAmelCase__=1_2							,  lowerCAmelCase__=1_2							,  lowerCAmelCase__=3_0_7_2							,  lowerCAmelCase__="gelu"							,  lowerCAmelCase__=0.1							,  lowerCAmelCase__=0.1							,  lowerCAmelCase__=5_1_2							,  lowerCAmelCase__=0.02							,  lowerCAmelCase__=1E-12							,  lowerCAmelCase__=0							,  lowerCAmelCase__="absolute"							,  lowerCAmelCase__=2							,  lowerCAmelCase__=1_4_0_8							,  **lowerCAmelCase__							,  ):
				super().__init__(pad_token_id=lowerCAmelCase__							,  **lowerCAmelCase__)

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




		@classmethod
		def 	snake_case_     (     cls							,  lowerCAmelCase__							,  **lowerCAmelCase__):
				cls._set_token_in_kwargs(lowerCAmelCase__)

				__SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE            = cls.get_config_dict(lowerCAmelCase__							,  **lowerCAmelCase__)

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



class 						SCREAMING_SNAKE_CASE_ (						__a ):




		"""simple docstring"""





		__lowercase							:  Optional[Any]						   =		'''blip-2'''
		__lowercase							:  Any						   =		True




		def __init__(     self							,  lowerCAmelCase__=None							,  lowerCAmelCase__=None							,  lowerCAmelCase__=None							,  lowerCAmelCase__=3_2							,  **lowerCAmelCase__):
				super().__init__(**lowerCAmelCase__)

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

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

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

				__SCREAMING_SNAKE_CASE            = BlipaVisionConfig(**lowerCAmelCase__)
				__SCREAMING_SNAKE_CASE            = BlipaQFormerConfig(**lowerCAmelCase__)
				__SCREAMING_SNAKE_CASE            = text_config["""model_type"""] if """model_type""" in text_config else """opt"""
				__SCREAMING_SNAKE_CASE            = CONFIG_MAPPING[text_model_type](**lowerCAmelCase__)

				__SCREAMING_SNAKE_CASE            = self.text_config.tie_word_embeddings
				__SCREAMING_SNAKE_CASE            = self.text_config.is_encoder_decoder

				__SCREAMING_SNAKE_CASE            = num_query_tokens
				__SCREAMING_SNAKE_CASE            = self.vision_config.hidden_size
				__SCREAMING_SNAKE_CASE            = self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
				__SCREAMING_SNAKE_CASE            = 1.0
				__SCREAMING_SNAKE_CASE            = 0.02




		@classmethod
		def 	snake_case_     (     cls							,  lowerCAmelCase__							,  lowerCAmelCase__							,  lowerCAmelCase__							,  **lowerCAmelCase__							,  ):
				return cls(
				    vision_config=vision_config.to_dict()							,  qformer_config=qformer_config.to_dict()							,  text_config=text_config.to_dict()							,  **lowerCAmelCase__							,  )




		def 	snake_case_     (     self):
				__SCREAMING_SNAKE_CASE            = copy.deepcopy(self.__dict__)
				__SCREAMING_SNAKE_CASE            = self.vision_config.to_dict()
				__SCREAMING_SNAKE_CASE            = self.qformer_config.to_dict()
				__SCREAMING_SNAKE_CASE            = self.text_config.to_dict()
				__SCREAMING_SNAKE_CASE            = self.__class__.model_type
				return output




