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_			(						SCREAMING_SNAKE_CASE						):






 '''simple docstring'''






 UpperCAmelCase_     			= 42






class     a_			(						SCREAMING_SNAKE_CASE	,					SCREAMING_SNAKE_CASE						):






 '''simple docstring'''






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







        '''simple docstring'''



        super().__init__()
        lowerCAmelCase__					     = num_attention_heads
        lowerCAmelCase__					     = attention_head_dim
        lowerCAmelCase__					     = num_attention_heads * attention_head_dim
        lowerCAmelCase__					     = additional_embeddings

        lowerCAmelCase__					     = time_embed_dim or inner_dim
        lowerCAmelCase__					     = embedding_proj_dim or embedding_dim
        lowerCAmelCase__					     = clip_embed_dim or embedding_dim

        lowerCAmelCase__					     = Timesteps(lowercase__      ,   lowercase__      ,   0)
        lowerCAmelCase__					     = TimestepEmbedding(lowercase__      ,   lowercase__      ,   out_dim=lowercase__      ,   act_fn=lowercase__)

        lowerCAmelCase__					     = nn.Linear(lowercase__      ,   lowercase__)

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

        lowerCAmelCase__					     = nn.Linear(lowercase__      ,   lowercase__)

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

        lowerCAmelCase__					     = nn.Parameter(torch.zeros(1      ,   num_embeddings + additional_embeddings      ,   lowercase__))

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

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

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

        lowerCAmelCase__					     = nn.LayerNorm(lowercase__)

        lowerCAmelCase__					     = nn.Linear(lowercase__      ,   lowercase__)

        lowerCAmelCase__					     = torch.full(
            [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]      ,   -10_000.0)
        causal_attention_mask.triu_(1)
        lowerCAmelCase__					     = causal_attention_mask[None, ...]
        self.register_buffer('causal_attention_mask'      ,   lowercase__      ,   persistent=lowercase__)

        lowerCAmelCase__					     = nn.Parameter(torch.zeros(1      ,   lowercase__))
        lowerCAmelCase__					     = nn.Parameter(torch.zeros(1      ,   lowercase__))




 @property
 # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
 def 		__snake_case						(      self :  Dict):







        '''simple docstring'''



        lowerCAmelCase__					     = {}

        def fn_recursive_add_processors(lowercase__ :  str      ,   lowercase__ :  torch.nn.Module      ,   lowercase__ :  Dict[str, AttentionProcessor]):
               if hasattr(lowercase__      ,   'set_processor'):
                      lowerCAmelCase__					     = module.processor

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

               return processors

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

        return processors




 def 		__snake_case						(      self :  List[Any]      ,   lowercase__ :  Union[AttentionProcessor, Dict[str, AttentionProcessor]]):







        '''simple docstring'''



        lowerCAmelCase__					     = len(self.attn_processors.keys())

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

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

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




 def 		__snake_case						(      self :  List[str]):







        '''simple docstring'''



        self.set_attn_processor(AttnProcessor())




 def 		__snake_case						(      self :  List[str]      ,   lowercase__ :  Union[str, Any]      ,   lowercase__ :  Union[torch.Tensor, float, int]      ,   lowercase__ :  torch.FloatTensor      ,   lowercase__ :  Optional[torch.FloatTensor] = None      ,   lowercase__ :  Optional[torch.BoolTensor] = None      ,   lowercase__ :  bool = True      ,   ):







        '''simple docstring'''



        lowerCAmelCase__					     = hidden_states.shape[0]

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

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

        lowerCAmelCase__					     = self.time_proj(lowercase__)

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

        if self.embedding_proj_norm is not None:
               lowerCAmelCase__					     = self.embedding_proj_norm(lowercase__)

        lowerCAmelCase__					     = self.embedding_proj(lowercase__)
        if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
               lowerCAmelCase__					     = self.encoder_hidden_states_proj(lowercase__)
        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')

        lowerCAmelCase__					     = self.proj_in(lowercase__)

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

        lowerCAmelCase__					     = []
        lowerCAmelCase__					     = 0

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

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

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

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

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

        lowerCAmelCase__					     = torch.cat(
            lowercase__      ,   dim=1      ,   )

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

        lowerCAmelCase__					     = hidden_states + positional_embeddings

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

        if self.norm_in is not None:
               lowerCAmelCase__					     = self.norm_in(lowercase__)

        for block in self.transformer_blocks:
               lowerCAmelCase__					     = block(lowercase__      ,   attention_mask=lowercase__)

        lowerCAmelCase__					     = self.norm_out(lowercase__)

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

        lowerCAmelCase__					     = self.proj_to_clip_embeddings(lowercase__)

        if not return_dict:
               return (predicted_image_embedding,)

        return PriorTransformerOutput(predicted_image_embedding=lowercase__)




 def 		__snake_case						(      self :  Optional[Any]      ,   lowercase__ :  Union[str, Any]):







        '''simple docstring'''



        lowerCAmelCase__					     = (prior_latents * self.clip_std) + self.clip_mean
        return prior_latents
