


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


 def __init__(      self							:	Optional[int]			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	str=0.0			,       __lowerCamelCase							:	Optional[int] = None			,       __lowerCamelCase							:	str = "geglu"			,       __lowerCamelCase							:	Optional[int] = None			,       __lowerCamelCase							:	bool = False			,       __lowerCamelCase							:	bool = False			,       __lowerCamelCase							:	bool = False			,       __lowerCamelCase							:	bool = False			,       __lowerCamelCase							:	bool = True			,       __lowerCamelCase							:	str = "layer_norm"			,       __lowerCamelCase							:	bool = False			,       ):
    super().__init__()
    UpperCAmelCase__	:Optional[int]      =      only_cross_attention

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

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

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

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


 def        __SCREAMING_SNAKE_CASE					(      self							:	Tuple			,       __lowerCamelCase							:	Optional[int]			,       __lowerCamelCase							:	int ):
    # Sets chunk feed-forward
    UpperCAmelCase__	:List[str]      =      chunk_size
    UpperCAmelCase__	:str      =      dim





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

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

    UpperCAmelCase__	:Any      =      self.attna(
        __lowerCamelCase			,       encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None			,       attention_mask=__lowerCamelCase			,       **__lowerCamelCase			,       )
    if self.use_ada_layer_norm_zero:
       UpperCAmelCase__	:List[str]      =      gate_msa.unsqueeze(1 ) * attn_output
    UpperCAmelCase__	:int      =      attn_output + hidden_states

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

       UpperCAmelCase__	:Any      =      self.attna(
           __lowerCamelCase			,       encoder_hidden_states=__lowerCamelCase			,       attention_mask=__lowerCamelCase			,       **__lowerCamelCase			,       )
       UpperCAmelCase__	:Tuple      =      attn_output + hidden_states

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

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

    if self.use_ada_layer_norm_zero:
       UpperCAmelCase__	:Union[str, Any]      =      gate_mlp.unsqueeze(1 ) * ff_output

    UpperCAmelCase__	:List[str]      =      ff_output + hidden_states

    return hidden_states




class 				UpperCAmelCase			(			nn.Module						):


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

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

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





 def        __SCREAMING_SNAKE_CASE					(      self							:	int			,       __lowerCamelCase							:	Union[str, Any] ):
    for module in self.net:
       UpperCAmelCase__	:List[Any]      =      module(__lowerCamelCase )
    return hidden_states




class 				UpperCAmelCase			(			nn.Module						):


 def __init__(      self							:	Union[str, Any]			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	str = "none" ):
    super().__init__()
    UpperCAmelCase__	:Tuple      =      nn.Linear(__lowerCamelCase			,       __lowerCamelCase )
    UpperCAmelCase__	:Union[str, Any]      =      approximate


 def        __SCREAMING_SNAKE_CASE					(      self							:	Dict			,       __lowerCamelCase							:	List[Any] ):
    if gate.device.type != "mps":
       return F.gelu(__lowerCamelCase			,       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							:	Optional[Any]			,       __lowerCamelCase							:	Tuple ):
    UpperCAmelCase__	:str      =      self.proj(__lowerCamelCase )
    UpperCAmelCase__	:List[Any]      =      self.gelu(__lowerCamelCase )
    return hidden_states




class 				UpperCAmelCase			(			nn.Module						):


 def __init__(      self							:	Any			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	int ):
    super().__init__()
    UpperCAmelCase__	:List[Any]      =      nn.Linear(__lowerCamelCase			,       dim_out * 2 )


 def        __SCREAMING_SNAKE_CASE					(      self							:	Union[str, Any]			,       __lowerCamelCase							:	Optional[Any] ):
    if gate.device.type != "mps":
       return F.gelu(__lowerCamelCase )
    # mps: gelu is not implemented for float16
    return F.gelu(gate.to(dtype=torch.floataa ) ).to(dtype=gate.dtype )





 def        __SCREAMING_SNAKE_CASE					(      self							:	int			,       __lowerCamelCase							:	Union[str, Any] ):
    UpperCAmelCase__   ,						UpperCAmelCase__	:int      =      self.proj(__lowerCamelCase ).chunk(2			,       dim=-1 )
    return hidden_states * self.gelu(__lowerCamelCase )




class 				UpperCAmelCase			(			nn.Module						):


 def __init__(      self							:	List[str]			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	int ):
    super().__init__()
    UpperCAmelCase__	:List[str]      =      nn.Linear(__lowerCamelCase			,       __lowerCamelCase )





 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[int]			,       __lowerCamelCase							:	Any ):
    UpperCAmelCase__	:Union[str, Any]      =      self.proj(__lowerCamelCase )
    return x * torch.sigmoid(1.7_02 * x )




