

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 _UpperCamelCase				(				__A ):
			'''simple docstring'''







			lowerCamelCase__							     =42






class _UpperCamelCase				(				__A				,						__A ):
			'''simple docstring'''







			@register_to_config
			def __init__(      self				:			Any   ,   a				:			int = 32   ,   a				:			int = 64   ,   a				:			int = 20   ,   a				:			int = 768   ,   a				:			int=77   ,   a				:			str=4   ,   a				:			float = 0.0   ,   a				:			str = "silu"   ,   a				:			Optional[str] = None   ,   a				:			Optional[str] = None   ,   a				:			Optional[str] = "linear"   ,   a				:			Optional[str] = "prd"   ,   a				:			Optional[int] = None   ,   a				:			Optional[int] = None   ,   a				:			Optional[int] = None   ,   )				->							int:

				"""simple docstring"""

				super().__init__()
				SCREAMING_SNAKE_CASE :   int					  =							num_attention_heads
				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							attention_head_dim
				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							num_attention_heads * attention_head_dim
				SCREAMING_SNAKE_CASE :   Any					  =							additional_embeddings

				SCREAMING_SNAKE_CASE :   int					  =							time_embed_dim or inner_dim
				SCREAMING_SNAKE_CASE :   List[Any]					  =							embedding_proj_dim or embedding_dim
				SCREAMING_SNAKE_CASE :   Any					  =							clip_embed_dim or embedding_dim

				SCREAMING_SNAKE_CASE :   Any					  =							Timesteps(a   ,   a   ,   0						)
				SCREAMING_SNAKE_CASE :   Dict					  =							TimestepEmbedding(a   ,   a   ,   out_dim=a   ,   act_fn=a						)

				SCREAMING_SNAKE_CASE :   str					  =							nn.Linear(a   ,   a						)

				if embedding_proj_norm_type is None:
					SCREAMING_SNAKE_CASE :   Dict					  =							None
				elif embedding_proj_norm_type == "layer":
					SCREAMING_SNAKE_CASE :   Tuple					  =							nn.LayerNorm(a						)
				else:
					raise ValueError(F"unsupported embedding_proj_norm_type: {embedding_proj_norm_type}"						)

				SCREAMING_SNAKE_CASE :   str					  =							nn.Linear(a   ,   a						)

				if encoder_hid_proj_type is None:
					SCREAMING_SNAKE_CASE :   Tuple					  =							None
				elif encoder_hid_proj_type == "linear":
					SCREAMING_SNAKE_CASE :   Any					  =							nn.Linear(a   ,   a						)
				else:
					raise ValueError(F"unsupported encoder_hid_proj_type: {encoder_hid_proj_type}"						)

				SCREAMING_SNAKE_CASE :   Any					  =							nn.Parameter(torch.zeros(1   ,   num_embeddings + additional_embeddings   ,   a						)						)

				if added_emb_type == "prd":
					SCREAMING_SNAKE_CASE :   Tuple					  =							nn.Parameter(torch.zeros(1   ,   1   ,   a						)						)
				elif added_emb_type is None:
					SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							None
				else:
					raise ValueError(
					    F"`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `'prd'` or `None`."						)

				SCREAMING_SNAKE_CASE :   Dict					  =							nn.ModuleList(
				    [
				        BasicTransformerBlock(
				            a   ,   a   ,   a   ,   dropout=a   ,   activation_fn="gelu"   ,   attention_bias=a   ,   )
				        for d in range(a						)
				    ]						)

				if norm_in_type == "layer":
					SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							nn.LayerNorm(a						)
				elif norm_in_type is None:
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							None
				else:
					raise ValueError(F"Unsupported norm_in_type: {norm_in_type}."						)

				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							nn.LayerNorm(a						)

				SCREAMING_SNAKE_CASE :   Tuple					  =							nn.Linear(a   ,   a						)

				SCREAMING_SNAKE_CASE :   Tuple					  =							torch.full(
				    [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]   ,   -1_0000.0						)
				causal_attention_mask.triu_(1						)
				SCREAMING_SNAKE_CASE :   Dict					  =							causal_attention_mask[None, ...]
				self.register_buffer("causal_attention_mask"   ,   a   ,   persistent=a						)

				SCREAMING_SNAKE_CASE :   List[Any]					  =							nn.Parameter(torch.zeros(1   ,   a						)						)
				SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							nn.Parameter(torch.zeros(1   ,   a						)						)
			@property
			# Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
			def 				__UpperCamelCase			(      self				:			Optional[Any]						)				->							Dict[str, AttentionProcessor]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Optional[int]					  =							{}

				def fn_recursive_add_processors(a				:			str   ,   a				:			torch.nn.Module   ,   a				:			Dict[str, AttentionProcessor]						):
					if hasattr(a   ,   "set_processor"						):
						SCREAMING_SNAKE_CASE :   int					  =							module.processor

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

					return processors

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

				return processors
			def 				__UpperCamelCase			(      self				:			Tuple   ,   a				:			Union[AttentionProcessor, Dict[str, AttentionProcessor]]						)				->							Optional[Any]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							len(self.attn_processors.keys()						)

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

				def fn_recursive_attn_processor(a				:			str   ,   a				:			torch.nn.Module   ,   a				:			int						):
					if hasattr(a   ,   "set_processor"						):
						if not isinstance(a   ,   a						):
							module.set_processor(a						)
						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}"   ,   a   ,   a						)

				for name, module in self.named_children():
					fn_recursive_attn_processor(a   ,   a   ,   a						)
			def 				__UpperCamelCase			(      self				:			List[str]						)				->							str:

				"""simple docstring"""

				self.set_attn_processor(AttnProcessor()						)
			def 				__UpperCamelCase			(      self				:			Optional[Any]   ,   a				:			Union[str, Any]   ,   a				:			Union[torch.Tensor, float, int]   ,   a				:			torch.FloatTensor   ,   a				:			Optional[torch.FloatTensor] = None   ,   a				:			Optional[torch.BoolTensor] = None   ,   a				:			bool = True   ,   )				->							Any:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Tuple					  =							hidden_states.shape[0]

				SCREAMING_SNAKE_CASE :   Any					  =							timestep
				if not torch.is_tensor(a						):
					SCREAMING_SNAKE_CASE :   Optional[Any]					  =							torch.tensor([timesteps]   ,   dtype=torch.long   ,   device=hidden_states.device						)
				elif torch.is_tensor(a						) and len(timesteps.shape						) == 0:
					SCREAMING_SNAKE_CASE :   List[str]					  =							timesteps[None].to(hidden_states.device						)

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

				SCREAMING_SNAKE_CASE :   Tuple					  =							self.time_proj(a						)

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

				if self.embedding_proj_norm is not None:
					SCREAMING_SNAKE_CASE :   List[Any]					  =							self.embedding_proj_norm(a						)

				SCREAMING_SNAKE_CASE :   str					  =							self.embedding_proj(a						)
				if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							self.encoder_hidden_states_proj(a						)
				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"						)

				SCREAMING_SNAKE_CASE :   List[Any]					  =							self.proj_in(a						)

				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							self.positional_embedding.to(hidden_states.dtype						)

				SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							[]
				SCREAMING_SNAKE_CASE :   Optional[int]					  =							0

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

				if len(proj_embeddings.shape						) == 2:
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							proj_embeddings[:, None, :]

				if len(hidden_states.shape						) == 2:
					SCREAMING_SNAKE_CASE :   List[Any]					  =							hidden_states[:, None, :]

				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							additional_embeds + [
				    proj_embeddings,
				    time_embeddings[:, None, :],
				    hidden_states,
				]

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

				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							torch.cat(
				    a   ,   dim=1   ,   )

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

				SCREAMING_SNAKE_CASE :   List[Any]					  =							hidden_states + positional_embeddings

				if attention_mask is not None:
					SCREAMING_SNAKE_CASE :   int					  =							(1 - attention_mask.to(hidden_states.dtype						)) * -1_0000.0
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							F.pad(a   ,   (0, self.additional_embeddings)   ,   value=0.0						)
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							(attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype						)
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							attention_mask.repeat_interleave(self.config.num_attention_heads   ,   dim=0						)

				if self.norm_in is not None:
					SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							self.norm_in(a						)

				for block in self.transformer_blocks:
					SCREAMING_SNAKE_CASE :   str					  =							block(a   ,   attention_mask=a						)

				SCREAMING_SNAKE_CASE :   List[str]					  =							self.norm_out(a						)

				if self.prd_embedding is not None:
					SCREAMING_SNAKE_CASE :   Optional[Any]					  =							hidden_states[:, -1]
				else:
					SCREAMING_SNAKE_CASE :   Any					  =							hidden_states[:, additional_embeddings_len:]

				SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							self.proj_to_clip_embeddings(a						)

				if not return_dict:
					return (predicted_image_embedding,)

				return PriorTransformerOutput(predicted_image_embedding=a						)






			def 				__UpperCamelCase			(      self				:			str   ,   a				:			Optional[Any]						)				->							Optional[Any]:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   str					  =							(prior_latents * self.clip_std) + self.clip_mean
				return prior_latents


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

