






"""simple docstring"""



from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin

@dataclass
class        a     (   UpperCAmelCase__	):
    UpperCAmelCase_ :     Union[str, Any]	     =42


class        a     (   UpperCAmelCase__,				UpperCAmelCase__	):







    @register_to_config
    def __init__(			self							,		_lowerCamelCase = 3_2							,		_lowerCamelCase = 6_4							,		_lowerCamelCase = 2_0							,		_lowerCamelCase = 7_6_8							,		_lowerCamelCase=7_7							,		_lowerCamelCase=4							,		_lowerCamelCase = 0.0							,		_lowerCamelCase = "silu"							,		_lowerCamelCase = None							,		_lowerCamelCase = None							,		_lowerCamelCase = "linear"							,		_lowerCamelCase = "prd"							,		_lowerCamelCase = None							,		_lowerCamelCase = None							,		_lowerCamelCase = None							,		):
      super().__init__()
      lowercase							    =       num_attention_heads
      lowercase							    =       attention_head_dim
      lowercase							    =       num_attention_heads * attention_head_dim
      lowercase							    =       additional_embeddings

      lowercase							    =       time_embed_dim or inner_dim
      lowercase							    =       embedding_proj_dim or embedding_dim
      lowercase							    =       clip_embed_dim or embedding_dim

      lowercase							    =       Timesteps(UpperCAmelCase__							,		UpperCAmelCase__							,		0       )
      lowercase							    =       TimestepEmbedding(UpperCAmelCase__							,		UpperCAmelCase__							,		out_dim=UpperCAmelCase__							,		act_fn=UpperCAmelCase__       )

      lowercase							    =       nn.Linear(UpperCAmelCase__							,		UpperCAmelCase__       )

      if embedding_proj_norm_type is None:
        lowercase							    =       None
      elif embedding_proj_norm_type == "layer":
        lowercase							    =       nn.LayerNorm(UpperCAmelCase__       )
      else:
        raise ValueError(F'unsupported embedding_proj_norm_type: {embedding_proj_norm_type}'       )

      lowercase							    =       nn.Linear(UpperCAmelCase__							,		UpperCAmelCase__       )

      if encoder_hid_proj_type is None:
        lowercase							    =       None
      elif encoder_hid_proj_type == "linear":
        lowercase							    =       nn.Linear(UpperCAmelCase__							,		UpperCAmelCase__       )
      else:
        raise ValueError(F'unsupported encoder_hid_proj_type: {encoder_hid_proj_type}'       )

      lowercase							    =       nn.Parameter(torch.zeros(1							,		num_embeddings + additional_embeddings							,		UpperCAmelCase__       )       )

      if added_emb_type == "prd":
        lowercase							    =       nn.Parameter(torch.zeros(1							,		1							,		UpperCAmelCase__       )       )
      elif added_emb_type is None:
        lowercase							    =       None
      else:
        raise ValueError(
            F'`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `\'prd\'` or `None`.'       )

      lowercase							    =       nn.ModuleList(
          [
              BasicTransformerBlock(
                  UpperCAmelCase__							,		UpperCAmelCase__							,		UpperCAmelCase__							,		dropout=UpperCAmelCase__							,		activation_fn='gelu'							,		attention_bias=UpperCAmelCase__							,		)
              for d in range(UpperCAmelCase__       )
          ]       )

      if norm_in_type == "layer":
        lowercase							    =       nn.LayerNorm(UpperCAmelCase__       )
      elif norm_in_type is None:
        lowercase							    =       None
      else:
        raise ValueError(F'Unsupported norm_in_type: {norm_in_type}.'       )

      lowercase							    =       nn.LayerNorm(UpperCAmelCase__       )

      lowercase							    =       nn.Linear(UpperCAmelCase__							,		UpperCAmelCase__       )

      lowercase							    =       torch.full(
          [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]							,		-1_0_0_0_0.0       )
      causal_attention_mask.triu_(1       )
      lowercase							    =       causal_attention_mask[None, ...]
      self.register_buffer('causal_attention_mask'							,		UpperCAmelCase__							,		persistent=UpperCAmelCase__       )

      lowercase							    =       nn.Parameter(torch.zeros(1							,		UpperCAmelCase__       )       )
      lowercase							    =       nn.Parameter(torch.zeros(1							,		UpperCAmelCase__       )       )







    @property
    # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
    def 			UpperCamelCase_      (			self       ):
      lowercase							    =       {}

      def fn_recursive_add_processors(_lowerCamelCase							,		_lowerCamelCase							,		_lowerCamelCase       ):
        if hasattr(UpperCAmelCase__							,		'set_processor'       ):
          lowercase							    =       module.processor

        for sub_name, child in module.named_children():
          fn_recursive_add_processors(F'{name}.{sub_name}'							,		UpperCAmelCase__							,		UpperCAmelCase__       )

        return processors

      for name, module in self.named_children():
        fn_recursive_add_processors(UpperCAmelCase__							,		UpperCAmelCase__							,		UpperCAmelCase__       )

      return processors







    def 			UpperCamelCase_      (			self							,		_lowerCamelCase       ):
      lowercase							    =       len(self.attn_processors.keys()       )

      if isinstance(UpperCAmelCase__							,		UpperCAmelCase__       ) and len(UpperCAmelCase__       ) != count:
        raise ValueError(
            F'A dict of processors was passed, but the number of processors {len(UpperCAmelCase__       )} does not match the'
            F' number of attention layers: {count}. Please make sure to pass {count} processor classes.'       )

      def fn_recursive_attn_processor(_lowerCamelCase							,		_lowerCamelCase							,		_lowerCamelCase       ):
        if hasattr(UpperCAmelCase__							,		'set_processor'       ):
          if not isinstance(UpperCAmelCase__							,		UpperCAmelCase__       ):
            module.set_processor(UpperCAmelCase__       )
          else:
            module.set_processor(processor.pop(F'{name}.processor'       )       )

        for sub_name, child in module.named_children():
          fn_recursive_attn_processor(F'{name}.{sub_name}'							,		UpperCAmelCase__							,		UpperCAmelCase__       )

      for name, module in self.named_children():
        fn_recursive_attn_processor(UpperCAmelCase__							,		UpperCAmelCase__							,		UpperCAmelCase__       )







    def 			UpperCamelCase_      (			self       ):
      self.set_attn_processor(AttnProcessor()       )







    def 			UpperCamelCase_      (			self							,		_lowerCamelCase							,		_lowerCamelCase							,		_lowerCamelCase							,		_lowerCamelCase = None							,		_lowerCamelCase = None							,		_lowerCamelCase = True							,		):
      lowercase							    =       hidden_states.shape[0]

      lowercase							    =       timestep
      if not torch.is_tensor(UpperCAmelCase__       ):
        lowercase							    =       torch.tensor([timesteps]							,		dtype=torch.long							,		device=hidden_states.device       )
      elif torch.is_tensor(UpperCAmelCase__       ) and len(timesteps.shape       ) == 0:
        lowercase							    =       timesteps[None].to(hidden_states.device       )

      # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
      lowercase							    =       timesteps * torch.ones(UpperCAmelCase__							,		dtype=timesteps.dtype							,		device=timesteps.device       )

      lowercase							    =       self.time_proj(UpperCAmelCase__       )

      # timesteps does not contain any weights and will always return f32 tensors
      # but time_embedding might be fp16, so we need to cast here.
      lowercase							    =       timesteps_projected.to(dtype=self.dtype       )
      lowercase							    =       self.time_embedding(UpperCAmelCase__       )

      if self.embedding_proj_norm is not None:
        lowercase							    =       self.embedding_proj_norm(UpperCAmelCase__       )

      lowercase							    =       self.embedding_proj(UpperCAmelCase__       )
      if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
        lowercase							    =       self.encoder_hidden_states_proj(UpperCAmelCase__       )
      elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
        raise ValueError('`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set'       )

      lowercase							    =       self.proj_in(UpperCAmelCase__       )

      lowercase							    =       self.positional_embedding.to(hidden_states.dtype       )

      lowercase							    =       []
      lowercase							    =       0

      if encoder_hidden_states is not None:
        additional_embeds.append(UpperCAmelCase__       )
        additional_embeddings_len += encoder_hidden_states.shape[1]

      if len(proj_embeddings.shape       ) == 2:
        lowercase							    =       proj_embeddings[:, None, :]

      if len(hidden_states.shape       ) == 2:
        lowercase							    =       hidden_states[:, None, :]

      lowercase							    =       additional_embeds + [
          proj_embeddings,
          time_embeddings[:, None, :],
          hidden_states,
      ]

      if self.prd_embedding is not None:
        lowercase							    =       self.prd_embedding.to(hidden_states.dtype       ).expand(UpperCAmelCase__							,		-1							,		-1       )
        additional_embeds.append(UpperCAmelCase__       )

      lowercase							    =       torch.cat(
          UpperCAmelCase__							,		dim=1							,		)

      # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
      lowercase							    =       additional_embeddings_len + proj_embeddings.shape[1] + 1
      if positional_embeddings.shape[1] < hidden_states.shape[1]:
        lowercase							    =       F.pad(
            UpperCAmelCase__							,		(
                0,
                0,
                additional_embeddings_len,
                self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
            )							,		value=0.0							,		)

      lowercase							    =       hidden_states + positional_embeddings

      if attention_mask is not None:
        lowercase							    =       (1 - attention_mask.to(hidden_states.dtype       )) * -1_0_0_0_0.0
        lowercase							    =       F.pad(UpperCAmelCase__							,		(0, self.additional_embeddings)							,		value=0.0       )
        lowercase							    =       (attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype       )
        lowercase							    =       attention_mask.repeat_interleave(self.config.num_attention_heads							,		dim=0       )

      if self.norm_in is not None:
        lowercase							    =       self.norm_in(UpperCAmelCase__       )

      for block in self.transformer_blocks:
        lowercase							    =       block(UpperCAmelCase__							,		attention_mask=UpperCAmelCase__       )

      lowercase							    =       self.norm_out(UpperCAmelCase__       )

      if self.prd_embedding is not None:
        lowercase							    =       hidden_states[:, -1]
      else:
        lowercase							    =       hidden_states[:, additional_embeddings_len:]

      lowercase							    =       self.proj_to_clip_embeddings(UpperCAmelCase__       )

      if not return_dict:
        return (predicted_image_embedding,)

      return PriorTransformerOutput(predicted_image_embedding=UpperCAmelCase__       )







    def 			UpperCamelCase_      (			self							,		_lowerCamelCase       ):
      lowercase							    =       (prior_latents * self.clip_std) + self.clip_mean
      return prior_latents








