

from typing import Any, Dict, Optional

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

from ..utils import maybe_allow_in_graph
from .activations import get_activation
from .attention_processor import Attention
from .embeddings import CombinedTimestepLabelEmbeddings


@maybe_allow_in_graph
class      lowerCAmelCase					(     nn.Module					):


       '''simple docstring'''







       def __init__(						self       :	Any    ,						__snake_case       :	Any    ,						__snake_case       :	Union[str, Any]    ,						__snake_case       :	Tuple    ,						__snake_case       :	int=0.0    ,						__snake_case       :	List[Any] = None    ,						__snake_case       :	Optional[Any] = "geglu"    ,						__snake_case       :	List[Any] = None    ,						__snake_case       :	Union[str, Any] = False    ,						__snake_case       :	str = False    ,						__snake_case       :	Union[str, Any] = False    ,						__snake_case       :	int = False    ,						__snake_case       :	str = True    ,						__snake_case       :	Dict = "layer_norm"    ,						__snake_case       :	Dict = False    ,						) ->			str:







        '''simple docstring'''

        super().__init__()
        lowerCamelCase       =     only_cross_attention

        lowerCamelCase       =     (num_embeds_ada_norm is not None) and norm_type == 'ada_norm_zero'
        lowerCamelCase       =     (num_embeds_ada_norm is not None) and norm_type == 'ada_norm'

        if norm_type in ("ada_norm", "ada_norm_zero") and num_embeds_ada_norm is None:
         raise ValueError(
             F'''`norm_type` is set to {norm_type}, but `num_embeds_ada_norm` is not defined. Please make sure to'''
             F''' define `num_embeds_ada_norm` if setting `norm_type` to {norm_type}.'''  )

        # Define 3 blocks. Each block has its own normalization layer.
        # 1. Self-Attn
        if self.use_ada_layer_norm:
         lowerCamelCase       =     AdaLayerNorm(__snake_case    ,						__snake_case  )
        elif self.use_ada_layer_norm_zero:
         lowerCamelCase       =     AdaLayerNormZero(__snake_case    ,						__snake_case  )
        else:
         lowerCamelCase       =     nn.LayerNorm(__snake_case    ,						elementwise_affine=__snake_case  )
        lowerCamelCase       =     Attention(
            query_dim=__snake_case    ,						heads=__snake_case    ,						dim_head=__snake_case    ,						dropout=__snake_case    ,						bias=__snake_case    ,						cross_attention_dim=cross_attention_dim if only_cross_attention else None    ,						upcast_attention=__snake_case    ,						)

        # 2. Cross-Attn
        if cross_attention_dim is not None or double_self_attention:
         # We currently only use AdaLayerNormZero for self attention where there will only be one attention block.
         # I.e. the number of returned modulation chunks from AdaLayerZero would not make sense if returned during
         # the second cross attention block.
         lowerCamelCase       =     (
             AdaLayerNorm(__snake_case    ,						__snake_case  )
             if self.use_ada_layer_norm
             else nn.LayerNorm(__snake_case    ,						elementwise_affine=__snake_case  )
         )
         lowerCamelCase       =     Attention(
             query_dim=__snake_case    ,						cross_attention_dim=cross_attention_dim if not double_self_attention else None    ,						heads=__snake_case    ,						dim_head=__snake_case    ,						dropout=__snake_case    ,						bias=__snake_case    ,						upcast_attention=__snake_case    ,						)  # is self-attn if encoder_hidden_states is none
        else:
         lowerCamelCase       =     None
         lowerCamelCase       =     None

        # 3. Feed-forward
        lowerCamelCase       =     nn.LayerNorm(__snake_case    ,						elementwise_affine=__snake_case  )
        lowerCamelCase       =     FeedForward(__snake_case    ,						dropout=__snake_case    ,						activation_fn=__snake_case    ,						final_dropout=__snake_case  )

        # let chunk size default to None
        lowerCamelCase       =     None
        lowerCamelCase       =     0

       def 	lowerCamelCase__    (						self       :	List[str]    ,						__snake_case       :	str    ,						__snake_case       :	Any  ) ->			str:







        '''simple docstring'''

        lowerCamelCase       =     chunk_size
        lowerCamelCase       =     dim





       def 	lowerCamelCase__    (						self       :	int    ,						__snake_case       :	Optional[int]    ,						__snake_case       :	Union[str, Any] = None    ,						__snake_case       :	str = None    ,						__snake_case       :	List[Any] = None    ,						__snake_case       :	Any = None    ,						__snake_case       :	Optional[int] = None    ,						__snake_case       :	str = None    ,						) ->			List[str]:







        '''simple docstring'''

        if self.use_ada_layer_norm:
         lowerCamelCase       =     self.norma(__snake_case    ,						__snake_case  )
        elif self.use_ada_layer_norm_zero:
         lowerCamelCase       =     self.norma(
             __snake_case    ,						__snake_case    ,						__snake_case    ,						hidden_dtype=hidden_states.dtype  )
        else:
         lowerCamelCase       =     self.norma(__snake_case  )

        lowerCamelCase       =     cross_attention_kwargs if cross_attention_kwargs is not None else {}

        lowerCamelCase       =     self.attna(
            __snake_case    ,						encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None    ,						attention_mask=__snake_case    ,						**__snake_case    ,						)
        if self.use_ada_layer_norm_zero:
         lowerCamelCase       =     gate_msa.unsqueeze(1  ) * attn_output
        lowerCamelCase       =     attn_output + hidden_states

        # 2. Cross-Attention
        if self.attna is not None:
         lowerCamelCase       =     (
             self.norma(__snake_case    ,						__snake_case  ) if self.use_ada_layer_norm else self.norma(__snake_case  )
         )

         lowerCamelCase       =     self.attna(
             __snake_case    ,						encoder_hidden_states=__snake_case    ,						attention_mask=__snake_case    ,						**__snake_case    ,						)
         lowerCamelCase       =     attn_output + hidden_states

        # 3. Feed-forward
        lowerCamelCase       =     self.norma(__snake_case  )

        if self.use_ada_layer_norm_zero:
         lowerCamelCase       =     norm_hidden_states * (1 + scale_mlp[:, None]) + shift_mlp[:, None]

        if self._chunk_size is not None:
         # "feed_forward_chunk_size" can be used to save memory
         if norm_hidden_states.shape[self._chunk_dim] % self._chunk_size != 0:
          raise ValueError(
              F'''`hidden_states` dimension to be chunked: {norm_hidden_states.shape[self._chunk_dim]} has to be divisible by chunk size: {self._chunk_size}. Make sure to set an appropriate `chunk_size` when calling `unet.enable_forward_chunking`.'''  )

         lowerCamelCase       =     norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
         lowerCamelCase       =     torch.cat(
             [self.ff(__snake_case  ) for hid_slice in norm_hidden_states.chunk(__snake_case    ,						dim=self._chunk_dim  )]    ,						dim=self._chunk_dim    ,						)
        else:
         lowerCamelCase       =     self.ff(__snake_case  )

        if self.use_ada_layer_norm_zero:
         lowerCamelCase       =     gate_mlp.unsqueeze(1  ) * ff_output

        lowerCamelCase       =     ff_output + hidden_states

        return hidden_states


