






'''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							):







   def __init__(self					,   __a					,   __a					,   __a					,   __a=0.0					,   __a = None					,   __a = "geglu"					,   __a = None					,   __a = False					,   __a = False					,   __a = False					,   __a = False					,   __a = True					,   __a = "layer_norm"					,   __a = False					,   )			->      List[str]:
      """simple docstring"""






      super().__init__()
      __snake_case				:      Tuple     							= only_cross_attention

      __snake_case				:      List[Any]     							= (num_embeds_ada_norm is not None) and norm_type == "ada_norm_zero"
      __snake_case				:      List[Any]     							= (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:
         __snake_case				:      Union[str, Any]     							= AdaLayerNorm(__a					,   __a)
      elif self.use_ada_layer_norm_zero:
         __snake_case				:      Any     							= AdaLayerNormZero(__a					,   __a)
      else:
         __snake_case				:      Optional[Any]     							= nn.LayerNorm(__a					,   elementwise_affine=__a)
      __snake_case				:      Union[str, Any]     							= Attention(
          query_dim=__a					,   heads=__a					,   dim_head=__a					,   dropout=__a					,   bias=__a					,   cross_attention_dim=cross_attention_dim if only_cross_attention else None					,   upcast_attention=__a					,   )

      # 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.
         __snake_case				:      int     							= (
             AdaLayerNorm(__a					,   __a)
             if self.use_ada_layer_norm
             else nn.LayerNorm(__a					,   elementwise_affine=__a)
         )
         __snake_case				:      Optional[Any]     							= Attention(
             query_dim=__a					,   cross_attention_dim=cross_attention_dim if not double_self_attention else None					,   heads=__a					,   dim_head=__a					,   dropout=__a					,   bias=__a					,   upcast_attention=__a					,   )  # is self-attn if encoder_hidden_states is none
      else:
         __snake_case				:      Optional[Any]     							= None
         __snake_case				:      Optional[int]     							= None

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

      # let chunk size default to None
      __snake_case				:      List[str]     							= None
      __snake_case				:      Optional[Any]     							= 0







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a)			->      Tuple:
      """simple docstring"""






      __snake_case				:      List[str]     							= chunk_size
      __snake_case				:      Tuple     							= dim







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   __a = None					,   )			->      Optional[Any]:
      """simple docstring"""






      if self.use_ada_layer_norm:
         __snake_case				:      int     							= self.norma(__a					,   __a)
      elif self.use_ada_layer_norm_zero:
         __snake_case				:      Dict     							= self.norma(
             __a					,   __a					,   __a					,   hidden_dtype=hidden_states.dtype)
      else:
         __snake_case				:      str     							= self.norma(__a)

      __snake_case				:      Optional[int]     							= cross_attention_kwargs if cross_attention_kwargs is not None else {}

      __snake_case				:      Optional[int]     							= self.attna(
          __a					,   encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None					,   attention_mask=__a					,   **__a					,   )
      if self.use_ada_layer_norm_zero:
         __snake_case				:      Optional[int]     							= gate_msa.unsqueeze(1) * attn_output
      __snake_case				:      Tuple     							= attn_output + hidden_states

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

         __snake_case				:      Any     							= self.attna(
             __a					,   encoder_hidden_states=__a					,   attention_mask=__a					,   **__a					,   )
         __snake_case				:      List[Any]     							= attn_output + hidden_states

      # 3. Feed-forward
      __snake_case				:      Optional[int]     							= self.norma(__a)

      if self.use_ada_layer_norm_zero:
         __snake_case				:      str     							= 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`.""")

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

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

      __snake_case				:      Tuple     							= ff_output + hidden_states

      return hidden_states




class a_    (    nn.Module							):







   def __init__(self					,   __a					,   __a = None					,   __a = 4					,   __a = 0.0					,   __a = "geglu"					,   __a = False					,   )			->      str:
      """simple docstring"""






      super().__init__()
      __snake_case				:      List[Any]     							= int(dim * mult)
      __snake_case				:      List[Any]     							= dim_out if dim_out is not None else dim

      if activation_fn == "gelu":
         __snake_case				:      Optional[Any]     							= GELU(__a					,   __a)
      if activation_fn == "gelu-approximate":
         __snake_case				:      Dict     							= GELU(__a					,   __a					,   approximate='tanh')
      elif activation_fn == "geglu":
         __snake_case				:      Dict     							= GEGLU(__a					,   __a)
      elif activation_fn == "geglu-approximate":
         __snake_case				:      List[str]     							= ApproximateGELU(__a					,   __a)

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







   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      List[str]:
      """simple docstring"""






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




class a_    (    nn.Module							):







   def __init__(self					,   __a					,   __a					,   __a = "none")			->      List[str]:
      """simple docstring"""






      super().__init__()
      __snake_case				:      Optional[Any]     							= nn.Linear(__a					,   __a)
      __snake_case				:      List[str]     							= approximate







   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      Dict:
      """simple docstring"""






      if gate.device.type != "mps":
         return F.gelu(__a					,   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     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      Tuple:
      """simple docstring"""






      __snake_case				:      int     							= self.proj(__a)
      __snake_case				:      List[Any]     							= self.gelu(__a)
      return hidden_states




class a_    (    nn.Module							):







   def __init__(self					,   __a					,   __a)			->      str:
      """simple docstring"""






      super().__init__()
      __snake_case				:      Optional[Any]     							= nn.Linear(__a					,   dim_out * 2)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      List[str]:
      """simple docstring"""






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







   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      Optional[int]:
      """simple docstring"""






      __snake_case				:      Optional[Any]     							= self.proj(__a).chunk(2					,   dim=-1)
      return hidden_states * self.gelu(__a)




class a_    (    nn.Module							):







   def __init__(self					,   __a					,   __a)			->      Any:
      """simple docstring"""






      super().__init__()
      __snake_case				:      str     							= nn.Linear(__a					,   __a)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      str:
      """simple docstring"""






      __snake_case				:      Dict     							= self.proj(__a)
      return x * torch.sigmoid(1.702 * x)




class a_    (    nn.Module							):







   def __init__(self					,   __a					,   __a)			->      Union[str, Any]:
      """simple docstring"""






      super().__init__()
      __snake_case				:      Dict     							= nn.Embedding(__a					,   __a)
      __snake_case				:      Optional[int]     							= nn.SiLU()
      __snake_case				:      Tuple     							= nn.Linear(__a					,   embedding_dim * 2)
      __snake_case				:      str     							= nn.LayerNorm(__a					,   elementwise_affine=__a)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a)			->      List[Any]:
      """simple docstring"""






      __snake_case				:      Optional[int]     							= self.linear(self.silu(self.emb(__a)))
      __snake_case				:      Optional[Any]     							= torch.chunk(__a					,   2)
      __snake_case				:      List[Any]     							= self.norm(__a) * (1 + scale) + shift
      return x




class a_    (    nn.Module							):







   def __init__(self					,   __a					,   __a)			->      List[Any]:
      """simple docstring"""






      super().__init__()

      __snake_case				:      Any     							= CombinedTimestepLabelEmbeddings(__a					,   __a)

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







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a					,   __a					,   __a=None)			->      int:
      """simple docstring"""






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




class a_    (    nn.Module							):







   def __init__(self					,   __a					,   __a					,   __a					,   __a = None					,   __a = 1E-5)			->      Dict:
      """simple docstring"""






      super().__init__()
      __snake_case				:      Optional[int]     							= num_groups
      __snake_case				:      List[str]     							= eps

      if act_fn is None:
         __snake_case				:      int     							= None
      else:
         __snake_case				:      List[str]     							= get_activation(__a)

      __snake_case				:      List[Any]     							= nn.Linear(__a					,   out_dim * 2)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a)			->      Tuple:
      """simple docstring"""






      if self.act:
         __snake_case				:      Any     							= self.act(__a)
      __snake_case				:      Optional[Any]     							= self.linear(__a)
      __snake_case				:      Tuple     							= emb[:, :, None, None]
      __snake_case				:      Optional[int]     							= emb.chunk(2					,   dim=1)

      __snake_case				:      Optional[int]     							= F.group_norm(__a					,   self.num_groups					,   eps=self.eps)
      __snake_case				:      Optional[Any]     							= x * (1 + scale) + shift
      return x







'''simple docstring'''






import unittest

from transformers import load_tool
from transformers.utils import is_torch_available


if is_torch_available():
      import torch

from transformers.testing_utils import require_torch

from .test_tools_common import ToolTesterMixin




@require_torch
class a_    (    unittest.TestCase		,      UpperCamelCase_							):







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[Any]:
      """simple docstring"""






      __snake_case				:      List[str]     							= load_tool('text-to-speech')
      self.tool.setup()







   def     SCREAMING_SNAKE_CASE__					(self)			->      Optional[Any]:
      """simple docstring"""






      torch.manual_seed(0)
      __snake_case				:      Dict     							= self.tool('hey')
      __snake_case				:      List[Any]     							= result.to_raw()
      self.assertTrue(
          torch.allclose(
              resulting_tensor[:3]					,   torch.tensor([-0.0_005_966_668_832_115_829, -0.0_003_657_640_190_795_064, -0.00_013_439_502_799_883_485])					,   ))







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[Any]:
      """simple docstring"""






      torch.manual_seed(0)
      __snake_case				:      Any     							= self.tool('hey')
      __snake_case				:      Any     							= result.to_raw()
      self.assertTrue(
          torch.allclose(
              resulting_tensor[:3]					,   torch.tensor([-0.0_005_966_668_832_115_829, -0.0_003_657_640_190_795_064, -0.00_013_439_502_799_883_485])					,   ))