

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					(		__SCREAMING_SNAKE_CASE			):
						_lowercase        =     4_2
						_lowercase        =     4_2







class 			a					(		nn.Module			):
						_lowercase        =     4_2
						_lowercase        =     (1_6, 3_2, 9_6, 2_5_6)
						_lowercase        =     jnp.floataa


						def     _UpperCAmelCase			(    self		):







										'''simple docstring'''


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

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

										_UpperCAmelCase    :						Tuple							=      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_		):







										'''simple docstring'''


										_UpperCAmelCase    :						Dict							=      self.conv_in(UpperCamelCase__		)
										_UpperCAmelCase    :						int							=      nn.silu(UpperCamelCase__		)

										for block in self.blocks:
														_UpperCAmelCase    :						str							=      block(UpperCamelCase__		)
														_UpperCAmelCase    :						Optional[Any]							=      nn.silu(UpperCamelCase__		)

										_UpperCAmelCase    :						Optional[Any]							=      self.conv_out(UpperCamelCase__		)

										return embedding







@flax_register_to_config
class 			a					(		nn.Module							,	__SCREAMING_SNAKE_CASE							,	__SCREAMING_SNAKE_CASE			):
						_lowercase        =     3_2
						_lowercase        =     4
						_lowercase        =     (
						    "CrossAttnDownBlock2D",
						    "CrossAttnDownBlock2D",
						    "CrossAttnDownBlock2D",
						    "DownBlock2D",
						)
						_lowercase        =     False
						_lowercase        =     (3_2_0, 6_4_0, 1_2_8_0, 1_2_8_0)
						_lowercase        =     2
						_lowercase        =     8
						_lowercase        =     None
						_lowercase        =     1_2_8_0
						_lowercase        =     0.0
						_lowercase        =     False
						_lowercase        =     jnp.floataa
						_lowercase        =     True
						_lowercase        =     0
						_lowercase        =     "rgb"
						_lowercase        =     (1_6, 3_2, 9_6, 2_5_6)


						def     _UpperCAmelCase			(    self  ,   A_		):







										'''simple docstring'''


										_UpperCAmelCase    :						Union[str, Any]							=      (1, self.in_channels, self.sample_size, self.sample_size)
										_UpperCAmelCase    :						Any							=      jnp.zeros(UpperCamelCase__  ,   dtype=jnp.floataa		)
										_UpperCAmelCase    :						Dict							=      jnp.ones((1,)  ,   dtype=jnp.intaa		)
										_UpperCAmelCase    :						List[str]							=      jnp.zeros((1, 1, self.cross_attention_dim)  ,   dtype=jnp.floataa		)
										_UpperCAmelCase    :						Optional[int]							=      (1, 3, self.sample_size * 8, self.sample_size * 8)
										_UpperCAmelCase    :						int							=      jnp.zeros(UpperCamelCase__  ,   dtype=jnp.floataa		)

										_UpperCAmelCase    :						Optional[int]							=      jax.random.split(UpperCamelCase__		)
										_UpperCAmelCase    :						Optional[int]							=      {'''params''': params_rng, '''dropout''': dropout_rng}

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


						def     _UpperCAmelCase			(    self		):







										'''simple docstring'''


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

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

										# time
										_UpperCAmelCase    :						Any							=      FlaxTimesteps(
										    block_out_channels[0]  ,   flip_sin_to_cos=self.flip_sin_to_cos  ,   freq_shift=self.config.freq_shift		)
										_UpperCAmelCase    :						List[Any]							=      FlaxTimestepEmbedding(UpperCamelCase__  ,   dtype=self.dtype		)

										_UpperCAmelCase    :						int							=      FlaxControlNetConditioningEmbedding(
										    conditioning_embedding_channels=block_out_channels[0]  ,   block_out_channels=self.conditioning_embedding_out_channels  ,   )

										_UpperCAmelCase    :						Any							=      self.only_cross_attention
										if isinstance(UpperCamelCase__  ,   UpperCamelCase__		):
														_UpperCAmelCase    :						Union[str, Any]							=      (only_cross_attention,) * len(self.down_block_types		)

										if isinstance(UpperCamelCase__  ,   UpperCamelCase__		):
														_UpperCAmelCase    :						str							=      (num_attention_heads,) * len(self.down_block_types		)

										# down
										_UpperCAmelCase    :						str							=      []
										_UpperCAmelCase    :						List[str]							=      []

										_UpperCAmelCase    :						Union[str, Any]							=      block_out_channels[0]

										_UpperCAmelCase    :						Tuple							=      nn.Conv(
										    UpperCamelCase__  ,   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(UpperCamelCase__		)

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

														if down_block_type == "CrossAttnDownBlock2D":
																		_UpperCAmelCase    :						List[Any]							=      FlaxCrossAttnDownBlockaD(
																		    in_channels=UpperCamelCase__  ,   out_channels=UpperCamelCase__  ,   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    :						str							=      FlaxDownBlockaD(
																		    in_channels=UpperCamelCase__  ,   out_channels=UpperCamelCase__  ,   dropout=self.dropout  ,   num_layers=self.layers_per_block  ,   add_downsample=not is_final_block  ,   dtype=self.dtype  ,   )

														down_blocks.append(UpperCamelCase__		)

														for _ in range(self.layers_per_block		):
																		_UpperCAmelCase    :						Union[str, Any]							=      nn.Conv(
																		    UpperCamelCase__  ,   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(UpperCamelCase__		)

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

										_UpperCAmelCase    :						List[Any]							=      down_blocks
										_UpperCAmelCase    :						List[Any]							=      controlnet_down_blocks

										# mid
										_UpperCAmelCase    :						Optional[int]							=      block_out_channels[-1]
										_UpperCAmelCase    :						Optional[Any]							=      FlaxUNetMidBlockaDCrossAttn(
										    in_channels=UpperCamelCase__  ,   dropout=self.dropout  ,   num_attention_heads=num_attention_heads[-1]  ,   use_linear_projection=self.use_linear_projection  ,   dtype=self.dtype  ,   )

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







										'''simple docstring'''


										_UpperCAmelCase    :						Optional[Any]							=      self.controlnet_conditioning_channel_order
										if channel_order == "bgr":
														_UpperCAmelCase    :						Dict							=      jnp.flip(UpperCamelCase__  ,   axis=1		)

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

										_UpperCAmelCase    :						int							=      self.time_proj(UpperCamelCase__		)
										_UpperCAmelCase    :						Tuple							=      self.time_embedding(UpperCamelCase__		)

										# 2. pre-process
										_UpperCAmelCase    :						Dict							=      jnp.transpose(UpperCamelCase__  ,   (0, 2, 3, 1)		)
										_UpperCAmelCase    :						Optional[int]							=      self.conv_in(UpperCamelCase__		)

										_UpperCAmelCase    :						str							=      jnp.transpose(UpperCamelCase__  ,   (0, 2, 3, 1)		)
										_UpperCAmelCase    :						Optional[int]							=      self.controlnet_cond_embedding(UpperCamelCase__		)
										sample += controlnet_cond

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

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

										# 5. contronet blocks
										_UpperCAmelCase    :						Tuple							=      ()
										for down_block_res_sample, controlnet_block in zip(UpperCamelCase__  ,   self.controlnet_down_blocks		):
														_UpperCAmelCase    :						Any							=      controlnet_block(UpperCamelCase__		)
														controlnet_down_block_res_samples += (down_block_res_sample,)

										_UpperCAmelCase    :						Optional[Any]							=      controlnet_down_block_res_samples

										_UpperCAmelCase    :						int							=      self.controlnet_mid_block(UpperCamelCase__		)

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





import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import is_speech_available, is_vision_available
from transformers.testing_utils import require_torch


if is_vision_available():
					from transformers import TvltImageProcessor

if is_speech_available():
					from transformers import TvltFeatureExtractor

from transformers import TvltProcessor

@require_torch
class 			a					(		unittest.TestCase			):


						def     _UpperCAmelCase			(    self		):







										'''simple docstring'''


										_UpperCAmelCase    :						int							=      "ZinengTang/tvlt-base"
										_UpperCAmelCase    :						int							=      tempfile.mkdtemp()


						def     _UpperCAmelCase			(    self  ,   **A_		):







										'''simple docstring'''


										return TvltImageProcessor.from_pretrained(self.checkpoint  ,   **A_		)


						def     _UpperCAmelCase			(    self  ,   **A_		):







										'''simple docstring'''


										return TvltFeatureExtractor.from_pretrained(self.checkpoint  ,   **A_		)


						def     _UpperCAmelCase			(    self		):







										'''simple docstring'''


										shutil.rmtree(self.tmpdirname		)


						def     _UpperCAmelCase			(    self		):







										'''simple docstring'''


										_UpperCAmelCase    :						Dict							=      self.get_image_processor()
										_UpperCAmelCase    :						Optional[int]							=      self.get_feature_extractor()

										_UpperCAmelCase    :						Tuple							=      TvltProcessor(image_processor=A_  ,   feature_extractor=A_		)
										processor.save_pretrained(self.tmpdirname		)
										_UpperCAmelCase    :						str							=      TvltProcessor.from_pretrained(self.tmpdirname		)

										self.assertIsInstance(processor.feature_extractor  ,   A_		)
										self.assertIsInstance(processor.image_processor  ,   A_		)


						def     _UpperCAmelCase			(    self		):







										'''simple docstring'''


										_UpperCAmelCase    :						List[Any]							=      self.get_image_processor()
										_UpperCAmelCase    :						int							=      self.get_feature_extractor()

										_UpperCAmelCase    :						Tuple							=      TvltProcessor(image_processor=A_  ,   feature_extractor=A_		)

										_UpperCAmelCase    :						List[str]							=      np.ones([12000]		)

										_UpperCAmelCase    :						int							=      feature_extractor(A_  ,   return_tensors="np"		)
										_UpperCAmelCase    :						int							=      processor(audio=A_  ,   return_tensors="np"		)

										for key in audio_dict.keys():
														self.assertAlmostEqual(audio_dict[key].sum()  ,   input_processor[key].sum()  ,   delta=1e-2		)


						def     _UpperCAmelCase			(    self		):







										'''simple docstring'''


										_UpperCAmelCase    :						List[str]							=      self.get_image_processor()
										_UpperCAmelCase    :						int							=      self.get_feature_extractor()

										_UpperCAmelCase    :						Optional[Any]							=      TvltProcessor(image_processor=A_  ,   feature_extractor=A_		)

										_UpperCAmelCase    :						Union[str, Any]							=      np.ones([3, 224, 224]		)

										_UpperCAmelCase    :						Tuple							=      image_processor(A_  ,   return_tensors="np"		)
										_UpperCAmelCase    :						List[str]							=      processor(images=A_  ,   return_tensors="np"		)

										for key in image_dict.keys():
														self.assertAlmostEqual(image_dict[key].sum()  ,   input_processor[key].sum()  ,   delta=1e-2		)


						def     _UpperCAmelCase			(    self		):







										'''simple docstring'''


										_UpperCAmelCase    :						List[str]							=      self.get_image_processor()
										_UpperCAmelCase    :						Any							=      self.get_feature_extractor()

										_UpperCAmelCase    :						Dict							=      TvltProcessor(image_processor=A_  ,   feature_extractor=A_		)

										_UpperCAmelCase    :						str							=      np.ones([12000]		)
										_UpperCAmelCase    :						Optional[Any]							=      np.ones([3, 224, 224]		)

										_UpperCAmelCase    :						List[Any]							=      processor(audio=A_  ,   images=A_		)

										self.assertListEqual(list(inputs.keys()		)  ,   ["audio_values", "audio_mask", "pixel_values", "pixel_mask"]		)

										# test if it raises when no input is passed
										with pytest.raises(A_		):
														processor()





						def     _UpperCAmelCase			(    self		):







										'''simple docstring'''


										_UpperCAmelCase    :						Optional[Any]							=      self.get_image_processor()
										_UpperCAmelCase    :						int							=      self.get_feature_extractor()

										_UpperCAmelCase    :						str							=      TvltProcessor(image_processor=A_  ,   feature_extractor=A_		)

										self.assertListEqual(
										    processor.model_input_names  ,   image_processor.model_input_names + feature_extractor.model_input_names  ,   msg="`processor` and `image_processor`+`feature_extractor` model input names do not match"  ,   )