from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_torch_available,
    is_vision_available,
)


_lowerCamelCase	:		int							       =   {
    """configuration_blip""": [
        """BLIP_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """BlipConfig""",
        """BlipTextConfig""",
        """BlipVisionConfig""",
    ],
    """processing_blip""": ["""BlipProcessor"""],
}

try:
  if not is_vision_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  _lowerCamelCase	:		Tuple							       =   ["""BlipImageProcessor"""]


try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  _lowerCamelCase	:		List[Any]							       =   [
      """BLIP_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """BlipModel""",
      """BlipPreTrainedModel""",
      """BlipForConditionalGeneration""",
      """BlipForQuestionAnswering""",
      """BlipVisionModel""",
      """BlipTextModel""",
      """BlipForImageTextRetrieval""",
  ]

try:
  if not is_tf_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  _lowerCamelCase	:		Optional[Any]							       =   [
      """TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST""",
      """TFBlipModel""",
      """TFBlipPreTrainedModel""",
      """TFBlipForConditionalGeneration""",
      """TFBlipForQuestionAnswering""",
      """TFBlipVisionModel""",
      """TFBlipTextModel""",
      """TFBlipForImageTextRetrieval""",
  ]

if TYPE_CHECKING:
  from .configuration_blip import BLIP_PRETRAINED_CONFIG_ARCHIVE_MAP, BlipConfig, BlipTextConfig, BlipVisionConfig
  from .processing_blip import BlipProcessor

  try:
    if not is_vision_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .image_processing_blip import BlipImageProcessor

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_blip import (
        BLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
        BlipForConditionalGeneration,
        BlipForImageTextRetrieval,
        BlipForQuestionAnswering,
        BlipModel,
        BlipPreTrainedModel,
        BlipTextModel,
        BlipVisionModel,
    )

  try:
    if not is_tf_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_tf_blip import (
        TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
        TFBlipForConditionalGeneration,
        TFBlipForImageTextRetrieval,
        TFBlipForQuestionAnswering,
        TFBlipModel,
        TFBlipPreTrainedModel,
        TFBlipTextModel,
        TFBlipVisionModel,
    )

else:
  import sys

  _lowerCamelCase	:		List[Any]							       =   _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)


