



'''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 	lowerCamelCase_    ( UpperCAmelCase__						):






						'''simple docstring'''
						a__   =  42
						a__   =  42







class 	lowerCamelCase_    ( nn.Module						):






						'''simple docstring'''
						a__   =  42
						a__   =  (16, 32, 96, 256)
						a__   =  jnp.floataa







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Tuple )      ->				Optional[int]:
									A							:					Optional[Any]              =						nn.Conv(
									    self.block_out_channels[0]	,	kernel_size=(3, 3)	,	padding=((1, 1), (1, 1))	,	dtype=self.dtype	,	)

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

									A							:					Optional[Any]              =						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      :			Optional[Any]	,	__lowerCamelCase      :			Tuple )      ->				int:
									A							:					Union[str, Any]              =						self.conv_in(lowerCamelCase__ )
									A							:					Optional[Any]              =						nn.silu(lowerCamelCase__ )

									for block in self.blocks:
												A							:					str              =						block(lowerCamelCase__ )
												A							:					Union[str, Any]              =						nn.silu(lowerCamelCase__ )

									A							:					Union[str, Any]              =						self.conv_out(lowerCamelCase__ )

									return embedding







@flax_register_to_config
class 	lowerCamelCase_    ( nn.Module						,UpperCAmelCase__						,UpperCAmelCase__						):






						'''simple docstring'''
						a__   =  32
						a__   =  4
						a__   =  (
						    "CrossAttnDownBlock2D",
						    "CrossAttnDownBlock2D",
						    "CrossAttnDownBlock2D",
						    "DownBlock2D",
						)
						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__   =  "rgb"
						a__   =  (16, 32, 96, 256)







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Tuple	,	__lowerCamelCase      :			jax.random.KeyArray )      ->				FrozenDict:
									A							:					str              =						(1, self.in_channels, self.sample_size, self.sample_size)
									A							:					Optional[Any]              =						jnp.zeros(lowerCamelCase__	,	dtype=jnp.floataa )
									A							:					Tuple              =						jnp.ones((1,)	,	dtype=jnp.intaa )
									A							:					Any              =						jnp.zeros((1, 1, self.cross_attention_dim)	,	dtype=jnp.floataa )
									A							:					Optional[Any]              =						(1, 3, self.sample_size * 8, self.sample_size * 8)
									A							:					int              =						jnp.zeros(lowerCamelCase__	,	dtype=jnp.floataa )

									A							:					List[Any]              =						jax.random.split(lowerCamelCase__ )
									A							:					Union[str, Any]              =						{"params": params_rng, "dropout": dropout_rng}

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







						def 			SCREAMING_SNAKE_CASE__		(	self      :			Dict )      ->				List[Any]:
									A							:					str              =						self.block_out_channels
									A							:					List[Any]              =						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.
									A							:					Union[str, Any]              =						self.num_attention_heads or self.attention_head_dim

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

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

									A							:					Union[str, Any]              =						FlaxControlNetConditioningEmbedding(
									    conditioning_embedding_channels=block_out_channels[0]	,	block_out_channels=self.conditioning_embedding_out_channels	,	)

									A							:					int              =						self.only_cross_attention
									if isinstance(lowerCamelCase__	,	lowerCamelCase__ ):
												A							:					str              =						(only_cross_attention,) * len(self.down_block_types )

									if isinstance(lowerCamelCase__	,	lowerCamelCase__ ):
												A							:					Optional[int]              =						(num_attention_heads,) * len(self.down_block_types )

									# down
									A							:					Dict              =						[]
									A							:					List[str]              =						[]

									A							:					Dict              =						block_out_channels[0]

									A							:					List[Any]              =						nn.Conv(
									    lowerCamelCase__	,	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(lowerCamelCase__ )

									for i, down_block_type in enumerate(self.down_block_types ):
												A							:					Dict              =						output_channel
												A							:					Optional[Any]              =						block_out_channels[i]
												A							:					int              =						i == len(lowerCamelCase__ ) - 1

												if down_block_type == "CrossAttnDownBlock2D":
															A							:					Union[str, 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]	,	dtype=self.dtype	,	)
												else:
															A							:					Union[str, 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__ )

												for _ in range(self.layers_per_block ):
															A							:					Dict              =						nn.Conv(
															    lowerCamelCase__	,	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(lowerCamelCase__ )

												if not is_final_block:
															A							:					List[Any]              =						nn.Conv(
															    lowerCamelCase__	,	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(lowerCamelCase__ )

									A							:					List[str]              =						down_blocks
									A							:					List[Any]              =						controlnet_down_blocks

									# mid
									A							:					Tuple              =						block_out_channels[-1]
									A							:					str              =						FlaxUNetMidBlockaDCrossAttn(
									    in_channels=lowerCamelCase__	,	dropout=self.dropout	,	num_attention_heads=num_attention_heads[-1]	,	use_linear_projection=self.use_linear_projection	,	dtype=self.dtype	,	)

									A							:					Optional[int]              =						nn.Conv(
									    lowerCamelCase__	,	kernel_size=(1, 1)	,	padding="VALID"	,	kernel_init=nn.initializers.zeros_init()	,	bias_init=nn.initializers.zeros_init()	,	dtype=self.dtype	,	)







						def __call__(	self      :			Union[str, Any]	,	__lowerCamelCase      :			List[str]	,	__lowerCamelCase      :			List[Any]	,	__lowerCamelCase      :			List[Any]	,	__lowerCamelCase      :			int	,	__lowerCamelCase      :			float = 1.0	,	__lowerCamelCase      :			bool = True	,	__lowerCamelCase      :			bool = False	,	)      ->				Union[FlaxControlNetOutput, Tuple]:
									A							:					List[Any]              =						self.controlnet_conditioning_channel_order
									if channel_order == "bgr":
												A							:					Optional[Any]              =						jnp.flip(lowerCamelCase__	,	axis=1 )

									# 1. time
									if not isinstance(lowerCamelCase__	,	jnp.ndarray ):
												A							:					Union[str, Any]              =						jnp.array([timesteps]	,	dtype=jnp.intaa )
									elif isinstance(lowerCamelCase__	,	jnp.ndarray ) and len(timesteps.shape ) == 0:
												A							:					List[str]              =						timesteps.astype(dtype=jnp.floataa )
												A							:					Optional[int]              =						jnp.expand_dims(lowerCamelCase__	,	0 )

									A							:					Any              =						self.time_proj(lowerCamelCase__ )
									A							:					Optional[Any]              =						self.time_embedding(lowerCamelCase__ )

									# 2. pre-process
									A							:					List[Any]              =						jnp.transpose(lowerCamelCase__	,	(0, 2, 3, 1) )
									A							:					Dict              =						self.conv_in(lowerCamelCase__ )

									A							:					List[Any]              =						jnp.transpose(lowerCamelCase__	,	(0, 2, 3, 1) )
									A							:					Optional[Any]              =						self.controlnet_cond_embedding(lowerCamelCase__ )
									sample += controlnet_cond

									# 3. down
									A							:					int              =						(sample,)
									for down_block in self.down_blocks:
												if isinstance(lowerCamelCase__	,	lowerCamelCase__ ):
															A							:					str              =						down_block(lowerCamelCase__	,	lowerCamelCase__	,	lowerCamelCase__	,	deterministic=not train )
												else:
															A							:					List[str]              =						down_block(lowerCamelCase__	,	lowerCamelCase__	,	deterministic=not train )
												down_block_res_samples += res_samples

									# 4. mid
									A							:					Optional[int]              =						self.mid_block(lowerCamelCase__	,	lowerCamelCase__	,	lowerCamelCase__	,	deterministic=not train )

									# 5. contronet blocks
									A							:					List[str]              =						()
									for down_block_res_sample, controlnet_block in zip(lowerCamelCase__	,	self.controlnet_down_blocks ):
												A							:					Dict              =						controlnet_block(lowerCamelCase__ )
												controlnet_down_block_res_samples += (down_block_res_sample,)

									A							:					Dict              =						controlnet_down_block_res_samples

									A							:					Dict              =						self.controlnet_mid_block(lowerCamelCase__ )

									# 6. scaling
									A							:					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=lowerCamelCase__	,	mid_block_res_sample=lowerCamelCase__ )




import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow


if is_torch_available():
							import torch

							from transformers import XLMRobertaModel

@require_sentencepiece
@require_tokenizers
@require_torch
class 	lowerCamelCase_    ( unittest.TestCase						):






						'''simple docstring'''







						@slow
						def 			SCREAMING_SNAKE_CASE__		(	self      :			Any )      ->				Union[str, Any]:
									A							:					Union[str, Any]              =						XLMRobertaModel.from_pretrained("xlm-roberta-base" )
									A							:					Tuple              =						torch.tensor([[0, 5_81, 1_02_69, 83, 9_99_42, 1_36, 6_07_42, 23, 70, 8_05_83, 1_82_76, 2]] )
									# The dog is cute and lives in the garden house

									A							:					Tuple              =						torch.Size((1, 12, 7_68) )  # batch_size, sequence_length, embedding_vector_dim
									A							:					List[str]              =						torch.tensor(
									    [[-0.0101, 0.1218, -0.0803, 0.0801, 0.1327, 0.0776, -0.1215, 0.2383, 0.3338, 0.3106, 0.0300, 0.0252]] )
									#  xlmr = torch.hub.load('pytorch/fairseq', 'xlmr.base')
									#  xlmr.eval()
									#  expected_output_values_last_dim = xlmr.extract_features(input_ids[0])[:, :, -1]
									with torch.no_grad():
												A							:					Tuple              =						model(__lowerCamelCase )["last_hidden_state"].detach()
									self.assertEqual(output.shape	,	__lowerCamelCase )
									# compare the actual values for a slice of last dim
									self.assertTrue(torch.allclose(output[:, :, -1]	,	__lowerCamelCase	,	atol=1e-3 ) )







						@slow
						def 			SCREAMING_SNAKE_CASE__		(	self      :			Optional[int] )      ->				int:
									A							:					str              =						XLMRobertaModel.from_pretrained("xlm-roberta-large" )
									A							:					List[Any]              =						torch.tensor([[0, 5_81, 1_02_69, 83, 9_99_42, 1_36, 6_07_42, 23, 70, 8_05_83, 1_82_76, 2]] )
									# The dog is cute and lives in the garden house

									A							:					Optional[Any]              =						torch.Size((1, 12, 10_24) )  # batch_size, sequence_length, embedding_vector_dim
									A							:					List[Any]              =						torch.tensor(
									    [[-0.0699, -0.0318, 0.0705, -0.1241, 0.0999, -0.0520, 0.1004, -0.1838, -0.4704, 0.1437, 0.0821, 0.0126]] )
									#  xlmr = torch.hub.load('pytorch/fairseq', 'xlmr.large')
									#  xlmr.eval()
									#  expected_output_values_last_dim = xlmr.extract_features(input_ids[0])[:, :, -1]
									with torch.no_grad():
												A							:					Optional[int]              =						model(__lowerCamelCase )["last_hidden_state"].detach()
									self.assertEqual(output.shape	,	__lowerCamelCase )
									# compare the actual values for a slice of last dim
									self.assertTrue(torch.allclose(output[:, :, -1]	,	__lowerCamelCase	,	atol=1e-3 ) )