
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,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
)







@flax.struct.dataclass
class 			A_  (     __lowerCamelCase  ):




	'''simple docstring'''


	_UpperCamelCase	:	jnp.ndarray
	_UpperCamelCase	:	jnp.ndarray







class 			A_  (     nn.Module  ):




	'''simple docstring'''


	_UpperCamelCase	:	int
	_UpperCamelCase	:	Tuple[int]									=			(16, 32, 96, 256)
	_UpperCamelCase	:	jnp.dtype									=			jnp.floataa





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       nn.Conv(
		    self.block_out_channels[0]						,       kernel_size=(3, 3)						,       padding=((1, 1), (1, 1))						,       dtype=self.dtype						,       )

		lowercase	    =       []
		for i in range(len(self.block_out_channels			) - 1			):
			lowercase	    =       self.block_out_channels[i]
			lowercase	    =       self.block_out_channels[i + 1]
			lowercase	    =       nn.Conv(
			    snake_case						,       kernel_size=(3, 3)						,       padding=((1, 1), (1, 1))						,       dtype=self.dtype						,       )
			blocks.append(snake_case			)
			lowercase	    =       nn.Conv(
			    snake_case						,       kernel_size=(3, 3)						,       strides=(2, 2)						,       padding=((1, 1), (1, 1))						,       dtype=self.dtype						,       )
			blocks.append(snake_case			)
		lowercase	    =       blocks

		lowercase	    =       nn.Conv(
		    self.conditioning_embedding_channels						,       kernel_size=(3, 3)						,       padding=((1, 1), (1, 1))						,       kernel_init=nn.initializers.zeros_init()						,       bias_init=nn.initializers.zeros_init()						,       dtype=self.dtype						,       )





	def __call__(			self						,       snake_case			):
		lowercase	    =       self.conv_in(snake_case			)
		lowercase	    =       nn.silu(snake_case			)

		for block in self.blocks:
			lowercase	    =       block(snake_case			)
			lowercase	    =       nn.silu(snake_case			)

		lowercase	    =       self.conv_out(snake_case			)

		return embedding







