



from dataclasses import dataclass
from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..models.embeddings import ImagePositionalEmbeddings
from ..utils import BaseOutput, deprecate
from .attention import BasicTransformerBlock
from .embeddings import PatchEmbed
from .modeling_utils import ModelMixin





@dataclass
class 							lowerCAmelCase_		(		lowercase_     ):
		SCREAMING_SNAKE_CASE_       :			torch.FloatTensor





class 							lowerCAmelCase_		(		lowercase_							,		lowercase_     ):







		@register_to_config
		def __init__(    self  :   Optional[int]       ,		UpperCAmelCase_  :   List[str] = 16       ,		UpperCAmelCase_  :   Tuple = 88       ,		UpperCAmelCase_  :   List[Any] = None       ,		UpperCAmelCase_  :   List[Any] = None       ,		UpperCAmelCase_  :   Dict = 1       ,		UpperCAmelCase_  :   List[Any] = 0.0       ,		UpperCAmelCase_  :   Any = 32       ,		UpperCAmelCase_  :   Tuple = None       ,		UpperCAmelCase_  :   List[str] = False       ,		UpperCAmelCase_  :   List[str] = None       ,		UpperCAmelCase_  :   str = None       ,		UpperCAmelCase_  :   Optional[int] = None       ,		UpperCAmelCase_  :   int = "geglu"       ,		UpperCAmelCase_  :   Optional[Any] = None       ,		UpperCAmelCase_  :   Optional[int] = False       ,		UpperCAmelCase_  :   Dict = False       ,		UpperCAmelCase_  :   Dict = False       ,		UpperCAmelCase_  :   Optional[Any] = "layer_norm"       ,		UpperCAmelCase_  :   str = True       ,		)       ->		Dict:


						'''simple docstring'''







						super().__init__()
						_UpperCAmelCase						:  Tuple     					=	use_linear_projection
						_UpperCAmelCase						:  Union[str, Any]     					=	num_attention_heads
						_UpperCAmelCase						:  str     					=	attention_head_dim
						_UpperCAmelCase						:  Optional[Any]     					=	num_attention_heads * attention_head_dim

						# 1. Transformer2DModel can process both standard continuous images of shape `(batch_size, num_channels, width, height)` as well as quantized image embeddings of shape `(batch_size, num_image_vectors)`
						# Define whether input is continuous or discrete depending on configuration
						_UpperCAmelCase						:  List[str]     					=	(in_channels is not None) and (patch_size is None)
						_UpperCAmelCase						:  Any     					=	num_vector_embeds is not None
						_UpperCAmelCase						:  Optional[Any]     					=	in_channels is not None and patch_size is not None

						if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
										_UpperCAmelCase						:  Any     					=	(
										    F'''The configuration file of this model: {self.__class__} is outdated. `norm_type` is either not set or'''
										    """ incorrectly set to `'layer_norm'`.Make sure to set `norm_type` to `'ada_norm'` in the config."""
										    """ Please make sure to update the config accordingly as leaving `norm_type` might led to incorrect"""
										    """ results in future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it"""
										    """ would be very nice if you could open a Pull request for the `transformer/config.json` file"""
										)
										deprecate('''norm_type!=num_embeds_ada_norm'''       ,		'''1.0.0'''       ,		UpperCAmelCase_       ,		standard_warn=UpperCAmelCase_					)
										_UpperCAmelCase						:  Optional[int]     					=	"""ada_norm"""

						if self.is_input_continuous and self.is_input_vectorized:
										raise ValueError(
										    F'''Cannot define both `in_channels`: {in_channels} and `num_vector_embeds`: {num_vector_embeds}. Make'''
										    ''' sure that either `in_channels` or `num_vector_embeds` is None.'''					)
						elif self.is_input_vectorized and self.is_input_patches:
										raise ValueError(
										    F'''Cannot define both `num_vector_embeds`: {num_vector_embeds} and `patch_size`: {patch_size}. Make'''
										    ''' sure that either `num_vector_embeds` or `num_patches` is None.'''					)
						elif not self.is_input_continuous and not self.is_input_vectorized and not self.is_input_patches:
										raise ValueError(
										    F'''Has to define `in_channels`: {in_channels}, `num_vector_embeds`: {num_vector_embeds}, or patch_size:'''
										    F''' {patch_size}. Make sure that `in_channels`, `num_vector_embeds` or `num_patches` is not None.'''					)

						# 2. Define input layers
						if self.is_input_continuous:
										_UpperCAmelCase						:  int     					=	in_channels

										_UpperCAmelCase						:  int     					=	torch.nn.GroupNorm(num_groups=UpperCAmelCase_       ,		num_channels=UpperCAmelCase_       ,		eps=1E-6       ,		affine=UpperCAmelCase_					)
										if use_linear_projection:
														_UpperCAmelCase						:  Tuple     					=	nn.Linear(UpperCAmelCase_       ,		UpperCAmelCase_					)
										else:
														_UpperCAmelCase						:  Tuple     					=	nn.Convad(UpperCAmelCase_       ,		UpperCAmelCase_       ,		kernel_size=1       ,		stride=1       ,		padding=0					)
						elif self.is_input_vectorized:
										assert sample_size is not None, "Transformer2DModel over discrete input must provide sample_size"
										assert num_vector_embeds is not None, "Transformer2DModel over discrete input must provide num_embed"

										_UpperCAmelCase						:  int     					=	sample_size
										_UpperCAmelCase						:  Dict     					=	sample_size
										_UpperCAmelCase						:  List[str]     					=	num_vector_embeds
										_UpperCAmelCase						:  str     					=	self.height * self.width

										_UpperCAmelCase						:  Any     					=	ImagePositionalEmbeddings(
										    num_embed=UpperCAmelCase_       ,		embed_dim=UpperCAmelCase_       ,		height=self.height       ,		width=self.width					)
						elif self.is_input_patches:
										assert sample_size is not None, "Transformer2DModel over patched input must provide sample_size"

										_UpperCAmelCase						:  List[str]     					=	sample_size
										_UpperCAmelCase						:  str     					=	sample_size

										_UpperCAmelCase						:  Tuple     					=	patch_size
										_UpperCAmelCase						:  List[str]     					=	PatchEmbed(
										    height=UpperCAmelCase_       ,		width=UpperCAmelCase_       ,		patch_size=UpperCAmelCase_       ,		in_channels=UpperCAmelCase_       ,		embed_dim=UpperCAmelCase_       ,		)

						# 3. Define transformers blocks
						_UpperCAmelCase						:  Tuple     					=	nn.ModuleList(
						    [
						        BasicTransformerBlock(
						            UpperCAmelCase_       ,		UpperCAmelCase_       ,		UpperCAmelCase_       ,		dropout=UpperCAmelCase_       ,		cross_attention_dim=UpperCAmelCase_       ,		activation_fn=UpperCAmelCase_       ,		num_embeds_ada_norm=UpperCAmelCase_       ,		attention_bias=UpperCAmelCase_       ,		only_cross_attention=UpperCAmelCase_       ,		upcast_attention=UpperCAmelCase_       ,		norm_type=UpperCAmelCase_       ,		norm_elementwise_affine=UpperCAmelCase_       ,		)
						        for d in range(UpperCAmelCase_					)
						    ]					)

						# 4. Define output layers
						_UpperCAmelCase						:  Optional[Any]     					=	in_channels if out_channels is None else out_channels
						if self.is_input_continuous:
										# TODO: should use out_channels for continuous projections
										if use_linear_projection:
														_UpperCAmelCase						:  Optional[Any]     					=	nn.Linear(UpperCAmelCase_       ,		UpperCAmelCase_					)
										else:
														_UpperCAmelCase						:  List[str]     					=	nn.Convad(UpperCAmelCase_       ,		UpperCAmelCase_       ,		kernel_size=1       ,		stride=1       ,		padding=0					)
						elif self.is_input_vectorized:
										_UpperCAmelCase						:  Tuple     					=	nn.LayerNorm(UpperCAmelCase_					)
										_UpperCAmelCase						:  Optional[Any]     					=	nn.Linear(UpperCAmelCase_       ,		self.num_vector_embeds - 1					)
						elif self.is_input_patches:
										_UpperCAmelCase						:  str     					=	nn.LayerNorm(UpperCAmelCase_       ,		elementwise_affine=UpperCAmelCase_       ,		eps=1E-6					)
										_UpperCAmelCase						:  str     					=	nn.Linear(UpperCAmelCase_       ,		2 * inner_dim					)
										_UpperCAmelCase						:  str     					=	nn.Linear(UpperCAmelCase_       ,		patch_size * patch_size * self.out_channels					)







		def 					a_  (    self  :   List[Any]       ,		UpperCAmelCase_  :   Optional[Any]       ,		UpperCAmelCase_  :   Union[str, Any] = None       ,		UpperCAmelCase_  :   Dict = None       ,		UpperCAmelCase_  :   Union[str, Any] = None       ,		UpperCAmelCase_  :   List[Any] = None       ,		UpperCAmelCase_  :   Optional[Any] = None       ,		UpperCAmelCase_  :   Any = None       ,		UpperCAmelCase_  :   Dict = True       ,		)       ->		Any:


						'''simple docstring'''







						if attention_mask is not None and attention_mask.ndim == 2:
										# assume that mask is expressed as:
										#   (1 = keep,      0 = discard)
										# convert mask into a bias that can be added to attention scores:
										#       (keep = +0,     discard = -10000.0)
										_UpperCAmelCase						:  Dict     					=	(1 - attention_mask.to(hidden_states.dtype					)) * -10_000.0
										_UpperCAmelCase						:  Any     					=	attention_mask.unsqueeze(1					)

						# convert encoder_attention_mask to a bias the same way we do for attention_mask
						if encoder_attention_mask is not None and encoder_attention_mask.ndim == 2:
										_UpperCAmelCase						:  List[str]     					=	(1 - encoder_attention_mask.to(hidden_states.dtype					)) * -10_000.0
										_UpperCAmelCase						:  List[Any]     					=	encoder_attention_mask.unsqueeze(1					)

						# 1. Input
						if self.is_input_continuous:
										_UpperCAmelCase						:  Optional[int]     					=	hidden_states.shape
										_UpperCAmelCase						:  str     					=	hidden_states

										_UpperCAmelCase						:  Tuple     					=	self.norm(UpperCAmelCase_					)
										if not self.use_linear_projection:
														_UpperCAmelCase						:  Tuple     					=	self.proj_in(UpperCAmelCase_					)
														_UpperCAmelCase						:  Any     					=	hidden_states.shape[1]
														_UpperCAmelCase						:  Tuple     					=	hidden_states.permute(0       ,		2       ,		3       ,		1					).reshape(UpperCAmelCase_       ,		height * width       ,		UpperCAmelCase_					)
										else:
														_UpperCAmelCase						:  List[str]     					=	hidden_states.shape[1]
														_UpperCAmelCase						:  Optional[int]     					=	hidden_states.permute(0       ,		2       ,		3       ,		1					).reshape(UpperCAmelCase_       ,		height * width       ,		UpperCAmelCase_					)
														_UpperCAmelCase						:  Union[str, Any]     					=	self.proj_in(UpperCAmelCase_					)
						elif self.is_input_vectorized:
										_UpperCAmelCase						:  Dict     					=	self.latent_image_embedding(UpperCAmelCase_					)
						elif self.is_input_patches:
										_UpperCAmelCase						:  int     					=	self.pos_embed(UpperCAmelCase_					)

						# 2. Blocks
						for block in self.transformer_blocks:
										_UpperCAmelCase						:  Any     					=	block(
										    UpperCAmelCase_       ,		attention_mask=UpperCAmelCase_       ,		encoder_hidden_states=UpperCAmelCase_       ,		encoder_attention_mask=UpperCAmelCase_       ,		timestep=UpperCAmelCase_       ,		cross_attention_kwargs=UpperCAmelCase_       ,		class_labels=UpperCAmelCase_       ,		)

						# 3. Output
						if self.is_input_continuous:
										if not self.use_linear_projection:
														_UpperCAmelCase						:  Optional[Any]     					=	hidden_states.reshape(UpperCAmelCase_       ,		UpperCAmelCase_       ,		UpperCAmelCase_       ,		UpperCAmelCase_					).permute(0       ,		3       ,		1       ,		2					).contiguous()
														_UpperCAmelCase						:  Optional[Any]     					=	self.proj_out(UpperCAmelCase_					)
										else:
														_UpperCAmelCase						:  List[str]     					=	self.proj_out(UpperCAmelCase_					)
														_UpperCAmelCase						:  Optional[int]     					=	hidden_states.reshape(UpperCAmelCase_       ,		UpperCAmelCase_       ,		UpperCAmelCase_       ,		UpperCAmelCase_					).permute(0       ,		3       ,		1       ,		2					).contiguous()

										_UpperCAmelCase						:  List[str]     					=	hidden_states + residual
						elif self.is_input_vectorized:
										_UpperCAmelCase						:  int     					=	self.norm_out(UpperCAmelCase_					)
										_UpperCAmelCase						:  List[str]     					=	self.out(UpperCAmelCase_					)
										# (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
										_UpperCAmelCase						:  List[Any]     					=	logits.permute(0       ,		2       ,		1					)

										# log(p(x_0))
										_UpperCAmelCase						:  List[Any]     					=	F.log_softmax(logits.double()       ,		dim=1					).float()
						elif self.is_input_patches:
										# TODO: cleanup!
										_UpperCAmelCase						:  List[Any]     					=	self.transformer_blocks[0].norma.emb(
										    UpperCAmelCase_       ,		UpperCAmelCase_       ,		hidden_dtype=hidden_states.dtype					)
										_UpperCAmelCase						:  Dict     					=	self.proj_out_a(F.silu(UpperCAmelCase_					)					).chunk(2       ,		dim=1					)
										_UpperCAmelCase						:  int     					=	self.norm_out(UpperCAmelCase_					) * (1 + scale[:, None]) + shift[:, None]
										_UpperCAmelCase						:  int     					=	self.proj_out_a(UpperCAmelCase_					)

										# unpatchify
										_UpperCAmelCase						:  Optional[Any]     					=	int(hidden_states.shape[1] ** 0.5					)
										_UpperCAmelCase						:  Union[str, Any]     					=	hidden_states.reshape(
										    shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels)					)
										_UpperCAmelCase						:  List[Any]     					=	torch.einsum('''nhwpqc->nchpwq'''       ,		UpperCAmelCase_					)
										_UpperCAmelCase						:  Optional[Any]     					=	hidden_states.reshape(
										    shape=(-1, self.out_channels, height * self.patch_size, width * self.patch_size)					)

						if not return_dict:
										return (output,)

						return TransformeraDModelOutput(sample=UpperCAmelCase_					)





