


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	:  List[str]   ,     lowerCamelCase	:  Dict   ,     lowerCamelCase	:  Optional[int]   ,     lowerCamelCase	:  Optional[int]   ,     lowerCamelCase	:  Tuple=0.0   ,     lowerCamelCase	:  Optional[Any] = None   ,     lowerCamelCase	:  Optional[Any] = "geglu"   ,     lowerCamelCase	:  Any = None   ,     lowerCamelCase	:  Any = False   ,     lowerCamelCase	:  Any = False   ,     lowerCamelCase	:  int = False   ,     lowerCamelCase	:  int = False   ,     lowerCamelCase	:  str = True   ,     lowerCamelCase	:  Optional[int] = "layer_norm"   ,     lowerCamelCase	:  Optional[int] = False   ,     )     ->   List[str]:
														super().__init__()
														__snake_case		:      Union[str, Any]										=       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		:      List[Any]										=       AdaLayerNorm(__A   ,     __A    )
														elif self.use_ada_layer_norm_zero:
																					__snake_case		:      List[str]										=       AdaLayerNormZero(__A   ,     __A    )
														else:
																					__snake_case		:      Dict										=       nn.LayerNorm(__A   ,     elementwise_affine=__A    )
														__snake_case		:      Optional[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		:      List[str]										=       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		:      str										=       None
																					__snake_case		:      List[str]										=       None

														# 3. Feed-forward
														__snake_case		:      Dict										=       nn.LayerNorm(__A   ,     elementwise_affine=__A    )
														__snake_case		:      Tuple										=       FeedForward(__A   ,     dropout=__A   ,     activation_fn=__A   ,     final_dropout=__A    )

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

							def    __snake_case (							self	:  List[str]   ,     lowerCamelCase	:  List[str]   ,     lowerCamelCase	:  List[Any]    )     ->   str:
														# Sets chunk feed-forward
														__snake_case		:      Any										=       chunk_size
														__snake_case		:      List[Any]										=       dim






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

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

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

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

																					__snake_case		:      Union[str, Any]										=       self.attna(
																					    __A   ,     encoder_hidden_states=__A   ,     attention_mask=__A   ,     **__A   ,     )
																					__snake_case		:      Dict										=       attn_output + hidden_states

														# 3. Feed-forward
														__snake_case		:      Dict										=       self.norma(__A    )

														if self.use_ada_layer_norm_zero:
																					__snake_case		:      Union[str, 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`.'    )

																					__snake_case		:      Optional[int]										=       norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
																					__snake_case		:      Tuple										=       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		:      List[str]										=       self.ff(__A    )

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

														__snake_case		:      List[Any]										=       ff_output + hidden_states

														return hidden_states

class     a		(nn.Module				):




							"""simple docstring"""


							def __init__(							self	:  Dict   ,     lowerCamelCase	:  List[Any]   ,     lowerCamelCase	:  str = None   ,     lowerCamelCase	:  int = 4   ,     lowerCamelCase	:  Optional[Any] = 0.0   ,     lowerCamelCase	:  Any = "geglu"   ,     lowerCamelCase	:  List[Any] = False   ,     )     ->   Dict:
														super().__init__()
														__snake_case		:      Optional[int]										=       int(dim * mult    )
														__snake_case		:      List[str]										=       dim_out if dim_out is not None else dim

														if activation_fn == "gelu":
																					__snake_case		:      str										=       GELU(__A   ,     __A    )
														if activation_fn == "gelu-approximate":
																					__snake_case		:      List[str]										=       GELU(__A   ,     __A   ,     approximate="tanh"    )
														elif activation_fn == "geglu":
																					__snake_case		:      Tuple										=       GEGLU(__A   ,     __A    )
														elif activation_fn == "geglu-approximate":
																					__snake_case		:      List[Any]										=       ApproximateGELU(__A   ,     __A    )

														__snake_case		:      Optional[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    __snake_case (							self	:  Optional[int]   ,     lowerCamelCase	:  Tuple    )     ->   List[Any]:
														for module in self.net:
																					__snake_case		:      str										=       module(__A    )
														return hidden_states

class     a		(nn.Module				):




							"""simple docstring"""


							def __init__(							self	:  Tuple   ,     lowerCamelCase	:  str   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  List[str] = "none"    )     ->   Dict:
														super().__init__()
														__snake_case		:      int										=       nn.Linear(__A   ,     __A    )
														__snake_case		:      Tuple										=       approximate

							def    __snake_case (							self	:  int   ,     lowerCamelCase	:  Union[str, Any]    )     ->   str:
														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    __snake_case (							self	:  List[str]   ,     lowerCamelCase	:  Any    )     ->   Tuple:
														__snake_case		:      List[Any]										=       self.proj(__A    )
														__snake_case		:      List[str]										=       self.gelu(__A    )
														return hidden_states

class     a		(nn.Module				):




							"""simple docstring"""


							def __init__(							self	:  List[Any]   ,     lowerCamelCase	:  str   ,     lowerCamelCase	:  Dict    )     ->   Union[str, Any]:
														super().__init__()
														__snake_case		:      Any										=       nn.Linear(__A   ,     dim_out * 2    )

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

class     a		(nn.Module				):




							"""simple docstring"""


							def __init__(							self	:  Any   ,     lowerCamelCase	:  str   ,     lowerCamelCase	:  List[Any]    )     ->   Optional[Any]:
														super().__init__()
														__snake_case		:      Optional[int]										=       nn.Linear(__A   ,     __A    )






							def    __snake_case (							self	:  List[Any]   ,     lowerCamelCase	:  Dict    )     ->   str:
														__snake_case		:      str										=       self.proj(__A    )
														return x * torch.sigmoid(1.7_02 * x    )

class     a		(nn.Module				):




							"""simple docstring"""


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






							def    __snake_case (							self	:  List[Any]   ,     lowerCamelCase	:  Optional[Any]   ,     lowerCamelCase	:  Union[str, Any]    )     ->   str:
														__snake_case		:      List[Any]										=       self.linear(self.silu(self.emb(__A    )    )    )
														__snake_case		:      str										=       torch.chunk(__A   ,     2    )
														__snake_case		:      Optional[Any]										=       self.norm(__A    ) * (1 + scale) + shift
														return x

class     a		(nn.Module				):




							"""simple docstring"""


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

														__snake_case		:      Optional[Any]										=       CombinedTimestepLabelEmbeddings(__A   ,     __A    )

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






							def    __snake_case (							self	:  int   ,     lowerCamelCase	:  List[Any]   ,     lowerCamelCase	:  str   ,     lowerCamelCase	:  Union[str, Any]   ,     lowerCamelCase	:  List[Any]=None    )     ->   Tuple:
														__snake_case		:      List[Any]										=       self.linear(self.silu(self.emb(__A   ,     __A   ,     hidden_dtype=__A    )    )    )
														__snake_case		:      Union[str, Any]										=       emb.chunk(6   ,     dim=1    )
														__snake_case		:      List[str]										=       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	:  List[Any]   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  Union[str, Any]   ,     lowerCamelCase	:  Dict   ,     lowerCamelCase	:  List[str] = None   ,     lowerCamelCase	:  Dict = 1E-5    )     ->   int:
														super().__init__()
														__snake_case		:      Tuple										=       num_groups
														__snake_case		:      Tuple										=       eps

														if act_fn is None:
																					__snake_case		:      int										=       None
														else:
																					__snake_case		:      List[Any]										=       get_activation(__A    )

														__snake_case		:      Dict										=       nn.Linear(__A   ,     out_dim * 2    )






							def    __snake_case (							self	:  Tuple   ,     lowerCamelCase	:  Optional[int]   ,     lowerCamelCase	:  Union[str, Any]    )     ->   Tuple:
														if self.act:
																					__snake_case		:      Optional[Any]										=       self.act(__A    )
														__snake_case		:      Optional[int]										=       self.linear(__A    )
														__snake_case		:      Optional[Any]										=       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		:      int										=       x * (1 + scale) + shift
														return x





# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from pathlib import Path

import torch

from ...utils import is_npu_available, is_xpu_available
from .config_args import ClusterConfig, default_json_config_file
from .config_utils import SubcommandHelpFormatter


_snake_case					:		Optional[Any]      =    "Create a default config file for Accelerate with only a few flags set."







def     lowerCAmelCase_					( __lowerCamelCase="no"							,     __lowerCamelCase = default_json_config_file							,     __lowerCamelCase = False     ):
							__snake_case		:      int										=       Path(__lowerCamelCase     )
							path.parent.mkdir(parents=__lowerCamelCase							,     exist_ok=__lowerCamelCase     )
							if path.exists():
														print(
														    F'Configuration already exists at {save_location}, will not override. Run `accelerate config` manually or pass a different `save_location`.'     )
														return False
							__snake_case		:      Any										=       mixed_precision.lower()
							if mixed_precision not in ["no", "fp16", "bf16", "fp8"]:
														raise ValueError(
														    F'`mixed_precision` should be one of \'no\', \'fp16\', \'bf16\', or \'fp8\'. Received {mixed_precision}'     )
							__snake_case		:      Optional[int]										=       {
							    "compute_environment": "LOCAL_MACHINE",
							    "mixed_precision": mixed_precision,
							}
							if torch.cuda.is_available():
														__snake_case		:      Dict										=       torch.cuda.device_count()
														__snake_case		:      Tuple										=       num_gpus
														__snake_case		:      List[str]										=       False
														if num_gpus > 1:
																					__snake_case		:      Optional[int]										=       "MULTI_GPU"
														else:
																					__snake_case		:      Dict										=       "NO"
							elif is_xpu_available() and use_xpu:
														__snake_case		:      List[str]										=       torch.xpu.device_count()
														__snake_case		:      str										=       num_xpus
														__snake_case		:      int										=       False
														if num_xpus > 1:
																					__snake_case		:      Optional[int]										=       "MULTI_XPU"
														else:
																					__snake_case		:      str										=       "NO"
							elif is_npu_available():
														__snake_case		:      Any										=       torch.npu.device_count()
														__snake_case		:      str										=       num_npus
														__snake_case		:      str										=       False
														if num_npus > 1:
																					__snake_case		:      Optional[int]										=       "MULTI_NPU"
														else:
																					__snake_case		:      int										=       "NO"
							else:
														__snake_case		:      List[Any]										=       0
														__snake_case		:      Dict										=       True
														__snake_case		:      Tuple										=       1
														__snake_case		:      Tuple										=       "NO"
							__snake_case		:      str										=       ClusterConfig(**__lowerCamelCase     )
							config.to_json_file(__lowerCamelCase     )
							return path







def     lowerCAmelCase_					( __lowerCamelCase							,     __lowerCamelCase     ):
							__snake_case		:      Optional[Any]										=       parser.add_parser("default"							,     parents=__lowerCamelCase							,     help=__lowerCamelCase							,     formatter_class=__lowerCamelCase     )
							parser.add_argument(
							    "--config_file"							,     default=__lowerCamelCase							,     help=(
							        "The path to use to store the config file. Will default to a file named default_config.yaml in the cache "
							        "location, which is the content of the environment `HF_HOME` suffixed with 'accelerate', or if you don't have "
							        "such an environment variable, your cache directory ('~/.cache' or the content of `XDG_CACHE_HOME`) suffixed "
							        "with 'huggingface'."
							    )							,     dest="save_location"							,     )

							parser.add_argument(
							    "--mixed_precision"							,     choices=["no", "fp16", "bf16"]							,     type=__lowerCamelCase							,     help="Whether or not to use mixed precision training. "
							    "Choose between FP16 and BF16 (bfloat16) training. "
							    "BF16 training is only supported on Nvidia Ampere GPUs and PyTorch 1.10 or later."							,     default="no"							,     )
							parser.set_defaults(func=__lowerCamelCase     )
							return parser







def     lowerCAmelCase_					( __lowerCamelCase     ):
							__snake_case		:      List[Any]										=       write_basic_config(args.mixed_precision							,     args.save_location     )
							if config_file:
														print(F'accelerate configuration saved at {config_file}'     )