class      lowerCAmelCase					(     nn.Module					):


       '''simple docstring'''







       def __init__(						self       :	Any    ,						__snake_case       :	Tuple    ,						__snake_case       :	Optional[int] = None    ,						__snake_case       :	Optional[Any] = 4    ,						__snake_case       :	Dict = 0.0    ,						__snake_case       :	List[str] = "geglu"    ,						__snake_case       :	Optional[Any] = False    ,						) ->			List[Any]:







        '''simple docstring'''

        super().__init__()
        lowerCamelCase       =     int(dim * mult  )
        lowerCamelCase       =     dim_out if dim_out is not None else dim

        if activation_fn == "gelu":
         lowerCamelCase       =     GELU(__snake_case    ,						__snake_case  )
        if activation_fn == "gelu-approximate":
         lowerCamelCase       =     GELU(__snake_case    ,						__snake_case    ,						approximate='tanh'  )
        elif activation_fn == "geglu":
         lowerCamelCase       =     GEGLU(__snake_case    ,						__snake_case  )
        elif activation_fn == "geglu-approximate":
         lowerCamelCase       =     ApproximateGELU(__snake_case    ,						__snake_case  )

        lowerCamelCase       =     nn.ModuleList([]  )
        # project in
        self.net.append(__snake_case  )
        # project dropout
        self.net.append(nn.Dropout(__snake_case  )  )
        # project out
        self.net.append(nn.Linear(__snake_case    ,						__snake_case  )  )
        # FF as used in Vision Transformer, MLP-Mixer, etc. have a final dropout
        if final_dropout:
         self.net.append(nn.Dropout(__snake_case  )  )





       def 	lowerCamelCase__    (						self       :	List[Any]    ,						__snake_case       :	Tuple  ) ->			int:







        '''simple docstring'''

        for module in self.net:
         lowerCamelCase       =     module(__snake_case  )
        return hidden_states