@flax_register_to_config
class 			A_  (     nn.Module ,      __lowerCamelCase ,      __lowerCamelCase  ):




	'''simple docstring'''


	_UpperCamelCase	:	int									=			32
	_UpperCamelCase	:	int									=			4
	_UpperCamelCase	:	Tuple[str]									=			(
	    "CrossAttnDownBlock2D",
	    "CrossAttnDownBlock2D",
	    "CrossAttnDownBlock2D",
	    "DownBlock2D",
	)
	_UpperCamelCase	:	Union[bool, Tuple[bool]]									=			False
	_UpperCamelCase	:	Tuple[int]									=			(320, 640, 1280, 1280)
	_UpperCamelCase	:	int									=			2
	_UpperCamelCase	:	Union[int, Tuple[int]]									=			8
	_UpperCamelCase	:	Optional[Union[int, Tuple[int]]]									=			None
	_UpperCamelCase	:	int									=			1280
	_UpperCamelCase	:	float									=			0.0
	_UpperCamelCase	:	bool									=			False
	_UpperCamelCase	:	jnp.dtype									=			jnp.floataa
	_UpperCamelCase	:	bool									=			True
	_UpperCamelCase	:	int									=			0
	_UpperCamelCase	:	str									=			"rgb"
	_UpperCamelCase	:	Tuple[int]									=			(16, 32, 96, 256)





	def       SCREAMING_SNAKE_CASE__		(			self						,       snake_case			):
		# init input tensors
		lowercase	    =       (1, self.in_channels, self.sample_size, self.sample_size)
		lowercase	    =       jnp.zeros(snake_case						,       dtype=jnp.floataa			)
		lowercase	    =       jnp.ones((1,)						,       dtype=jnp.intaa			)
		lowercase	    =       jnp.zeros((1, 1, self.cross_attention_dim)						,       dtype=jnp.floataa			)
		lowercase	    =       (1, 3, self.sample_size * 8, self.sample_size * 8)
		lowercase	    =       jnp.zeros(snake_case						,       dtype=jnp.floataa			)

		lowercase    ,   lowercase	    =       jax.random.split(snake_case			)
		lowercase	    =       {'params': params_rng, 'dropout': dropout_rng}

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





	def       SCREAMING_SNAKE_CASE__		(			self			):
		lowercase	    =       self.block_out_channels
		lowercase	    =       block_out_channels[0] * 4

		# 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	    =       self.num_attention_heads or self.attention_head_dim

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

		# time
		lowercase	    =       FlaxTimesteps(
		    block_out_channels[0]						,       flip_sin_to_cos=self.flip_sin_to_cos						,       freq_shift=self.config.freq_shift			)
		lowercase	    =       FlaxTimestepEmbedding(snake_case						,       dtype=self.dtype			)

		lowercase	    =       FlaxControlNetConditioningEmbedding(
		    conditioning_embedding_channels=block_out_channels[0]						,       block_out_channels=self.conditioning_embedding_out_channels						,       )

		lowercase	    =       self.only_cross_attention
		if isinstance(snake_case						,       snake_case			):
			lowercase	    =       (only_cross_attention,) * len(self.down_block_types			)

		if isinstance(snake_case						,       snake_case			):
			lowercase	    =       (num_attention_heads,) * len(self.down_block_types			)

		# down
		lowercase	    =       []
		lowercase	    =       []

		lowercase	    =       block_out_channels[0]

		lowercase	    =       nn.Conv(
		    snake_case						,       kernel_size=(1, 1)						,       padding='VALID'						,       kernel_init=nn.initializers.zeros_init()						,       bias_init=nn.initializers.zeros_init()						,       dtype=self.dtype						,       )
		controlnet_down_blocks.append(snake_case			)

		for i, down_block_type in enumerate(self.down_block_types			):
			lowercase	    =       output_channel
			lowercase	    =       block_out_channels[i]
			lowercase	    =       i == len(snake_case			) - 1

			if down_block_type == "CrossAttnDownBlock2D":
				lowercase	    =       FlaxCrossAttnDownBlockaD(
				    in_channels=snake_case						,       out_channels=snake_case						,       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]						,       dtype=self.dtype						,       )
			else:
				lowercase	    =       FlaxDownBlockaD(
				    in_channels=snake_case						,       out_channels=snake_case						,       dropout=self.dropout						,       num_layers=self.layers_per_block						,       add_downsample=not is_final_block						,       dtype=self.dtype						,       )

			down_blocks.append(snake_case			)

			for _ in range(self.layers_per_block			):
				lowercase	    =       nn.Conv(
				    snake_case						,       kernel_size=(1, 1)						,       padding='VALID'						,       kernel_init=nn.initializers.zeros_init()						,       bias_init=nn.initializers.zeros_init()						,       dtype=self.dtype						,       )
				controlnet_down_blocks.append(snake_case			)

			if not is_final_block:
				lowercase	    =       nn.Conv(
				    snake_case						,       kernel_size=(1, 1)						,       padding='VALID'						,       kernel_init=nn.initializers.zeros_init()						,       bias_init=nn.initializers.zeros_init()						,       dtype=self.dtype						,       )
				controlnet_down_blocks.append(snake_case			)

		lowercase	    =       down_blocks
		lowercase	    =       controlnet_down_blocks

		# mid
		lowercase	    =       block_out_channels[-1]
		lowercase	    =       FlaxUNetMidBlockaDCrossAttn(
		    in_channels=snake_case						,       dropout=self.dropout						,       num_attention_heads=num_attention_heads[-1]						,       use_linear_projection=self.use_linear_projection						,       dtype=self.dtype						,       )

		lowercase	    =       nn.Conv(
		    snake_case						,       kernel_size=(1, 1)						,       padding='VALID'						,       kernel_init=nn.initializers.zeros_init()						,       bias_init=nn.initializers.zeros_init()						,       dtype=self.dtype						,       )





	def __call__(			self						,       snake_case						,       snake_case						,       snake_case						,       snake_case						,       snake_case = 1.0						,       snake_case = True						,       snake_case = False						,       ):
		lowercase	    =       self.controlnet_conditioning_channel_order
		if channel_order == "bgr":
			lowercase	    =       jnp.flip(snake_case						,       axis=1			)

		# 1. time
		if not isinstance(snake_case						,       jnp.ndarray			):
			lowercase	    =       jnp.array([timesteps]						,       dtype=jnp.intaa			)
		elif isinstance(snake_case						,       jnp.ndarray			) and len(timesteps.shape			) == 0:
			lowercase	    =       timesteps.astype(dtype=jnp.floataa			)
			lowercase	    =       jnp.expand_dims(snake_case						,       0			)

		lowercase	    =       self.time_proj(snake_case			)
		lowercase	    =       self.time_embedding(snake_case			)

		# 2. pre-process
		lowercase	    =       jnp.transpose(snake_case						,       (0, 2, 3, 1)			)
		lowercase	    =       self.conv_in(snake_case			)

		lowercase	    =       jnp.transpose(snake_case						,       (0, 2, 3, 1)			)
		lowercase	    =       self.controlnet_cond_embedding(snake_case			)
		sample += controlnet_cond

		# 3. down
		lowercase	    =       (sample,)
		for down_block in self.down_blocks:
			if isinstance(snake_case						,       snake_case			):
				lowercase    ,   lowercase	    =       down_block(snake_case						,       snake_case						,       snake_case						,       deterministic=not train			)
			else:
				lowercase    ,   lowercase	    =       down_block(snake_case						,       snake_case						,       deterministic=not train			)
			down_block_res_samples += res_samples

		# 4. mid
		lowercase	    =       self.mid_block(snake_case						,       snake_case						,       snake_case						,       deterministic=not train			)

		# 5. contronet blocks
		lowercase	    =       ()
		for down_block_res_sample, controlnet_block in zip(snake_case						,       self.controlnet_down_blocks			):
			lowercase	    =       controlnet_block(snake_case			)
			controlnet_down_block_res_samples += (down_block_res_sample,)

		lowercase	    =       controlnet_down_block_res_samples

		lowercase	    =       self.controlnet_mid_block(snake_case			)

		# 6. scaling
		lowercase	    =       [sample * conditioning_scale for sample in down_block_res_samples]
		mid_block_res_sample *= conditioning_scale

		if not return_dict:
			return (down_block_res_samples, mid_block_res_sample)

		return FlaxControlNetOutput(
		    down_block_res_samples=snake_case						,       mid_block_res_sample=snake_case			)









