

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      _a				( snake_case_							):




      """simple docstring"""

      _lowerCamelCase					: torch.FloatTensor




class      _a				( snake_case_    ,							snake_case_							):




      """simple docstring"""

      @register_to_config
      def __init__(      self       :							Tuple	, UpperCAmelCase       :							int = 16	, UpperCAmelCase       :							int = 88	, UpperCAmelCase       :							Optional[int] = None	, UpperCAmelCase       :							Optional[int] = None	, UpperCAmelCase       :							int = 1	, UpperCAmelCase       :							float = 0.0	, UpperCAmelCase       :							int = 32	, UpperCAmelCase       :							Optional[int] = None	, UpperCAmelCase       :							bool = False	, UpperCAmelCase       :							Optional[int] = None	, UpperCAmelCase       :							Optional[int] = None	, UpperCAmelCase       :							Optional[int] = None	, UpperCAmelCase       :							str = "geglu"	, UpperCAmelCase       :							Optional[int] = None	, UpperCAmelCase       :							bool = False	, UpperCAmelCase       :							bool = False	, UpperCAmelCase       :							bool = False	, UpperCAmelCase       :							str = "layer_norm"	, UpperCAmelCase       :							bool = True	, ):
            super().__init__()
            A_	     =							use_linear_projection
            A_	     =							num_attention_heads
            A_	     =							attention_head_dim
            A_	     =							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
            A_	     =							(in_channels is not None) and (patch_size is None)
            A_	     =							num_vector_embeds is not None
            A_	     =							in_channels is not None and patch_size is not None

            if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
                  A_	     =							(
                      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  )
                  A_	     =							"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:
                  A_	     =							in_channels

                  A_	     =							torch.nn.GroupNorm(num_groups=UpperCAmelCase	, num_channels=UpperCAmelCase	, eps=1E-6	, affine=UpperCAmelCase  )
                  if use_linear_projection:
                        A_	     =							nn.Linear(UpperCAmelCase	, UpperCAmelCase  )
                  else:
                        A_	     =							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"

                  A_	     =							sample_size
                  A_	     =							sample_size
                  A_	     =							num_vector_embeds
                  A_	     =							self.height * self.width

                  A_	     =							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"

                  A_	     =							sample_size
                  A_	     =							sample_size

                  A_	     =							patch_size
                  A_	     =							PatchEmbed(
                      height=UpperCAmelCase	, width=UpperCAmelCase	, patch_size=UpperCAmelCase	, in_channels=UpperCAmelCase	, embed_dim=UpperCAmelCase	, )

            # 3. Define transformers blocks
            A_	     =							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
            A_	     =							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:
                        A_	     =							nn.Linear(UpperCAmelCase	, UpperCAmelCase  )
                  else:
                        A_	     =							nn.Convad(UpperCAmelCase	, UpperCAmelCase	, kernel_size=1	, stride=1	, padding=0  )
            elif self.is_input_vectorized:
                  A_	     =							nn.LayerNorm(UpperCAmelCase  )
                  A_	     =							nn.Linear(UpperCAmelCase	, self.num_vector_embeds - 1  )
            elif self.is_input_patches:
                  A_	     =							nn.LayerNorm(UpperCAmelCase	, elementwise_affine=UpperCAmelCase	, eps=1E-6  )
                  A_	     =							nn.Linear(UpperCAmelCase	, 2 * inner_dim  )
                  A_	     =							nn.Linear(UpperCAmelCase	, patch_size * patch_size * self.out_channels  )







      def 		__A  (      self       :							str	, UpperCAmelCase       :							torch.Tensor	, UpperCAmelCase       :							Optional[torch.Tensor] = None	, UpperCAmelCase       :							Optional[torch.LongTensor] = None	, UpperCAmelCase       :							Optional[torch.LongTensor] = None	, UpperCAmelCase       :							Dict[str, Any] = None	, UpperCAmelCase       :							Optional[torch.Tensor] = None	, UpperCAmelCase       :							Optional[torch.Tensor] = None	, UpperCAmelCase       :							bool = True	, ):
            # ensure attention_mask is a bias, and give it a singleton query_tokens dimension.
            #   we may have done this conversion already, e.g. if we came here via UNet2DConditionModel#forward.
            #   we can tell by counting dims; if ndim == 2: it's a mask rather than a bias.
            # expects mask of shape:
            #   [batch, key_tokens]
            # adds singleton query_tokens dimension:
            #   [batch,                    1, key_tokens]
            # this helps to broadcast it as a bias over attention scores, which will be in one of the following shapes:
            #   [batch,  heads, query_tokens, key_tokens] (e.g. torch sdp attn)
            #   [batch * heads, query_tokens, key_tokens] (e.g. xformers or classic attn)
            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)
                  A_	     =							(1 - attention_mask.to(hidden_states.dtype  )) * -10_000.0
                  A_	     =							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:
                  A_	     =							(1 - encoder_attention_mask.to(hidden_states.dtype  )) * -10_000.0
                  A_	     =							encoder_attention_mask.unsqueeze(1  )

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

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

            # 2. Blocks
            for block in self.transformer_blocks:
                  A_	     =							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:
                        A_	     =							hidden_states.reshape(UpperCAmelCase	, UpperCAmelCase	, UpperCAmelCase	, UpperCAmelCase  ).permute(0	, 3	, 1	, 2  ).contiguous()
                        A_	     =							self.proj_out(UpperCAmelCase  )
                  else:
                        A_	     =							self.proj_out(UpperCAmelCase  )
                        A_	     =							hidden_states.reshape(UpperCAmelCase	, UpperCAmelCase	, UpperCAmelCase	, UpperCAmelCase  ).permute(0	, 3	, 1	, 2  ).contiguous()

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

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

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


