


import functools
import math

import flax.linen as nn
import jax
import jax.numpy as jnp



def      lowerCAmelCase	(						lowerCAmelCase_	,				lowerCAmelCase_	,				lowerCAmelCase_	,				lowerCAmelCase_	,				lowerCAmelCase_ = 4_096	)->	Optional[int]:
				lowerCAmelCase_	: str      			=			key.shape[-3:]
				lowerCAmelCase_	: Dict      			=			value.shape[-1]
				lowerCAmelCase_	: int      			=			min(lowerCAmelCase_	,				lowerCAmelCase_	)
				lowerCAmelCase_	: List[Any]      			=			query / jnp.sqrt(lowerCAmelCase_	)

				@functools.partial(jax.checkpoint	,				prevent_cse=lowerCAmelCase_	)
				def summarize_chunk(lowerCAmelCase_	,				lowerCAmelCase_	,				lowerCAmelCase_	):
								lowerCAmelCase_	: int      			=			jnp.einsum('''...qhd,...khd->...qhk'''	,				lowerCAmelCase_	,				lowerCAmelCase_	,				precision=lowerCAmelCase_	)

								lowerCAmelCase_	: int      			=			jnp.max(lowerCAmelCase_	,				axis=-1	,				keepdims=lowerCAmelCase_	)
								lowerCAmelCase_	: Optional[int]      			=			jax.lax.stop_gradient(lowerCAmelCase_	)
								lowerCAmelCase_	: Tuple      			=			jnp.exp(attn_weights - max_score	)

								lowerCAmelCase_	: Tuple      			=			jnp.einsum('''...vhf,...qhv->...qhf'''	,				lowerCAmelCase_	,				lowerCAmelCase_	,				precision=lowerCAmelCase_	)
								lowerCAmelCase_	: str      			=			jnp.einsum('''...qhk->...qh'''	,				lowerCAmelCase_	)

								return (exp_values, exp_weights.sum(axis=-1	), max_score)

				def chunk_scanner(lowerCAmelCase_	):
								# julienne key array
								lowerCAmelCase_	: Union[str, Any]      			=			jax.lax.dynamic_slice(
								    operand=lowerCAmelCase_	,				start_indices=[0] * (key.ndim - 3) + [chunk_idx, 0, 0]	,				slice_sizes=list(key.shape[:-3]	) + [key_chunk_size, num_heads, k_features]	,				)

								# julienne value array
								lowerCAmelCase_	: Optional[int]      			=			jax.lax.dynamic_slice(
								    operand=lowerCAmelCase_	,				start_indices=[0] * (value.ndim - 3) + [chunk_idx, 0, 0]	,				slice_sizes=list(value.shape[:-3]	) + [key_chunk_size, num_heads, v_features]	,				)

								return summarize_chunk(lowerCAmelCase_	,				lowerCAmelCase_	,				lowerCAmelCase_	)

				lowerCAmelCase_	: Optional[Any]      			=			jax.lax.map(f=lowerCAmelCase_	,				xs=jnp.arange(0	,				lowerCAmelCase_	,				lowerCAmelCase_	)	)

				lowerCAmelCase_	: List[str]      			=			jnp.max(lowerCAmelCase_	,				axis=0	,				keepdims=lowerCAmelCase_	)
				lowerCAmelCase_	: List[str]      			=			jnp.exp(chunk_max - global_max	)

				chunk_values *= jnp.expand_dims(lowerCAmelCase_	,				axis=-1	)
				chunk_weights *= max_diffs

				lowerCAmelCase_	: Dict      			=			chunk_values.sum(axis=0	)
				lowerCAmelCase_	: Union[str, Any]      			=			jnp.expand_dims(lowerCAmelCase_	,				-1	).sum(axis=0	)

				return all_values / all_weights



