


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






@flax.struct.dataclass
class      A(  lowerCamelCase__		):

		"""simple docstring"""





		A    =							42
		A    =							42






class      A(  nn.Module		):

		"""simple docstring"""





		A    =							42
		A    =							(16, 32, 96, 2_56)
		A    =							jnp.floataa
		def   _UpperCamelCase( self   )      ->	Optional[int]:

					"""simple docstring"""



					_UpperCamelCase					:int					     =	nn.Conv(
					    self.block_out_channels[0]					,				kernel_size=(3, 3)					,				padding=((1, 1), (1, 1))					,				dtype=self.dtype					,				)

					_UpperCamelCase					:List[str]					     =	[]
					for i in range(len(self.block_out_channels   ) - 1   ):
								_UpperCamelCase					:List[str]					     =	self.block_out_channels[i]
								_UpperCamelCase					:Union[str, Any]					     =	self.block_out_channels[i + 1]
								_UpperCamelCase					:Any					     =	nn.Conv(
								    SCREAMING_SNAKE_CASE__					,				kernel_size=(3, 3)					,				padding=((1, 1), (1, 1))					,				dtype=self.dtype					,				)
								blocks.append(SCREAMING_SNAKE_CASE__   )
								_UpperCamelCase					:int					     =	nn.Conv(
								    SCREAMING_SNAKE_CASE__					,				kernel_size=(3, 3)					,				strides=(2, 2)					,				padding=((1, 1), (1, 1))					,				dtype=self.dtype					,				)
								blocks.append(SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:List[Any]					     =	blocks

					_UpperCamelCase					:List[str]					     =	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					,				SCREAMING_SNAKE_CASE__   )      ->	List[Any]:

					"""simple docstring"""



					_UpperCamelCase					:Tuple					     =	self.conv_in(SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:Optional[int]					     =	nn.silu(SCREAMING_SNAKE_CASE__   )

					for block in self.blocks:
								_UpperCamelCase					:List[str]					     =	block(SCREAMING_SNAKE_CASE__   )
								_UpperCamelCase					:Dict					     =	nn.silu(SCREAMING_SNAKE_CASE__   )

					_UpperCamelCase					:List[str]					     =	self.conv_out(SCREAMING_SNAKE_CASE__   )

					return embedding






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

		"""simple docstring"""





		A    =							32
		A    =							4
		A    =							(
		    "CrossAttnDownBlock2D",
		    "CrossAttnDownBlock2D",
		    "CrossAttnDownBlock2D",
		    "DownBlock2D",
		)
		A    =							False
		A    =							(3_20, 6_40, 12_80, 12_80)
		A    =							2
		A    =							8
		A    =							None
		A    =							12_80
		A    =							0.0
		A    =							False
		A    =							jnp.floataa
		A    =							True
		A    =							0
		A    =							"rgb"
		A    =							(16, 32, 96, 2_56)
		def   _UpperCamelCase( self					,				SCREAMING_SNAKE_CASE__   )      ->	FrozenDict:

					"""simple docstring"""



					_UpperCamelCase					:List[Any]					     =	(1, self.in_channels, self.sample_size, self.sample_size)
					_UpperCamelCase					:Any					     =	jnp.zeros(SCREAMING_SNAKE_CASE__					,				dtype=jnp.floataa   )
					_UpperCamelCase					:List[str]					     =	jnp.ones((1,)					,				dtype=jnp.intaa   )
					_UpperCamelCase					:Tuple					     =	jnp.zeros((1, 1, self.cross_attention_dim)					,				dtype=jnp.floataa   )
					_UpperCamelCase					:List[str]					     =	(1, 3, self.sample_size * 8, self.sample_size * 8)
					_UpperCamelCase					:Dict					     =	jnp.zeros(SCREAMING_SNAKE_CASE__					,				dtype=jnp.floataa   )

					_UpperCamelCase     ,      _UpperCamelCase					:Tuple					     =	jax.random.split(SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:Union[str, Any]					     =	{'''params''': params_rng, '''dropout''': dropout_rng}

					return self.init(SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__   )["params"]
		def   _UpperCamelCase( self   )      ->	Dict:

					"""simple docstring"""



					_UpperCamelCase					:Union[str, Any]					     =	self.block_out_channels
					_UpperCamelCase					:Tuple					     =	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.
					_UpperCamelCase					:int					     =	self.num_attention_heads or self.attention_head_dim

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

					# time
					_UpperCamelCase					:Optional[Any]					     =	FlaxTimesteps(
					    block_out_channels[0]					,				flip_sin_to_cos=self.flip_sin_to_cos					,				freq_shift=self.config.freq_shift   )
					_UpperCamelCase					:Any					     =	FlaxTimestepEmbedding(SCREAMING_SNAKE_CASE__					,				dtype=self.dtype   )

					_UpperCamelCase					:str					     =	FlaxControlNetConditioningEmbedding(
					    conditioning_embedding_channels=block_out_channels[0]					,				block_out_channels=self.conditioning_embedding_out_channels					,				)

					_UpperCamelCase					:Dict					     =	self.only_cross_attention
					if isinstance(SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__   ):
								_UpperCamelCase					:str					     =	(only_cross_attention,) * len(self.down_block_types   )

					if isinstance(SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__   ):
								_UpperCamelCase					:Union[str, Any]					     =	(num_attention_heads,) * len(self.down_block_types   )

					# down
					_UpperCamelCase					:Tuple					     =	[]
					_UpperCamelCase					:Optional[int]					     =	[]

					_UpperCamelCase					:str					     =	block_out_channels[0]

					_UpperCamelCase					:Tuple					     =	nn.Conv(
					    SCREAMING_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(SCREAMING_SNAKE_CASE__   )

					for i, down_block_type in enumerate(self.down_block_types   ):
								_UpperCamelCase					:List[str]					     =	output_channel
								_UpperCamelCase					:str					     =	block_out_channels[i]
								_UpperCamelCase					:List[Any]					     =	i == len(SCREAMING_SNAKE_CASE__   ) - 1

								if down_block_type == "CrossAttnDownBlock2D":
											_UpperCamelCase					:int					     =	FlaxCrossAttnDownBlockaD(
											    in_channels=SCREAMING_SNAKE_CASE__					,				out_channels=SCREAMING_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:
											_UpperCamelCase					:Tuple					     =	FlaxDownBlockaD(
											    in_channels=SCREAMING_SNAKE_CASE__					,				out_channels=SCREAMING_SNAKE_CASE__					,				dropout=self.dropout					,				num_layers=self.layers_per_block					,				add_downsample=not is_final_block					,				dtype=self.dtype					,				)

								down_blocks.append(SCREAMING_SNAKE_CASE__   )

								for _ in range(self.layers_per_block   ):
											_UpperCamelCase					:Dict					     =	nn.Conv(
											    SCREAMING_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(SCREAMING_SNAKE_CASE__   )

								if not is_final_block:
											_UpperCamelCase					:Union[str, Any]					     =	nn.Conv(
											    SCREAMING_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(SCREAMING_SNAKE_CASE__   )

					_UpperCamelCase					:int					     =	down_blocks
					_UpperCamelCase					:List[Any]					     =	controlnet_down_blocks

					# mid
					_UpperCamelCase					:Union[str, Any]					     =	block_out_channels[-1]
					_UpperCamelCase					:Optional[Any]					     =	FlaxUNetMidBlockaDCrossAttn(
					    in_channels=SCREAMING_SNAKE_CASE__					,				dropout=self.dropout					,				num_attention_heads=num_attention_heads[-1]					,				use_linear_projection=self.use_linear_projection					,				dtype=self.dtype					,				)

					_UpperCamelCase					:Tuple					     =	nn.Conv(
					    SCREAMING_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					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__ = 1.0					,				SCREAMING_SNAKE_CASE__ = True					,				SCREAMING_SNAKE_CASE__ = False					,				)      ->	Union[FlaxControlNetOutput, Tuple]:

					"""simple docstring"""



					_UpperCamelCase					:List[Any]					     =	self.controlnet_conditioning_channel_order
					if channel_order == "bgr":
								_UpperCamelCase					:Union[str, Any]					     =	jnp.flip(SCREAMING_SNAKE_CASE__					,				axis=1   )

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

					_UpperCamelCase					:Any					     =	self.time_proj(SCREAMING_SNAKE_CASE__   )
					_UpperCamelCase					:Optional[Any]					     =	self.time_embedding(SCREAMING_SNAKE_CASE__   )

					# 2. pre-process
					_UpperCamelCase					:Optional[int]					     =	jnp.transpose(SCREAMING_SNAKE_CASE__					,				(0, 2, 3, 1)   )
					_UpperCamelCase					:Tuple					     =	self.conv_in(SCREAMING_SNAKE_CASE__   )

					_UpperCamelCase					:str					     =	jnp.transpose(SCREAMING_SNAKE_CASE__					,				(0, 2, 3, 1)   )
					_UpperCamelCase					:List[Any]					     =	self.controlnet_cond_embedding(SCREAMING_SNAKE_CASE__   )
					sample += controlnet_cond

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

					# 4. mid
					_UpperCamelCase					:Dict					     =	self.mid_block(SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__					,				deterministic=not train   )

					# 5. contronet blocks
					_UpperCamelCase					:Union[str, Any]					     =	()
					for down_block_res_sample, controlnet_block in zip(SCREAMING_SNAKE_CASE__					,				self.controlnet_down_blocks   ):
								_UpperCamelCase					:Tuple					     =	controlnet_block(SCREAMING_SNAKE_CASE__   )
								controlnet_down_block_res_samples += (down_block_res_sample,)

					_UpperCamelCase					:List[str]					     =	controlnet_down_block_res_samples

					_UpperCamelCase					:Tuple					     =	self.controlnet_mid_block(SCREAMING_SNAKE_CASE__   )

					# 6. scaling
					_UpperCamelCase					:Optional[int]					     =	[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=SCREAMING_SNAKE_CASE__					,				mid_block_res_sample=SCREAMING_SNAKE_CASE__   )





"""simple docstring"""






import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
					import torch

if is_vision_available():
					from PIL import Image

					from transformers import MobileNetVaImageProcessor






class      A(  unittest.TestCase		):

		"""simple docstring"""





		def __init__( self					,				SCREAMING_SNAKE_CASE__					,				SCREAMING_SNAKE_CASE__=7					,				SCREAMING_SNAKE_CASE__=3					,				SCREAMING_SNAKE_CASE__=18					,				SCREAMING_SNAKE_CASE__=30					,				SCREAMING_SNAKE_CASE__=4_00					,				SCREAMING_SNAKE_CASE__=True					,				SCREAMING_SNAKE_CASE__=None					,				SCREAMING_SNAKE_CASE__=True					,				SCREAMING_SNAKE_CASE__=None					,				)      ->	Dict:

					"""simple docstring"""



					_UpperCamelCase					:Dict					     =	size if size is not None else {'''shortest_edge''': 20}
					_UpperCamelCase					:Union[str, Any]					     =	crop_size if crop_size is not None else {'''height''': 18, '''width''': 18}
					_UpperCamelCase					:str					     =	parent
					_UpperCamelCase					:str					     =	batch_size
					_UpperCamelCase					:str					     =	num_channels
					_UpperCamelCase					:Optional[Any]					     =	image_size
					_UpperCamelCase					:List[str]					     =	min_resolution
					_UpperCamelCase					:Tuple					     =	max_resolution
					_UpperCamelCase					:List[Any]					     =	do_resize
					_UpperCamelCase					:Tuple					     =	size
					_UpperCamelCase					:int					     =	do_center_crop
					_UpperCamelCase					:Optional[int]					     =	crop_size





		def   _UpperCamelCase( self   )      ->	int:

					"""simple docstring"""



					return {
					    "do_resize": self.do_resize,
					    "size": self.size,
					    "do_center_crop": self.do_center_crop,
					    "crop_size": self.crop_size,
					}






@require_torch
@require_vision
class      A(  lowerCamelCase__				,							unittest.TestCase		):

		"""simple docstring"""





		A    =							MobileNetVaImageProcessor if is_vision_available() else None
		def   _UpperCamelCase( self   )      ->	List[Any]:

					"""simple docstring"""



					_UpperCamelCase					:int					     =	MobileNetVaImageProcessingTester(self   )
		@property
		def   _UpperCamelCase( self   )      ->	List[str]:

					"""simple docstring"""



					return self.image_processor_tester.prepare_image_processor_dict()
		def   _UpperCamelCase( self   )      ->	List[Any]:

					"""simple docstring"""



					_UpperCamelCase					:Union[str, Any]					     =	self.image_processing_class(**self.image_processor_dict   )
					self.assertTrue(hasattr(SCREAMING_SNAKE_CASE__					,				'''do_resize'''   )   )
					self.assertTrue(hasattr(SCREAMING_SNAKE_CASE__					,				'''size'''   )   )
					self.assertTrue(hasattr(SCREAMING_SNAKE_CASE__					,				'''do_center_crop'''   )   )
					self.assertTrue(hasattr(SCREAMING_SNAKE_CASE__					,				'''crop_size'''   )   )
		def   _UpperCamelCase( self   )      ->	Dict:

					"""simple docstring"""



					_UpperCamelCase					:List[str]					     =	self.image_processing_class.from_dict(self.image_processor_dict   )
					self.assertEqual(image_processor.size					,				{'''shortest_edge''': 20}   )
					self.assertEqual(image_processor.crop_size					,				{'''height''': 18, '''width''': 18}   )

					_UpperCamelCase					:List[Any]					     =	self.image_processing_class.from_dict(self.image_processor_dict					,				size=42					,				crop_size=84   )
					self.assertEqual(image_processor.size					,				{'''shortest_edge''': 42}   )
					self.assertEqual(image_processor.crop_size					,				{'''height''': 84, '''width''': 84}   )
		def   _UpperCamelCase( self   )      ->	List[Any]:

					"""simple docstring"""



					pass
		def   _UpperCamelCase( self   )      ->	Dict:

					"""simple docstring"""



					_UpperCamelCase					:Any					     =	self.image_processing_class(**self.image_processor_dict   )
					# create random PIL images
					_UpperCamelCase					:Union[str, Any]					     =	prepare_image_inputs(self.image_processor_tester					,				equal_resolution=SCREAMING_SNAKE_CASE__   )
					for image in image_inputs:
								self.assertIsInstance(SCREAMING_SNAKE_CASE__					,				Image.Image   )

					# Test not batched input
					_UpperCamelCase					:Any					     =	image_processing(image_inputs[0]					,				return_tensors='''pt'''   ).pixel_values
					self.assertEqual(
					    encoded_images.shape					,				(
					        1,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size['''height'''],
					        self.image_processor_tester.crop_size['''width'''],
					    )					,				)

					# Test batched
					_UpperCamelCase					:Any					     =	image_processing(SCREAMING_SNAKE_CASE__					,				return_tensors='''pt'''   ).pixel_values
					self.assertEqual(
					    encoded_images.shape					,				(
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size['''height'''],
					        self.image_processor_tester.crop_size['''width'''],
					    )					,				)
		def   _UpperCamelCase( self   )      ->	Tuple:

					"""simple docstring"""



					_UpperCamelCase					:int					     =	self.image_processing_class(**self.image_processor_dict   )
					# create random numpy tensors
					_UpperCamelCase					:Any					     =	prepare_image_inputs(self.image_processor_tester					,				equal_resolution=SCREAMING_SNAKE_CASE__					,				numpify=SCREAMING_SNAKE_CASE__   )
					for image in image_inputs:
								self.assertIsInstance(SCREAMING_SNAKE_CASE__					,				np.ndarray   )

					# Test not batched input
					_UpperCamelCase					:Any					     =	image_processing(image_inputs[0]					,				return_tensors='''pt'''   ).pixel_values
					self.assertEqual(
					    encoded_images.shape					,				(
					        1,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size['''height'''],
					        self.image_processor_tester.crop_size['''width'''],
					    )					,				)

					# Test batched
					_UpperCamelCase					:List[str]					     =	image_processing(SCREAMING_SNAKE_CASE__					,				return_tensors='''pt'''   ).pixel_values
					self.assertEqual(
					    encoded_images.shape					,				(
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size['''height'''],
					        self.image_processor_tester.crop_size['''width'''],
					    )					,				)





		def   _UpperCamelCase( self   )      ->	Any:

					"""simple docstring"""



					_UpperCamelCase					:List[Any]					     =	self.image_processing_class(**self.image_processor_dict   )
					# create random PyTorch tensors
					_UpperCamelCase					:Union[str, Any]					     =	prepare_image_inputs(self.image_processor_tester					,				equal_resolution=SCREAMING_SNAKE_CASE__					,				torchify=SCREAMING_SNAKE_CASE__   )
					for image in image_inputs:
								self.assertIsInstance(SCREAMING_SNAKE_CASE__					,				torch.Tensor   )

					# Test not batched input
					_UpperCamelCase					:Union[str, Any]					     =	image_processing(image_inputs[0]					,				return_tensors='''pt'''   ).pixel_values
					self.assertEqual(
					    encoded_images.shape					,				(
					        1,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size['''height'''],
					        self.image_processor_tester.crop_size['''width'''],
					    )					,				)

					# Test batched
					_UpperCamelCase					:Optional[int]					     =	image_processing(SCREAMING_SNAKE_CASE__					,				return_tensors='''pt'''   ).pixel_values
					self.assertEqual(
					    encoded_images.shape					,				(
					        self.image_processor_tester.batch_size,
					        self.image_processor_tester.num_channels,
					        self.image_processor_tester.crop_size['''height'''],
					        self.image_processor_tester.crop_size['''width'''],
					    )					,				)

