





'''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     lowercase_		(						a	):

  '''simple docstring'''





  __lowerCAmelCase   :	torch.FloatTensor




class     lowercase_		(						a	,    a	):

  '''simple docstring'''





  @register_to_config
  def __init__(       self				,	a_ = 1_6				,	a_ = 8_8				,	a_ = None				,	a_ = None				,	a_ = 1				,	a_ = 0.0				,	a_ = 3_2				,	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   snake_case_     (       self				,	a_				,	a_ = None				,	a_ = None				,	a_ = None				,	a_ = None				,	a_ = None				,	a_ = None				,	a_ = True				,	)							->			Optional[int]:


    """simple docstring"""
    # 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)
      UpperCAmelCase  							=		(1 - attention_mask.to(hidden_states.dtype       )) * -1_0_0_0_0.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_0_0_0_0.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_       )





'''simple docstring'''




import os
import pathlib
import tempfile
import uuid

import numpy as np

from ..utils import is_soundfile_availble, is_torch_available, is_vision_available, logging


_a			:	int								=  logging.get_logger(__name__)

if is_vision_available():
  import PIL.Image
  from PIL import Image
  from PIL.Image import Image as ImageType
else:
  _a			:	List[str]								=  object

if is_torch_available():
  import torch

if is_soundfile_availble():
  import soundfile as sf



class     lowercase_		:

  '''simple docstring'''





  def __init__(       self				,	a_       )							->			Dict:


    """simple docstring"""
    UpperCAmelCase  							=		value
  def __str__(       self       )							->			Any:


    """simple docstring"""
    return self.to_string()
  def   snake_case_     (       self       )							->			Union[str, Any]:


    """simple docstring"""
    logger.error(
        'This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable'       )
    return self._value





  def   snake_case_     (       self       )							->			str:


    """simple docstring"""
    logger.error(
        'This is a raw AgentType of unknown type. Display in notebooks and string conversion will be unreliable'       )
    return str(self._value       )




class     lowercase_		(						a	,    a	):

  '''simple docstring'''





  def   snake_case_     (       self       )							->			Dict:


    """simple docstring"""
    return self._value





  def   snake_case_     (       self       )							->			List[Any]:


    """simple docstring"""
    return self._value




class     lowercase_		(						a	,    a	):

  '''simple docstring'''





  def __init__(       self				,	a_       )							->			int:


    """simple docstring"""
    super().__init__(a_       )

    if not is_vision_available():
      raise ImportError('PIL must be installed in order to handle images.'       )

    UpperCAmelCase  							=		None
    UpperCAmelCase  							=		None
    UpperCAmelCase  							=		None

    if isinstance(a_				,	a_       ):
      UpperCAmelCase  							=		value
    elif isinstance(a_				,	(str, pathlib.Path)       ):
      UpperCAmelCase  							=		value
    elif isinstance(a_				,	torch.Tensor       ):
      UpperCAmelCase  							=		value
    else:
      raise ValueError(F'''Unsupported type for {self.__class__.__name__}: {type(a_       )}'''       )
  def   snake_case_     (       self				,	a_=None				,	a_=None       )							->			Dict:


    """simple docstring"""
    from IPython.display import Image, display

    display(Image(self.to_string()       )       )
  def   snake_case_     (       self       )							->			List[str]:


    """simple docstring"""
    if self._raw is not None:
      return self._raw

    if self._path is not None:
      UpperCAmelCase  							=		Image.open(self._path       )
      return self._raw





  def   snake_case_     (       self       )							->			List[Any]:


    """simple docstring"""
    if self._path is not None:
      return self._path

    if self._raw is not None:
      UpperCAmelCase  							=		tempfile.mkdtemp()
      UpperCAmelCase  							=		os.path.join(a_				,	str(uuid.uuida()       ) + '.png'       )
      self._raw.save(self._path       )

      return self._path

    if self._tensor is not None:
      UpperCAmelCase  							=		self._tensor.cpu().detach().numpy()

      # There is likely simpler than load into image into save
      UpperCAmelCase  							=		Image.fromarray((array * 2_5_5).astype(np.uinta       )       )

      UpperCAmelCase  							=		tempfile.mkdtemp()
      UpperCAmelCase  							=		os.path.join(a_				,	str(uuid.uuida()       ) + '.png'       )

      img.save(self._path       )

      return self._path




class     lowercase_		(						a	):

  '''simple docstring'''





  def __init__(       self				,	a_				,	a_=1_6_0_0_0       )							->			str:


    """simple docstring"""
    super().__init__(a_       )

    if not is_soundfile_availble():
      raise ImportError('soundfile must be installed in order to handle audio.'       )

    UpperCAmelCase  							=		None
    UpperCAmelCase  							=		None

    UpperCAmelCase  							=		samplerate

    if isinstance(a_				,	(str, pathlib.Path)       ):
      UpperCAmelCase  							=		value
    elif isinstance(a_				,	torch.Tensor       ):
      UpperCAmelCase  							=		value
    else:
      raise ValueError(F'''Unsupported audio type: {type(a_       )}'''       )
  def   snake_case_     (       self				,	a_=None				,	a_=None       )							->			Dict:


    """simple docstring"""
    from IPython.display import Audio, display

    display(Audio(self.to_string()				,	rate=self.samplerate       )       )
  def   snake_case_     (       self       )							->			List[str]:


    """simple docstring"""
    if self._tensor is not None:
      return self._tensor

    if self._path is not None:
      UpperCAmelCase							, UpperCAmelCase  							=		sf.read(self._path       )
      UpperCAmelCase  							=		torch.tensor(a_       )
      return self._tensor





  def   snake_case_     (       self       )							->			int:


    """simple docstring"""
    if self._path is not None:
      return self._path

    if self._tensor is not None:
      UpperCAmelCase  							=		tempfile.mkdtemp()
      UpperCAmelCase  							=		os.path.join(a_				,	str(uuid.uuida()       ) + '.wav'       )
      sf.write(self._path				,	self._tensor				,	samplerate=self.samplerate       )
      return self._path




_a			:	Optional[int]								=  {'text': AgentText, 'image': AgentImage, 'audio': AgentAudio}
_a			:	Tuple								=  {str: AgentText}

if is_vision_available():
  _a			:	List[str]								=  AgentImage


def  lowerCamelCase__						(  *SCREAMING_SNAKE_CASE :				List[str]    ,				**SCREAMING_SNAKE_CASE :				Tuple  ):
  UpperCAmelCase  							=		[(arg.to_raw() if isinstance(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE  ) else arg) for arg in args]
  UpperCAmelCase  							=		{k: (v.to_raw() if isinstance(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE  ) else v) for k, v in kwargs.items()}
  return args, kwargs



def  lowerCamelCase__						(  SCREAMING_SNAKE_CASE :				Tuple    ,				SCREAMING_SNAKE_CASE :				Optional[Any]=None  ):
  if isinstance(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE  ):
    UpperCAmelCase  							=		{}
    for i, (k, v) in enumerate(outputs.items()  ):
      if output_types is not None:
        # If the class has defined outputs, we can map directly according to the class definition
        if output_types[i] in AGENT_TYPE_MAPPING:
          UpperCAmelCase  							=		AGENT_TYPE_MAPPING[output_types[i]](SCREAMING_SNAKE_CASE  )
        else:
          UpperCAmelCase  							=		AgentType(SCREAMING_SNAKE_CASE  )

      else:
        # If the class does not have defined output, then we map according to the type
        for _k, _v in INSTANCE_TYPE_MAPPING.items():
          if isinstance(SCREAMING_SNAKE_CASE    ,				_k  ):
            UpperCAmelCase  							=		_v(SCREAMING_SNAKE_CASE  )
        if k not in decoded_outputs:
          UpperCAmelCase  							=		AgentType[v]

  elif isinstance(SCREAMING_SNAKE_CASE    ,				(list, tuple)  ):
    UpperCAmelCase  							=		type(SCREAMING_SNAKE_CASE  )()
    for i, v in enumerate(SCREAMING_SNAKE_CASE  ):
      if output_types is not None:
        # If the class has defined outputs, we can map directly according to the class definition
        if output_types[i] in AGENT_TYPE_MAPPING:
          decoded_outputs.append(AGENT_TYPE_MAPPING[output_types[i]](SCREAMING_SNAKE_CASE  )  )
        else:
          decoded_outputs.append(AgentType(SCREAMING_SNAKE_CASE  )  )
      else:
        # If the class does not have defined output, then we map according to the type
        UpperCAmelCase  							=		False
        for _k, _v in INSTANCE_TYPE_MAPPING.items():
          if isinstance(SCREAMING_SNAKE_CASE    ,				_k  ):
            decoded_outputs.append(_v(SCREAMING_SNAKE_CASE  )  )
            UpperCAmelCase  							=		True

        if not found:
          decoded_outputs.append(AgentType(SCREAMING_SNAKE_CASE  )  )

  else:
    if output_types[0] in AGENT_TYPE_MAPPING:
      # If the class has defined outputs, we can map directly according to the class definition
      UpperCAmelCase  							=		AGENT_TYPE_MAPPING[output_types[0]](SCREAMING_SNAKE_CASE  )

    else:
      # If the class does not have defined output, then we map according to the type
      for _k, _v in INSTANCE_TYPE_MAPPING.items():
        if isinstance(SCREAMING_SNAKE_CASE    ,				_k  ):
          return _v(SCREAMING_SNAKE_CASE  )
      return AgentType(SCREAMING_SNAKE_CASE  )

  return decoded_outputs


