




'''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_	(       UpperCAmelCase_							):







  '''simple docstring'''

  UpperCamelCase_			:      torch.FloatTensor






class 						lowerCAmelCase_	(       UpperCAmelCase_       ,					UpperCAmelCase_							):







  '''simple docstring'''

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







       '''simple docstring'''



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

            A:		List[str]          =		torch.nn.GroupNorm(num_groups=SCREAMING_SNAKE_CASE_					,  num_channels=SCREAMING_SNAKE_CASE_					,  eps=1E-6					,  affine=SCREAMING_SNAKE_CASE_	)
            if use_linear_projection:
                 A:		Optional[int]          =		nn.Linear(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
            else:
                 A:		Any          =		nn.Convad(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  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:		Dict          =		sample_size
            A:		str          =		sample_size
            A:		Any          =		num_vector_embeds
            A:		Optional[int]          =		self.height * self.width

            A:		str          =		ImagePositionalEmbeddings(
                num_embed=SCREAMING_SNAKE_CASE_					,  embed_dim=SCREAMING_SNAKE_CASE_					,  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:		Dict          =		sample_size
            A:		Optional[int]          =		sample_size

            A:		List[str]          =		patch_size
            A:		Optional[int]          =		PatchEmbed(
                height=SCREAMING_SNAKE_CASE_					,  width=SCREAMING_SNAKE_CASE_					,  patch_size=SCREAMING_SNAKE_CASE_					,  in_channels=SCREAMING_SNAKE_CASE_					,  embed_dim=SCREAMING_SNAKE_CASE_					,  )

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

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




  def 					_snake_case     ( self				:							str					,  SCREAMING_SNAKE_CASE_				:							torch.Tensor					,  SCREAMING_SNAKE_CASE_				:							Optional[torch.Tensor] = None					,  SCREAMING_SNAKE_CASE_				:							Optional[torch.LongTensor] = None					,  SCREAMING_SNAKE_CASE_				:							Optional[torch.LongTensor] = None					,  SCREAMING_SNAKE_CASE_				:							Dict[str, Any] = None					,  SCREAMING_SNAKE_CASE_				:							Optional[torch.Tensor] = None					,  SCREAMING_SNAKE_CASE_				:							Optional[torch.Tensor] = None					,  SCREAMING_SNAKE_CASE_				:							bool = True					,  )			->		int:







       '''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)
            A:		str          =		(1 - attention_mask.to(hidden_states.dtype	)) * -1_0000.0
            A:		List[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:
            A:		Union[str, Any]          =		(1 - encoder_attention_mask.to(hidden_states.dtype	)) * -1_0000.0
            A:		int          =		encoder_attention_mask.unsqueeze(1	)

       # 1. Input
       if self.is_input_continuous:
            A     ,    A     ,    A     ,    A:		Dict          =		hidden_states.shape
            A:		str          =		hidden_states

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

       # 2. Blocks
       for block in self.transformer_blocks:
            A:		List[str]          =		block(
                SCREAMING_SNAKE_CASE_					,  attention_mask=SCREAMING_SNAKE_CASE_					,  encoder_hidden_states=SCREAMING_SNAKE_CASE_					,  encoder_attention_mask=SCREAMING_SNAKE_CASE_					,  timestep=SCREAMING_SNAKE_CASE_					,  cross_attention_kwargs=SCREAMING_SNAKE_CASE_					,  class_labels=SCREAMING_SNAKE_CASE_					,  )

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

            A:		List[Any]          =		hidden_states + residual
       elif self.is_input_vectorized:
            A:		Optional[int]          =		self.norm_out(SCREAMING_SNAKE_CASE_	)
            A:		Tuple          =		self.out(SCREAMING_SNAKE_CASE_	)
            # (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:		Tuple          =		self.transformer_blocks[0].norma.emb(
                SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  hidden_dtype=hidden_states.dtype	)
            A     ,    A:		int          =		self.proj_out_a(F.silu(SCREAMING_SNAKE_CASE_	)	).chunk(2					,  dim=1	)
            A:		List[Any]          =		self.norm_out(SCREAMING_SNAKE_CASE_	) * (1 + scale[:, None]) + shift[:, None]
            A:		Tuple          =		self.proj_out_a(SCREAMING_SNAKE_CASE_	)

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












