



"""simple docstring"""





from typing import Optional, Tuple, Union

import flax
import flax.linen as nn
import jax
import jax.numpy as jnp
from flax.core.frozen_dict import FrozenDict

from ..configuration_utils import ConfigMixin, flax_register_to_config
from ..utils import BaseOutput
from .embeddings_flax import FlaxTimestepEmbedding, FlaxTimesteps
from .modeling_flax_utils import FlaxModelMixin
from .unet_ad_blocks_flax import (
    FlaxCrossAttnDownBlockaD,
    FlaxCrossAttnUpBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
    FlaxUpBlockaD,
)



@flax.struct.dataclass
class  _lowerCAmelCase     (					__UpperCamelCase		):





		"""simple docstring"""






		_lowerCamelCase      	=							42




@flax_register_to_config
class  _lowerCAmelCase     (					nn.Module	,					__UpperCamelCase	,					__UpperCamelCase		):





		"""simple docstring"""






		_lowerCamelCase      	=							32
		_lowerCamelCase      	=							4
		_lowerCamelCase      	=							4
		_lowerCamelCase      	=							(
		    '''CrossAttnDownBlock2D''',
		    '''CrossAttnDownBlock2D''',
		    '''CrossAttnDownBlock2D''',
		    '''DownBlock2D''',
		)
		_lowerCamelCase      	=							('''UpBlock2D''', '''CrossAttnUpBlock2D''', '''CrossAttnUpBlock2D''', '''CrossAttnUpBlock2D''')
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							(320, 640, 1_280, 1_280)
		_lowerCamelCase      	=							2
		_lowerCamelCase      	=							8
		_lowerCamelCase      	=							None
		_lowerCamelCase      	=							1_280
		_lowerCamelCase      	=							0.0
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							jnp.floataa
		_lowerCamelCase      	=							True
		_lowerCamelCase      	=							0
		_lowerCamelCase      	=							False





		def  UpperCAmelCase__				(       self				,     _lowercase )			->    FrozenDict:







								'''simple docstring'''

								snake_case_ :			str            =      (1, self.in_channels, self.sample_size, self.sample_size)
								snake_case_ :			Tuple            =      jnp.zeros(__UpperCAmelCase				,     dtype=jnp.floataa )
								snake_case_ :			Optional[int]            =      jnp.ones((1,)				,     dtype=jnp.intaa )
								snake_case_ :			int            =      jnp.zeros((1, 1, self.cross_attention_dim)				,     dtype=jnp.floataa )

								snake_case_ :			int            =      jax.random.split(__UpperCAmelCase )
								snake_case_ :			List[str]            =      {"params": params_rng, "dropout": dropout_rng}

								return self.init(__UpperCAmelCase				,     __UpperCAmelCase				,     __UpperCAmelCase				,     __UpperCAmelCase )["params"]





		def  UpperCAmelCase__				(       self )			->    Dict:







								'''simple docstring'''

								snake_case_ :			Dict            =      self.block_out_channels
								snake_case_ :			Union[str, Any]            =      block_out_channels[0] * 4

								if self.num_attention_heads is not None:
														raise ValueError(
														    """At the moment it is not possible to define the number of attention heads via `num_attention_heads` because of a naming issue as described in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131. Passing `num_attention_heads` will only be supported in diffusers v0.19.""" )

								# If `num_attention_heads` is not defined (which is the case for most models)
								# it will default to `attention_head_dim`. This looks weird upon first reading it and it is.
								# The reason for this behavior is to correct for incorrectly named variables that were introduced
								# when this library was created. The incorrect naming was only discovered much later in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131
								# Changing `attention_head_dim` to `num_attention_heads` for 40,000+ configurations is too backwards breaking
								# which is why we correct for the naming here.
								snake_case_ :			List[Any]            =      self.num_attention_heads or self.attention_head_dim

								# input
								snake_case_ :			Optional[int]            =      nn.Conv(
								    block_out_channels[0]				,     kernel_size=(3, 3)				,     strides=(1, 1)				,     padding=((1, 1), (1, 1))				,     dtype=self.dtype				,     )

								# time
								snake_case_ :			Optional[int]            =      FlaxTimesteps(
								    block_out_channels[0]				,     flip_sin_to_cos=self.flip_sin_to_cos				,     freq_shift=self.config.freq_shift )
								snake_case_ :			Optional[int]            =      FlaxTimestepEmbedding(__UpperCAmelCase				,     dtype=self.dtype )

								snake_case_ :			int            =      self.only_cross_attention
								if isinstance(__UpperCAmelCase				,     __UpperCAmelCase ):
														snake_case_ :			Union[str, Any]            =      (only_cross_attention,) * len(self.down_block_types )

								if isinstance(__UpperCAmelCase				,     __UpperCAmelCase ):
														snake_case_ :			Optional[Any]            =      (num_attention_heads,) * len(self.down_block_types )

								# down
								snake_case_ :			Union[str, Any]            =      []
								snake_case_ :			Union[str, Any]            =      block_out_channels[0]
								for i, down_block_type in enumerate(self.down_block_types ):
														snake_case_ :			List[str]            =      output_channel
														snake_case_ :			int            =      block_out_channels[i]
														snake_case_ :			Optional[int]            =      i == len(__UpperCAmelCase ) - 1

														if down_block_type == "CrossAttnDownBlock2D":
																				snake_case_ :			Optional[int]            =      FlaxCrossAttnDownBlockaD(
																				    in_channels=__UpperCAmelCase				,     out_channels=__UpperCAmelCase				,     dropout=self.dropout				,     num_layers=self.layers_per_block				,     num_attention_heads=num_attention_heads[i]				,     add_downsample=not is_final_block				,     use_linear_projection=self.use_linear_projection				,     only_cross_attention=only_cross_attention[i]				,     use_memory_efficient_attention=self.use_memory_efficient_attention				,     dtype=self.dtype				,     )
														else:
																				snake_case_ :			Optional[int]            =      FlaxDownBlockaD(
																				    in_channels=__UpperCAmelCase				,     out_channels=__UpperCAmelCase				,     dropout=self.dropout				,     num_layers=self.layers_per_block				,     add_downsample=not is_final_block				,     dtype=self.dtype				,     )

														down_blocks.append(__UpperCAmelCase )
								snake_case_ :			List[str]            =      down_blocks

								# mid
								snake_case_ :			List[str]            =      FlaxUNetMidBlockaDCrossAttn(
								    in_channels=block_out_channels[-1]				,     dropout=self.dropout				,     num_attention_heads=num_attention_heads[-1]				,     use_linear_projection=self.use_linear_projection				,     use_memory_efficient_attention=self.use_memory_efficient_attention				,     dtype=self.dtype				,     )

								# up
								snake_case_ :			Dict            =      []
								snake_case_ :			List[str]            =      list(reversed(__UpperCAmelCase ) )
								snake_case_ :			Tuple            =      list(reversed(__UpperCAmelCase ) )
								snake_case_ :			str            =      list(reversed(__UpperCAmelCase ) )
								snake_case_ :			List[str]            =      reversed_block_out_channels[0]
								for i, up_block_type in enumerate(self.up_block_types ):
														snake_case_ :			Union[str, Any]            =      output_channel
														snake_case_ :			Optional[Any]            =      reversed_block_out_channels[i]
														snake_case_ :			Tuple            =      reversed_block_out_channels[min(i + 1				,     len(__UpperCAmelCase ) - 1 )]

														snake_case_ :			Union[str, Any]            =      i == len(__UpperCAmelCase ) - 1

														if up_block_type == "CrossAttnUpBlock2D":
																				snake_case_ :			str            =      FlaxCrossAttnUpBlockaD(
																				    in_channels=__UpperCAmelCase				,     out_channels=__UpperCAmelCase				,     prev_output_channel=__UpperCAmelCase				,     num_layers=self.layers_per_block + 1				,     num_attention_heads=reversed_num_attention_heads[i]				,     add_upsample=not is_final_block				,     dropout=self.dropout				,     use_linear_projection=self.use_linear_projection				,     only_cross_attention=only_cross_attention[i]				,     use_memory_efficient_attention=self.use_memory_efficient_attention				,     dtype=self.dtype				,     )
														else:
																				snake_case_ :			Optional[Any]            =      FlaxUpBlockaD(
																				    in_channels=__UpperCAmelCase				,     out_channels=__UpperCAmelCase				,     prev_output_channel=__UpperCAmelCase				,     num_layers=self.layers_per_block + 1				,     add_upsample=not is_final_block				,     dropout=self.dropout				,     dtype=self.dtype				,     )

														up_blocks.append(__UpperCAmelCase )
														snake_case_ :			Dict            =      output_channel
								snake_case_ :			Any            =      up_blocks

								# out
								snake_case_ :			Tuple            =      nn.GroupNorm(num_groups=3_2				,     epsilon=1E-5 )
								snake_case_ :			List[Any]            =      nn.Conv(
								    self.out_channels				,     kernel_size=(3, 3)				,     strides=(1, 1)				,     padding=((1, 1), (1, 1))				,     dtype=self.dtype				,     )





		def __call__(       self				,     _lowercase				,     _lowercase				,     _lowercase				,     _lowercase=None				,     _lowercase=None				,     _lowercase = True				,     _lowercase = False				,     )			->    Union[FlaxUNetaDConditionOutput, Tuple]:







								'''simple docstring'''

								if not isinstance(__UpperCAmelCase				,     jnp.ndarray ):
														snake_case_ :			Tuple            =      jnp.array([timesteps]				,     dtype=jnp.intaa )
								elif isinstance(__UpperCAmelCase				,     jnp.ndarray ) and len(timesteps.shape ) == 0:
														snake_case_ :			List[str]            =      timesteps.astype(dtype=jnp.floataa )
														snake_case_ :			Tuple            =      jnp.expand_dims(__UpperCAmelCase				,     0 )

								snake_case_ :			Any            =      self.time_proj(__UpperCAmelCase )
								snake_case_ :			str            =      self.time_embedding(__UpperCAmelCase )

								# 2. pre-process
								snake_case_ :			str            =      jnp.transpose(__UpperCAmelCase				,     (0, 2, 3, 1) )
								snake_case_ :			Dict            =      self.conv_in(__UpperCAmelCase )

								# 3. down
								snake_case_ :			Union[str, Any]            =      (sample,)
								for down_block in self.down_blocks:
														if isinstance(__UpperCAmelCase				,     __UpperCAmelCase ):
																				snake_case_ :			Any            =      down_block(__UpperCAmelCase				,     __UpperCAmelCase				,     __UpperCAmelCase				,     deterministic=not train )
														else:
																				snake_case_ :			Tuple            =      down_block(__UpperCAmelCase				,     __UpperCAmelCase				,     deterministic=not train )
														down_block_res_samples += res_samples

								if down_block_additional_residuals is not None:
														snake_case_ :			Optional[Any]            =      ()

														for down_block_res_sample, down_block_additional_residual in zip(
														    __UpperCAmelCase				,     __UpperCAmelCase ):
																				down_block_res_sample += down_block_additional_residual
																				new_down_block_res_samples += (down_block_res_sample,)

														snake_case_ :			Any            =      new_down_block_res_samples

								# 4. mid
								snake_case_ :			Union[str, Any]            =      self.mid_block(__UpperCAmelCase				,     __UpperCAmelCase				,     __UpperCAmelCase				,     deterministic=not train )

								if mid_block_additional_residual is not None:
														sample += mid_block_additional_residual

								# 5. up
								for up_block in self.up_blocks:
														snake_case_ :			Any            =      down_block_res_samples[-(self.layers_per_block + 1) :]
														snake_case_ :			List[str]            =      down_block_res_samples[: -(self.layers_per_block + 1)]
														if isinstance(__UpperCAmelCase				,     __UpperCAmelCase ):
																				snake_case_ :			List[str]            =      up_block(
																				    __UpperCAmelCase				,     temb=__UpperCAmelCase				,     encoder_hidden_states=__UpperCAmelCase				,     res_hidden_states_tuple=__UpperCAmelCase				,     deterministic=not train				,     )
														else:
																				snake_case_ :			Union[str, Any]            =      up_block(__UpperCAmelCase				,     temb=__UpperCAmelCase				,     res_hidden_states_tuple=__UpperCAmelCase				,     deterministic=not train )

        # 6. post-process
								snake_case_ :			Optional[int]            =      self.conv_norm_out(__UpperCAmelCase )
								snake_case_ :			Dict            =      nn.silu(__UpperCAmelCase )
								snake_case_ :			Union[str, Any]            =      self.conv_out(__UpperCAmelCase )
								snake_case_ :			Optional[int]            =      jnp.transpose(__UpperCAmelCase				,     (0, 3, 1, 2) )

								if not return_dict:
														return (sample,)

								return FlaxUNetaDConditionOutput(sample=__UpperCAmelCase )