"""simple docstring"""







import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


UpperCAmelCase:  Union[str, Any]			     =				namedtuple("""ExampleNamedTuple""", """a b c""")





class   UpperCamelCase	(     unittest.TestCase       ):




	"""simple docstring"""






	def    lowerCamelCase__					(     self   ):
			_lowercase				: List[Any]     =   torch.randn(5				,2   )
			_lowercase				: List[Any]     =   torch.device("""cuda"""   ) if torch.cuda.is_available() else torch.device("""cpu"""   )

			_lowercase				: int     =   send_to_device(UpperCAmelCase_				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta.cpu()				,UpperCAmelCase_   )   )

			_lowercase				: int     =   send_to_device((tensor, [tensor, tensor], 1)				,UpperCAmelCase_   )
			self.assertIsInstance(UpperCAmelCase_				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta[0].cpu()				,UpperCAmelCase_   )   )
			self.assertIsInstance(resulta[1]				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta[1][0].cpu()				,UpperCAmelCase_   )   )
			self.assertTrue(torch.equal(resulta[1][1].cpu()				,UpperCAmelCase_   )   )
			self.assertEqual(resulta[2]				,1   )

			_lowercase				: Any     =   send_to_device({"""a""": tensor, """b""": [tensor, tensor], """c""": 1}				,UpperCAmelCase_   )
			self.assertIsInstance(UpperCAmelCase_				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta["""a"""].cpu()				,UpperCAmelCase_   )   )
			self.assertIsInstance(resulta["""b"""]				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta["""b"""][0].cpu()				,UpperCAmelCase_   )   )
			self.assertTrue(torch.equal(resulta["""b"""][1].cpu()				,UpperCAmelCase_   )   )
			self.assertEqual(resulta["""c"""]				,1   )

			_lowercase				: Optional[int]     =   send_to_device(ExampleNamedTuple(a=UpperCAmelCase_				,b=[tensor, tensor]				,c=1   )				,UpperCAmelCase_   )
			self.assertIsInstance(UpperCAmelCase_				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta.a.cpu()				,UpperCAmelCase_   )   )
			self.assertIsInstance(resulta.b				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta.b[0].cpu()				,UpperCAmelCase_   )   )
			self.assertTrue(torch.equal(resulta.b[1].cpu()				,UpperCAmelCase_   )   )
			self.assertEqual(resulta.c				,1   )

			_lowercase				: str     =   send_to_device(UserDict({"""a""": tensor, """b""": [tensor, tensor], """c""": 1}   )				,UpperCAmelCase_   )
			self.assertIsInstance(UpperCAmelCase_				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta["""a"""].cpu()				,UpperCAmelCase_   )   )
			self.assertIsInstance(resulta["""b"""]				,UpperCAmelCase_   )
			self.assertTrue(torch.equal(resulta["""b"""][0].cpu()				,UpperCAmelCase_   )   )
			self.assertTrue(torch.equal(resulta["""b"""][1].cpu()				,UpperCAmelCase_   )   )
			self.assertEqual(resulta["""c"""]				,1   )






	def    lowerCamelCase__					(     self   ):
			with self.assertRaises(UpperCAmelCase_   ) as cm:
					_lowercase				: str     =   recursively_apply(torch.tensor				,(torch.tensor(1   ), 1)				,error_on_other_type=UpperCAmelCase_   )
			self.assertEqual(
			    str(cm.exception   )				,"""Unsupported types (<class 'int'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed."""				,)






	def    lowerCamelCase__					(     self   ):
			_lowercase				: Optional[int]     =   torch.tensor([1, 2, 3, 4, 5]   )
			self.assertEqual(listify(UpperCAmelCase_   )				,[1, 2, 3, 4, 5]   )

			_lowercase				: Optional[int]     =   torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]   )
			self.assertEqual(listify(UpperCAmelCase_   )				,[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]   )

			_lowercase				: Optional[Any]     =   torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]   )
			self.assertEqual(
			    listify(UpperCAmelCase_   )				,[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]   )






	def    lowerCamelCase__					(     self   ):
			with patch_environment(aa=1				,BB=2   ):
					self.assertEqual(os.environ.get("""AA"""   )				,"""1"""   )
					self.assertEqual(os.environ.get("""BB"""   )				,"""2"""   )

			self.assertNotIn("""AA"""				,os.environ   )
			self.assertNotIn("""BB"""				,os.environ   )






	def    lowerCamelCase__					(     self   ):
			_lowercase				: Optional[Any]     =   RegressionModel()
			_lowercase				: List[Any]     =   model.forward
			_lowercase				: Optional[Any]     =   convert_outputs_to_fpaa(model.forward   )
			_lowercase				: Any     =   extract_model_from_parallel(UpperCAmelCase_				,keep_fpaa_wrapper=UpperCAmelCase_   )
			_lowercase				: Tuple     =   pickle.dumps(UpperCAmelCase_   )






	@require_cuda
	def    lowerCamelCase__					(     self   ):
			_lowercase				: str     =   RegressionModel()
			_lowercase				: Any     =   model.forward
			_lowercase				: Any     =   torch.cuda.amp.autocast(dtype=torch.floataa   )(model.forward   )
			_lowercase				: Optional[Any]     =   convert_outputs_to_fpaa(model.forward   )
			_lowercase				: Any     =   extract_model_from_parallel(UpperCAmelCase_				,keep_fpaa_wrapper=UpperCAmelCase_   )
			_lowercase				: Tuple     =   pickle.dumps(UpperCAmelCase_   )






	@require_cuda
	@require_torch_min_version(version="""2.0"""   )
	def    lowerCamelCase__					(     self   ):
			_lowercase				: str     =   RegressionModel()
			_lowercase				: Any     =   model.forward
			_lowercase				: Union[str, Any]     =   torch.cuda.amp.autocast(dtype=torch.floataa   )(model.forward   )
			_lowercase				: Optional[Any]     =   convert_outputs_to_fpaa(model.forward   )
			_lowercase				: Union[str, Any]     =   torch.compile(model.forward				,backend="""inductor"""   )
			_lowercase				: Any     =   torch.randn(4				,10   ).cuda()
			_lowercase				: int     =   model(UpperCAmelCase_   )






	def    lowerCamelCase__					(     self   ):
			_lowercase				: List[str]     =   RegressionModel()
			# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
			_lowercase				: Any     =   torch.nn.parallel.DataParallel(UpperCAmelCase_   )
			_lowercase				: List[Any]     =   extract_model_from_parallel(UpperCAmelCase_   )

			self.assertEqual(UpperCAmelCase_				,UpperCAmelCase_   )






	@require_torch_min_version(version="""2.0"""   )
	def    lowerCamelCase__					(     self   ):
			_lowercase				: int     =   RegressionModel()
			_lowercase				: List[str]     =   torch.compile(UpperCAmelCase_   )

			# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
			_lowercase				: Optional[int]     =   torch.nn.parallel.DataParallel(UpperCAmelCase_   )
			_lowercase				: int     =   torch.compile(UpperCAmelCase_   )
			_lowercase				: Dict     =   extract_model_from_parallel(UpperCAmelCase_   )

			self.assertEqual(compiled_model._orig_mod				,compiled_model_unwrapped._orig_mod   )






	def    lowerCamelCase__					(     self   ):
			self.assertEqual(find_device([1, """a""", torch.tensor([1, 2, 3]   )]   )				,torch.device("""cpu"""   )   )
			self.assertEqual(find_device({"""a""": 1, """b""": torch.tensor([1, 2, 3]   )}   )				,torch.device("""cpu"""   )   )
			self.assertIsNone(find_device([1, """a"""]   )   )



