

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     lowercase						(       _UpperCamelCase					):



 '''simple docstring'''


 __SCREAMING_SNAKE_CASE												=				42







class     lowercase						(       _UpperCamelCase ,				_UpperCamelCase					):



 '''simple docstring'''


 @register_to_config
 def __init__(self       ,       __a = 16       ,       __a = 88       ,       __a = None       ,       __a = None       ,       __a = 1       ,       __a = 0.0       ,       __a = 32       ,       __a = None       ,       __a = False       ,       __a = None       ,       __a = None       ,       __a = None       ,       __a = "geglu"       ,       __a = None       ,       __a = False       ,       __a = False       ,       __a = False       ,       __a = "layer_norm"       ,       __a = True       ,       ) ->					List[str]:






  """simple docstring"""





  super().__init__()
  UpperCAmelCase__							=				use_linear_projection
  UpperCAmelCase__							=				num_attention_heads
  UpperCAmelCase__							=				attention_head_dim
  UpperCAmelCase__							=				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__							=				(in_channels is not None) and (patch_size is None)
  UpperCAmelCase__							=				num_vector_embeds is not None
  UpperCAmelCase__							=				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__							=				(
       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'       ,       __a       ,       standard_warn=__a )
   UpperCAmelCase__							=				'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__							=				in_channels

   UpperCAmelCase__							=				torch.nn.GroupNorm(num_groups=__a       ,       num_channels=__a       ,       eps=1E-6       ,       affine=__a )
   if use_linear_projection:
    UpperCAmelCase__							=				nn.Linear(__a       ,       __a )
   else:
    UpperCAmelCase__							=				nn.Convad(__a       ,       __a       ,       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__							=				sample_size
   UpperCAmelCase__							=				sample_size
   UpperCAmelCase__							=				num_vector_embeds
   UpperCAmelCase__							=				self.height * self.width

   UpperCAmelCase__							=				ImagePositionalEmbeddings(
       num_embed=__a       ,       embed_dim=__a       ,       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__							=				sample_size
   UpperCAmelCase__							=				sample_size

   UpperCAmelCase__							=				patch_size
   UpperCAmelCase__							=				PatchEmbed(
       height=__a       ,       width=__a       ,       patch_size=__a       ,       in_channels=__a       ,       embed_dim=__a       ,       )

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

  # 4. Define output layers
  UpperCAmelCase__							=				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__							=				nn.Linear(__a       ,       __a )
   else:
    UpperCAmelCase__							=				nn.Convad(__a       ,       __a       ,       kernel_size=1       ,       stride=1       ,       padding=0 )
  elif self.is_input_vectorized:
   UpperCAmelCase__							=				nn.LayerNorm(__a )
   UpperCAmelCase__							=				nn.Linear(__a       ,       self.num_vector_embeds - 1 )
  elif self.is_input_patches:
   UpperCAmelCase__							=				nn.LayerNorm(__a       ,       elementwise_affine=__a       ,       eps=1E-6 )
   UpperCAmelCase__							=				nn.Linear(__a       ,       2 * inner_dim )
   UpperCAmelCase__							=				nn.Linear(__a       ,       patch_size * patch_size * self.out_channels )




 def   UpperCamelCase__							(self       ,       __a       ,       __a = None       ,       __a = None       ,       __a = None       ,       __a = None       ,       __a = None       ,       __a = None       ,       __a = 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__							=				(1 - attention_mask.to(hidden_states.dtype )) * -1_00_00.0
   UpperCAmelCase__							=				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__							=				(1 - encoder_attention_mask.to(hidden_states.dtype )) * -1_00_00.0
   UpperCAmelCase__							=				encoder_attention_mask.unsqueeze(1 )

  # 1. Input
  if self.is_input_continuous:
   UpperCAmelCase__				,    UpperCAmelCase__				,    UpperCAmelCase__				,    UpperCAmelCase__							=				hidden_states.shape
   UpperCAmelCase__							=				hidden_states

   UpperCAmelCase__							=				self.norm(__a )
   if not self.use_linear_projection:
    UpperCAmelCase__							=				self.proj_in(__a )
    UpperCAmelCase__							=				hidden_states.shape[1]
    UpperCAmelCase__							=				hidden_states.permute(0       ,       2       ,       3       ,       1 ).reshape(__a       ,       height * width       ,       __a )
   else:
    UpperCAmelCase__							=				hidden_states.shape[1]
    UpperCAmelCase__							=				hidden_states.permute(0       ,       2       ,       3       ,       1 ).reshape(__a       ,       height * width       ,       __a )
    UpperCAmelCase__							=				self.proj_in(__a )
  elif self.is_input_vectorized:
   UpperCAmelCase__							=				self.latent_image_embedding(__a )
  elif self.is_input_patches:
   UpperCAmelCase__							=				self.pos_embed(__a )

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

  # 3. Output
  if self.is_input_continuous:
   if not self.use_linear_projection:
    UpperCAmelCase__							=				hidden_states.reshape(__a       ,       __a       ,       __a       ,       __a ).permute(0       ,       3       ,       1       ,       2 ).contiguous()
    UpperCAmelCase__							=				self.proj_out(__a )
   else:
    UpperCAmelCase__							=				self.proj_out(__a )
    UpperCAmelCase__							=				hidden_states.reshape(__a       ,       __a       ,       __a       ,       __a ).permute(0       ,       3       ,       1       ,       2 ).contiguous()

   UpperCAmelCase__							=				hidden_states + residual
  elif self.is_input_vectorized:
   UpperCAmelCase__							=				self.norm_out(__a )
   UpperCAmelCase__							=				self.out(__a )
   # (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
   UpperCAmelCase__							=				logits.permute(0       ,       2       ,       1 )

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

   # unpatchify
   UpperCAmelCase__							=				UpperCAmelCase__							=				int(hidden_states.shape[1] ** 0.5 )
   UpperCAmelCase__							=				hidden_states.reshape(
       shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels) )
   UpperCAmelCase__							=				torch.einsum('nhwpqc->nchpwq'       ,       __a )
   UpperCAmelCase__							=				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=__a )






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			       =			namedtuple('''ExampleNamedTuple''', '''a b c''')



class     lowercase						(       unittest.TestCase					):



 '''simple docstring'''


 def   UpperCamelCase__							(self ) ->					Tuple:






  """simple docstring"""





  UpperCAmelCase__							=				torch.randn(5       ,       2 )
  UpperCAmelCase__							=				torch.device('cuda' ) if torch.cuda.is_available() else torch.device('cpu' )

  UpperCAmelCase__							=				send_to_device(__a       ,       __a )
  self.assertTrue(torch.equal(resulta.cpu()       ,       __a ) )

  UpperCAmelCase__							=				send_to_device((tensor, [tensor, tensor], 1)       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta[0].cpu()       ,       __a ) )
  self.assertIsInstance(resulta[1]       ,       __a )
  self.assertTrue(torch.equal(resulta[1][0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta[1][1].cpu()       ,       __a ) )
  self.assertEqual(resulta[2]       ,       1 )

  UpperCAmelCase__							=				send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta['a'].cpu()       ,       __a ) )
  self.assertIsInstance(resulta['b']       ,       __a )
  self.assertTrue(torch.equal(resulta['b'][0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta['b'][1].cpu()       ,       __a ) )
  self.assertEqual(resulta['c']       ,       1 )

  UpperCAmelCase__							=				send_to_device(ExampleNamedTuple(a=__a       ,       b=[tensor, tensor]       ,       c=1 )       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta.a.cpu()       ,       __a ) )
  self.assertIsInstance(resulta.b       ,       __a )
  self.assertTrue(torch.equal(resulta.b[0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta.b[1].cpu()       ,       __a ) )
  self.assertEqual(resulta.c       ,       1 )

  UpperCAmelCase__							=				send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1} )       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta['a'].cpu()       ,       __a ) )
  self.assertIsInstance(resulta['b']       ,       __a )
  self.assertTrue(torch.equal(resulta['b'][0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta['b'][1].cpu()       ,       __a ) )
  self.assertEqual(resulta['c']       ,       1 )




 def   UpperCamelCase__							(self ) ->					List[str]:






  """simple docstring"""





  with self.assertRaises(__a ) as cm:
   UpperCAmelCase__							=				recursively_apply(torch.tensor       ,       (torch.tensor(1 ), 1)       ,       error_on_other_type=__a )
  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   UpperCamelCase__							(self ) ->					List[str]:






  """simple docstring"""





  UpperCAmelCase__							=				torch.tensor([1, 2, 3, 4, 5] )
  self.assertEqual(listify(__a )       ,       [1, 2, 3, 4, 5] )

  UpperCAmelCase__							=				torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]] )
  self.assertEqual(listify(__a )       ,       [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]] )

  UpperCAmelCase__							=				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(__a )       ,       [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]] )




 def   UpperCamelCase__							(self ) ->					int:






  """simple docstring"""





  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   UpperCamelCase__							(self ) ->					Dict:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				model.forward
  UpperCAmelCase__							=				convert_outputs_to_fpaa(model.forward )
  UpperCAmelCase__							=				extract_model_from_parallel(__a       ,       keep_fpaa_wrapper=__a )
  UpperCAmelCase__							=				pickle.dumps(__a )




 @require_cuda
 def   UpperCamelCase__							(self ) ->					Tuple:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				model.forward
  UpperCAmelCase__							=				torch.cuda.amp.autocast(dtype=torch.floataa )(model.forward )
  UpperCAmelCase__							=				convert_outputs_to_fpaa(model.forward )
  UpperCAmelCase__							=				extract_model_from_parallel(__a       ,       keep_fpaa_wrapper=__a )
  UpperCAmelCase__							=				pickle.dumps(__a )




 @require_cuda
 @require_torch_min_version(version='2.0' )
 def   UpperCamelCase__							(self ) ->					List[str]:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				model.forward
  UpperCAmelCase__							=				torch.cuda.amp.autocast(dtype=torch.floataa )(model.forward )
  UpperCAmelCase__							=				convert_outputs_to_fpaa(model.forward )
  UpperCAmelCase__							=				torch.compile(model.forward       ,       backend='inductor' )
  UpperCAmelCase__							=				torch.randn(4       ,       10 ).cuda()
  UpperCAmelCase__							=				model(__a )




 def   UpperCamelCase__							(self ) ->					Optional[Any]:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
  UpperCAmelCase__							=				torch.nn.parallel.DataParallel(__a )
  UpperCAmelCase__							=				extract_model_from_parallel(__a )

  self.assertEqual(__a       ,       __a )




 @require_torch_min_version(version='2.0' )
 def   UpperCamelCase__							(self ) ->					Optional[Any]:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				torch.compile(__a )

  # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
  UpperCAmelCase__							=				torch.nn.parallel.DataParallel(__a )
  UpperCAmelCase__							=				torch.compile(__a )
  UpperCAmelCase__							=				extract_model_from_parallel(__a )

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




 def   UpperCamelCase__							(self ) ->					str:






  """simple docstring"""





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



