


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 							lowercase  (						_UpperCAmelCase			):
     lowerCamelCase			:			jnp.ndarray
     lowerCamelCase			:			jnp.ndarray




class 							lowercase  (						nn.Module			):
     lowerCamelCase			:			int
     lowerCamelCase			:			Tuple[int]      = (16, 32, 96, 256)
     lowerCamelCase			:			jnp.dtype      = jnp.floataa







     def 							lowercase__       (						self :						Union[str, Any]      ):
           SCREAMING_SNAKE_CASE__       :  Dict						     =		nn.Conv(
               self.block_out_channels[0]   ,   kernel_size=(3, 3)   ,   padding=((1, 1), (1, 1))   ,   dtype=self.dtype   ,   )

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

           SCREAMING_SNAKE_CASE__       :  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 :						Union[str, Any]   ,   _lowercase :						List[str]      ):
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.conv_in(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		nn.silu(_lowercase      )

           for block in self.blocks:
                 SCREAMING_SNAKE_CASE__       :  Dict						     =		block(_lowercase      )
                 SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		nn.silu(_lowercase      )

           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		self.conv_out(_lowercase      )

           return embedding




@flax_register_to_config
class 							lowercase  (						nn.Module    ,       _UpperCAmelCase    ,       _UpperCAmelCase			):
     lowerCamelCase			:			int      = 32
     lowerCamelCase			:			int      = 4
     lowerCamelCase			:			Tuple[str]      = (
         "CrossAttnDownBlock2D",
         "CrossAttnDownBlock2D",
         "CrossAttnDownBlock2D",
         "DownBlock2D",
     )
     lowerCamelCase			:			Union[bool, Tuple[bool]]      = False
     lowerCamelCase			:			Tuple[int]      = (320, 640, 1_280, 1_280)
     lowerCamelCase			:			int      = 2
     lowerCamelCase			:			Union[int, Tuple[int]]      = 8
     lowerCamelCase			:			Optional[Union[int, Tuple[int]]]      = None
     lowerCamelCase			:			int      = 1_280
     lowerCamelCase			:			float      = 0.0
     lowerCamelCase			:			bool      = False
     lowerCamelCase			:			jnp.dtype      = jnp.floataa
     lowerCamelCase			:			bool      = True
     lowerCamelCase			:			int      = 0
     lowerCamelCase			:			str      = "rgb"
     lowerCamelCase			:			Tuple[int]      = (16, 32, 96, 256)







     def 							lowercase__       (						self :						int   ,   _lowercase :						jax.random.KeyArray      ):
           # init input tensors
           SCREAMING_SNAKE_CASE__       :  Any						     =		(1, self.in_channels, self.sample_size, self.sample_size)
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		jnp.zeros(_lowercase   ,   dtype=jnp.floataa      )
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		jnp.ones((1,)   ,   dtype=jnp.intaa      )
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		jnp.zeros((1, 1, self.cross_attention_dim)   ,   dtype=jnp.floataa      )
           SCREAMING_SNAKE_CASE__       :  Any						     =		(1, 3, self.sample_size * 8, self.sample_size * 8)
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		jnp.zeros(_lowercase   ,   dtype=jnp.floataa      )

           SCREAMING_SNAKE_CASE__     ,       SCREAMING_SNAKE_CASE__       :  List[Any]						     =		jax.random.split(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  Any						     =		{'''params''': params_rng, '''dropout''': dropout_rng}

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







     def 							lowercase__       (						self :						Optional[Any]      ):
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		self.block_out_channels
           SCREAMING_SNAKE_CASE__       :  Optional[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.
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.num_attention_heads or self.attention_head_dim

           # input
           SCREAMING_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
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		FlaxTimesteps(
               block_out_channels[0]   ,   flip_sin_to_cos=self.flip_sin_to_cos   ,   freq_shift=self.config.freq_shift      )
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		FlaxTimestepEmbedding(_lowercase   ,   dtype=self.dtype      )

           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		FlaxControlNetConditioningEmbedding(
               conditioning_embedding_channels=block_out_channels[0]   ,   block_out_channels=self.conditioning_embedding_out_channels   ,   )

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

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

           # down
           SCREAMING_SNAKE_CASE__       :  Any						     =		[]
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		[]

           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		block_out_channels[0]

           SCREAMING_SNAKE_CASE__       :  List[str]						     =		nn.Conv(
               _lowercase   ,   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(_lowercase      )

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

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

                 down_blocks.append(_lowercase      )

                 for _ in range(self.layers_per_block      ):
                       SCREAMING_SNAKE_CASE__       :  int						     =		nn.Conv(
                           _lowercase   ,   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(_lowercase      )

                 if not is_final_block:
                       SCREAMING_SNAKE_CASE__       :  str						     =		nn.Conv(
                           _lowercase   ,   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(_lowercase      )

           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		down_blocks
           SCREAMING_SNAKE_CASE__       :  List[str]						     =		controlnet_down_blocks

           # mid
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		block_out_channels[-1]
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		FlaxUNetMidBlockaDCrossAttn(
               in_channels=_lowercase   ,   dropout=self.dropout   ,   num_attention_heads=num_attention_heads[-1]   ,   use_linear_projection=self.use_linear_projection   ,   dtype=self.dtype   ,   )

           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		nn.Conv(
               _lowercase   ,   kernel_size=(1, 1)   ,   padding='''VALID'''   ,   kernel_init=nn.initializers.zeros_init()   ,   bias_init=nn.initializers.zeros_init()   ,   dtype=self.dtype   ,   )







     def __call__(						self :						Tuple   ,   _lowercase :						Dict   ,   _lowercase :						List[Any]   ,   _lowercase :						str   ,   _lowercase :						Union[str, Any]   ,   _lowercase :						float = 1.0   ,   _lowercase :						bool = True   ,   _lowercase :						bool = False   ,   ):
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.controlnet_conditioning_channel_order
           if channel_order == "bgr":
                 SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		jnp.flip(_lowercase   ,   axis=1      )

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

           SCREAMING_SNAKE_CASE__       :  Dict						     =		self.time_proj(_lowercase      )
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		self.time_embedding(_lowercase      )

           # 2. pre-process
           SCREAMING_SNAKE_CASE__       :  Any						     =		jnp.transpose(_lowercase   ,   (0, 2, 3, 1)      )
           SCREAMING_SNAKE_CASE__       :  str						     =		self.conv_in(_lowercase      )

           SCREAMING_SNAKE_CASE__       :  Any						     =		jnp.transpose(_lowercase   ,   (0, 2, 3, 1)      )
           SCREAMING_SNAKE_CASE__       :  Dict						     =		self.controlnet_cond_embedding(_lowercase      )
           sample += controlnet_cond

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

           # 4. mid
           SCREAMING_SNAKE_CASE__       :  List[str]						     =		self.mid_block(_lowercase   ,   _lowercase   ,   _lowercase   ,   deterministic=not train      )

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

           SCREAMING_SNAKE_CASE__       :  List[str]						     =		controlnet_down_block_res_samples

           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.controlnet_mid_block(_lowercase      )

           # 6. scaling
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		[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=_lowercase   ,   mid_block_res_sample=_lowercase      )













'''simple docstring'''




import os
import sys
import unittest


__UpperCAmelCase            =       os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, """utils"""))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
__UpperCAmelCase            =       os.path.join(git_repo_path, """src""", """transformers""")

__UpperCAmelCase            =       """
{0} = None
"""

__UpperCAmelCase            =       """
class {0}(metaclass=DummyObject):
    _backends = {1}

    def __init__(self, *args, **kwargs):
        requires_backends(self, {1})
"""


__UpperCAmelCase            =       """
def {0}(*args, **kwargs):
    requires_backends({0}, {1})
"""




class      UpperCamelCase__					(					unittest.TestCase			):




					"""simple docstring"""


					def lowerCamelCase_			(						self				:			List[str]						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				Dict		=			find_backend("""    _import_structure[\"models.albert\"].append(\"AlbertTokenizerFast\")"""						)
							self.assertIsNone(lowerCamelCase_						)

							SCREAMING_SNAKE_CASE :				str		=			find_backend("""    if not is_tokenizers_available():"""						)
							self.assertEqual(lowerCamelCase_      ,						"""tokenizers"""						)

							SCREAMING_SNAKE_CASE :				Union[str, Any]		=			find_backend("""    if not is_tensorflow_text_available():"""						)
							self.assertEqual(lowerCamelCase_      ,						"""tensorflow_text"""						)

							SCREAMING_SNAKE_CASE :				List[str]		=			find_backend("""    if not (is_sentencepiece_available() and is_tokenizers_available()):"""						)
							self.assertEqual(lowerCamelCase_      ,						"""sentencepiece_and_tokenizers"""						)

							SCREAMING_SNAKE_CASE :				Tuple		=			find_backend(
							    """    if not (is_sentencepiece_available() and is_tensorflow_text_available()):"""						)
							self.assertEqual(lowerCamelCase_      ,						"""sentencepiece_and_tensorflow_text"""						)

							SCREAMING_SNAKE_CASE :				Union[str, Any]		=			find_backend(
							    """    if not (is_sentencepiece_available() and is_tokenizers_available() and is_vision_available()):"""						)
							self.assertEqual(lowerCamelCase_      ,						"""sentencepiece_and_tokenizers_and_vision"""						)


					def lowerCamelCase_			(						self				:			Any						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				Optional[Any]		=			read_init()
							# We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
							self.assertIn("""torch"""      ,						lowerCamelCase_						)
							self.assertIn("""tensorflow_text"""      ,						lowerCamelCase_						)
							self.assertIn("""sentencepiece_and_tokenizers"""      ,						lowerCamelCase_						)

							# Likewise, we can't assert on the exact content of a key
							self.assertIn("""BertModel"""      ,						objects["""torch"""]						)
							self.assertIn("""TFBertModel"""      ,						objects["""tf"""]						)
							self.assertIn("""FlaxBertModel"""      ,						objects["""flax"""]						)
							self.assertIn("""BertModel"""      ,						objects["""torch"""]						)
							self.assertIn("""TFBertTokenizer"""      ,						objects["""tensorflow_text"""]						)
							self.assertIn("""convert_slow_tokenizer"""      ,						objects["""sentencepiece_and_tokenizers"""]						)


					def lowerCamelCase_			(						self				:			int						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				str		=			create_dummy_object("""CONSTANT"""      ,						"""'torch'"""						)
							self.assertEqual(lowerCamelCase_      ,						"""\nCONSTANT = None\n"""						)

							SCREAMING_SNAKE_CASE :				List[Any]		=			create_dummy_object("""function"""      ,						"""'torch'"""						)
							self.assertEqual(
							    lowerCamelCase_      ,						"""\ndef function(*args, **kwargs):\n    requires_backends(function, 'torch')\n"""						)

							SCREAMING_SNAKE_CASE :				str		=			"""
class FakeClass(metaclass=DummyObject):
    _backends = 'torch'

    def __init__(self, *args, **kwargs):
        requires_backends(self, 'torch')
"""
							SCREAMING_SNAKE_CASE :				Union[str, Any]		=			create_dummy_object("""FakeClass"""      ,						"""'torch'"""						)
							self.assertEqual(lowerCamelCase_      ,						lowerCamelCase_						)



					def lowerCamelCase_			(						self				:			int						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				Union[str, Any]		=			"""# This file is autogenerated by the command `make fix-copies`, do not edit.
from ..utils import DummyObject, requires_backends


CONSTANT = None


def function(*args, **kwargs):
    requires_backends(function, [\"torch\"])


class FakeClass(metaclass=DummyObject):
    _backends = [\"torch\"]

    def __init__(self, *args, **kwargs):
        requires_backends(self, [\"torch\"])
"""
							SCREAMING_SNAKE_CASE :				List[Any]		=			create_dummy_files({"""torch""": ["""CONSTANT""", """function""", """FakeClass"""]}						)
							self.assertEqual(dummy_files["""torch"""]      ,						lowerCamelCase_						)



