'''simple docstring'''




from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin






@dataclass
class 		__A       (__magic_name__   ):
       snake_case       :torch.FloatTensor






class 		__A       (__magic_name__ ,					__magic_name__   ):






       @register_to_config
       def __init__(      self				,     UpperCamelCase_ = 32				,     UpperCamelCase_ = 64				,     UpperCamelCase_ = 20				,     UpperCamelCase_ = 7_68				,     UpperCamelCase_=77				,     UpperCamelCase_=4				,     UpperCamelCase_ = 0.0				,     UpperCamelCase_ = "silu"				,     UpperCamelCase_ = None				,     UpperCamelCase_ = None				,     UpperCamelCase_ = "linear"				,     UpperCamelCase_ = "prd"				,     UpperCamelCase_ = None				,     UpperCamelCase_ = None				,     UpperCamelCase_ = None				,     ):
              super().__init__()
              __UpperCAmelCase  :				List[str]       				=		num_attention_heads
              __UpperCAmelCase  :				Dict       				=		attention_head_dim
              __UpperCAmelCase  :				str       				=		num_attention_heads * attention_head_dim
              __UpperCAmelCase  :				Any       				=		additional_embeddings

              __UpperCAmelCase  :				Union[str, Any]       				=		time_embed_dim or inner_dim
              __UpperCAmelCase  :				List[str]       				=		embedding_proj_dim or embedding_dim
              __UpperCAmelCase  :				Any       				=		clip_embed_dim or embedding_dim

              __UpperCAmelCase  :				List[str]       				=		Timesteps(UpperCamelCase_				,     UpperCamelCase_				,     0      )
              __UpperCAmelCase  :				Any       				=		TimestepEmbedding(UpperCamelCase_				,     UpperCamelCase_				,     out_dim=UpperCamelCase_				,     act_fn=UpperCamelCase_      )

              __UpperCAmelCase  :				Tuple       				=		nn.Linear(UpperCamelCase_				,     UpperCamelCase_      )

              if embedding_proj_norm_type is None:
                     __UpperCAmelCase  :				List[str]       				=		None
              elif embedding_proj_norm_type == "layer":
                     __UpperCAmelCase  :				Dict       				=		nn.LayerNorm(UpperCamelCase_      )
              else:
                     raise ValueError(f"""unsupported embedding_proj_norm_type: {embedding_proj_norm_type}"""      )

              __UpperCAmelCase  :				List[Any]       				=		nn.Linear(UpperCamelCase_				,     UpperCamelCase_      )

              if encoder_hid_proj_type is None:
                     __UpperCAmelCase  :				Union[str, Any]       				=		None
              elif encoder_hid_proj_type == "linear":
                     __UpperCAmelCase  :				Dict       				=		nn.Linear(UpperCamelCase_				,     UpperCamelCase_      )
              else:
                     raise ValueError(f"""unsupported encoder_hid_proj_type: {encoder_hid_proj_type}"""      )

              __UpperCAmelCase  :				str       				=		nn.Parameter(torch.zeros(1				,     num_embeddings + additional_embeddings				,     UpperCamelCase_      )      )

              if added_emb_type == "prd":
                     __UpperCAmelCase  :				Optional[Any]       				=		nn.Parameter(torch.zeros(1				,     1				,     UpperCamelCase_      )      )
              elif added_emb_type is None:
                     __UpperCAmelCase  :				Any       				=		None
              else:
                     raise ValueError(
                         f"""`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `'prd'` or `None`."""      )

              __UpperCAmelCase  :				Optional[Any]       				=		nn.ModuleList(
                  [
                      BasicTransformerBlock(
                          UpperCamelCase_				,     UpperCamelCase_				,     UpperCamelCase_				,     dropout=UpperCamelCase_				,     activation_fn="gelu"				,     attention_bias=UpperCamelCase_				,     )
                      for d in range(UpperCamelCase_      )
                  ]      )

              if norm_in_type == "layer":
                     __UpperCAmelCase  :				Union[str, Any]       				=		nn.LayerNorm(UpperCamelCase_      )
              elif norm_in_type is None:
                     __UpperCAmelCase  :				Tuple       				=		None
              else:
                     raise ValueError(f"""Unsupported norm_in_type: {norm_in_type}."""      )

              __UpperCAmelCase  :				Dict       				=		nn.LayerNorm(UpperCamelCase_      )

              __UpperCAmelCase  :				List[Any]       				=		nn.Linear(UpperCamelCase_				,     UpperCamelCase_      )

              __UpperCAmelCase  :				Dict       				=		torch.full(
                  [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]				,     -1_0_0_0_0.0      )
              causal_attention_mask.triu_(1      )
              __UpperCAmelCase  :				List[Any]       				=		causal_attention_mask[None, ...]
              self.register_buffer("causal_attention_mask"				,     UpperCamelCase_				,     persistent=UpperCamelCase_      )

              __UpperCAmelCase  :				int       				=		nn.Parameter(torch.zeros(1				,     UpperCamelCase_      )      )
              __UpperCAmelCase  :				Tuple       				=		nn.Parameter(torch.zeros(1				,     UpperCamelCase_      )      )






       @property
       # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
       def 			_snake_case						(      self      ):
              __UpperCAmelCase  :				Union[str, Any]       				=		{}

              def fn_recursive_add_processors(UpperCamelCase_				,     UpperCamelCase_				,     UpperCamelCase_      ):
                     if hasattr(UpperCamelCase_				,     "set_processor"      ):
                            __UpperCAmelCase  :				Dict       				=		module.processor

                     for sub_name, child in module.named_children():
                            fn_recursive_add_processors(f"""{name}.{sub_name}"""				,     UpperCamelCase_				,     UpperCamelCase_      )

                     return processors

              for name, module in self.named_children():
                     fn_recursive_add_processors(UpperCamelCase_				,     UpperCamelCase_				,     UpperCamelCase_      )

              return processors






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              __UpperCAmelCase  :				Any       				=		len(self.attn_processors.keys()      )

              if isinstance(UpperCamelCase_				,     UpperCamelCase_      ) and len(UpperCamelCase_      ) != count:
                     raise ValueError(
                         f"""A dict of processors was passed, but the number of processors {len(UpperCamelCase_      )} does not match the"""
                         f""" number of attention layers: {count}. Please make sure to pass {count} processor classes."""      )

              def fn_recursive_attn_processor(UpperCamelCase_				,     UpperCamelCase_				,     UpperCamelCase_      ):
                     if hasattr(UpperCamelCase_				,     "set_processor"      ):
                            if not isinstance(UpperCamelCase_				,     UpperCamelCase_      ):
                                   module.set_processor(UpperCamelCase_      )
                            else:
                                   module.set_processor(processor.pop(f"""{name}.processor"""      )      )

                     for sub_name, child in module.named_children():
                            fn_recursive_attn_processor(f"""{name}.{sub_name}"""				,     UpperCamelCase_				,     UpperCamelCase_      )

              for name, module in self.named_children():
                     fn_recursive_attn_processor(UpperCamelCase_				,     UpperCamelCase_				,     UpperCamelCase_      )






       def 			_snake_case						(      self      ):
              self.set_attn_processor(AttnProcessor()      )






       def 			_snake_case						(      self				,     UpperCamelCase_				,     UpperCamelCase_				,     UpperCamelCase_				,     UpperCamelCase_ = None				,     UpperCamelCase_ = None				,     UpperCamelCase_ = True				,     ):
              __UpperCAmelCase  :				Tuple       				=		hidden_states.shape[0]

              __UpperCAmelCase  :				Optional[Any]       				=		timestep
              if not torch.is_tensor(UpperCamelCase_      ):
                     __UpperCAmelCase  :				Dict       				=		torch.tensor([timesteps]				,     dtype=torch.long				,     device=hidden_states.device      )
              elif torch.is_tensor(UpperCamelCase_      ) and len(timesteps.shape      ) == 0:
                     __UpperCAmelCase  :				int       				=		timesteps[None].to(hidden_states.device      )

              # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
              __UpperCAmelCase  :				int       				=		timesteps * torch.ones(UpperCamelCase_				,     dtype=timesteps.dtype				,     device=timesteps.device      )

              __UpperCAmelCase  :				Union[str, Any]       				=		self.time_proj(UpperCamelCase_      )

              # timesteps does not contain any weights and will always return f32 tensors
              # but time_embedding might be fp16, so we need to cast here.
              __UpperCAmelCase  :				Union[str, Any]       				=		timesteps_projected.to(dtype=self.dtype      )
              __UpperCAmelCase  :				Optional[Any]       				=		self.time_embedding(UpperCamelCase_      )

              if self.embedding_proj_norm is not None:
                     __UpperCAmelCase  :				Optional[int]       				=		self.embedding_proj_norm(UpperCamelCase_      )

              __UpperCAmelCase  :				int       				=		self.embedding_proj(UpperCamelCase_      )
              if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
                     __UpperCAmelCase  :				Optional[Any]       				=		self.encoder_hidden_states_proj(UpperCamelCase_      )
              elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
                     raise ValueError("`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set"      )

              __UpperCAmelCase  :				List[str]       				=		self.proj_in(UpperCamelCase_      )

              __UpperCAmelCase  :				Optional[int]       				=		self.positional_embedding.to(hidden_states.dtype      )

              __UpperCAmelCase  :				Any       				=		[]
              __UpperCAmelCase  :				Dict       				=		0

              if encoder_hidden_states is not None:
                     additional_embeds.append(UpperCamelCase_      )
                     additional_embeddings_len += encoder_hidden_states.shape[1]

              if len(proj_embeddings.shape      ) == 2:
                     __UpperCAmelCase  :				List[Any]       				=		proj_embeddings[:, None, :]

              if len(hidden_states.shape      ) == 2:
                     __UpperCAmelCase  :				Dict       				=		hidden_states[:, None, :]

              __UpperCAmelCase  :				int       				=		additional_embeds + [
                  proj_embeddings,
                  time_embeddings[:, None, :],
                  hidden_states,
              ]

              if self.prd_embedding is not None:
                     __UpperCAmelCase  :				Any       				=		self.prd_embedding.to(hidden_states.dtype      ).expand(UpperCamelCase_				,     -1				,     -1      )
                     additional_embeds.append(UpperCamelCase_      )

              __UpperCAmelCase  :				int       				=		torch.cat(
                  UpperCamelCase_				,     dim=1				,     )

              # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
              __UpperCAmelCase  :				int       				=		additional_embeddings_len + proj_embeddings.shape[1] + 1
              if positional_embeddings.shape[1] < hidden_states.shape[1]:
                     __UpperCAmelCase  :				int       				=		F.pad(
                         UpperCamelCase_				,     (
                             0,
                             0,
                             additional_embeddings_len,
                             self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
                         )				,     value=0.0				,     )

              __UpperCAmelCase  :				str       				=		hidden_states + positional_embeddings

              if attention_mask is not None:
                     __UpperCAmelCase  :				Dict       				=		(1 - attention_mask.to(hidden_states.dtype      )) * -1_0_0_0_0.0
                     __UpperCAmelCase  :				List[str]       				=		F.pad(UpperCamelCase_				,     (0, self.additional_embeddings)				,     value=0.0      )
                     __UpperCAmelCase  :				Any       				=		(attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype      )
                     __UpperCAmelCase  :				str       				=		attention_mask.repeat_interleave(self.config.num_attention_heads				,     dim=0      )

              if self.norm_in is not None:
                     __UpperCAmelCase  :				Dict       				=		self.norm_in(UpperCamelCase_      )

              for block in self.transformer_blocks:
                     __UpperCAmelCase  :				Optional[Any]       				=		block(UpperCamelCase_				,     attention_mask=UpperCamelCase_      )

              __UpperCAmelCase  :				Any       				=		self.norm_out(UpperCamelCase_      )

              if self.prd_embedding is not None:
                     __UpperCAmelCase  :				List[Any]       				=		hidden_states[:, -1]
              else:
                     __UpperCAmelCase  :				Dict       				=		hidden_states[:, additional_embeddings_len:]

              __UpperCAmelCase  :				List[Any]       				=		self.proj_to_clip_embeddings(UpperCamelCase_      )

              if not return_dict:
                     return (predicted_image_embedding,)

              return PriorTransformerOutput(predicted_image_embedding=UpperCamelCase_      )







       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              __UpperCAmelCase  :				Optional[Any]       				=		(prior_latents * self.clip_std) + self.clip_mean
              return prior_latents







