
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		(					SCREAMING_SNAKE_CASE__		):
							lowercase_       :       jnp.ndarray
							lowercase_       :       jnp.ndarray







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




							def  A__      (				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(
																	    A__   ,kernel_size=(3, 3)   ,padding=((1, 1), (1, 1))   ,dtype=self.dtype   ,)
																	blocks.append(A__)
																	lowercase							=						nn.Conv(
																	    A__   ,kernel_size=(3, 3)   ,strides=(2, 2)   ,padding=((1, 1), (1, 1))   ,dtype=self.dtype   ,)
																	blocks.append(A__)
												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   ,A__):
												lowercase							=						self.conv_in(A__)
												lowercase							=						nn.silu(A__)

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

												lowercase							=						self.conv_out(A__)

												return embedding







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




							def  A__      (				self   ,A__):
												# init input tensors
												lowercase							=						(1, self.in_channels, self.sample_size, self.sample_size)
												lowercase							=						jnp.zeros(A__   ,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(A__   ,dtype=jnp.floataa)

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

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




							def  A__      (				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(A__   ,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(A__   ,A__):
																	lowercase							=						(only_cross_attention,) * len(self.down_block_types)

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

												# down
												lowercase							=						[]
												lowercase							=						[]

												lowercase							=						block_out_channels[0]

												lowercase							=						nn.Conv(
												    A__   ,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(A__)

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

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

																	down_blocks.append(A__)

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

																	if not is_final_block:
																						lowercase							=						nn.Conv(
																						    A__   ,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(A__)

												lowercase							=						down_blocks
												lowercase							=						controlnet_down_blocks

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

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





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

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

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

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

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

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

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

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

												lowercase							=						controlnet_down_block_res_samples

												lowercase							=						self.controlnet_mid_block(A__)

												# 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=A__   ,mid_block_res_sample=A__)










from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


__UpperCamelCase					:       Optional[Any]    					=      {
    "configuration_jukebox": [
        "JUKEBOX_PRETRAINED_CONFIG_ARCHIVE_MAP",
        "JukeboxConfig",
        "JukeboxPriorConfig",
        "JukeboxVQVAEConfig",
    ],
    "tokenization_jukebox": ["JukeboxTokenizer"],
}

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		__UpperCamelCase					:       Tuple    					=      [
		    "JUKEBOX_PRETRAINED_MODEL_ARCHIVE_LIST",
		    "JukeboxModel",
		    "JukeboxPreTrainedModel",
		    "JukeboxVQVAE",
		    "JukeboxPrior",
		]

if TYPE_CHECKING:
		from .configuration_jukebox import (
		    JUKEBOX_PRETRAINED_CONFIG_ARCHIVE_MAP,
		    JukeboxConfig,
		    JukeboxPriorConfig,
		    JukeboxVQVAEConfig,
		)
		from .tokenization_jukebox import JukeboxTokenizer

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_jukebox import (
				    JUKEBOX_PRETRAINED_MODEL_ARCHIVE_LIST,
				    JukeboxModel,
				    JukeboxPreTrainedModel,
				    JukeboxPrior,
				    JukeboxVQVAE,
				)

else:
		import sys

		__UpperCamelCase					:       List[Any]    					=      _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)

