'''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__      (						A     ):




					'''simple docstring'''



					A_						    =						42






@flax_register_to_config
class 		lowerCamelCase__      (						nn.Module	,   A	,   A     ):




					'''simple docstring'''



					A_						    =						32
					A_						    =						4
					A_						    =						4
					A_						    =						(
					    """CrossAttnDownBlock2D""",
					    """CrossAttnDownBlock2D""",
					    """CrossAttnDownBlock2D""",
					    """DownBlock2D""",
					)
					A_						    =						("""UpBlock2D""", """CrossAttnUpBlock2D""", """CrossAttnUpBlock2D""", """CrossAttnUpBlock2D""")
					A_						    =						False
					A_						    =						(320, 640, 1280, 1280)
					A_						    =						2
					A_						    =						8
					A_						    =						None
					A_						    =						1280
					A_						    =						0.0
					A_						    =						False
					A_						    =						jnp.floataa
					A_						    =						True
					A_						    =						0
					A_						    =						False




					def 						__UpperCAmelCase     (  self				:	int     ,  UpperCamelCase_				:	Optional[int]					)  ->       Optional[int]:



						'''simple docstring'''





						_lowercase					:       Optional[Any]       =		(1, self.in_channels, self.sample_size, self.sample_size)
						_lowercase					:       Tuple       =		jnp.zeros(__lowerCAmelCase     ,  dtype=jnp.floataa					)
						_lowercase					:       int       =		jnp.ones((1,)     ,  dtype=jnp.intaa					)
						_lowercase					:       Union[str, Any]       =		jnp.zeros((1, 1, self.cross_attention_dim)     ,  dtype=jnp.floataa					)

						_lowercase  ,    _lowercase					:       int       =		jax.random.split(__lowerCAmelCase					)
						_lowercase					:       Any       =		{'params': params_rng, 'dropout': dropout_rng}

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




					def 						__UpperCAmelCase     (  self				:	Optional[Any]					)  ->       int:



						'''simple docstring'''





						_lowercase					:       Optional[int]       =		self.block_out_channels
						_lowercase					:       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.
						_lowercase					:       Dict       =		self.num_attention_heads or self.attention_head_dim

						# input
						_lowercase					:       List[str]       =		nn.Conv(
						    block_out_channels[0]     ,  kernel_size=(3, 3)     ,  strides=(1, 1)     ,  padding=((1, 1), (1, 1))     ,  dtype=self.dtype     ,  )

						# time
						_lowercase					:       Optional[int]       =		FlaxTimesteps(
						    block_out_channels[0]     ,  flip_sin_to_cos=self.flip_sin_to_cos     ,  freq_shift=self.config.freq_shift					)
						_lowercase					:       List[str]       =		FlaxTimestepEmbedding(__lowerCAmelCase     ,  dtype=self.dtype					)

						_lowercase					:       List[str]       =		self.only_cross_attention
						if isinstance(__lowerCAmelCase     ,  __lowerCAmelCase					):
							_lowercase					:       str       =		(only_cross_attention,) * len(self.down_block_types					)

						if isinstance(__lowerCAmelCase     ,  __lowerCAmelCase					):
							_lowercase					:       List[Any]       =		(num_attention_heads,) * len(self.down_block_types					)

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

							if down_block_type == "CrossAttnDownBlock2D":
								_lowercase					:       Any       =		FlaxCrossAttnDownBlockaD(
								    in_channels=__lowerCAmelCase     ,  out_channels=__lowerCAmelCase     ,  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:
								_lowercase					:       Optional[Any]       =		FlaxDownBlockaD(
								    in_channels=__lowerCAmelCase     ,  out_channels=__lowerCAmelCase     ,  dropout=self.dropout     ,  num_layers=self.layers_per_block     ,  add_downsample=not is_final_block     ,  dtype=self.dtype     ,  )

							down_blocks.append(__lowerCAmelCase					)
						_lowercase					:       Optional[int]       =		down_blocks

						# mid
						_lowercase					:       Optional[int]       =		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
						_lowercase					:       int       =		[]
						_lowercase					:       List[Any]       =		list(reversed(__lowerCAmelCase					)					)
						_lowercase					:       int       =		list(reversed(__lowerCAmelCase					)					)
						_lowercase					:       Dict       =		list(reversed(__lowerCAmelCase					)					)
						_lowercase					:       Optional[int]       =		reversed_block_out_channels[0]
						for i, up_block_type in enumerate(self.up_block_types					):
							_lowercase					:       Dict       =		output_channel
							_lowercase					:       Optional[int]       =		reversed_block_out_channels[i]
							_lowercase					:       List[Any]       =		reversed_block_out_channels[min(i + 1     ,  len(__lowerCAmelCase					) - 1					)]

							_lowercase					:       str       =		i == len(__lowerCAmelCase					) - 1

							if up_block_type == "CrossAttnUpBlock2D":
								_lowercase					:       List[Any]       =		FlaxCrossAttnUpBlockaD(
								    in_channels=__lowerCAmelCase     ,  out_channels=__lowerCAmelCase     ,  prev_output_channel=__lowerCAmelCase     ,  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:
								_lowercase					:       List[str]       =		FlaxUpBlockaD(
								    in_channels=__lowerCAmelCase     ,  out_channels=__lowerCAmelCase     ,  prev_output_channel=__lowerCAmelCase     ,  num_layers=self.layers_per_block + 1     ,  add_upsample=not is_final_block     ,  dropout=self.dropout     ,  dtype=self.dtype     ,  )

							up_blocks.append(__lowerCAmelCase					)
							_lowercase					:       Union[str, Any]       =		output_channel
						_lowercase					:       Union[str, Any]       =		up_blocks

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




					def __call__(  self				:	Union[str, Any]     ,  UpperCamelCase_				:	Optional[Any]     ,  UpperCamelCase_				:	List[str]     ,  UpperCamelCase_				:	str     ,  UpperCamelCase_				:	List[Any]=None     ,  UpperCamelCase_				:	str=None     ,  UpperCamelCase_				:	Tuple = True     ,  UpperCamelCase_				:	int = False     ,  )  ->       int:



						'''simple docstring'''





						if not isinstance(__lowerCAmelCase     ,  jnp.ndarray					):
							_lowercase					:       List[str]       =		jnp.array([timesteps]     ,  dtype=jnp.intaa					)
						elif isinstance(__lowerCAmelCase     ,  jnp.ndarray					) and len(timesteps.shape					) == 0:
							_lowercase					:       Union[str, Any]       =		timesteps.astype(dtype=jnp.floataa					)
							_lowercase					:       Dict       =		jnp.expand_dims(__lowerCAmelCase     ,  0					)

						_lowercase					:       Union[str, Any]       =		self.time_proj(__lowerCAmelCase					)
						_lowercase					:       Union[str, Any]       =		self.time_embedding(__lowerCAmelCase					)

						# 2. pre-process
						_lowercase					:       Union[str, Any]       =		jnp.transpose(__lowerCAmelCase     ,  (0, 2, 3, 1)					)
						_lowercase					:       List[Any]       =		self.conv_in(__lowerCAmelCase					)

						# 3. down
						_lowercase					:       Any       =		(sample,)
						for down_block in self.down_blocks:
							if isinstance(__lowerCAmelCase     ,  __lowerCAmelCase					):
								_lowercase  ,    _lowercase					:       Optional[int]       =		down_block(__lowerCAmelCase     ,  __lowerCAmelCase     ,  __lowerCAmelCase     ,  deterministic=not train					)
							else:
								_lowercase  ,    _lowercase					:       Optional[Any]       =		down_block(__lowerCAmelCase     ,  __lowerCAmelCase     ,  deterministic=not train					)
							down_block_res_samples += res_samples

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

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

							_lowercase					:       Dict       =		new_down_block_res_samples

						# 4. mid
						_lowercase					:       List[Any]       =		self.mid_block(__lowerCAmelCase     ,  __lowerCAmelCase     ,  __lowerCAmelCase     ,  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:
							_lowercase					:       Optional[Any]       =		down_block_res_samples[-(self.layers_per_block + 1) :]
							_lowercase					:       Tuple       =		down_block_res_samples[: -(self.layers_per_block + 1)]
							if isinstance(__lowerCAmelCase     ,  __lowerCAmelCase					):
								_lowercase					:       str       =		up_block(
								    __lowerCAmelCase     ,  temb=__lowerCAmelCase     ,  encoder_hidden_states=__lowerCAmelCase     ,  res_hidden_states_tuple=__lowerCAmelCase     ,  deterministic=not train     ,  )
							else:
								_lowercase					:       Optional[Any]       =		up_block(__lowerCAmelCase     ,  temb=__lowerCAmelCase     ,  res_hidden_states_tuple=__lowerCAmelCase     ,  deterministic=not train					)

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

						if not return_dict:
							return (sample,)

						return FlaxUNetaDConditionOutput(sample=__lowerCAmelCase					)





'''simple docstring'''

_A				:    Optional[Any]     						='''ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/'''







def 						__UpperCamelCase				(							_lowercase					) ->			bytes:
	# Make sure the supplied data is a bytes-like object
	if not isinstance(_lowercase,				_lowercase					):
		_lowercase					:       Union[str, Any]       =		f'''a bytes-like object is required, not \'{data.__class__.__name__}\''''
		raise TypeError(_lowercase					)

	_lowercase					:       int       =		''.join(bin(_lowercase					)[2:].zfill(8					) for byte in data					)

	_lowercase					:       Dict       =		len(_lowercase					) % 6 != 0

	if padding_needed:
		# The padding that will be added later
		_lowercase					:       Optional[Any]       =		B'=' * ((6 - len(_lowercase					) % 6) // 2)

		# Append binary_stream with arbitrary binary digits (0's by default) to make its
		# length a multiple of 6.
		binary_stream += "0" * (6 - len(_lowercase					) % 6)
	else:
		_lowercase					:       Optional[int]       =		B''

	# Encode every 6 binary digits to their corresponding Base64 character
	return (
	    "".join(
	        B64_CHARSET[int(binary_stream[index : index + 6],				2					)]
	        for index in range(0,				len(_lowercase					),				6					)					).encode()
	    + padding
	)







def 						__UpperCamelCase				(							_lowercase					) ->			bytes:
	# Make sure encoded_data is either a string or a bytes-like object
	if not isinstance(_lowercase,				_lowercase					) and not isinstance(_lowercase,				_lowercase					):
		_lowercase					:       int       =		(
		    'argument should be a bytes-like object or ASCII string, '
		    f'''not \'{encoded_data.__class__.__name__}\''''
		)
		raise TypeError(_lowercase					)

	# In case encoded_data is a bytes-like object, make sure it contains only
	# ASCII characters so we convert it to a string object
	if isinstance(_lowercase,				_lowercase					):
		try:
			_lowercase					:       Optional[int]       =		encoded_data.decode('utf-8'					)
		except UnicodeDecodeError:
			raise ValueError('base64 encoded data should only contain ASCII characters'					)

	_lowercase					:       Optional[int]       =		encoded_data.count('='					)

	# Check if the encoded string contains non base64 characters
	if padding:
		assert all(
		    char in B64_CHARSET for char in encoded_data[:-padding]					), "Invalid base64 character(s) found."
	else:
		assert all(
		    char in B64_CHARSET for char in encoded_data					), "Invalid base64 character(s) found."

	# Check the padding
	assert len(_lowercase					) % 4 == 0 and padding < 3, "Incorrect padding"

	if padding:
		# Remove padding if there is one
		_lowercase					:       str       =		encoded_data[:-padding]

		_lowercase					:       Tuple       =		''.join(
		    bin(B64_CHARSET.index(_lowercase					)					)[2:].zfill(6					) for char in encoded_data					)[: -padding * 2]
	else:
		_lowercase					:       Union[str, Any]       =		''.join(
		    bin(B64_CHARSET.index(_lowercase					)					)[2:].zfill(6					) for char in encoded_data					)

	_lowercase					:       List[str]       =		[
	    int(binary_stream[index : index + 8],				2					)
	    for index in range(0,				len(_lowercase					),				8					)
	]

	return bytes(_lowercase					)


if __name__ == "__main__":
		import doctest

		doctest.testmod()