'''simple docstring'''




# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
   import torch






class 		__A       (TensorFormatter[Mapping, "torch.Tensor", Mapping]   ):






       def __init__(      self				,     UpperCamelCase_=None				,     **UpperCamelCase_      ):
              super().__init__(features=UpperCamelCase_      )
              __UpperCAmelCase  :				Union[str, Any]       				=		torch_tensor_kwargs
              import torch  # noqa import torch at initialization






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              import torch

              if isinstance(UpperCamelCase_				,     UpperCamelCase_      ) and column:
                     if all(
                         isinstance(UpperCamelCase_				,     torch.Tensor      ) and x.shape == column[0].shape and x.dtype == column[0].dtype
                         for x in column      ):
                            return torch.stack(UpperCamelCase_      )
              return column






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              import torch

              if isinstance(UpperCamelCase_				,     (str, bytes, type(UpperCamelCase_      ))      ):
                     return value
              elif isinstance(UpperCamelCase_				,     (np.character, np.ndarray)      ) and np.issubdtype(value.dtype				,     np.character      ):
                     return value.tolist()

              __UpperCAmelCase  :				int       				=		{}

              if isinstance(UpperCamelCase_				,     (np.number, np.ndarray)      ) and np.issubdtype(value.dtype				,     np.integer      ):
                     __UpperCAmelCase  :				Optional[int]       				=		{"dtype": torch.intaa}
              elif isinstance(UpperCamelCase_				,     (np.number, np.ndarray)      ) and np.issubdtype(value.dtype				,     np.floating      ):
                     __UpperCAmelCase  :				str       				=		{"dtype": torch.floataa}
              elif config.PIL_AVAILABLE and "PIL" in sys.modules:
                     import PIL.Image

                     if isinstance(UpperCamelCase_				,     PIL.Image.Image      ):
                            __UpperCAmelCase  :				str       				=		np.asarray(UpperCamelCase_      )
              return torch.tensor(UpperCamelCase_				,     **{**default_dtype, **self.torch_tensor_kwargs}      )






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              import torch

              # support for torch, tf, jax etc.
              if hasattr(UpperCamelCase_				,     "__array__"      ) and not isinstance(UpperCamelCase_				,     torch.Tensor      ):
                     __UpperCAmelCase  :				Dict       				=		data_struct.__array__()
              # support for nested types like struct of list of struct
              if isinstance(UpperCamelCase_				,     np.ndarray      ):
                     if data_struct.dtype == object:  # torch tensors cannot be instantied from an array of objects
                            return self._consolidate([self.recursive_tensorize(UpperCamelCase_      ) for substruct in data_struct]      )
              elif isinstance(UpperCamelCase_				,     (list, tuple)      ):
                     return self._consolidate([self.recursive_tensorize(UpperCamelCase_      ) for substruct in data_struct]      )
              return self._tensorize(UpperCamelCase_      )






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              return map_nested(self._recursive_tensorize				,     UpperCamelCase_				,     map_list=UpperCamelCase_      )






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              __UpperCAmelCase  :				List[str]       				=		self.numpy_arrow_extractor().extract_row(UpperCamelCase_      )
              __UpperCAmelCase  :				Union[str, Any]       				=		self.python_features_decoder.decode_row(UpperCamelCase_      )
              return self.recursive_tensorize(UpperCamelCase_      )






       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              __UpperCAmelCase  :				Union[str, Any]       				=		self.numpy_arrow_extractor().extract_column(UpperCamelCase_      )
              __UpperCAmelCase  :				Optional[Any]       				=		self.python_features_decoder.decode_column(UpperCamelCase_				,     pa_table.column_names[0]      )
              __UpperCAmelCase  :				List[Any]       				=		self.recursive_tensorize(UpperCamelCase_      )
              __UpperCAmelCase  :				List[str]       				=		self._consolidate(UpperCamelCase_      )
              return column







       def 			_snake_case						(      self				,     UpperCamelCase_      ):
              __UpperCAmelCase  :				int       				=		self.numpy_arrow_extractor().extract_batch(UpperCamelCase_      )
              __UpperCAmelCase  :				Any       				=		self.python_features_decoder.decode_batch(UpperCamelCase_      )
              __UpperCAmelCase  :				Optional[int]       				=		self.recursive_tensorize(UpperCamelCase_      )
              for column_name in batch:
                     __UpperCAmelCase  :				Tuple       				=		self._consolidate(batch[column_name]      )
              return batch






