


'''simple docstring'''



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 			a       (	nn.Module ):


     """simple docstring"""






     def __init__(       self  :	Tuple							,				snake_case  :	int							,				snake_case  :	int							,				snake_case  :	int							,				snake_case  :	Optional[int]=0.0							,				snake_case  :	Optional[int] = None							,				snake_case  :	str = "geglu"							,				snake_case  :	Optional[int] = None							,				snake_case  :	bool = False							,				snake_case  :	bool = False							,				snake_case  :	bool = False							,				snake_case  :	bool = False							,				snake_case  :	bool = True							,				snake_case  :	str = "layer_norm"							,				snake_case  :	bool = False							,				)		->							str:
          super().__init__()
          __UpperCAmelCase			:				List[str]													=    only_cross_attention

          __UpperCAmelCase			:				Any													=    (num_embeds_ada_norm is not None) and norm_type == '''ada_norm_zero'''
          __UpperCAmelCase			:				Dict													=    (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:
               __UpperCAmelCase			:				Dict													=    AdaLayerNorm(snake_case							,				snake_case    )
          elif self.use_ada_layer_norm_zero:
               __UpperCAmelCase			:				Union[str, Any]													=    AdaLayerNormZero(snake_case							,				snake_case    )
          else:
               __UpperCAmelCase			:				Optional[int]													=    nn.LayerNorm(snake_case							,				elementwise_affine=snake_case    )
          __UpperCAmelCase			:				Optional[int]													=    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.
               __UpperCAmelCase			:				List[Any]													=    (
                   AdaLayerNorm(snake_case							,				snake_case    )
                   if self.use_ada_layer_norm
                   else nn.LayerNorm(snake_case							,				elementwise_affine=snake_case    )
               )
               __UpperCAmelCase			:				List[str]													=    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:
               __UpperCAmelCase			:				Union[str, Any]													=    None
               __UpperCAmelCase			:				Any													=    None

          # 3. Feed-forward
          __UpperCAmelCase			:				Tuple													=    nn.LayerNorm(snake_case							,				elementwise_affine=snake_case    )
          __UpperCAmelCase			:				List[Any]													=    FeedForward(snake_case							,				dropout=snake_case							,				activation_fn=snake_case							,				final_dropout=snake_case    )

          # let chunk size default to None
          __UpperCAmelCase			:				Union[str, Any]													=    None
          __UpperCAmelCase			:				int													=    0

     def 				lowerCamelCase__				(       self  :	Optional[int]							,				snake_case  :	Optional[int]							,				snake_case  :	int    )		->							Optional[Any]:
          # Sets chunk feed-forward
          __UpperCAmelCase			:				int													=    chunk_size
          __UpperCAmelCase			:				int													=    dim






     def 				lowerCamelCase__				(       self  :	str							,				snake_case  :	torch.FloatTensor							,				snake_case  :	Optional[torch.FloatTensor] = None							,				snake_case  :	Optional[torch.FloatTensor] = None							,				snake_case  :	Optional[torch.FloatTensor] = None							,				snake_case  :	Optional[torch.LongTensor] = None							,				snake_case  :	Dict[str, Any] = None							,				snake_case  :	Optional[torch.LongTensor] = None							,				)		->							Tuple:
          # Notice that normalization is always applied before the real computation in the following blocks.
          # 1. Self-Attention
          if self.use_ada_layer_norm:
               __UpperCAmelCase			:				Tuple													=    self.norma(snake_case							,				snake_case    )
          elif self.use_ada_layer_norm_zero:
               __UpperCAmelCase							,     __UpperCAmelCase							,     __UpperCAmelCase							,     __UpperCAmelCase							,     __UpperCAmelCase			:				Dict													=    self.norma(
                   snake_case							,				snake_case							,				snake_case							,				hidden_dtype=hidden_states.dtype    )
          else:
               __UpperCAmelCase			:				Tuple													=    self.norma(snake_case    )

          __UpperCAmelCase			:				Union[str, Any]													=    cross_attention_kwargs if cross_attention_kwargs is not None else {}

          __UpperCAmelCase			:				Union[str, Any]													=    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:
               __UpperCAmelCase			:				Optional[int]													=    gate_msa.unsqueeze(1    ) * attn_output
          __UpperCAmelCase			:				Any													=    attn_output + hidden_states

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

               __UpperCAmelCase			:				Any													=    self.attna(
                   snake_case							,				encoder_hidden_states=snake_case							,				attention_mask=snake_case							,				**snake_case							,				)
               __UpperCAmelCase			:				Any													=    attn_output + hidden_states

          # 3. Feed-forward
          __UpperCAmelCase			:				int													=    self.norma(snake_case    )

          if self.use_ada_layer_norm_zero:
               __UpperCAmelCase			:				Tuple													=    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`.'    )

               __UpperCAmelCase			:				Tuple													=    norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
               __UpperCAmelCase			:				int													=    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:
               __UpperCAmelCase			:				Union[str, Any]													=    self.ff(snake_case    )

          if self.use_ada_layer_norm_zero:
               __UpperCAmelCase			:				List[Any]													=    gate_mlp.unsqueeze(1    ) * ff_output

          __UpperCAmelCase			:				Union[str, Any]													=    ff_output + hidden_states

          return hidden_states



class 			a       (	nn.Module ):


     """simple docstring"""






     def __init__(       self  :	List[str]							,				snake_case  :	int							,				snake_case  :	Optional[int] = None							,				snake_case  :	int = 4							,				snake_case  :	float = 0.0							,				snake_case  :	str = "geglu"							,				snake_case  :	bool = False							,				)		->							Tuple:
          super().__init__()
          __UpperCAmelCase			:				int													=    int(dim * mult    )
          __UpperCAmelCase			:				Union[str, Any]													=    dim_out if dim_out is not None else dim

          if activation_fn == "gelu":
               __UpperCAmelCase			:				Tuple													=    GELU(snake_case							,				snake_case    )
          if activation_fn == "gelu-approximate":
               __UpperCAmelCase			:				Union[str, Any]													=    GELU(snake_case							,				snake_case							,				approximate='''tanh'''    )
          elif activation_fn == "geglu":
               __UpperCAmelCase			:				List[str]													=    GEGLU(snake_case							,				snake_case    )
          elif activation_fn == "geglu-approximate":
               __UpperCAmelCase			:				Optional[Any]													=    ApproximateGELU(snake_case							,				snake_case    )

          __UpperCAmelCase			:				List[Any]													=    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  :	Dict							,				snake_case  :	int    )		->							Dict:
          for module in self.net:
               __UpperCAmelCase			:				Any													=    module(snake_case    )
          return hidden_states



class 			a       (	nn.Module ):


     """simple docstring"""






     def __init__(       self  :	int							,				snake_case  :	int							,				snake_case  :	int							,				snake_case  :	str = "none"    )		->							str:
          super().__init__()
          __UpperCAmelCase			:				Union[str, Any]													=    nn.Linear(snake_case							,				snake_case    )
          __UpperCAmelCase			:				Optional[Any]													=    approximate

     def 				lowerCamelCase__				(       self  :	str							,				snake_case  :	Any    )		->							Any:
          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  :	Any							,				snake_case  :	Union[str, Any]    )		->							Union[str, Any]:
          __UpperCAmelCase			:				Optional[int]													=    self.proj(snake_case    )
          __UpperCAmelCase			:				int													=    self.gelu(snake_case    )
          return hidden_states



class 			a       (	nn.Module ):


     """simple docstring"""






     def __init__(       self  :	Optional[Any]							,				snake_case  :	int							,				snake_case  :	int    )		->							List[Any]:
          super().__init__()
          __UpperCAmelCase			:				str													=    nn.Linear(snake_case							,				dim_out * 2    )

     def 				lowerCamelCase__				(       self  :	str							,				snake_case  :	List[str]    )		->							Dict:
          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  :	Optional[Any]							,				snake_case  :	Any    )		->							Tuple:
          __UpperCAmelCase							,     __UpperCAmelCase			:				Optional[int]													=    self.proj(snake_case    ).chunk(2							,				dim=-1    )
          return hidden_states * self.gelu(snake_case    )



class 			a       (	nn.Module ):


     """simple docstring"""






     def __init__(       self  :	Optional[int]							,				snake_case  :	int							,				snake_case  :	int    )		->							Dict:
          super().__init__()
          __UpperCAmelCase			:				Dict													=    nn.Linear(snake_case							,				snake_case    )






     def 				lowerCamelCase__				(       self  :	Optional[int]							,				snake_case  :	List[str]    )		->							str:
          __UpperCAmelCase			:				Optional[int]													=    self.proj(snake_case    )
          return x * torch.sigmoid(1.702 * x    )



class 			a       (	nn.Module ):


     """simple docstring"""






     def __init__(       self  :	int							,				snake_case  :	Tuple							,				snake_case  :	Optional[Any]    )		->							Optional[Any]:
          super().__init__()
          __UpperCAmelCase			:				List[str]													=    nn.Embedding(snake_case							,				snake_case    )
          __UpperCAmelCase			:				Union[str, Any]													=    nn.SiLU()
          __UpperCAmelCase			:				Optional[int]													=    nn.Linear(snake_case							,				embedding_dim * 2    )
          __UpperCAmelCase			:				List[Any]													=    nn.LayerNorm(snake_case							,				elementwise_affine=snake_case    )






     def 				lowerCamelCase__				(       self  :	Dict							,				snake_case  :	Dict							,				snake_case  :	Dict    )		->							Any:
          __UpperCAmelCase			:				Optional[Any]													=    self.linear(self.silu(self.emb(snake_case    )    )    )
          __UpperCAmelCase							,     __UpperCAmelCase			:				List[str]													=    torch.chunk(snake_case							,				2    )
          __UpperCAmelCase			:				Tuple													=    self.norm(snake_case    ) * (1 + scale) + shift
          return x



class 			a       (	nn.Module ):


     """simple docstring"""






     def __init__(       self  :	str							,				snake_case  :	Optional[int]							,				snake_case  :	List[str]    )		->							Any:
          super().__init__()

          __UpperCAmelCase			:				Union[str, Any]													=    CombinedTimestepLabelEmbeddings(snake_case							,				snake_case    )

          __UpperCAmelCase			:				List[Any]													=    nn.SiLU()
          __UpperCAmelCase			:				str													=    nn.Linear(snake_case							,				6 * embedding_dim							,				bias=snake_case    )
          __UpperCAmelCase			:				Optional[int]													=    nn.LayerNorm(snake_case							,				elementwise_affine=snake_case							,				eps=1E-6    )






     def 				lowerCamelCase__				(       self  :	Any							,				snake_case  :	Dict							,				snake_case  :	Optional[int]							,				snake_case  :	List[str]							,				snake_case  :	Any=None    )		->							Optional[Any]:
          __UpperCAmelCase			:				Union[str, Any]													=    self.linear(self.silu(self.emb(snake_case							,				snake_case							,				hidden_dtype=snake_case    )    )    )
          __UpperCAmelCase							,     __UpperCAmelCase							,     __UpperCAmelCase							,     __UpperCAmelCase							,     __UpperCAmelCase							,     __UpperCAmelCase			:				List[Any]													=    emb.chunk(6							,				dim=1    )
          __UpperCAmelCase			:				Any													=    self.norm(snake_case    ) * (1 + scale_msa[:, None]) + shift_msa[:, None]
          return x, gate_msa, shift_mlp, scale_mlp, gate_mlp



class 			a       (	nn.Module ):


     """simple docstring"""






     def __init__(       self  :	Union[str, Any]							,				snake_case  :	int							,				snake_case  :	int							,				snake_case  :	int							,				snake_case  :	Optional[str] = None							,				snake_case  :	float = 1E-5    )		->							Tuple:
          super().__init__()
          __UpperCAmelCase			:				List[str]													=    num_groups
          __UpperCAmelCase			:				Optional[int]													=    eps

          if act_fn is None:
               __UpperCAmelCase			:				Optional[int]													=    None
          else:
               __UpperCAmelCase			:				int													=    get_activation(snake_case    )

          __UpperCAmelCase			:				int													=    nn.Linear(snake_case							,				out_dim * 2    )






     def 				lowerCamelCase__				(       self  :	str							,				snake_case  :	List[str]							,				snake_case  :	Optional[Any]    )		->							Tuple:
          if self.act:
               __UpperCAmelCase			:				int													=    self.act(snake_case    )
          __UpperCAmelCase			:				Optional[int]													=    self.linear(snake_case    )
          __UpperCAmelCase			:				List[str]													=    emb[:, :, None, None]
          __UpperCAmelCase							,     __UpperCAmelCase			:				Dict													=    emb.chunk(2							,				dim=1    )

          __UpperCAmelCase			:				Optional[int]													=    F.group_norm(snake_case							,				self.num_groups							,				eps=self.eps    )
          __UpperCAmelCase			:				Union[str, Any]													=    x * (1 + scale) + shift
          return x



'''simple docstring'''



from collections.abc import Iterator, MutableMapping
from dataclasses import dataclass
from typing import Generic, TypeVar

__UpperCAmelCase    :Optional[int]				       =     TypeVar("KEY")
__UpperCAmelCase    :Tuple				       =     TypeVar("VAL")



@dataclass(frozen=_a			,						slots=_a )
class 			a       (	Generic[KEY, VAL] ):


     """simple docstring"""






     SCREAMING_SNAKE_CASE :				KEY
     SCREAMING_SNAKE_CASE :				VAL



class 			a       (	_Item ):


     """simple docstring"""






     def __init__(       self  :	Optional[int]    )		->							None:
          super().__init__(snake_case							,				snake_case    )






     def __bool__(       self  :	Dict    )		->							bool:
          return False



__UpperCAmelCase    :Optional[Any]				       =     _DeletedItem()



class 			a       (	MutableMapping[KEY, VAL] ):


     """simple docstring"""






     def __init__(       self  :	str							,				snake_case  :	int = 8							,				snake_case  :	float = 0.75    )		->							None:
          __UpperCAmelCase			:				Dict													=    initial_block_size
          __UpperCAmelCase			:				list[_Item | None]													=    [None] * initial_block_size
          assert 0.0 < capacity_factor < 1.0
          __UpperCAmelCase			:				str													=    capacity_factor
          __UpperCAmelCase			:				Optional[int]													=    0

     def 				lowerCamelCase__				(       self  :	Optional[Any]							,				snake_case  :	KEY    )		->							int:
          return hash(snake_case    ) % len(self._buckets    )

     def 				lowerCamelCase__				(       self  :	Union[str, Any]							,				snake_case  :	int    )		->							int:
          return (ind + 1) % len(self._buckets    )

     def 				lowerCamelCase__				(       self  :	List[str]							,				snake_case  :	int							,				snake_case  :	KEY							,				snake_case  :	VAL    )		->							bool:
          __UpperCAmelCase			:				int													=    self._buckets[ind]
          if not stored:
               __UpperCAmelCase			:				List[str]													=    _Item(snake_case							,				snake_case    )
               self._len += 1
               return True
          elif stored.key == key:
               __UpperCAmelCase			:				int													=    _Item(snake_case							,				snake_case    )
               return True
          else:
               return False

     def 				lowerCamelCase__				(       self  :	str    )		->							bool:
          __UpperCAmelCase			:				Union[str, Any]													=    len(self._buckets    ) * self._capacity_factor
          return len(self    ) >= int(snake_case    )

     def 				lowerCamelCase__				(       self  :	List[str]    )		->							bool:
          if len(self._buckets    ) <= self._initial_block_size:
               return False
          __UpperCAmelCase			:				List[str]													=    len(self._buckets    ) * self._capacity_factor / 2
          return len(self    ) < limit

     def 				lowerCamelCase__				(       self  :	Optional[int]							,				snake_case  :	int    )		->							None:
          __UpperCAmelCase			:				Union[str, Any]													=    self._buckets
          __UpperCAmelCase			:				Union[str, Any]													=    [None] * new_size
          __UpperCAmelCase			:				List[str]													=    0
          for item in old_buckets:
               if item:
                    self._add_item(item.key							,				item.val    )

     def 				lowerCamelCase__				(       self  :	Optional[int]    )		->							None:
          self._resize(len(self._buckets    ) * 2    )

     def 				lowerCamelCase__				(       self  :	List[str]    )		->							None:
          self._resize(len(self._buckets    ) // 2    )

     def 				lowerCamelCase__				(       self  :	Union[str, Any]							,				snake_case  :	KEY    )		->							Iterator[int]:
          __UpperCAmelCase			:				Tuple													=    self._get_bucket_index(snake_case    )
          for _ in range(len(self._buckets    )    ):
               yield ind
               __UpperCAmelCase			:				Dict													=    self._get_next_ind(snake_case    )

     def 				lowerCamelCase__				(       self  :	str							,				snake_case  :	KEY							,				snake_case  :	VAL    )		->							None:
          for ind in self._iterate_buckets(snake_case    ):
               if self._try_set(snake_case							,				snake_case							,				snake_case    ):
                    break

     def __setitem__(       self  :	List[str]							,				snake_case  :	KEY							,				snake_case  :	VAL    )		->							None:
          if self._is_full():
               self._size_up()

          self._add_item(snake_case							,				snake_case    )

     def __delitem__(       self  :	Any							,				snake_case  :	KEY    )		->							None:
          for ind in self._iterate_buckets(snake_case    ):
               __UpperCAmelCase			:				List[Any]													=    self._buckets[ind]
               if item is None:
                    raise KeyError(snake_case    )
               if item is _deleted:
                    continue
               if item.key == key:
                    __UpperCAmelCase			:				Optional[Any]													=    _deleted
                    self._len -= 1
                    break
          if self._is_sparse():
               self._size_down()

     def __getitem__(       self  :	List[str]							,				snake_case  :	KEY    )		->							VAL:
          for ind in self._iterate_buckets(snake_case    ):
               __UpperCAmelCase			:				Optional[int]													=    self._buckets[ind]
               if item is None:
                    break
               if item is _deleted:
                    continue
               if item.key == key:
                    return item.val
          raise KeyError(snake_case    )

     def __len__(       self  :	List[Any]    )		->							int:
          return self._len

     def __iter__(       self  :	str    )		->							Iterator[KEY]:
          yield from (item.key for item in self._buckets if item)






     def __repr__(       self  :	Optional[Any]    )		->							str:
          __UpperCAmelCase			:				str													=    ''' ,'''.join(
              f'{item.key}: {item.val}' for item in self._buckets if item    )
          return f'HashMap({val_string})'