

'''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 				__lowerCAmelCase  (      snake_case__							):
  _a		    =						42





class 				__lowerCAmelCase  (      snake_case__    ,    snake_case__							):




  @register_to_config
  def __init__(       self							,     lowerCAmelCase = 32							,     lowerCAmelCase = 64							,     lowerCAmelCase = 20							,     lowerCAmelCase = 768							,     lowerCAmelCase=77							,     lowerCAmelCase=4							,     lowerCAmelCase = 0.0							,     lowerCAmelCase = "silu"							,     lowerCAmelCase = None							,     lowerCAmelCase = None							,     lowerCAmelCase = "linear"							,     lowerCAmelCase = "prd"							,     lowerCAmelCase = None							,     lowerCAmelCase = None							,     lowerCAmelCase = None							,     )						->				str:
        '''simple docstring'''






        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(_A							,     _A							,     0      )
        _lowercase			 =TimestepEmbedding(_A							,     _A							,     out_dim=_A							,     act_fn=_A      )

        _lowercase			 =nn.Linear(_A							,     _A      )

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

        _lowercase			 =nn.Linear(_A							,     _A      )

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

        _lowercase			 =nn.Parameter(torch.zeros(1							,     num_embeddings + additional_embeddings							,     _A      )      )

        if added_emb_type == "prd":
              _lowercase			 =nn.Parameter(torch.zeros(1							,     1							,     _A      )      )
        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(
                    _A							,     _A							,     _A							,     dropout=_A							,     activation_fn='gelu'							,     attention_bias=_A							,     )
                for d in range(_A      )
            ]      )

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

        _lowercase			 =nn.LayerNorm(_A      )

        _lowercase			 =nn.Linear(_A							,     _A      )

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

        _lowercase			 =nn.Parameter(torch.zeros(1							,     _A      )      )
        _lowercase			 =nn.Parameter(torch.zeros(1							,     _A      )      )




  @property
  # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
  def 						A__  (       self      )						->				Tuple:
        '''simple docstring'''






        _lowercase			 ={}

        def fn_recursive_add_processors(lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase      ):
              if hasattr(_A							,     'set_processor'      ):
                    _lowercase			 =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							,     lowerCAmelCase      )						->				List[str]:
        '''simple docstring'''






        _lowercase			 =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(lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase      ):
              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]:
        '''simple docstring'''






        self.set_attn_processor(AttnProcessor()      )




  def 						A__  (       self							,     lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase = None							,     lowerCAmelCase = None							,     lowerCAmelCase = True							,     )						->				str:
        '''simple docstring'''






        _lowercase			 =hidden_states.shape[0]

        _lowercase			 =timestep
        if not torch.is_tensor(_A      ):
              _lowercase			 =torch.tensor([timesteps]							,     dtype=torch.long							,     device=hidden_states.device      )
        elif torch.is_tensor(_A      ) 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(_A							,     dtype=timesteps.dtype							,     device=timesteps.device      )

        _lowercase			 =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			 =timesteps_projected.to(dtype=self.dtype      )
        _lowercase			 =self.time_embedding(_A      )

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

        _lowercase			 =self.embedding_proj(_A      )
        if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
              _lowercase			 =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			 =self.proj_in(_A      )

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

        _lowercase			 =[]
        _lowercase			 =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			 =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(_A							,     -1							,     -1      )
              additional_embeds.append(_A      )

        _lowercase			 =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			 =additional_embeddings_len + proj_embeddings.shape[1] + 1
        if positional_embeddings.shape[1] < hidden_states.shape[1]:
              _lowercase			 =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			 =hidden_states + positional_embeddings

        if attention_mask is not None:
              _lowercase			 =(1 - attention_mask.to(hidden_states.dtype      )) * -10_000.0
              _lowercase			 =F.pad(_A							,     (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(_A      )

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

        _lowercase			 =self.norm_out(_A      )

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

        _lowercase			 =self.proj_to_clip_embeddings(_A      )

        if not return_dict:
              return (predicted_image_embedding,)

        return PriorTransformerOutput(predicted_image_embedding=_A      )




  def 						A__  (       self							,     lowerCAmelCase      )						->				int:
        '''simple docstring'''






        _lowercase			 =(prior_latents * self.clip_std) + self.clip_mean
        return prior_latents





from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, require_tokenizers, slow


if is_tf_available():
  import numpy as np
  import tensorflow as tf

  from transformers import TFCamembertModel





@require_tf
@require_sentencepiece
@require_tokenizers
class 				__lowerCAmelCase  (      unittest.TestCase							):




  @slow
  def 						A__  (       self      )						->				str:
        '''simple docstring'''






        _lowercase			 =TFCamembertModel.from_pretrained('jplu/tf-camembert-base'      )

        _lowercase			 =tf.convert_to_tensor(
            [[5, 121, 11, 660, 16, 730, 25_543, 110, 83, 6]]							,     dtype=tf.intaa							,     )  # J'aime le camembert !"

        _lowercase			 =model(lowerCAmelCase      )['last_hidden_state']
        _lowercase			 =tf.TensorShape((1, 10, 768)      )
        self.assertEqual(output.shape							,     lowerCAmelCase      )
        # compare the actual values for a slice.
        _lowercase			 =tf.convert_to_tensor(
            [[[-0.0254, 0.0235, 0.1027], [0.0606, -0.1811, -0.0418], [-0.1561, -0.1127, 0.2687]]]							,     dtype=tf.floataa							,     )
        # camembert = torch.hub.load('pytorch/fairseq', 'camembert.v0')
        # camembert.eval()
        # expected_slice = roberta.model.forward(input_ids)[0][:, :3, :3].detach()

        self.assertTrue(np.allclose(output[:, :3, :3].numpy()							,     expected_slice.numpy()							,     atol=1e-4      )      )