import json
import os
import tempfile

import datasets
from datasets.arrow_writer import ArrowWriter
from datasets.features import ArrayaD
from utils import generate_examples, get_duration


__a						:Dict            = (30, 487)
__a						:int            = (36, 1024)
__a						:Tuple            = (100, 100)
__a						:Dict            = 100

__a						:List[str]            = datasets.Features(
    {'text': ArrayaD(SHAPE_TEST_1, dtype='float32'), 'image': ArrayaD(SHAPE_TEST_2, dtype='float32')}
)

__a      ,			__a						:str            = os.path.split(__file__)
__a						:Optional[Any]            = os.path.join(RESULTS_BASEPATH, 'results', RESULTS_FILENAME.replace('.py', '.json'))

@get_duration
def 			__snake_case	(   __UpperCamelCase  :   Optional[int]							,__UpperCamelCase  :   Union[str, Any]							,__UpperCamelCase  :   Dict						):



      """simple docstring"""

      with ArrowWriter(features=__UpperCamelCase							,path=os.path.join(__UpperCamelCase							,"beta.arrow"						)						) as writer:
            for key, record in dummy_data:
                  A_	     =							my_features.encode_example(__UpperCamelCase						)
                  writer.write(__UpperCamelCase						)
            A_     ,    A_	     =							writer.finalize()

@get_duration
def 			__snake_case	(   __UpperCamelCase  :   int							,__UpperCamelCase  :   Any						):



      """simple docstring"""

      A_	     =							datasets.Dataset.from_file(
          filename=os.path.join(__UpperCamelCase							,"beta.arrow"						)							,info=datasets.DatasetInfo(features=__UpperCamelCase						)						)
      for _ in dataset:
            pass

@get_duration
def 			__snake_case	(   __UpperCamelCase  :   Optional[Any]							,__UpperCamelCase  :   List[str]						):



      """simple docstring"""

      A_	     =							datasets.Dataset.from_file(
          filename=os.path.join(__UpperCamelCase							,"beta.arrow"						)							,info=datasets.DatasetInfo(features=__UpperCamelCase						)						)
      dataset.set_format("numpy"						)
      for _ in dataset:
            pass

@get_duration
def 			__snake_case	(   __UpperCamelCase  :   List[str]							,__UpperCamelCase  :   Tuple						):



      """simple docstring"""

      A_	     =							10
      A_	     =							datasets.Dataset.from_file(
          filename=os.path.join(__UpperCamelCase							,"beta.arrow"						)							,info=datasets.DatasetInfo(features=__UpperCamelCase						)						)
      for i in range(0							,len(__UpperCamelCase						)							,__UpperCamelCase						):
            A_	     =							dataset[i : i + batch_size]