'''simple docstring'''


from __future__ import annotations

from collections.abc import Callable
from typing import Generic, TypeVar

UpperCamelCase			     =					TypeVar('''T''')
UpperCamelCase			     =					TypeVar('''U''')






class 						lowerCAmelCase_	(       Generic[T, U]							):







  '''simple docstring'''

  def __init__( self				:							Tuple					,  SCREAMING_SNAKE_CASE_				:							T | None					,  SCREAMING_SNAKE_CASE_				:							U | None	)			->		Any:







       '''simple docstring'''



       A:		Union[str, Any]          =		key
       A:		Union[str, Any]          =		val
       A:		int          =		0
       A:		DoubleLinkedListNode[T, U] | None          =		None
       A:		DoubleLinkedListNode[T, U] | None          =		None




  def __repr__( self				:							Any	)			->		str:







       '''simple docstring'''



       return "Node: key: {}, val: {}, freq: {}, has next: {}, has prev: {}".format(
           self.key					,  self.val					,  self.freq					,  self.next is not None					,  self.prev is not None	)






class 						lowerCAmelCase_	(       Generic[T, U]							):







  '''simple docstring'''

  def __init__( self				:							int	)			->		None:







       '''simple docstring'''



       A:		DoubleLinkedListNode[T, U]          =		DoubleLinkedListNode(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
       A:		DoubleLinkedListNode[T, U]          =		DoubleLinkedListNode(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
       A     ,    A:		Any          =		self.rear, self.head



  def __repr__( self				:							Optional[int]	)			->		str:







       '''simple docstring'''



       A:		int          =		['''DoubleLinkedList''']
       A:		Dict          =		self.head
       while node.next is not None:
            rep.append(str(SCREAMING_SNAKE_CASE_	)	)
            A:		int          =		node.next
       rep.append(str(self.rear	)	)
       return ",\n    ".join(SCREAMING_SNAKE_CASE_	)



  def 					_snake_case     ( self				:							Any					,  SCREAMING_SNAKE_CASE_				:							DoubleLinkedListNode[T, U]	)			->		None:







       '''simple docstring'''



       A:		Union[str, Any]          =		self.rear.prev

       # All nodes other than self.head are guaranteed to have non-None previous
       assert previous is not None

       A:		Optional[int]          =		node
       A:		int          =		previous
       A:		Dict          =		node
       A:		int          =		self.rear
       node.freq += 1
       self._position_node(SCREAMING_SNAKE_CASE_	)



  def 					_snake_case     ( self				:							int					,  SCREAMING_SNAKE_CASE_				:							DoubleLinkedListNode[T, U]	)			->		None:







       '''simple docstring'''



       while node.prev is not None and node.prev.freq > node.freq:
            # swap node with previous node
            A:		Any          =		node.prev

            A:		Optional[Any]          =		previous_node.prev
            A:		int          =		node.prev
            A:		int          =		previous_node
            A:		Optional[int]          =		node




  def 					_snake_case     ( self				:							Dict					,  SCREAMING_SNAKE_CASE_				:							DoubleLinkedListNode[T, U]	)			->		DoubleLinkedListNode[T, U] | None:







       '''simple docstring'''



       if node.prev is None or node.next is None:
            return None

       A:		List[Any]          =		node.next
       A:		Dict          =		node.prev
       A:		str          =		None
       A:		Union[str, Any]          =		None
       return node






class 						lowerCAmelCase_	(       Generic[T, U]							):







  '''simple docstring'''

  UpperCamelCase_			:      dict[Callable[[T], U], LFUCache[T, U]]										=				{}



  def __init__( self				:							List[str]					,  SCREAMING_SNAKE_CASE_				:							int	)			->		int:







       '''simple docstring'''



       A:		DoubleLinkedList[T, U]          =		DoubleLinkedList()
       A:		Any          =		capacity
       A:		List[Any]          =		0
       A:		Dict          =		0
       A:		Dict          =		0
       A:		dict[T, DoubleLinkedListNode[T, U]]          =		{}



  def __repr__( self				:							Optional[int]	)			->		str:







       '''simple docstring'''



       return (
           f"""CacheInfo(hits={self.hits}, misses={self.miss}, """
           f"""capacity={self.capacity}, current_size={self.num_keys})"""
       )



  def __contains__( self				:							str					,  SCREAMING_SNAKE_CASE_				:							T	)			->		bool:







       '''simple docstring'''



       return key in self.cache



  def 					_snake_case     ( self				:							Dict					,  SCREAMING_SNAKE_CASE_				:							T	)			->		U | None:







       '''simple docstring'''



       if key in self.cache:
            self.hits += 1
            A:		DoubleLinkedListNode[T, U]          =		self.cache[key]
            A:		List[str]          =		self.list.remove(self.cache[key]	)
            assert node == value_node

            # node is guaranteed not None because it is in self.cache
            assert node is not None
            self.list.add(SCREAMING_SNAKE_CASE_	)
            return node.val
       self.miss += 1
       return None



  def 					_snake_case     ( self				:							List[str]					,  SCREAMING_SNAKE_CASE_				:							T					,  SCREAMING_SNAKE_CASE_				:							U	)			->		None:







       '''simple docstring'''



       if key not in self.cache:
            if self.num_keys >= self.capacity:
                 # delete first node when over capacity
                 A:		List[Any]          =		self.list.head.next

                 # guaranteed to have a non-None first node when num_keys > 0
                 # explain to type checker via assertions
                 assert first_node is not None
                 assert first_node.key is not None
                 assert self.list.remove(SCREAMING_SNAKE_CASE_	) is not None
                 # first_node guaranteed to be in list

                 del self.cache[first_node.key]
                 self.num_keys -= 1
            A:		Tuple          =		DoubleLinkedListNode(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
            self.list.add(self.cache[key]	)
            self.num_keys += 1

       else:
            A:		Optional[Any]          =		self.list.remove(self.cache[key]	)
            assert node is not None  # node guaranteed to be in list
            A:		List[str]          =		value
            self.list.add(SCREAMING_SNAKE_CASE_	)




  @classmethod
  def 					_snake_case     ( cls				:							type[LFUCache[T, U]]					,  SCREAMING_SNAKE_CASE_				:							int = 1_28	)			->		Callable[[Callable[[T], U]], Callable[..., U]]:







       '''simple docstring'''



       def cache_decorator_inner(SCREAMING_SNAKE_CASE_				:							Callable[[T], U]	) -> Callable[..., U]:
            def cache_decorator_wrapper(*SCREAMING_SNAKE_CASE_				:							T	) -> U:
                 if func not in cls.decorator_function_to_instance_map:
                      A:		Tuple          =		LFUCache(SCREAMING_SNAKE_CASE_	)

                 A:		List[Any]          =		cls.decorator_function_to_instance_map[func].get(args[0]	)
                 if result is None:
                      A:		List[str]          =		func(*SCREAMING_SNAKE_CASE_	)
                      cls.decorator_function_to_instance_map[func].put(args[0]					,  SCREAMING_SNAKE_CASE_	)
                 return result

            def cache_info() -> LFUCache[T, U]:
                 return cls.decorator_function_to_instance_map[func]

            setattr(SCREAMING_SNAKE_CASE_					,  '''cache_info'''					,  SCREAMING_SNAKE_CASE_	)  # noqa: B010

            return cache_decorator_wrapper

       return cache_decorator_inner


if __name__ == "__main__":
     import doctest

     doctest.testmod()