def      lowerCAmelCase	(						lowerCAmelCase_	,				lowerCAmelCase_	,				lowerCAmelCase_	,				lowerCAmelCase_=jax.lax.Precision.HIGHEST	,				lowerCAmelCase_ = 1_024	,				lowerCAmelCase_ = 4_096	)->	int:
				lowerCAmelCase_	: List[Any]      			=			query.shape[-3:]

				def chunk_scanner(lowerCAmelCase_	,				lowerCAmelCase_	):
								# julienne query array
								lowerCAmelCase_	: Tuple      			=			jax.lax.dynamic_slice(
								    operand=lowerCAmelCase_	,				start_indices=([0] * (query.ndim - 3)) + [chunk_idx, 0, 0]	,				slice_sizes=list(query.shape[:-3]	) + [min(lowerCAmelCase_	,				lowerCAmelCase_	), num_heads, q_features]	,				)

								return (
								    chunk_idx + query_chunk_size,  # unused ignore it
								    _query_chunk_attention(
								        query=lowerCAmelCase_	,				key=lowerCAmelCase_	,				value=lowerCAmelCase_	,				precision=lowerCAmelCase_	,				key_chunk_size=lowerCAmelCase_	),
								)

				lowerCAmelCase_	: List[Any]      			=			jax.lax.scan(
				    f=lowerCAmelCase_	,				init=0	,				xs=lowerCAmelCase_	,				length=math.ceil(num_q / query_chunk_size	)	)

				return jnp.concatenate(lowerCAmelCase_	,				axis=-3	)  # fuse the chunked result back