lowerCAmelCase__     						=			'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/'






def 				__lowerCamelCase		(						lowerCAmelCase__     ):
       # Make sure the supplied data is a bytes-like object
       if not isinstance(lowerCAmelCase__     ,  lowerCAmelCase__     ):
              lowerCAmelCase__					     = F"""a bytes-like object is required, not '{data.__class__.__name__}'"""
              raise TypeError(lowerCAmelCase__     )

       lowerCAmelCase__					     = ''.join(bin(lowerCAmelCase__     )[2:].zfill(8     ) for byte in data     )

       lowerCAmelCase__					     = len(lowerCAmelCase__     ) % 6 != 0

       if padding_needed:
              # The padding that will be added later
              lowerCAmelCase__					     = b'=' * ((6 - len(lowerCAmelCase__     ) % 6) // 2)

              # Append binary_stream with arbitrary binary digits (0's by default) to make its
              # length a multiple of 6.
              binary_stream += "0" * (6 - len(lowerCAmelCase__     ) % 6)
       else:
              lowerCAmelCase__					     = b''

       # Encode every 6 binary digits to their corresponding Base64 character
       return (
           "".join(
               B64_CHARSET[int(binary_stream[index : index + 6]     ,  2     )]
               for index in range(0     ,  len(lowerCAmelCase__     )     ,  6     )     ).encode()
           + padding
       )






def 				__lowerCamelCase		(						lowerCAmelCase__     ):
       # Make sure encoded_data is either a string or a bytes-like object
       if not isinstance(lowerCAmelCase__     ,  lowerCAmelCase__     ) and not isinstance(lowerCAmelCase__     ,  lowerCAmelCase__     ):
              lowerCAmelCase__					     = (
                  'argument should be a bytes-like object or ASCII string, '
                  F"""not '{encoded_data.__class__.__name__}'"""
              )
              raise TypeError(lowerCAmelCase__     )

       # In case encoded_data is a bytes-like object, make sure it contains only
       # ASCII characters so we convert it to a string object
       if isinstance(lowerCAmelCase__     ,  lowerCAmelCase__     ):
              try:
                     lowerCAmelCase__					     = encoded_data.decode('utf-8'     )
              except UnicodeDecodeError:
                     raise ValueError('base64 encoded data should only contain ASCII characters'     )

       lowerCAmelCase__					     = encoded_data.count('='     )

       # Check if the encoded string contains non base64 characters
       if padding:
              assert all(
                  char in B64_CHARSET for char in encoded_data[:-padding]     ), "Invalid base64 character(s) found."
       else:
              assert all(
                  char in B64_CHARSET for char in encoded_data     ), "Invalid base64 character(s) found."

       # Check the padding
       assert len(lowerCAmelCase__     ) % 4 == 0 and padding < 3, "Incorrect padding"

       if padding:
              # Remove padding if there is one
              lowerCAmelCase__					     = encoded_data[:-padding]

              lowerCAmelCase__					     = ''.join(
                  bin(B64_CHARSET.index(lowerCAmelCase__     )     )[2:].zfill(6     ) for char in encoded_data     )[: -padding * 2]
       else:
              lowerCAmelCase__					     = ''.join(
                  bin(B64_CHARSET.index(lowerCAmelCase__     )     )[2:].zfill(6     ) for char in encoded_data     )

       lowerCAmelCase__					     = [
           int(binary_stream[index : index + 8]     ,  2     )
           for index in range(0     ,  len(lowerCAmelCase__     )     ,  8     )
       ]

       return bytes(lowerCAmelCase__     )


if __name__ == "__main__":
    import doctest

    doctest.testmod()
