





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


	"""simple docstring"""





	__magic_name__							   =		42



class  lowerCamelCase__				(						snake_case_  ,							snake_case_						):


	"""simple docstring"""





	@register_to_config
	def __init__(					self   ,     UpperCAmelCase__ = 1_6   ,     UpperCAmelCase__ = 8_8   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = 1   ,     UpperCAmelCase__ = 0.0   ,     UpperCAmelCase__ = 3_2   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = False   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = "geglu"   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = False   ,     UpperCAmelCase__ = False   ,     UpperCAmelCase__ = False   ,     UpperCAmelCase__ = "layer_norm"   ,     UpperCAmelCase__ = True   ,     )       ->				Union[str, Any]:
					super().__init__()
					_A :				str				=  use_linear_projection
					_A :				str				=  num_attention_heads
					_A :				Optional[Any]				=  attention_head_dim
					_A :				int				=  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 :				Dict				=  (in_channels is not None) and (patch_size is None)
					_A :				Union[str, Any]				=  num_vector_embeds is not None
					_A :				str				=  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 :				Dict				=  (
									    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 :				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:
									_A :				str				=  in_channels

									_A :				Any				=  torch.nn.GroupNorm(num_groups=UpperCAmelCase__   ,     num_channels=UpperCAmelCase__   ,     eps=1e-6   ,     affine=UpperCAmelCase__			)
									if use_linear_projection:
													_A :				Dict				=  nn.Linear(UpperCAmelCase__   ,     UpperCAmelCase__			)
									else:
													_A :				int				=  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 :				str				=  sample_size
									_A :				Any				=  sample_size
									_A :				str				=  num_vector_embeds
									_A :				Tuple				=  self.height * self.width

									_A :				Dict				=  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 :				Optional[int]				=  sample_size
									_A :				Dict				=  sample_size

									_A :				Dict				=  patch_size
									_A :				str				=  PatchEmbed(
									    height=UpperCAmelCase__   ,     width=UpperCAmelCase__   ,     patch_size=UpperCAmelCase__   ,     in_channels=UpperCAmelCase__   ,     embed_dim=UpperCAmelCase__   ,     )

					# 3. Define transformers blocks
					_A :				Optional[Any]				=  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 :				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:
													_A :				List[str]				=  nn.Linear(UpperCAmelCase__   ,     UpperCAmelCase__			)
									else:
													_A :				Dict				=  nn.Convad(UpperCAmelCase__   ,     UpperCAmelCase__   ,     kernel_size=1   ,     stride=1   ,     padding=0			)
					elif self.is_input_vectorized:
									_A :				Optional[Any]				=  nn.LayerNorm(UpperCAmelCase__			)
									_A :				Any				=  nn.Linear(UpperCAmelCase__   ,     self.num_vector_embeds - 1			)
					elif self.is_input_patches:
									_A :				Optional[int]				=  nn.LayerNorm(UpperCAmelCase__   ,     elementwise_affine=UpperCAmelCase__   ,     eps=1e-6			)
									_A :				Optional[int]				=  nn.Linear(UpperCAmelCase__   ,     2 * inner_dim			)
									_A :				Optional[Any]				=  nn.Linear(UpperCAmelCase__   ,     patch_size * patch_size * self.out_channels			)




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = None   ,     UpperCAmelCase__ = True   ,     )       ->				Tuple:
					# 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 :				Any				=  (1 - attention_mask.to(hidden_states.dtype			)) * -1_0_0_0_0.0
									_A :				str				=  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 :				int				=  (1 - encoder_attention_mask.to(hidden_states.dtype			)) * -1_0_0_0_0.0
									_A :				List[str]				=  encoder_attention_mask.unsqueeze(1			)

					# 1. Input
					if self.is_input_continuous:
									_A	,       _A	,       _A	,       _A :				Optional[Any]				=  hidden_states.shape
									_A :				Any				=  hidden_states

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

					# 2. Blocks
					for block in self.transformer_blocks:
									_A :				Union[str, 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:
													_A :				Any				=  hidden_states.reshape(UpperCAmelCase__   ,     UpperCAmelCase__   ,     UpperCAmelCase__   ,     UpperCAmelCase__			).permute(0   ,     3   ,     1   ,     2			).contiguous()
													_A :				Optional[Any]				=  self.proj_out(UpperCAmelCase__			)
									else:
													_A :				int				=  self.proj_out(UpperCAmelCase__			)
													_A :				Union[str, Any]				=  hidden_states.reshape(UpperCAmelCase__   ,     UpperCAmelCase__   ,     UpperCAmelCase__   ,     UpperCAmelCase__			).permute(0   ,     3   ,     1   ,     2			).contiguous()

									_A :				Optional[Any]				=  hidden_states + residual
					elif self.is_input_vectorized:
									_A :				Union[str, Any]				=  self.norm_out(UpperCAmelCase__			)
									_A :				Tuple				=  self.out(UpperCAmelCase__			)
									# (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
									_A :				str				=  logits.permute(0   ,     2   ,     1			)

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

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






from .hash_table import HashTable
from .number_theory.prime_numbers import is_prime, next_prime



class  lowerCamelCase__				(						snake_case_						):


	"""simple docstring"""





	def __init__(					self   ,     *UpperCAmelCase__   ,     **UpperCAmelCase__			)       ->				Optional[int]:
					super().__init__(*UpperCAmelCase__   ,     **UpperCAmelCase__			)




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__   ,     UpperCAmelCase__			)       ->				Optional[int]:
					_A :				Union[str, Any]				=  (
					    next_prime(value % self.size_table			)
					    if not is_prime(value % self.size_table			)
					    else value % self.size_table
					)  # gt = bigger than
					return next_prime_gt - (data % next_prime_gt)




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__   ,     UpperCAmelCase__   ,     UpperCAmelCase__			)       ->				Union[str, Any]:
					return (increment * self.__hash_function_a(UpperCAmelCase__   ,     UpperCAmelCase__			)) % self.size_table




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__   ,     UpperCAmelCase__=None			)       ->				List[Any]:
					_A :				Optional[int]				=  1
					_A :				Optional[int]				=  self.hash_function(UpperCAmelCase__			)

					while self.values[new_key] is not None and self.values[new_key] != key:
									_A :				int				=  (
									    self.__hash_double_function(UpperCAmelCase__   ,     UpperCAmelCase__   ,     UpperCAmelCase__			)
									    if self.balanced_factor() >= self.lim_charge
									    else None
									)
									if new_key is None:
													break
									else:
													i += 1

					return new_key