class 				UpperCAmelCase			(			nn.Module						):


 def __init__(      self							:	Dict			,       __lowerCamelCase							:	Any			,       __lowerCamelCase							:	str ):
    super().__init__()
    UpperCAmelCase__	:str      =      nn.Embedding(__lowerCamelCase			,       __lowerCamelCase )
    UpperCAmelCase__	:Dict      =      nn.SiLU()
    UpperCAmelCase__	:Any      =      nn.Linear(__lowerCamelCase			,       embedding_dim * 2 )
    UpperCAmelCase__	:Optional[Any]      =      nn.LayerNorm(__lowerCamelCase			,       elementwise_affine=__lowerCamelCase )





 def        __SCREAMING_SNAKE_CASE					(      self							:	Optional[int]			,       __lowerCamelCase							:	Tuple			,       __lowerCamelCase							:	Dict ):
    UpperCAmelCase__	:List[Any]      =      self.linear(self.silu(self.emb(__lowerCamelCase ) ) )
    UpperCAmelCase__   ,						UpperCAmelCase__	:Optional[int]      =      torch.chunk(__lowerCamelCase			,       2 )
    UpperCAmelCase__	:Dict      =      self.norm(__lowerCamelCase ) * (1 + scale) + shift
    return x




class 				UpperCAmelCase			(			nn.Module						):


 def __init__(      self							:	Union[str, Any]			,       __lowerCamelCase							:	Tuple			,       __lowerCamelCase							:	Optional[Any] ):
    super().__init__()

    UpperCAmelCase__	:Dict      =      CombinedTimestepLabelEmbeddings(__lowerCamelCase			,       __lowerCamelCase )

    UpperCAmelCase__	:str      =      nn.SiLU()
    UpperCAmelCase__	:Union[str, Any]      =      nn.Linear(__lowerCamelCase			,       6 * embedding_dim			,       bias=__lowerCamelCase )
    UpperCAmelCase__	:Optional[int]      =      nn.LayerNorm(__lowerCamelCase			,       elementwise_affine=__lowerCamelCase			,       eps=1e-6 )





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




class 				UpperCAmelCase			(			nn.Module						):


 def __init__(      self							:	Union[str, Any]			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	int			,       __lowerCamelCase							:	Optional[str] = None			,       __lowerCamelCase							:	float = 1e-5 ):
    super().__init__()
    UpperCAmelCase__	:str      =      num_groups
    UpperCAmelCase__	:Any      =      eps

    if act_fn is None:
       UpperCAmelCase__	:List[str]      =      None
    else:
       UpperCAmelCase__	:Dict      =      get_activation(__lowerCamelCase )

    UpperCAmelCase__	:Any      =      nn.Linear(__lowerCamelCase			,       out_dim * 2 )





 def        __SCREAMING_SNAKE_CASE					(      self							:	Tuple			,       __lowerCamelCase							:	str			,       __lowerCamelCase							:	List[Any] ):
    if self.act:
       UpperCAmelCase__	:List[Any]      =      self.act(__lowerCamelCase )
    UpperCAmelCase__	:Dict      =      self.linear(__lowerCamelCase )
    UpperCAmelCase__	:Dict      =      emb[:, :, None, None]
    UpperCAmelCase__   ,						UpperCAmelCase__	:int      =      emb.chunk(2			,       dim=1 )

    UpperCAmelCase__	:Union[str, Any]      =      F.group_norm(__lowerCamelCase			,       self.num_groups			,       eps=self.eps )
    UpperCAmelCase__	:List[Any]      =      x * (1 + scale) + shift
    return x







'''simple docstring'''





import pytest

from datasets.splits import SplitDict, SplitInfo
from datasets.utils.py_utils import asdict


@pytest.mark.parametrize(
    '''split_dict''',   [
        SplitDict(),
        SplitDict({'''train''': SplitInfo(name='''train''',   num_bytes=1_337,   num_examples=42,   dataset_name='''my_dataset'''						)}						),
        SplitDict({'''train''': SplitInfo(name='''train''',   num_bytes=1_337,   num_examples=42						)}						),
        SplitDict({'''train''': SplitInfo()}						),
    ],   )
def        a__     (					UpperCamelCase_						:      SplitDict						):
   UpperCAmelCase__	:List[Any]      =      split_dict._to_yaml_list()
   assert len(UpperCamelCase_						) == len(UpperCamelCase_						)
   UpperCAmelCase__	:Tuple      =      SplitDict._from_yaml_list(UpperCamelCase_						)
   for split_name, split_info in split_dict.items():
      # dataset_name field is deprecated, and is therefore not part of the YAML dump
      UpperCAmelCase__	:Union[str, Any]      =      None
      # the split name of split_dict takes over the name of the split info object
      UpperCAmelCase__	:Union[str, Any]      =      split_name
   assert split_dict == reloaded


@pytest.mark.parametrize(
    '''split_info''',   [SplitInfo(), SplitInfo(dataset_name=UpperCamelCase_						), SplitInfo(dataset_name='''my_dataset'''						)]						)
def        a__     (					UpperCamelCase_						:      List[Any]						):
   # For backward compatibility, we need asdict(split_dict) to return split info dictrionaries with the "dataset_name"
   # field even if it's deprecated. This way old versionso of `datasets` can still reload dataset_infos.json files
   UpperCAmelCase__	:List[Any]      =      asdict(SplitDict({'''train''': split_info}						)						)
   assert "dataset_name" in split_dict_asdict["train"]
   assert split_dict_asdict["train"]["dataset_name"] == split_info.dataset_name



