






"""simple docstring"""


from __future__ import annotations

from abc import abstractmethod
from collections.abc import Iterable
from typing import Generic, Protocol, TypeVar


class    snake_case						(				_lowerCAmelCase		):


   '''simple docstring'''

   @abstractmethod
   def __lt__(      self					:	T,			_lowerCamelCase					:	T    ):

     '''simple docstring'''
     pass


   @abstractmethod
   def __gt__(      self					:	T,			_lowerCamelCase					:	T    ):

     '''simple docstring'''
     pass






   @abstractmethod
   def __eq__(      self					:	T,			_lowerCamelCase					:	object    ):

     '''simple docstring'''
     pass





lowercase_								=	TypeVar('T', bound=Comparable)


class    snake_case						(				Generic[T]		):


   '''simple docstring'''

   def __init__(      self					:	Union[str, Any]    ):

     '''simple docstring'''
     __A		     =					[]
     __A		     =					0


   def __repr__(      self					:	Dict    ):

     '''simple docstring'''
     return str(self.h    )


   def 							_SCREAMING_SNAKE_CASE							(      self					:	List[str],			_lowerCamelCase					:	int    ):

     '''simple docstring'''
     if child_idx > 0:
       return (child_idx - 1) // 2
     return None


   def 							_SCREAMING_SNAKE_CASE							(      self					:	Union[str, Any],			_lowerCamelCase					:	int    ):

     '''simple docstring'''
     __A		     =					2 * parent_idx + 1
     if left_child_index < self.heap_size:
       return left_child_index
     return None


   def 							_SCREAMING_SNAKE_CASE							(      self					:	Tuple,			_lowerCamelCase					:	int    ):

     '''simple docstring'''
     __A		     =					2 * parent_idx + 2
     if right_child_index < self.heap_size:
       return right_child_index
     return None


   def 							_SCREAMING_SNAKE_CASE							(      self					:	Tuple,			_lowerCamelCase					:	int    ):

     '''simple docstring'''
     if index < self.heap_size:
       __A		     =					index
       __A		     =					self.left_child_idx(_lowerCamelCase    )
       __A		     =					self.right_child_idx(_lowerCamelCase    )
       # check which child is larger than its parent
       if left_child is not None and self.h[left_child] > self.h[violation]:
         __A		     =					left_child
       if right_child is not None and self.h[right_child] > self.h[violation]:
         __A		     =					right_child
       # if violation indeed exists
       if violation != index:
         # swap to fix the violation
         __A     ,       __A		     =					self.h[index], self.h[violation]
         # fix the subsequent violation recursively if any
         self.max_heapify(_lowerCamelCase    )


   def 							_SCREAMING_SNAKE_CASE							(      self					:	Dict,			_lowerCamelCase					:	Iterable[T]    ):

     '''simple docstring'''
     __A		     =					list(_lowerCamelCase    )
     __A		     =					len(self.h    )
     if self.heap_size > 1:
       # max_heapify from right to left but exclude leaves (last level)
       for i in range(self.heap_size // 2 - 1,			-1,			-1    ):
         self.max_heapify(_lowerCamelCase    )


   def 							_SCREAMING_SNAKE_CASE							(      self					:	Any    ):

     '''simple docstring'''
     if self.heap_size >= 2:
       __A		     =					self.h[0]
       __A		     =					self.h.pop(-1    )
       self.heap_size -= 1
       self.max_heapify(0    )
       return me
     elif self.heap_size == 1:
       self.heap_size -= 1
       return self.h.pop(-1    )
     else:
       raise Exception('''Empty heap'''    )


   def 							_SCREAMING_SNAKE_CASE							(      self					:	List[Any],			_lowerCamelCase					:	T    ):

     '''simple docstring'''
     self.h.append(_lowerCamelCase    )
     __A		     =					(self.heap_size - 1) // 2
     self.heap_size += 1
     while idx >= 0:
       self.max_heapify(_lowerCamelCase    )
       __A		     =					(idx - 1) // 2






   def 							_SCREAMING_SNAKE_CASE							(      self					:	str    ):

     '''simple docstring'''
     __A		     =					self.heap_size
     for j in range(size - 1,			0,			-1    ):
       __A     ,       __A		     =					self.h[j], self.h[0]
       self.heap_size -= 1
       self.max_heapify(0    )
     __A		     =					size


if __name__ == "__main__":
       import doctest

       # run doc test
       doctest.testmod()

       # demo
       for unsorted in [
           [0],
           [2],
           [3, 5],
           [5, 3],
           [5, 5],
           [0, 0, 0, 0],
           [1, 1, 1, 1],
           [2, 2, 3, 5],
           [0, 2, 2, 3, 5],
           [2, 5, 3, 0, 2, 3, 0, 3],
           [6, 1, 2, 7, 9, 3, 4, 5, 10, 8],
           [103, 9, 1, 7, 11, 15, 25, 201, 209, 107, 5],
           [-45, -2, -5],
       ]:
              print(F'''unsorted array: {unsorted}''')

              lowercase_								=	Heap()
              heap.build_max_heap(unsorted)
              print(F'''after build heap: {heap}''')

              print(F'''max value: {heap.extract_max()}''')
              print(F'''after max value removed: {heap}''')

              heap.insert(100)
              print(F'''after new value 100 inserted: {heap}''')

              heap.heap_sort()
              print(F'''heap-sorted array: {heap}\n''')





"""simple docstring"""







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      (__lowerCamelCase  ):
				_snake_case  =	42



class 			_lowerCamelCase      (__lowerCamelCase    ,				__lowerCamelCase  ):







				@register_to_config
				def __init__(			self     :							Dict      ,       lowerCamelCase_     :							int = 1_6      ,       lowerCamelCase_     :							int = 8_8      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							int = 1      ,       lowerCamelCase_     :							float = 0.0      ,       lowerCamelCase_     :							int = 3_2      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							bool = False      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							str = "geglu"      ,       lowerCamelCase_     :							Optional[int] = None      ,       lowerCamelCase_     :							bool = False      ,       lowerCamelCase_     :							bool = False      ,       lowerCamelCase_     :							bool = False      ,       lowerCamelCase_     :							str = "layer_norm"      ,       lowerCamelCase_     :							bool = True      ,       ):

					"""simple docstring"""






					super().__init__()
					_lowercase			:     Tuple      					=				use_linear_projection
					_lowercase			:     List[Any]      					=				num_attention_heads
					_lowercase			:     Optional[int]      					=				attention_head_dim
					_lowercase			:     Union[str, 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
					_lowercase			:     Dict      					=				(in_channels is not None) and (patch_size is None)
					_lowercase			:     Any      					=				num_vector_embeds is not None
					_lowercase			:     List[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:
						_lowercase			:     Tuple      					=				(
						    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'      ,       lowerCamelCase_      ,       standard_warn=lowerCamelCase_    )
						_lowercase			:     Dict      					=				'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:
						_lowercase			:     int      					=				in_channels

						_lowercase			:     List[Any]      					=				torch.nn.GroupNorm(num_groups=lowerCamelCase_      ,       num_channels=lowerCamelCase_      ,       eps=1E-6      ,       affine=lowerCamelCase_    )
						if use_linear_projection:
							_lowercase			:     Any      					=				nn.Linear(lowerCamelCase_      ,       lowerCamelCase_    )
						else:
							_lowercase			:     List[str]      					=				nn.Convad(lowerCamelCase_      ,       lowerCamelCase_      ,       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"

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

						_lowercase			:     Optional[int]      					=				ImagePositionalEmbeddings(
						    num_embed=lowerCamelCase_      ,       embed_dim=lowerCamelCase_      ,       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"

						_lowercase			:     Optional[Any]      					=				sample_size
						_lowercase			:     Optional[int]      					=				sample_size

						_lowercase			:     str      					=				patch_size
						_lowercase			:     Union[str, Any]      					=				PatchEmbed(
						    height=lowerCamelCase_      ,       width=lowerCamelCase_      ,       patch_size=lowerCamelCase_      ,       in_channels=lowerCamelCase_      ,       embed_dim=lowerCamelCase_      ,       )

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

					# 4. Define output layers
					_lowercase			:     Dict      					=				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:
							_lowercase			:     str      					=				nn.Linear(lowerCamelCase_      ,       lowerCamelCase_    )
						else:
							_lowercase			:     List[Any]      					=				nn.Convad(lowerCamelCase_      ,       lowerCamelCase_      ,       kernel_size=1      ,       stride=1      ,       padding=0    )
					elif self.is_input_vectorized:
						_lowercase			:     Dict      					=				nn.LayerNorm(lowerCamelCase_    )
						_lowercase			:     Union[str, Any]      					=				nn.Linear(lowerCamelCase_      ,       self.num_vector_embeds - 1    )
					elif self.is_input_patches:
						_lowercase			:     Dict      					=				nn.LayerNorm(lowerCamelCase_      ,       elementwise_affine=lowerCamelCase_      ,       eps=1E-6    )
						_lowercase			:     Union[str, Any]      					=				nn.Linear(lowerCamelCase_      ,       2 * inner_dim    )
						_lowercase			:     List[Any]      					=				nn.Linear(lowerCamelCase_      ,       patch_size * patch_size * self.out_channels    )







				def 				__UpperCAmelCase       (			self     :							Optional[Any]      ,       lowerCamelCase_     :							torch.Tensor      ,       lowerCamelCase_     :							Optional[torch.Tensor] = None      ,       lowerCamelCase_     :							Optional[torch.LongTensor] = None      ,       lowerCamelCase_     :							Optional[torch.LongTensor] = None      ,       lowerCamelCase_     :							Dict[str, Any] = None      ,       lowerCamelCase_     :							Optional[torch.Tensor] = None      ,       lowerCamelCase_     :							Optional[torch.Tensor] = None      ,       lowerCamelCase_     :							bool = True      ,       ):

					"""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)
						_lowercase			:     Optional[Any]      					=				(1 - attention_mask.to(hidden_states.dtype    )) * -1_0000.0
						_lowercase			:     Optional[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:
						_lowercase			:     List[Any]      					=				(1 - encoder_attention_mask.to(hidden_states.dtype    )) * -1_0000.0
						_lowercase			:     Any      					=				encoder_attention_mask.unsqueeze(1    )

					# 1. Input
					if self.is_input_continuous:
						_lowercase					,   _lowercase					,   _lowercase					,   _lowercase			:     List[Any]      					=				hidden_states.shape
						_lowercase			:     int      					=				hidden_states

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

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

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

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

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

						# unpatchify
						_lowercase			:     Union[str, Any]      					=				int(hidden_states.shape[1] ** 0.5    )
						_lowercase			:     int      					=				hidden_states.reshape(
						    shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels)    )
						_lowercase			:     str      					=				torch.einsum('nhwpqc->nchpwq'      ,       lowerCamelCase_    )
						_lowercase			:     List[str]      					=				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=lowerCamelCase_    )



