





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





@flax.struct.dataclass
class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):
  __lowerCAmelCase   :						jnp.ndarray





@flax_register_to_config
class UpperCAmelCase_				( nn.Module     ,		_SCREAMING_SNAKE_CASE     ,		_SCREAMING_SNAKE_CASE ):
  __lowerCAmelCase   :						int							=   32
  __lowerCAmelCase   :						int							=   4
  __lowerCAmelCase   :						int							=   4
  __lowerCAmelCase   :						Tuple[str]							=   (
      "CrossAttnDownBlock2D",
      "CrossAttnDownBlock2D",
      "CrossAttnDownBlock2D",
      "DownBlock2D",
  )
  __lowerCAmelCase   :						Tuple[str]							=   ("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
  __lowerCAmelCase   :						Union[bool, Tuple[bool]]							=   False
  __lowerCAmelCase   :						Tuple[int]							=   (320, 640, 1280, 1280)
  __lowerCAmelCase   :						int							=   2
  __lowerCAmelCase   :						Union[int, Tuple[int]]							=   8
  __lowerCAmelCase   :						Optional[Union[int, Tuple[int]]]							=   None
  __lowerCAmelCase   :						int							=   1280
  __lowerCAmelCase   :						float							=   0.0
  __lowerCAmelCase   :						bool							=   False
  __lowerCAmelCase   :						jnp.dtype							=   jnp.floataa
  __lowerCAmelCase   :						bool							=   True
  __lowerCAmelCase   :						int							=   0
  __lowerCAmelCase   :						bool							=   False


  def 						__UpperCAmelCase     (							self ,__snake_case ):



       """simple docstring"""





       A_          =       (1, self.in_channels, self.sample_size, self.sample_size)
       A_          =       jnp.zeros(__snake_case ,dtype=jnp.floataa )
       A_          =       jnp.ones((1,) ,dtype=jnp.intaa )
       A_          =       jnp.zeros((1, 1, self.cross_attention_dim) ,dtype=jnp.floataa )

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

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


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       self.block_out_channels
       A_          =       block_out_channels[0] * 4

       if self.num_attention_heads is not None:
            raise ValueError(
                '''At the moment it is not possible to define the number of attention heads via `num_attention_heads` because of a naming issue as described in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131. Passing `num_attention_heads` will only be supported in diffusers v0.19.''' )

       # 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_          =       self.num_attention_heads or self.attention_head_dim

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

       # time
       A_          =       FlaxTimesteps(
           block_out_channels[0] ,flip_sin_to_cos=self.flip_sin_to_cos ,freq_shift=self.config.freq_shift )
       A_          =       FlaxTimestepEmbedding(__snake_case ,dtype=self.dtype )

       A_          =       self.only_cross_attention
       if isinstance(__snake_case ,__snake_case ):
            A_          =       (only_cross_attention,) * len(self.down_block_types )

       if isinstance(__snake_case ,__snake_case ):
            A_          =       (num_attention_heads,) * len(self.down_block_types )

       # down
       A_          =       []
       A_          =       block_out_channels[0]
       for i, down_block_type in enumerate(self.down_block_types ):
            A_          =       output_channel
            A_          =       block_out_channels[i]
            A_          =       i == len(__snake_case ) - 1

            if down_block_type == "CrossAttnDownBlock2D":
                 A_          =       FlaxCrossAttnDownBlockaD(
                     in_channels=__snake_case ,out_channels=__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] ,use_memory_efficient_attention=self.use_memory_efficient_attention ,dtype=self.dtype ,)
            else:
                 A_          =       FlaxDownBlockaD(
                     in_channels=__snake_case ,out_channels=__snake_case ,dropout=self.dropout ,num_layers=self.layers_per_block ,add_downsample=not is_final_block ,dtype=self.dtype ,)

            down_blocks.append(__snake_case )
       A_          =       down_blocks

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

       # up
       A_          =       []
       A_          =       list(reversed(__snake_case ) )
       A_          =       list(reversed(__snake_case ) )
       A_          =       list(reversed(__snake_case ) )
       A_          =       reversed_block_out_channels[0]
       for i, up_block_type in enumerate(self.up_block_types ):
            A_          =       output_channel
            A_          =       reversed_block_out_channels[i]
            A_          =       reversed_block_out_channels[min(i + 1 ,len(__snake_case ) - 1 )]

            A_          =       i == len(__snake_case ) - 1

            if up_block_type == "CrossAttnUpBlock2D":
                 A_          =       FlaxCrossAttnUpBlockaD(
                     in_channels=__snake_case ,out_channels=__snake_case ,prev_output_channel=__snake_case ,num_layers=self.layers_per_block + 1 ,num_attention_heads=reversed_num_attention_heads[i] ,add_upsample=not is_final_block ,dropout=self.dropout ,use_linear_projection=self.use_linear_projection ,only_cross_attention=only_cross_attention[i] ,use_memory_efficient_attention=self.use_memory_efficient_attention ,dtype=self.dtype ,)
            else:
                 A_          =       FlaxUpBlockaD(
                     in_channels=__snake_case ,out_channels=__snake_case ,prev_output_channel=__snake_case ,num_layers=self.layers_per_block + 1 ,add_upsample=not is_final_block ,dropout=self.dropout ,dtype=self.dtype ,)

            up_blocks.append(__snake_case )
            A_          =       output_channel
       A_          =       up_blocks

       # out
       A_          =       nn.GroupNorm(num_groups=3_2 ,epsilon=1E-5 )
       A_          =       nn.Conv(
           self.out_channels ,kernel_size=(3, 3) ,strides=(1, 1) ,padding=((1, 1), (1, 1)) ,dtype=self.dtype ,)




  def __call__(							self ,__snake_case ,__snake_case ,__snake_case ,__snake_case=None ,__snake_case=None ,__snake_case = True ,__snake_case = False ,):



       """simple docstring"""





       if not isinstance(__snake_case ,jnp.ndarray ):
            A_          =       jnp.array([timesteps] ,dtype=jnp.intaa )
       elif isinstance(__snake_case ,jnp.ndarray ) and len(timesteps.shape ) == 0:
            A_          =       timesteps.astype(dtype=jnp.floataa )
            A_          =       jnp.expand_dims(__snake_case ,0 )

       A_          =       self.time_proj(__snake_case )
       A_          =       self.time_embedding(__snake_case )

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

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

       if down_block_additional_residuals is not None:
            A_          =       ()

            for down_block_res_sample, down_block_additional_residual in zip(
                __snake_case ,__snake_case ):
                 down_block_res_sample += down_block_additional_residual
                 new_down_block_res_samples += (down_block_res_sample,)

            A_          =       new_down_block_res_samples

       # 4. mid
       A_          =       self.mid_block(__snake_case ,__snake_case ,__snake_case ,deterministic=not train )

       if mid_block_additional_residual is not None:
            sample += mid_block_additional_residual

       # 5. up
       for up_block in self.up_blocks:
            A_          =       down_block_res_samples[-(self.layers_per_block + 1) :]
            A_          =       down_block_res_samples[: -(self.layers_per_block + 1)]
            if isinstance(__snake_case ,__snake_case ):
                 A_          =       up_block(
                     __snake_case ,temb=__snake_case ,encoder_hidden_states=__snake_case ,res_hidden_states_tuple=__snake_case ,deterministic=not train ,)
            else:
                 A_          =       up_block(__snake_case ,temb=__snake_case ,res_hidden_states_tuple=__snake_case ,deterministic=not train )

        # 6. post-process
       A_          =       self.conv_norm_out(__snake_case )
       A_          =       nn.silu(__snake_case )
       A_          =       self.conv_out(__snake_case )
       A_          =       jnp.transpose(__snake_case ,(0, 3, 1, 2) )

       if not return_dict:
            return (sample,)

       return FlaxUNetaDConditionOutput(sample=__snake_case )










import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionImageVariationPipeline
from diffusers.utils.testing_utils import load_image, require_torch_gpu, slow, torch_device


a__      :     Optional[int]         =		False





class UpperCAmelCase_				( unittest.TestCase ):
  pass





@slow
@require_torch_gpu
class UpperCAmelCase_				( unittest.TestCase ):


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       VersatileDiffusionImageVariationPipeline.from_pretrained('''shi-labs/versatile-diffusion''' )
       pipe.to(__snake_case )
       pipe.set_progress_bar_config(disable=__snake_case )

       A_          =       load_image(
           '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg''' )
       A_          =       torch.manual_seed(0 )
       A_          =       pipe(
           image=__snake_case ,generator=__snake_case ,guidance_scale=7.5 ,num_inference_steps=5_0 ,output_type='''numpy''' ,).images

       A_          =       image[0, 2_5_3:2_5_6, 2_5_3:2_5_6, -1]

       assert image.shape == (1, 5_1_2, 5_1_2, 3)
       A_          =       np.array([0.0441, 0.0469, 0.0507, 0.0575, 0.0632, 0.0650, 0.0865, 0.0909, 0.0945] )

       assert np.abs(image_slice.flatten() - expected_slice ).max() < 1E-2