class      lowerCAmelCase					(     nn.Module					):


       '''simple docstring'''







       def __init__(						self       :	int    ,						__snake_case       :	Optional[int]    ,						__snake_case       :	Any    ,						__snake_case       :	int = "none"  ) ->			Tuple:







        '''simple docstring'''

        super().__init__()
        lowerCamelCase       =     nn.Linear(__snake_case    ,						__snake_case  )
        lowerCamelCase       =     approximate

       def 	lowerCamelCase__    (						self       :	Optional[int]    ,						__snake_case       :	Optional[int]  ) ->			Optional[Any]:







        '''simple docstring'''

        if gate.device.type != "mps":
         return F.gelu(__snake_case    ,						approximate=self.approximate  )
        # mps: gelu is not implemented for float16
        return F.gelu(gate.to(dtype=torch.floataa  )    ,						approximate=self.approximate  ).to(dtype=gate.dtype  )





       def 	lowerCamelCase__    (						self       :	Optional[Any]    ,						__snake_case       :	Tuple  ) ->			Dict:







        '''simple docstring'''

        lowerCamelCase       =     self.proj(__snake_case  )
        lowerCamelCase       =     self.gelu(__snake_case  )
        return hidden_states


class      lowerCAmelCase					(     nn.Module					):


       '''simple docstring'''







       def __init__(						self       :	Any    ,						__snake_case       :	Optional[Any]    ,						__snake_case       :	Optional[Any]  ) ->			Optional[int]:







        '''simple docstring'''

        super().__init__()
        lowerCamelCase       =     nn.Linear(__snake_case    ,						dim_out * 2  )

       def 	lowerCamelCase__    (						self       :	List[str]    ,						__snake_case       :	str  ) ->			Dict:







        '''simple docstring'''

        if gate.device.type != "mps":
         return F.gelu(__snake_case  )
        # mps: gelu is not implemented for float16
        return F.gelu(gate.to(dtype=torch.floataa  )  ).to(dtype=gate.dtype  )





       def 	lowerCamelCase__    (						self       :	List[Any]    ,						__snake_case       :	List[Any]  ) ->			str:







        '''simple docstring'''

        lowerCamelCase       =     self.proj(__snake_case  ).chunk(2    ,						dim=-1  )
        return hidden_states * self.gelu(__snake_case  )


class      lowerCAmelCase					(     nn.Module					):


       '''simple docstring'''







       def __init__(						self       :	Union[str, Any]    ,						__snake_case       :	Optional[int]    ,						__snake_case       :	str  ) ->			List[str]:







        '''simple docstring'''

        super().__init__()
        lowerCamelCase       =     nn.Linear(__snake_case    ,						__snake_case  )





       def 	lowerCamelCase__    (						self       :	Dict    ,						__snake_case       :	Dict  ) ->			int:







        '''simple docstring'''

        lowerCamelCase       =     self.proj(__snake_case  )
        return x * torch.sigmoid(1.702 * x  )


class      lowerCAmelCase					(     nn.Module					):


       '''simple docstring'''







       def __init__(						self       :	int    ,						__snake_case       :	Dict    ,						__snake_case       :	Optional[Any]  ) ->			Dict:







        '''simple docstring'''

        super().__init__()
        lowerCamelCase       =     nn.Embedding(__snake_case    ,						__snake_case  )
        lowerCamelCase       =     nn.SiLU()
        lowerCamelCase       =     nn.Linear(__snake_case    ,						embedding_dim * 2  )
        lowerCamelCase       =     nn.LayerNorm(__snake_case    ,						elementwise_affine=__snake_case  )





       def 	lowerCamelCase__    (						self       :	List[Any]    ,						__snake_case       :	Optional[int]    ,						__snake_case       :	str  ) ->			Any:







        '''simple docstring'''

        lowerCamelCase       =     self.linear(self.silu(self.emb(__snake_case  )  )  )
        lowerCamelCase       =     torch.chunk(__snake_case    ,						2  )
        lowerCamelCase       =     self.norm(__snake_case  ) * (1 + scale) + shift
        return x


class      lowerCAmelCase					(     nn.Module					):


       '''simple docstring'''







       def __init__(						self       :	Union[str, Any]    ,						__snake_case       :	Tuple    ,						__snake_case       :	Union[str, Any]  ) ->			List[Any]:







        '''simple docstring'''

        super().__init__()

        lowerCamelCase       =     CombinedTimestepLabelEmbeddings(__snake_case    ,						__snake_case  )

        lowerCamelCase       =     nn.SiLU()
        lowerCamelCase       =     nn.Linear(__snake_case    ,						6 * embedding_dim    ,						bias=__snake_case  )
        lowerCamelCase       =     nn.LayerNorm(__snake_case    ,						elementwise_affine=__snake_case    ,						eps=1e-6  )





       def 	lowerCamelCase__    (						self       :	Dict    ,						__snake_case       :	Optional[Any]    ,						__snake_case       :	int    ,						__snake_case       :	Dict    ,						__snake_case       :	Dict=None  ) ->			Dict:







        '''simple docstring'''

        lowerCamelCase       =     self.linear(self.silu(self.emb(__snake_case    ,						__snake_case    ,						hidden_dtype=__snake_case  )  )  )
        lowerCamelCase       =     emb.chunk(6    ,						dim=1  )
        lowerCamelCase       =     self.norm(__snake_case  ) * (1 + scale_msa[:, None]) + shift_msa[:, None]
        return x, gate_msa, shift_mlp, scale_mlp, gate_mlp


