

'''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 :	int     ,						A_ :	int     ,						A_ :	int     ,						A_ :	int     ,						A_ :	int=0.0     ,						A_ :	Optional[int] = None     ,						A_ :	str = "geglu"     ,						A_ :	Optional[int] = None     ,						A_ :	bool = False     ,						A_ :	bool = False     ,						A_ :	bool = False     ,						A_ :	bool = False     ,						A_ :	bool = True     ,						A_ :	str = "layer_norm"     ,						A_ :	bool = False     ,						):






      '''simple docstring'''




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

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

      # 3. Feed-forward
      _lowerCAmelCase   :					str        =    nn.LayerNorm(A_     ,						elementwise_affine=A_       )
      _lowerCAmelCase   :					List[str]        =    FeedForward(A_     ,						dropout=A_     ,						activation_fn=A_     ,						final_dropout=A_       )

      # let chunk size default to None
      _lowerCAmelCase   :					List[str]        =    None
      _lowerCAmelCase   :					int        =    0

 def    __magic_name__       (		self :	Optional[Any]     ,						A_ :	Optional[int]     ,						A_ :	int       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Optional[Any]        =    chunk_size
      _lowerCAmelCase   :					Dict        =    dim


 def    __magic_name__       (		self :	Optional[int]     ,						A_ :	torch.FloatTensor     ,						A_ :	Optional[torch.FloatTensor] = None     ,						A_ :	Optional[torch.FloatTensor] = None     ,						A_ :	Optional[torch.FloatTensor] = None     ,						A_ :	Optional[torch.LongTensor] = None     ,						A_ :	Dict[str, Any] = None     ,						A_ :	Optional[torch.LongTensor] = None     ,						):






      '''simple docstring'''




      if self.use_ada_layer_norm:
           _lowerCAmelCase   :					List[Any]        =    self.norma(A_     ,						A_       )
      elif self.use_ada_layer_norm_zero:
           _lowerCAmelCase   :					Optional[int]        =    self.norma(
               A_     ,						A_     ,						A_     ,						hidden_dtype=hidden_states.dtype       )
      else:
           _lowerCAmelCase   :					str        =    self.norma(A_       )

      _lowerCAmelCase   :					List[Any]        =    cross_attention_kwargs if cross_attention_kwargs is not None else {}

      _lowerCAmelCase   :					Optional[Any]        =    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:
           _lowerCAmelCase   :					Dict        =    gate_msa.unsqueeze(1       ) * attn_output
      _lowerCAmelCase   :					Optional[Any]        =    attn_output + hidden_states

      # 2. Cross-Attention
      if self.attna is not None:
           _lowerCAmelCase   :					List[str]        =    (
               self.norma(A_     ,						A_       ) if self.use_ada_layer_norm else self.norma(A_       )
           )

           _lowerCAmelCase   :					Tuple        =    self.attna(
               A_     ,						encoder_hidden_states=A_     ,						attention_mask=A_     ,						**A_     ,						)
           _lowerCAmelCase   :					List[str]        =    attn_output + hidden_states

      # 3. Feed-forward
      _lowerCAmelCase   :					Union[str, Any]        =    self.norma(A_       )

      if self.use_ada_layer_norm_zero:
           _lowerCAmelCase   :					Any        =    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   :					Optional[Any]        =    norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
           _lowerCAmelCase   :					Any        =    torch.cat(
               [self.ff(A_       ) for hid_slice in norm_hidden_states.chunk(A_     ,						dim=self._chunk_dim       )]     ,						dim=self._chunk_dim     ,						)
      else:
           _lowerCAmelCase   :					Dict        =    self.ff(A_       )

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

      _lowerCAmelCase   :					Dict        =    ff_output + hidden_states

      return hidden_states





class A__    ( nn.Module ):



 """simple docstring"""







 def __init__(		self :	Union[str, Any]     ,						A_ :	int     ,						A_ :	Optional[int] = None     ,						A_ :	int = 4     ,						A_ :	float = 0.0     ,						A_ :	str = "geglu"     ,						A_ :	bool = False     ,						):






      '''simple docstring'''




      super().__init__()
      _lowerCAmelCase   :					Optional[Any]        =    int(dim * mult       )
      _lowerCAmelCase   :					Union[str, Any]        =    dim_out if dim_out is not None else dim

      if activation_fn == "gelu":
           _lowerCAmelCase   :					Optional[Any]        =    GELU(A_     ,						A_       )
      if activation_fn == "gelu-approximate":
           _lowerCAmelCase   :					List[str]        =    GELU(A_     ,						A_     ,						approximate="tanh"       )
      elif activation_fn == "geglu":
           _lowerCAmelCase   :					List[Any]        =    GEGLU(A_     ,						A_       )
      elif activation_fn == "geglu-approximate":
           _lowerCAmelCase   :					int        =    ApproximateGELU(A_     ,						A_       )

      _lowerCAmelCase   :					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    __magic_name__       (		self :	Optional[Any]     ,						A_ :	Dict       ):






      '''simple docstring'''




      for module in self.net:
           _lowerCAmelCase   :					List[Any]        =    module(A_       )
      return hidden_states





class A__    ( nn.Module ):



 """simple docstring"""







 def __init__(		self :	int     ,						A_ :	int     ,						A_ :	int     ,						A_ :	str = "none"       ):






      '''simple docstring'''




      super().__init__()
      _lowerCAmelCase   :					Union[str, Any]        =    nn.Linear(A_     ,						A_       )
      _lowerCAmelCase   :					List[str]        =    approximate

 def    __magic_name__       (		self :	List[Any]     ,						A_ :	Any       ):






      '''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    __magic_name__       (		self :	int     ,						A_ :	int       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Optional[Any]        =    self.proj(A_       )
      _lowerCAmelCase   :					Dict        =    self.gelu(A_       )
      return hidden_states





class A__    ( nn.Module ):



 """simple docstring"""







 def __init__(		self :	List[Any]     ,						A_ :	int     ,						A_ :	int       ):






      '''simple docstring'''




      super().__init__()
      _lowerCAmelCase   :					Tuple        =    nn.Linear(A_     ,						dim_out * 2       )

 def    __magic_name__       (		self :	Any     ,						A_ :	Dict       ):






      '''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    __magic_name__       (		self :	str     ,						A_ :	Tuple       ):






      '''simple docstring'''




      _lowerCAmelCase   :					List[str]        =    self.proj(A_       ).chunk(2     ,						dim=-1       )
      return hidden_states * self.gelu(A_       )





class A__    ( nn.Module ):



 """simple docstring"""







 def __init__(		self :	Optional[Any]     ,						A_ :	int     ,						A_ :	int       ):






      '''simple docstring'''




      super().__init__()
      _lowerCAmelCase   :					Optional[int]        =    nn.Linear(A_     ,						A_       )


 def    __magic_name__       (		self :	Dict     ,						A_ :	List[Any]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Union[str, Any]        =    self.proj(A_       )
      return x * torch.sigmoid(1.702 * x       )





class A__    ( nn.Module ):



 """simple docstring"""







 def __init__(		self :	Dict     ,						A_ :	List[str]     ,						A_ :	Optional[Any]       ):






      '''simple docstring'''




      super().__init__()
      _lowerCAmelCase   :					List[Any]        =    nn.Embedding(A_     ,						A_       )
      _lowerCAmelCase   :					Dict        =    nn.SiLU()
      _lowerCAmelCase   :					int        =    nn.Linear(A_     ,						embedding_dim * 2       )
      _lowerCAmelCase   :					Optional[int]        =    nn.LayerNorm(A_     ,						elementwise_affine=A_       )


 def    __magic_name__       (		self :	Tuple     ,						A_ :	List[Any]     ,						A_ :	Optional[int]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Any        =    self.linear(self.silu(self.emb(A_       )       )       )
      _lowerCAmelCase   :					Union[str, Any]        =    torch.chunk(A_     ,						2       )
      _lowerCAmelCase   :					Optional[Any]        =    self.norm(A_       ) * (1 + scale) + shift
      return x





class A__    ( nn.Module ):



 """simple docstring"""







 def __init__(		self :	Union[str, Any]     ,						A_ :	str     ,						A_ :	str       ):






      '''simple docstring'''




      super().__init__()

      _lowerCAmelCase   :					Union[str, Any]        =    CombinedTimestepLabelEmbeddings(A_     ,						A_       )

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


 def    __magic_name__       (		self :	str     ,						A_ :	Any     ,						A_ :	Dict     ,						A_ :	int     ,						A_ :	List[str]=None       ):






      '''simple docstring'''




      _lowerCAmelCase   :					int        =    self.linear(self.silu(self.emb(A_     ,						A_     ,						hidden_dtype=A_       )       )       )
      _lowerCAmelCase   :					str        =    emb.chunk(6     ,						dim=1       )
      _lowerCAmelCase   :					Any        =    self.norm(A_       ) * (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 :	int     ,						A_ :	int     ,						A_ :	int     ,						A_ :	int     ,						A_ :	Optional[str] = None     ,						A_ :	float = 1E-5       ):






      '''simple docstring'''




      super().__init__()
      _lowerCAmelCase   :					List[Any]        =    num_groups
      _lowerCAmelCase   :					str        =    eps

      if act_fn is None:
           _lowerCAmelCase   :					str        =    None
      else:
           _lowerCAmelCase   :					List[str]        =    get_activation(A_       )

      _lowerCAmelCase   :					Dict        =    nn.Linear(A_     ,						out_dim * 2       )


 def    __magic_name__       (		self :	Dict     ,						A_ :	Any     ,						A_ :	Tuple       ):






      '''simple docstring'''




      if self.act:
           _lowerCAmelCase   :					str        =    self.act(A_       )
      _lowerCAmelCase   :					Dict        =    self.linear(A_       )
      _lowerCAmelCase   :					str        =    emb[:, :, None, None]
      _lowerCAmelCase   :					List[str]        =    emb.chunk(2     ,						dim=1       )

      _lowerCAmelCase   :					Dict        =    F.group_norm(A_     ,						self.num_groups     ,						eps=self.eps       )
      _lowerCAmelCase   :					Optional[Any]        =    x * (1 + scale) + shift
      return x









from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_tf_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_tf_available():
  import tensorflow as tf


__UpperCAmelCase    				=     logging.get_logger(__name__)





@dataclass
class A__    ( A ):



 """simple docstring"""







 _lowercase     :					List[Any]      =						[
     '''no_inference''',
     '''no_cuda''',
     '''no_tpu''',
     '''no_speed''',
     '''no_memory''',
     '''no_env_print''',
     '''no_multi_process''',
 ]

 def __init__(		self :	List[Any]     ,						**A_ :	str       ):






      '''simple docstring'''




      for deprecated_arg in self.deprecated_args:
           if deprecated_arg in kwargs:
                _lowerCAmelCase   :					str        =    deprecated_arg[3:]
                _lowerCAmelCase   :					List[Any]        =    not kwargs.pop(A_       )
                logger.warning(
                    F'''{deprecated_arg} is depreciated. Please use --no-{positive_arg} or'''
                    F''' {positive_arg}={kwargs[positive_arg]}'''       )
      _lowerCAmelCase   :					Any        =    kwargs.pop("tpu_name"     ,						self.tpu_name       )
      _lowerCAmelCase   :					str        =    kwargs.pop("device_idx"     ,						self.device_idx       )
      _lowerCAmelCase   :					str        =    kwargs.pop("eager_mode"     ,						self.eager_mode       )
      _lowerCAmelCase   :					Tuple        =    kwargs.pop("use_xla"     ,						self.use_xla       )
      super().__init__(**A_       )

 _lowercase     :					str      =						field(
     default=A  ,		metadata={'''help''': '''Name of TPU'''}  ,		)
 _lowercase     :					int      =						field(
     default=0  ,		metadata={'''help''': '''CPU / GPU device index. Defaults to 0.'''}  ,		)
 _lowercase     :					bool      =						field(default=A  ,		metadata={'''help''': '''Benchmark models in eager model.'''} )
 _lowercase     :					bool      =						field(
     default=A  ,		metadata={
         '''help''': '''Benchmark models using XLA JIT compilation. Note that `eager_model` has to be set to `False`.'''
     }  ,		)

 @cached_property
 def    __magic_name__       (		self :	Tuple       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      _lowerCAmelCase   :					Any        =    None
      if self.tpu:
           try:
                if self.tpu_name:
                     _lowerCAmelCase   :					Union[str, Any]        =    tf.distribute.cluster_resolver.TPUClusterResolver(self.tpu_name       )
                else:
                     _lowerCAmelCase   :					List[str]        =    tf.distribute.cluster_resolver.TPUClusterResolver()
           except ValueError:
                _lowerCAmelCase   :					Optional[int]        =    None
      return tpu

 @cached_property
 def    __magic_name__       (		self :	Optional[int]       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      if self.is_tpu:
           tf.config.experimental_connect_to_cluster(self._setup_tpu       )
           tf.tpu.experimental.initialize_tpu_system(self._setup_tpu       )

           _lowerCAmelCase   :					List[Any]        =    tf.distribute.TPUStrategy(self._setup_tpu       )
      else:
           # currently no multi gpu is allowed
           if self.is_gpu:
                # TODO: Currently only single GPU is supported
                tf.config.set_visible_devices(self.gpu_list[self.device_idx]     ,						"GPU"       )
                _lowerCAmelCase   :					Dict        =    tf.distribute.OneDeviceStrategy(device=F'''/gpu:{self.device_idx}'''       )
           else:
                tf.config.set_visible_devices([]     ,						"GPU"       )  # disable GPU
                _lowerCAmelCase   :					Optional[Any]        =    tf.distribute.OneDeviceStrategy(device=F'''/cpu:{self.device_idx}'''       )

      return strategy

 @property
 def    __magic_name__       (		self :	int       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      return self._setup_tpu is not None

 @property
 def    __magic_name__       (		self :	List[Any]       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      return self._setup_strategy

 @property
 def    __magic_name__       (		self :	List[str]       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      return tf.config.list_physical_devices("GPU"       )

 @property
 def    __magic_name__       (		self :	Optional[Any]       ):






      '''simple docstring'''




      requires_backends(self     ,						["tf"]       )
      if self.cuda:
           return len(self.gpu_list       )
      return 0


 @property
 def    __magic_name__       (		self :	Optional[Any]       ):






      '''simple docstring'''




      return self.n_gpu > 0