class 			snake_case__(							nn.Module							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					Any   =							42
						SCREAMING_SNAKE_CASE__ :					List[Any]   =							8
						SCREAMING_SNAKE_CASE__ :					Tuple   =							64
						SCREAMING_SNAKE_CASE__ :					Any   =							0.0
						SCREAMING_SNAKE_CASE__ :					List[Any]   =							False
						SCREAMING_SNAKE_CASE__ :					int   =							jnp.floataa




						def   lowercase_	(			self							)   ->      Any:
										lowerCAmelCase_	: Optional[Any]      			=			self.dim_head * self.heads
										lowerCAmelCase_	: List[Any]      			=			self.dim_head**-0.5

										# Weights were exported with old names {to_q, to_k, to_v, to_out}
										lowerCAmelCase_	: Dict      			=			nn.Dense(_snake_case				,  use_bias=_snake_case				,  dtype=self.dtype				,  name='''to_q'''							)
										lowerCAmelCase_	: List[str]      			=			nn.Dense(_snake_case				,  use_bias=_snake_case				,  dtype=self.dtype				,  name='''to_k'''							)
										lowerCAmelCase_	: str      			=			nn.Dense(_snake_case				,  use_bias=_snake_case				,  dtype=self.dtype				,  name='''to_v'''							)

										lowerCAmelCase_	: List[str]      			=			nn.Dense(self.query_dim				,  dtype=self.dtype				,  name='''to_out_0'''							)
										lowerCAmelCase_	: str      			=			nn.Dropout(rate=self.dropout							)




						def   lowercase_	(			self				,  __lowercase							)   ->      Dict:
										lowerCAmelCase_	: List[Any]      			=			tensor.shape
										lowerCAmelCase_	: Tuple      			=			self.heads
										lowerCAmelCase_	: int      			=			tensor.reshape(_snake_case				,  _snake_case				,  _snake_case				,  dim // head_size							)
										lowerCAmelCase_	: Tuple      			=			jnp.transpose(_snake_case				,  (0, 2, 1, 3)							)
										lowerCAmelCase_	: List[str]      			=			tensor.reshape(batch_size * head_size				,  _snake_case				,  dim // head_size							)
										return tensor




						def   lowercase_	(			self				,  __lowercase							)   ->      Optional[Any]:
										lowerCAmelCase_	: Dict      			=			tensor.shape
										lowerCAmelCase_	: List[Any]      			=			self.heads
										lowerCAmelCase_	: Optional[Any]      			=			tensor.reshape(batch_size // head_size				,  _snake_case				,  _snake_case				,  _snake_case							)
										lowerCAmelCase_	: Tuple      			=			jnp.transpose(_snake_case				,  (0, 2, 1, 3)							)
										lowerCAmelCase_	: List[Any]      			=			tensor.reshape(batch_size // head_size				,  _snake_case				,  dim * head_size							)
										return tensor




						def __call__(			self				,  __lowercase				,  __lowercase=None				,  __lowercase=True							)   ->      Optional[Any]:
										lowerCAmelCase_	: Any      			=			hidden_states if context is None else context

										lowerCAmelCase_	: Optional[Any]      			=			self.query(_snake_case							)
										lowerCAmelCase_	: Tuple      			=			self.key(_snake_case							)
										lowerCAmelCase_	: Optional[Any]      			=			self.value(_snake_case							)

										lowerCAmelCase_	: List[str]      			=			self.reshape_heads_to_batch_dim(_snake_case							)
										lowerCAmelCase_	: Any      			=			self.reshape_heads_to_batch_dim(_snake_case							)
										lowerCAmelCase_	: str      			=			self.reshape_heads_to_batch_dim(_snake_case							)

										if self.use_memory_efficient_attention:
														lowerCAmelCase_	: Optional[Any]      			=			query_states.transpose(1				,  0				,  2							)
														lowerCAmelCase_	: str      			=			key_states.transpose(1				,  0				,  2							)
														lowerCAmelCase_	: List[Any]      			=			value_states.transpose(1				,  0				,  2							)

														# this if statement create a chunk size for each layer of the unet
														# the chunk size is equal to the query_length dimension of the deepest layer of the unet

														lowerCAmelCase_	: Optional[Any]      			=			query_states.shape[-3]
														if flatten_latent_dim % 6_4 == 0:
																		lowerCAmelCase_	: List[Any]      			=			int(flatten_latent_dim / 6_4							)
														elif flatten_latent_dim % 1_6 == 0:
																		lowerCAmelCase_	: List[str]      			=			int(flatten_latent_dim / 1_6							)
														elif flatten_latent_dim % 4 == 0:
																		lowerCAmelCase_	: Optional[Any]      			=			int(flatten_latent_dim / 4							)
														else:
																		lowerCAmelCase_	: Dict      			=			int(_snake_case							)

														lowerCAmelCase_	: Union[str, Any]      			=			jax_memory_efficient_attention(
														    _snake_case				,  _snake_case				,  _snake_case				,  query_chunk_size=_snake_case				,  key_chunk_size=4_0_9_6 * 4							)

														lowerCAmelCase_	: int      			=			hidden_states.transpose(1				,  0				,  2							)
										else:
														# compute attentions
														lowerCAmelCase_	: List[Any]      			=			jnp.einsum('''b i d, b j d->b i j'''				,  _snake_case				,  _snake_case							)
														lowerCAmelCase_	: Tuple      			=			attention_scores * self.scale
														lowerCAmelCase_	: Optional[Any]      			=			nn.softmax(_snake_case				,  axis=2							)

														# attend to values
														lowerCAmelCase_	: Optional[int]      			=			jnp.einsum('''b i j, b j d -> b i d'''				,  _snake_case				,  _snake_case							)

										lowerCAmelCase_	: List[Any]      			=			self.reshape_batch_dim_to_heads(_snake_case							)
										lowerCAmelCase_	: Optional[int]      			=			self.proj_attn(_snake_case							)
										return self.dropout_layer(_snake_case				,  deterministic=_snake_case							)







class 			snake_case__(							nn.Module							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					int   =							42
						SCREAMING_SNAKE_CASE__ :					Tuple   =							42
						SCREAMING_SNAKE_CASE__ :					Tuple   =							42
						SCREAMING_SNAKE_CASE__ :					Optional[int]   =							0.0
						SCREAMING_SNAKE_CASE__ :					Optional[int]   =							False
						SCREAMING_SNAKE_CASE__ :					Tuple   =							jnp.floataa
						SCREAMING_SNAKE_CASE__ :					Any   =							False




						def   lowercase_	(			self							)   ->      Tuple:
										# self attention (or cross_attention if only_cross_attention is True)
										lowerCAmelCase_	: List[str]      			=			FlaxAttention(
										    self.dim				,  self.n_heads				,  self.d_head				,  self.dropout				,  self.use_memory_efficient_attention				,  dtype=self.dtype							)
										# cross attention
										lowerCAmelCase_	: str      			=			FlaxAttention(
										    self.dim				,  self.n_heads				,  self.d_head				,  self.dropout				,  self.use_memory_efficient_attention				,  dtype=self.dtype							)
										lowerCAmelCase_	: Dict      			=			FlaxFeedForward(dim=self.dim				,  dropout=self.dropout				,  dtype=self.dtype							)
										lowerCAmelCase_	: Optional[Any]      			=			nn.LayerNorm(epsilon=1e-5				,  dtype=self.dtype							)
										lowerCAmelCase_	: int      			=			nn.LayerNorm(epsilon=1e-5				,  dtype=self.dtype							)
										lowerCAmelCase_	: Tuple      			=			nn.LayerNorm(epsilon=1e-5				,  dtype=self.dtype							)
										lowerCAmelCase_	: str      			=			nn.Dropout(rate=self.dropout							)




						def __call__(			self				,  __lowercase				,  __lowercase				,  __lowercase=True							)   ->      str:
										# self attention
										lowerCAmelCase_	: List[Any]      			=			hidden_states
										if self.only_cross_attention:
														lowerCAmelCase_	: Optional[Any]      			=			self.attna(self.norma(_snake_case							)				,  _snake_case				,  deterministic=_snake_case							)
										else:
														lowerCAmelCase_	: List[Any]      			=			self.attna(self.norma(_snake_case							)				,  deterministic=_snake_case							)
										lowerCAmelCase_	: Optional[Any]      			=			hidden_states + residual

										# cross attention
										lowerCAmelCase_	: str      			=			hidden_states
										lowerCAmelCase_	: int      			=			self.attna(self.norma(_snake_case							)				,  _snake_case				,  deterministic=_snake_case							)
										lowerCAmelCase_	: Any      			=			hidden_states + residual

										# feed forward
										lowerCAmelCase_	: Union[str, Any]      			=			hidden_states
										lowerCAmelCase_	: Optional[int]      			=			self.ff(self.norma(_snake_case							)				,  deterministic=_snake_case							)
										lowerCAmelCase_	: int      			=			hidden_states + residual

										return self.dropout_layer(_snake_case				,  deterministic=_snake_case							)







class 			snake_case__(							nn.Module							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					Union[str, Any]   =							42
						SCREAMING_SNAKE_CASE__ :					Optional[int]   =							42
						SCREAMING_SNAKE_CASE__ :					int   =							42
						SCREAMING_SNAKE_CASE__ :					Optional[Any]   =							1
						SCREAMING_SNAKE_CASE__ :					Dict   =							0.0
						SCREAMING_SNAKE_CASE__ :					Dict   =							False
						SCREAMING_SNAKE_CASE__ :					Tuple   =							False
						SCREAMING_SNAKE_CASE__ :					Optional[int]   =							jnp.floataa
						SCREAMING_SNAKE_CASE__ :					Dict   =							False




						def   lowercase_	(			self							)   ->      List[str]:
										lowerCAmelCase_	: List[Any]      			=			nn.GroupNorm(num_groups=3_2				,  epsilon=1e-5							)

										lowerCAmelCase_	: Optional[Any]      			=			self.n_heads * self.d_head
										if self.use_linear_projection:
														lowerCAmelCase_	: Optional[int]      			=			nn.Dense(_snake_case				,  dtype=self.dtype							)
										else:
														lowerCAmelCase_	: int      			=			nn.Conv(
														    _snake_case				,  kernel_size=(1, 1)				,  strides=(1, 1)				,  padding='''VALID'''				,  dtype=self.dtype				,  )

										lowerCAmelCase_	: Dict      			=			[
										    FlaxBasicTransformerBlock(
										        _snake_case				,  self.n_heads				,  self.d_head				,  dropout=self.dropout				,  only_cross_attention=self.only_cross_attention				,  dtype=self.dtype				,  use_memory_efficient_attention=self.use_memory_efficient_attention				,  )
										    for _ in range(self.depth							)
										]

										if self.use_linear_projection:
														lowerCAmelCase_	: Dict      			=			nn.Dense(_snake_case				,  dtype=self.dtype							)
										else:
														lowerCAmelCase_	: Optional[Any]      			=			nn.Conv(
														    _snake_case				,  kernel_size=(1, 1)				,  strides=(1, 1)				,  padding='''VALID'''				,  dtype=self.dtype				,  )

										lowerCAmelCase_	: Optional[Any]      			=			nn.Dropout(rate=self.dropout							)




						def __call__(			self				,  __lowercase				,  __lowercase				,  __lowercase=True							)   ->      Any:
										lowerCAmelCase_	: List[Any]      			=			hidden_states.shape
										lowerCAmelCase_	: Union[str, Any]      			=			hidden_states
										lowerCAmelCase_	: int      			=			self.norm(_snake_case							)
										if self.use_linear_projection:
														lowerCAmelCase_	: Optional[int]      			=			hidden_states.reshape(_snake_case				,  height * width				,  _snake_case							)
														lowerCAmelCase_	: Any      			=			self.proj_in(_snake_case							)
										else:
														lowerCAmelCase_	: Union[str, Any]      			=			self.proj_in(_snake_case							)
														lowerCAmelCase_	: str      			=			hidden_states.reshape(_snake_case				,  height * width				,  _snake_case							)

										for transformer_block in self.transformer_blocks:
														lowerCAmelCase_	: Optional[int]      			=			transformer_block(_snake_case				,  _snake_case				,  deterministic=_snake_case							)

										if self.use_linear_projection:
														lowerCAmelCase_	: Dict      			=			self.proj_out(_snake_case							)
														lowerCAmelCase_	: Union[str, Any]      			=			hidden_states.reshape(_snake_case				,  _snake_case				,  _snake_case				,  _snake_case							)
										else:
														lowerCAmelCase_	: List[Any]      			=			hidden_states.reshape(_snake_case				,  _snake_case				,  _snake_case				,  _snake_case							)
														lowerCAmelCase_	: Optional[Any]      			=			self.proj_out(_snake_case							)

										lowerCAmelCase_	: Any      			=			hidden_states + residual
										return self.dropout_layer(_snake_case				,  deterministic=_snake_case							)







class 			snake_case__(							nn.Module							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					Optional[Any]   =							42
						SCREAMING_SNAKE_CASE__ :					Any   =							0.0
						SCREAMING_SNAKE_CASE__ :					Any   =							jnp.floataa




						def   lowercase_	(			self							)   ->      Any:
										# The second linear layer needs to be called
										# net_2 for now to match the index of the Sequential layer
										lowerCAmelCase_	: int      			=			FlaxGEGLU(self.dim				,  self.dropout				,  self.dtype							)
										lowerCAmelCase_	: Tuple      			=			nn.Dense(self.dim				,  dtype=self.dtype							)




						def __call__(			self				,  __lowercase				,  __lowercase=True							)   ->      str:
										lowerCAmelCase_	: Optional[int]      			=			self.net_a(_snake_case				,  deterministic=_snake_case							)
										lowerCAmelCase_	: Optional[Any]      			=			self.net_a(_snake_case							)
										return hidden_states







class 			snake_case__(							nn.Module							):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__ :					List[str]   =							42
						SCREAMING_SNAKE_CASE__ :					Any   =							0.0
						SCREAMING_SNAKE_CASE__ :					Tuple   =							jnp.floataa




						def   lowercase_	(			self							)   ->      List[Any]:
										lowerCAmelCase_	: List[str]      			=			self.dim * 4
										lowerCAmelCase_	: str      			=			nn.Dense(inner_dim * 2				,  dtype=self.dtype							)
										lowerCAmelCase_	: str      			=			nn.Dropout(rate=self.dropout							)




						def __call__(			self				,  __lowercase				,  __lowercase=True							)   ->      Optional[Any]:
										lowerCAmelCase_	: int      			=			self.proj(_snake_case							)
										lowerCAmelCase_	: int      			=			jnp.split(_snake_case				,  2				,  axis=2							)
										return self.dropout_layer(hidden_linear * nn.gelu(_snake_case							)				,  deterministic=_snake_case							)



from typing import List, Optional, Tuple, Union

from . import logging
from .import_utils import is_torch_available, is_torch_version


if is_torch_available():
						import torch

_UpperCAmelCase     :					List[Any]       =logging.get_logger(__name__)  # pylint: disable=invalid-name

try:
						from torch._dynamo import allow_in_graph as maybe_allow_in_graph
except (ImportError, ModuleNotFoundError):



						def      lowerCAmelCase	(						cls	)->	int:
										return cls



def      lowerCAmelCase	(						lowerCAmelCase_	,				lowerCAmelCase_ = None	,				lowerCAmelCase_ = None	,				lowerCAmelCase_ = None	,				lowerCAmelCase_ = None	,				)->	int:
				lowerCAmelCase_	: Tuple      			=			device
				lowerCAmelCase_	: List[str]      			=			shape[0]

				lowerCAmelCase_	: Union[str, Any]      			=			layout or torch.strided
				lowerCAmelCase_	: Union[str, Any]      			=			device or torch.device('''cpu'''	)

				if generator is not None:
								lowerCAmelCase_	: List[str]      			=			generator.device.type if not isinstance(lowerCAmelCase_	,				lowerCAmelCase_	) else generator[0].device.type
								if gen_device_type != device.type and gen_device_type == "cpu":
												lowerCAmelCase_	: Optional[Any]      			=			'''cpu'''
												if device != "mps":
																logger.info(
																    f"""The passed generator was created on 'cpu' even though a tensor on {device} was expected."""
																    f""" Tensors will be created on 'cpu' and then moved to {device}. Note that one can probably"""
																    f""" slighly speed up this function by passing a generator that was created on the {device} device."""	)
								elif gen_device_type != device.type and gen_device_type == "cuda":
												raise ValueError(f"""Cannot generate a {device} tensor from a generator of type {gen_device_type}."""	)

				if isinstance(lowerCAmelCase_	,				lowerCAmelCase_	):
								lowerCAmelCase_	: Dict      			=			(1,) + shape[1:]
								lowerCAmelCase_	: Union[str, Any]      			=			[
								    torch.randn(lowerCAmelCase_	,				generator=generator[i]	,				device=lowerCAmelCase_	,				dtype=lowerCAmelCase_	,				layout=lowerCAmelCase_	)
								    for i in range(lowerCAmelCase_	)
								]
								lowerCAmelCase_	: Tuple      			=			torch.cat(lowerCAmelCase_	,				dim=0	).to(lowerCAmelCase_	)
				else:
								lowerCAmelCase_	: int      			=			torch.randn(lowerCAmelCase_	,				generator=lowerCAmelCase_	,				device=lowerCAmelCase_	,				dtype=lowerCAmelCase_	,				layout=lowerCAmelCase_	).to(lowerCAmelCase_	)

				return latents




def      lowerCAmelCase	(						lowerCAmelCase_	)->	Optional[Any]:
				if is_torch_version('''<'''	,				'''2.0.0'''	) or not hasattr(lowerCAmelCase_	,				'''_dynamo'''	):
								return False
				return isinstance(lowerCAmelCase_	,				torch._dynamo.eval_frame.OptimizedModule	)