def     a_   (__A      )				->						Dict:



					"""simple docstring"""
					if not head:
										return True
					# split the list to two parts
					__a      ,			__a					:						Any      =					head.next, head
					while fast and fast.next:
										__a					:						Optional[int]      =					fast.next.next
										__a					:						Optional[int]      =					slow.next
					__a					:						Optional[int]      =					slow.next
					__a					:						Tuple      =					None  # Don't forget here! But forget still works!
					# reverse the second part
					__a					:						Any      =					None
					while second:
										__a					:						int      =					second.next
										__a					:						int      =					node
										__a					:						Union[str, Any]      =					second
										__a					:						Union[str, Any]      =					nxt
					# compare two parts
					# second part has the same or one less node
					while node:
										if node.val != head.val:
															return False
										__a					:						int      =					node.next
										__a					:						Dict      =					head.next
					return True







def     a_   (__A      )				->						Union[str, Any]:



					"""simple docstring"""
					if not head or not head.next:
										return True

					# 1. Get the midpoint (slow)
					__a					:						Union[str, Any]      =					head
					while fast and fast.next:
										__a      ,			__a					:						List[str]      =					fast.next.next, slow.next

					# 2. Push the second half into the stack
					__a					:						Dict      =					[slow.val]
					while slow.next:
										__a					:						Optional[Any]      =					slow.next
										stack.append(slow.val      )

					# 3. Comparison
					while stack:
										if stack.pop() != cur.val:
															return False
										__a					:						List[str]      =					cur.next

					return True







def     a_   (__A      )				->						List[str]:



					"""simple docstring"""
					if not head or not head.next:
										return True
					__a					:						Optional[int]      =					{}
					__a					:						Optional[int]      =					0
					while head:
										if head.val in d:
															d[head.val].append(__A      )
										else:
															__a					:						Dict      =					[pos]
										__a					:						Dict      =					head.next
										pos += 1
					__a					:						Dict      =					pos - 1
					__a					:						Optional[int]      =					0
					for v in d.values():
										if len(__A      ) % 2 != 0:
															middle += 1
										else:
															__a					:						Dict      =					0
															for i in range(0				,       len(__A      )      ):
																				if v[i] + v[len(__A      ) - 1 - step] != checksum:
																									return False
																				step += 1
										if middle > 1:
															return False
					return True