class      lowerCAmelCase					(     nn.Module					):


       '''simple docstring'''







       def __init__(						self       :	Any    ,						__snake_case       :	List[str]    ,						__snake_case       :	Tuple    ,						__snake_case       :	Dict    ,						__snake_case       :	Any = None    ,						__snake_case       :	Tuple = 1e-5  ) ->			List[str]:







        '''simple docstring'''

        super().__init__()
        lowerCamelCase       =     num_groups
        lowerCamelCase       =     eps

        if act_fn is None:
         lowerCamelCase       =     None
        else:
         lowerCamelCase       =     get_activation(__snake_case  )

        lowerCamelCase       =     nn.Linear(__snake_case    ,						out_dim * 2  )





       def 	lowerCamelCase__    (						self       :	Union[str, Any]    ,						__snake_case       :	str    ,						__snake_case       :	Union[str, Any]  ) ->			List[Any]:







        '''simple docstring'''

        if self.act:
         lowerCamelCase       =     self.act(__snake_case  )
        lowerCamelCase       =     self.linear(__snake_case  )
        lowerCamelCase       =     emb[:, :, None, None]
        lowerCamelCase       =     emb.chunk(2    ,						dim=1  )

        lowerCamelCase       =     F.group_norm(__snake_case    ,						self.num_groups    ,						eps=self.eps  )
        lowerCamelCase       =     x * (1 + scale) + shift
        return x










"""simple docstring"""






import os
from typing import Dict, List, Union

import tensorflow as tf
from keras_nlp.tokenizers import BytePairTokenizer
from tensorflow_text import pad_model_inputs

from .tokenization_gpta import GPTaTokenizer






class lowerCAmelCase__      (	tf.keras.layers.Layer		):






			'''simple docstring'''



			def __init__(    self    , lowercase    , lowercase    , lowercase = None    , lowercase = None    ):
							super().__init__()
							_lowerCamelCase	:     Optional[Any]       =       pad_token_id
							_lowerCamelCase	:     Dict       =       max_length
							_lowerCamelCase	:     Tuple       =       vocab
							_lowerCamelCase	:     List[str]       =       merges
							_lowerCamelCase	:     int       =       BytePairTokenizer(lowercase    , lowercase    , sequence_length=lowercase    )
			@classmethod
			def    A_       (    cls    , lowercase    , *lowercase    , **lowercase    ):
							_lowerCamelCase	:     str       =       [' '.join(lowercase    ) for m in tokenizer.bpe_ranks.keys()]
							_lowerCamelCase	:     Union[str, Any]       =       tokenizer.get_vocab()
							return cls(lowercase    , lowercase    , *lowercase    , **lowercase    )
			@classmethod
			def    A_       (    cls    , lowercase    , *lowercase    , **lowercase    ):
							_lowerCamelCase	:     Optional[int]       =       GPTaTokenizer.from_pretrained(lowercase    , *lowercase    , **lowercase    )
							return cls.from_tokenizer(lowercase    , *lowercase    , **lowercase    )
			@classmethod
			def    A_       (    cls    , lowercase    ):
							return cls(**lowercase    )
			def    A_       (    self    ):
							return {
							    "vocab": self.vocab,
							    "merges": self.merges,
							    "max_length": self.max_length,
							    "pad_token_id": self.pad_token_id,
							}






			def    A_       (    self    , lowercase    , lowercase = None    ):
							_lowerCamelCase	:     int       =       self.tf_tokenizer(lowercase    )
							_lowerCamelCase	:     Optional[int]       =       tf.ones_like(lowercase    )

							if self.pad_token_id is not None:
											# pad the tokens up to max length
											_lowerCamelCase	:     str       =       max_length if max_length is not None else self.max_length

											if max_length is not None:
															_lowerCamelCase,					_lowerCamelCase	:     Union[str, Any]       =       pad_model_inputs(
															    lowercase    , max_seq_length=lowercase    , pad_value=self.pad_token_id    )

							return {"attention_mask": attention_mask, "input_ids": input_ids}