




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 (  _lowerCamelCase				):
  _UpperCamelCase       :      torch.FloatTensor


class 						_A (  _lowerCamelCase      ,       _lowerCamelCase				):






  @register_to_config
  def __init__(  self		:     List[str]				,						_A		:     int = 32				,						_A		:     int = 64				,						_A		:     int = 20				,						_A		:     int = 768				,						_A		:     Optional[Any]=77				,						_A		:     str=4				,						_A		:     float = 0.0				,						_A		:     str = "silu"				,						_A		:     Optional[str] = None				,						_A		:     Optional[str] = None				,						_A		:     Optional[str] = "linear"				,						_A		:     Optional[str] = "prd"				,						_A		:     Optional[int] = None				,						_A		:     Optional[int] = None				,						_A		:     Optional[int] = None				,						)		->   Any:






       """simple docstring"""







       super().__init__()
       lowercase :       int       =			num_attention_heads
       lowercase :       Dict       =			attention_head_dim
       lowercase :       Dict       =			num_attention_heads * attention_head_dim
       lowercase :       int       =			additional_embeddings

       lowercase :       Dict       =			time_embed_dim or inner_dim
       lowercase :       str       =			embedding_proj_dim or embedding_dim
       lowercase :       str       =			clip_embed_dim or embedding_dim

       lowercase :       Dict       =			Timesteps(_A				,						_A				,						0				)
       lowercase :       Optional[Any]       =			TimestepEmbedding(_A				,						_A				,						out_dim=_A				,						act_fn=_A				)

       lowercase :       List[Any]       =			nn.Linear(_A				,						_A				)

       if embedding_proj_norm_type is None:
            lowercase :       Dict       =			None
       elif embedding_proj_norm_type == "layer":
            lowercase :       List[str]       =			nn.LayerNorm(_A				)
       else:
            raise ValueError(f"""unsupported embedding_proj_norm_type: {embedding_proj_norm_type}"""				)

       lowercase :       str       =			nn.Linear(_A				,						_A				)

       if encoder_hid_proj_type is None:
            lowercase :       Tuple       =			None
       elif encoder_hid_proj_type == "linear":
            lowercase :       int       =			nn.Linear(_A				,						_A				)
       else:
            raise ValueError(f"""unsupported encoder_hid_proj_type: {encoder_hid_proj_type}"""				)

       lowercase :       str       =			nn.Parameter(torch.zeros(1				,						num_embeddings + additional_embeddings				,						_A				)				)

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

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

       if norm_in_type == "layer":
            lowercase :       Any       =			nn.LayerNorm(_A				)
       elif norm_in_type is None:
            lowercase :       List[Any]       =			None
       else:
            raise ValueError(f"""Unsupported norm_in_type: {norm_in_type}."""				)

       lowercase :       Any       =			nn.LayerNorm(_A				)

       lowercase :       Optional[int]       =			nn.Linear(_A				,						_A				)

       lowercase :       Union[str, Any]       =			torch.full(
           [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]				,						-10_000.0				)
       causal_attention_mask.triu_(1				)
       lowercase :       Tuple       =			causal_attention_mask[None, ...]
       self.register_buffer('''causal_attention_mask'''				,						_A				,						persistent=_A				)

       lowercase :       Optional[Any]       =			nn.Parameter(torch.zeros(1				,						_A				)				)
       lowercase :       Union[str, Any]       =			nn.Parameter(torch.zeros(1				,						_A				)				)






  @property
  # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
  def  __a      (  self		:     str				)		->   Dict[str, AttentionProcessor]:






       """simple docstring"""







       lowercase :       Optional[Any]       =			{}

       def fn_recursive_add_processors(_A		:     str				,						_A		:     torch.nn.Module				,						_A		:     Dict[str, AttentionProcessor]				):
            if hasattr(_A				,						'''set_processor'''				):
                 lowercase :       Optional[int]       =			module.processor

            for sub_name, child in module.named_children():
                 fn_recursive_add_processors(f"""{name}.{sub_name}"""				,						_A				,						_A				)

            return processors

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

       return processors






  def  __a      (  self		:     List[str]				,						_A		:     Union[AttentionProcessor, Dict[str, AttentionProcessor]]				)		->   Optional[Any]:






       """simple docstring"""







       lowercase :       Optional[int]       =			len(self.attn_processors.keys()				)

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

       def fn_recursive_attn_processor(_A		:     str				,						_A		:     torch.nn.Module				,						_A		:     str				):
            if hasattr(_A				,						'''set_processor'''				):
                 if not isinstance(_A				,						_A				):
                      module.set_processor(_A				)
                 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}"""				,						_A				,						_A				)

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






  def  __a      (  self		:     Optional[int]				)		->   Optional[int]:






       """simple docstring"""







       self.set_attn_processor(AttnProcessor()				)






  def  __a      (  self		:     Union[str, Any]				,						_A		:     Union[str, Any]				,						_A		:     Union[torch.Tensor, float, int]				,						_A		:     torch.FloatTensor				,						_A		:     Optional[torch.FloatTensor] = None				,						_A		:     Optional[torch.BoolTensor] = None				,						_A		:     bool = True				,						)		->   Any:






       """simple docstring"""







       lowercase :       Optional[Any]       =			hidden_states.shape[0]

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

       # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
       lowercase :       Union[str, Any]       =			timesteps * torch.ones(_A				,						dtype=timesteps.dtype				,						device=timesteps.device				)

       lowercase :       Any       =			self.time_proj(_A				)

       # 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 :       int       =			timesteps_projected.to(dtype=self.dtype				)
       lowercase :       List[str]       =			self.time_embedding(_A				)

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

       lowercase :       int       =			self.embedding_proj(_A				)
       if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
            lowercase :       Union[str, Any]       =			self.encoder_hidden_states_proj(_A				)
       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 :       Dict       =			self.proj_in(_A				)

       lowercase :       Optional[Any]       =			self.positional_embedding.to(hidden_states.dtype				)

       lowercase :       Dict       =			[]
       lowercase :       Any       =			0

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

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

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

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

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

       lowercase :       List[Any]       =			torch.cat(
           _A				,						dim=1				,						)

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

       lowercase :       Any       =			hidden_states + positional_embeddings

       if attention_mask is not None:
            lowercase :       Any       =			(1 - attention_mask.to(hidden_states.dtype				)) * -10_000.0
            lowercase :       Optional[int]       =			F.pad(_A				,						(0, self.additional_embeddings)				,						value=0.0				)
            lowercase :       Tuple       =			(attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype				)
            lowercase :       int       =			attention_mask.repeat_interleave(self.config.num_attention_heads				,						dim=0				)

       if self.norm_in is not None:
            lowercase :       Union[str, Any]       =			self.norm_in(_A				)

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

       lowercase :       int       =			self.norm_out(_A				)

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

       lowercase :       Optional[Any]       =			self.proj_to_clip_embeddings(_A				)

       if not return_dict:
            return (predicted_image_embedding,)

       return PriorTransformerOutput(predicted_image_embedding=_A				)






  def  __a      (  self		:     List[str]				,						_A		:     List[Any]				)		->   List[Any]:






       """simple docstring"""







       lowercase :       Tuple       =			(prior_latents * self.clip_std) + self.clip_mean
       return prior_latents





import os
import sys
import unittest


lowerCAmelCase_						=			os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, 'utils'))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
lowerCAmelCase_						=			os.path.join(git_repo_path, 'src', 'diffusers')
class 						_A (  unittest.TestCase				):






  def  __a      (  self		:     Any				)		->   str:






       """simple docstring"""







       lowercase :       List[str]       =			find_backend('''    if not is_torch_available():'''				)
       self.assertEqual(_A				,						'''torch'''				)

       # backend_with_underscore = find_backend("    if not is_tensorflow_text_available():")
       # self.assertEqual(backend_with_underscore, "tensorflow_text")

       lowercase :       str       =			find_backend('''    if not (is_torch_available() and is_transformers_available()):'''				)
       self.assertEqual(_A				,						'''torch_and_transformers'''				)

       # double_backend_with_underscore = find_backend(
       #    "    if not (is_sentencepiece_available() and is_tensorflow_text_available()):"
       # )
       # self.assertEqual(double_backend_with_underscore, "sentencepiece_and_tensorflow_text")

       lowercase :       str       =			find_backend(
           '''    if not (is_torch_available() and is_transformers_available() and is_onnx_available()):'''				)
       self.assertEqual(_A				,						'''torch_and_transformers_and_onnx'''				)






  def  __a      (  self		:     Optional[int]				)		->   Any:






       """simple docstring"""







       lowercase :       Dict       =			read_init()
       # We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
       self.assertIn('''torch'''				,						_A				)
       self.assertIn('''torch_and_transformers'''				,						_A				)
       self.assertIn('''flax_and_transformers'''				,						_A				)
       self.assertIn('''torch_and_transformers_and_onnx'''				,						_A				)

       # Likewise, we can't assert on the exact content of a key
       self.assertIn('''UNet2DModel'''				,						objects['''torch''']				)
       self.assertIn('''FlaxUNet2DConditionModel'''				,						objects['''flax''']				)
       self.assertIn('''StableDiffusionPipeline'''				,						objects['''torch_and_transformers''']				)
       self.assertIn('''FlaxStableDiffusionPipeline'''				,						objects['''flax_and_transformers''']				)
       self.assertIn('''LMSDiscreteScheduler'''				,						objects['''torch_and_scipy''']				)
       self.assertIn('''OnnxStableDiffusionPipeline'''				,						objects['''torch_and_transformers_and_onnx''']				)






  def  __a      (  self		:     Union[str, Any]				)		->   str:






       """simple docstring"""







       lowercase :       List[str]       =			create_dummy_object('''CONSTANT'''				,						'''\'torch\''''				)
       self.assertEqual(_A				,						'''\nCONSTANT = None\n'''				)

       lowercase :       List[Any]       =			create_dummy_object('''function'''				,						'''\'torch\''''				)
       self.assertEqual(
           _A				,						'''\ndef function(*args, **kwargs):\n    requires_backends(function, \'torch\')\n'''				)

       lowercase :       Tuple       =			'''
class FakeClass(metaclass=DummyObject):
    _backends = \'torch\'

    def __init__(self, *args, **kwargs):
        requires_backends(self, \'torch\')

    @classmethod
    def from_config(cls, *args, **kwargs):
        requires_backends(cls, \'torch\')

    @classmethod
    def from_pretrained(cls, *args, **kwargs):
        requires_backends(cls, \'torch\')
'''
       lowercase :       List[Any]       =			create_dummy_object('''FakeClass'''				,						'''\'torch\''''				)
       self.assertEqual(_A				,						_A				)






  def  __a      (  self		:     Union[str, Any]				)		->   str:






       """simple docstring"""







       lowercase :       Optional[int]       =			'''# This file is autogenerated by the command `make fix-copies`, do not edit.
from ..utils import DummyObject, requires_backends


CONSTANT = None


def function(*args, **kwargs):
    requires_backends(function, ["torch"])


class FakeClass(metaclass=DummyObject):
    _backends = ["torch"]

    def __init__(self, *args, **kwargs):
        requires_backends(self, ["torch"])

    @classmethod
    def from_config(cls, *args, **kwargs):
        requires_backends(cls, ["torch"])

    @classmethod
    def from_pretrained(cls, *args, **kwargs):
        requires_backends(cls, ["torch"])
'''
       lowercase :       int       =			create_dummy_files({'''torch''': ['''CONSTANT''', '''function''', '''FakeClass''']}				)
       self.assertEqual(dummy_files['''torch''']				,						_A				)