"""simple docstring"""
import timeit

import numpy as np

import datasets
from datasets.arrow_writer import ArrowWriter
from datasets.features.features import _ArrayXD

def 							snake_case (  A__		):
			def wrapper(*A__      ,**A__		):
						UpperCAmelCase_	:					Union[str, Any]     	=   timeit.default_timer()
						UpperCAmelCase_	:					Union[str, Any]     	=   func(*A__      ,**A__		)
						UpperCAmelCase_	:					Dict     	=   timeit.default_timer() - starttime
						return delta

			UpperCAmelCase_	:					Optional[Any]     	=   func.__name__

			return wrapper

def 							snake_case (  A__      ,A__=1_00      ,A__=None		):
			UpperCAmelCase_	:					str     	=   []
			UpperCAmelCase_	:					List[str]     	=   seq_shapes or {}
			for i in range(A__		):
						UpperCAmelCase_	:					int     	=   {}
						for col_id, (k, v) in enumerate(features.items()		):
									if isinstance(A__      ,_ArrayXD		):
												UpperCAmelCase_	:					Optional[int]     	=   np.random.rand(*v.shape		).astype(v.dtype		)
									elif isinstance(A__      ,datasets.Value		):
												if v.dtype == "string":
															UpperCAmelCase_	:					List[str]     	=   "The small grey turtle was surprisingly fast when challenged."
												else:
															UpperCAmelCase_	:					List[str]     	=   np.random.randint(10      ,size=1		).astype(v.dtype		).item()
									elif isinstance(A__      ,datasets.Sequence		):
												while isinstance(A__      ,datasets.Sequence		):
															UpperCAmelCase_	:					List[str]     	=   v.feature
												UpperCAmelCase_	:					List[Any]     	=   seq_shapes[k]
												UpperCAmelCase_	:					Dict     	=   np.random.rand(*A__		).astype(v.dtype		)
									UpperCAmelCase_	:					Dict     	=   data

						dummy_data.append((i, example)		)

			return dummy_data




def 							snake_case (  A__      ,A__      ,A__=1_00      ,A__=None		):
			UpperCAmelCase_	:					Optional[Any]     	=   generate_examples(A__      ,num_examples=A__      ,seq_shapes=A__		)

			with ArrowWriter(features=A__      ,path=A__		) as writer:
						for key, record in dummy_data:
									UpperCAmelCase_	:					Any     	=   features.encode_example(A__		)
									writer.write(A__		)

						UpperCAmelCase_	,      UpperCAmelCase_	:					Optional[int]     	=   writer.finalize()

			if not num_final_examples == num_examples:
						raise ValueError(
						    F"""Error writing the dataset, wrote {num_final_examples} examples but should have written {num_examples}."""		)

			UpperCAmelCase_	:					List[Any]     	=   datasets.Dataset.from_file(filename=A__      ,info=datasets.DatasetInfo(features=A__		)		)

			return dataset