import numpy as np
import pyarrow as pa

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


if TYPE_CHECKING:
			import jax
			import jaxlib

a_      =     get_logger()

a_      =     None






class _UpperCamelCase				(				TensorFormatter[Mapping, 'jax.Array', Mapping] ):
			'''simple docstring'''







			def __init__(      self				:			Optional[Any]   ,   a				:			str=None   ,   a				:			List[Any]=None   ,   **a				:			Any						)				->							Optional[Any]:

				"""simple docstring"""

				super().__init__(features=a						)
				import jax
				from jaxlib.xla_client import Device

				if isinstance(a   ,   a						):
					raise ValueError(
					    F"Expected {device} to be a `str` not {type(a						)}, as `jaxlib.xla_extension.Device` "
					    "is not serializable neither with `pickle` nor with `dill`. Instead you can surround "
					    "the device with `str()` to get its string identifier that will be internally mapped "
					    "to the actual `jaxlib.xla_extension.Device`."						)
				SCREAMING_SNAKE_CASE :   List[str]					  =							device if isinstance(a   ,   a						) else str(jax.devices()[0]						)
				# using global variable since `jaxlib.xla_extension.Device` is not serializable neither
				# with `pickle` nor with `dill`, so we need to use a global variable instead
				global DEVICE_MAPPING
				if DEVICE_MAPPING is None:
					SCREAMING_SNAKE_CASE :   str					  =							self._map_devices_to_str()
				if self.device not in list(DEVICE_MAPPING.keys()						):
					logger.warning(
					    F"Device with string identifier {self.device} not listed among the available "
					    F"devices: {list(DEVICE_MAPPING.keys()						)}, so falling back to the default "
					    F"device: {str(jax.devices()[0]						)}."						)
					SCREAMING_SNAKE_CASE :   Any					  =							str(jax.devices()[0]						)
				SCREAMING_SNAKE_CASE :   Any					  =							jnp_array_kwargs
			@staticmethod
			def 				__UpperCamelCase			(      )				->							Dict[str, "jaxlib.xla_extension.Device"]:

				"""simple docstring"""

				import jax

				return {str(a						): device for device in jax.devices()}
			def 				__UpperCamelCase			(      self				:			Dict   ,   a				:			Tuple						)				->							str:

				"""simple docstring"""

				import jax
				import jax.numpy as jnp

				if isinstance(a   ,   a						) and column:
					if all(
					    isinstance(a   ,   jax.Array						) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column						):
						return jnp.stack(a   ,   axis=0						)
				return column
			def 				__UpperCamelCase			(      self				:			Dict   ,   a				:			str						)				->							str:

				"""simple docstring"""

				import jax
				import jax.numpy as jnp

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

				SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							{}

				if isinstance(a   ,   (np.number, np.ndarray)						) and np.issubdtype(value.dtype   ,   np.integer						):
					# the default int precision depends on the jax config
					# see https://jax.readthedocs.io/en/latest/notebooks/Common_Gotchas_in_JAX.html#double-64bit-precision
					if jax.config.jax_enable_xaa:
						SCREAMING_SNAKE_CASE :   Dict					  =							{"dtype": jnp.intaa}
					else:
						SCREAMING_SNAKE_CASE :   str					  =							{"dtype": jnp.intaa}
				elif isinstance(a   ,   (np.number, np.ndarray)						) and np.issubdtype(value.dtype   ,   np.floating						):
					SCREAMING_SNAKE_CASE :   int					  =							{"dtype": jnp.floataa}
				elif config.PIL_AVAILABLE and "PIL" in sys.modules:
					import PIL.Image

					if isinstance(a   ,   PIL.Image.Image						):
						SCREAMING_SNAKE_CASE :   Dict					  =							np.asarray(a						)

        # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
        # with `pickle` nor with `dill`, so we need to use a global variable instead
				global DEVICE_MAPPING
				if DEVICE_MAPPING is None:
					SCREAMING_SNAKE_CASE :   Optional[Any]					  =							self._map_devices_to_str()

				with jax.default_device(DEVICE_MAPPING[self.device]						):
					# calling jnp.array on a np.ndarray does copy the data
					# see https://github.com/google/jax/issues/4486
					return jnp.array(a   ,   **{**default_dtype, **self.jnp_array_kwargs}						)
			def 				__UpperCamelCase			(      self				:			Any   ,   a				:			List[str]						)				->							Dict:

				"""simple docstring"""

				import jax

				# support for torch, tf, jax etc.
				if config.TORCH_AVAILABLE and "torch" in sys.modules:
					import torch

					if isinstance(a   ,   torch.Tensor						):
						return self._tensorize(data_struct.detach().cpu().numpy()[()]						)
				if hasattr(a   ,   "__array__"						) and not isinstance(a   ,   jax.Array						):
					SCREAMING_SNAKE_CASE :   Optional[int]					  =							data_struct.__array__()
				# support for nested types like struct of list of struct
				if isinstance(a   ,   np.ndarray						):
					if data_struct.dtype == object:  # jax arrays cannot be instantied from an array of objects
						return self._consolidate([self.recursive_tensorize(a						) for substruct in data_struct]						)
				elif isinstance(a   ,   (list, tuple)						):
					return self._consolidate([self.recursive_tensorize(a						) for substruct in data_struct]						)
				return self._tensorize(a						)
			def 				__UpperCamelCase			(      self				:			Optional[Any]   ,   a				:			dict						)				->							Dict:

				"""simple docstring"""

				return map_nested(self._recursive_tensorize   ,   a   ,   map_list=a						)
			def 				__UpperCamelCase			(      self				:			Dict   ,   a				:			pa.Table						)				->							Mapping:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							self.numpy_arrow_extractor().extract_row(a						)
				SCREAMING_SNAKE_CASE :   List[Any]					  =							self.python_features_decoder.decode_row(a						)
				return self.recursive_tensorize(a						)
			def 				__UpperCamelCase			(      self				:			Optional[int]   ,   a				:			pa.Table						)				->							"jax.Array":

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   Optional[int]					  =							self.numpy_arrow_extractor().extract_column(a						)
				SCREAMING_SNAKE_CASE :   Optional[Any]					  =							self.python_features_decoder.decode_column(a   ,   pa_table.column_names[0]						)
				SCREAMING_SNAKE_CASE :   Tuple					  =							self.recursive_tensorize(a						)
				SCREAMING_SNAKE_CASE :   Optional[int]					  =							self._consolidate(a						)
				return column






			def 				__UpperCamelCase			(      self				:			List[Any]   ,   a				:			pa.Table						)				->							Mapping:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE :   str					  =							self.numpy_arrow_extractor().extract_batch(a						)
				SCREAMING_SNAKE_CASE :   str					  =							self.python_features_decoder.decode_batch(a						)
				SCREAMING_SNAKE_CASE :   List[Any]					  =							self.recursive_tensorize(a						)
				for column_name in batch:
					SCREAMING_SNAKE_CASE :   List[Any]					  =							self._consolidate(batch[column_name]						)
				return batch