@get_duration
def 			__snake_case	(   __UpperCamelCase  :   int							,__UpperCamelCase  :   List[Any]						):



      """simple docstring"""

      A_	     =							10
      A_	     =							datasets.Dataset.from_file(
          filename=os.path.join(__UpperCamelCase							,"beta.arrow"						)							,info=datasets.DatasetInfo(features=__UpperCamelCase						)						)
      dataset.set_format("numpy"						)
      for i in range(0							,len(__UpperCamelCase						)							,__UpperCamelCase						):
            A_	     =							dataset[i : i + batch_size]

@get_duration
def 			__snake_case	(   __UpperCamelCase  :   int							,__UpperCamelCase  :   Tuple						):



      """simple docstring"""

      A_	     =							datasets.Dataset.from_file(
          filename=os.path.join(__UpperCamelCase							,"beta.arrow"						)							,info=datasets.DatasetInfo(features=__UpperCamelCase						)						)
      for col in feats:
            A_	     =							dataset[col]

@get_duration
def 			__snake_case	(   __UpperCamelCase  :   Optional[int]							,__UpperCamelCase  :   Optional[Any]						):



      """simple docstring"""

      A_	     =							datasets.Dataset.from_file(
          filename=os.path.join(__UpperCamelCase							,"beta.arrow"						)							,info=datasets.DatasetInfo(features=__UpperCamelCase						)						)
      dataset.set_format("numpy"						)
      for col in feats:
            A_	     =							dataset[col]

def 			__snake_case	(   ):



      """simple docstring"""

      A_	     =							{}
      A_	     =							(
          read_unformated,
          read_formatted_as_numpy,
          read_batch_unformated,
          read_batch_formatted_as_numpy,
          read_col_unformated,
          read_col_formatted_as_numpy,
      )
      with tempfile.TemporaryDirectory() as tmp_dir:
            A_	     =							datasets.Features({"image": ArrayaD(__UpperCamelCase							,dtype="float32"						)}						)
            A_	     =							generate_examples(features=__UpperCamelCase							,num_examples=__UpperCamelCase						)
            A_	     =							write(__UpperCamelCase							,__UpperCamelCase							,__UpperCamelCase						)
            for read_func in read_functions:
                  A_	     =							read_func(__UpperCamelCase							,__UpperCamelCase						)

      with tempfile.TemporaryDirectory() as tmp_dir:
            # don't use fixed length for fair comparison
            # feats = datasets.Features(
            #     {"image": datasets.Sequence(datasets.Sequence(datasets.Value("float32"), SPEED_TEST_SHAPE[1]), SPEED_TEST_SHAPE[0])}
            # )
            A_	     =							datasets.Features({"image": datasets.Sequence(datasets.Sequence(datasets.Value("float32"						)						)						)}						)
            A_	     =							generate_examples(
                features=__UpperCamelCase							,num_examples=__UpperCamelCase							,seq_shapes={"image": SPEED_TEST_SHAPE}						)
            A_	     =							write(__UpperCamelCase							,__UpperCamelCase							,__UpperCamelCase						)
            for read_func in read_functions:
                  A_	     =							read_func(__UpperCamelCase							,__UpperCamelCase						)

      with tempfile.TemporaryDirectory() as tmp_dir:
            # don't use fixed length for fair comparison
            # feats = datasets.Features(
            #     {"image": datasets.Sequence(datasets.Value("float32"), SPEED_TEST_SHAPE[0] * SPEED_TEST_SHAPE[1])}
            # )
            A_	     =							datasets.Features({"image": datasets.Sequence(datasets.Value("float32"						)						)}						)
            A_	     =							generate_examples(
                features=__UpperCamelCase							,num_examples=__UpperCamelCase							,seq_shapes={"image": [SPEED_TEST_SHAPE[0] * SPEED_TEST_SHAPE[1]]}							,)
            A_	     =							write(__UpperCamelCase							,__UpperCamelCase							,__UpperCamelCase						)
            for read_func in read_functions:
                  A_	     =							read_func(__UpperCamelCase							,__UpperCamelCase						)

      with open(__UpperCamelCase							,"wb"						) as f:
            f.write(json.dumps(__UpperCamelCase						).encode("utf-8"						)						)


if __name__ == "__main__":  # useful to run the profiler
      benchmark_array_xd()