






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






@flax.struct.dataclass
class      __SCREAMING_SNAKE_CASE      (			lowerCamelCase				):
 snake_case_  							=							42






@flax_register_to_config
class      __SCREAMING_SNAKE_CASE      (			nn.Module	,				lowerCamelCase	,				lowerCamelCase				):
 snake_case_  							=							32
 snake_case_  							=							4
 snake_case_  							=							4
 snake_case_  							=							(
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "DownBlock2D",
 )
 snake_case_  							=							("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
 snake_case_  							=							False
 snake_case_  							=							(320, 640, 1280, 1280)
 snake_case_  							=							2
 snake_case_  							=							8
 snake_case_  							=							None
 snake_case_  							=							1280
 snake_case_  							=							0.0
 snake_case_  							=							False
 snake_case_  							=							jnp.floataa
 snake_case_  							=							True
 snake_case_  							=							0
 snake_case_  							=							False




 def      __magic_name__	(      self	:							List[Any]					,      __lowercase	:							jax.random.KeyArray      )	->						FrozenDict:
       # init input tensors
       SCREAMING_SNAKE_CASE__     :				int									=(1, self.in_channels, self.sample_size, self.sample_size)
       SCREAMING_SNAKE_CASE__     :				List[Any]									=jnp.zeros(__lowercase					,      dtype=jnp.floataa      )
       SCREAMING_SNAKE_CASE__     :				int									=jnp.ones((1,)					,      dtype=jnp.intaa      )
       SCREAMING_SNAKE_CASE__     :				List[Any]									=jnp.zeros((1, 1, self.cross_attention_dim)					,      dtype=jnp.floataa      )

       SCREAMING_SNAKE_CASE__		,	SCREAMING_SNAKE_CASE__     :				List[str]									=jax.random.split(__lowercase      )
       SCREAMING_SNAKE_CASE__     :				Optional[int]									={'''params''': params_rng, '''dropout''': dropout_rng}

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




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

       # input
       SCREAMING_SNAKE_CASE__     :				Tuple									=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__     :				Optional[int]									=FlaxTimesteps(
           block_out_channels[0]					,      flip_sin_to_cos=self.flip_sin_to_cos					,      freq_shift=self.config.freq_shift      )
       SCREAMING_SNAKE_CASE__     :				Optional[int]									=FlaxTimestepEmbedding(__lowercase					,      dtype=self.dtype      )

       SCREAMING_SNAKE_CASE__     :				Optional[int]									=self.only_cross_attention
       if isinstance(__lowercase					,      __lowercase      ):
             SCREAMING_SNAKE_CASE__     :				List[Any]									=(only_cross_attention,) * len(self.down_block_types      )

       if isinstance(__lowercase					,      __lowercase      ):
             SCREAMING_SNAKE_CASE__     :				Any									=(num_attention_heads,) * len(self.down_block_types      )

       # down
       SCREAMING_SNAKE_CASE__     :				Union[str, Any]									=[]
       SCREAMING_SNAKE_CASE__     :				Optional[Any]									=block_out_channels[0]
       for i, down_block_type in enumerate(self.down_block_types      ):
             SCREAMING_SNAKE_CASE__     :				List[str]									=output_channel
             SCREAMING_SNAKE_CASE__     :				Tuple									=block_out_channels[i]
             SCREAMING_SNAKE_CASE__     :				Optional[int]									=i == len(__lowercase      ) - 1

             if down_block_type == "CrossAttnDownBlock2D":
                   SCREAMING_SNAKE_CASE__     :				Tuple									=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]					,      use_memory_efficient_attention=self.use_memory_efficient_attention					,      dtype=self.dtype					,      )
             else:
                   SCREAMING_SNAKE_CASE__     :				str									=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      )
       SCREAMING_SNAKE_CASE__     :				int									=down_blocks

       # mid
       SCREAMING_SNAKE_CASE__     :				List[Any]									=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
       SCREAMING_SNAKE_CASE__     :				int									=[]
       SCREAMING_SNAKE_CASE__     :				Any									=list(reversed(__lowercase      )      )
       SCREAMING_SNAKE_CASE__     :				List[Any]									=list(reversed(__lowercase      )      )
       SCREAMING_SNAKE_CASE__     :				Dict									=list(reversed(__lowercase      )      )
       SCREAMING_SNAKE_CASE__     :				List[str]									=reversed_block_out_channels[0]
       for i, up_block_type in enumerate(self.up_block_types      ):
             SCREAMING_SNAKE_CASE__     :				Optional[int]									=output_channel
             SCREAMING_SNAKE_CASE__     :				List[Any]									=reversed_block_out_channels[i]
             SCREAMING_SNAKE_CASE__     :				List[Any]									=reversed_block_out_channels[min(i + 1					,      len(__lowercase      ) - 1      )]

             SCREAMING_SNAKE_CASE__     :				Optional[int]									=i == len(__lowercase      ) - 1

             if up_block_type == "CrossAttnUpBlock2D":
                   SCREAMING_SNAKE_CASE__     :				List[Any]									=FlaxCrossAttnUpBlockaD(
                       in_channels=__lowercase					,      out_channels=__lowercase					,      prev_output_channel=__lowercase					,      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:
                   SCREAMING_SNAKE_CASE__     :				List[Any]									=FlaxUpBlockaD(
                       in_channels=__lowercase					,      out_channels=__lowercase					,      prev_output_channel=__lowercase					,      num_layers=self.layers_per_block + 1					,      add_upsample=not is_final_block					,      dropout=self.dropout					,      dtype=self.dtype					,      )

             up_blocks.append(__lowercase      )
             SCREAMING_SNAKE_CASE__     :				Dict									=output_channel
       SCREAMING_SNAKE_CASE__     :				Optional[Any]									=up_blocks

       # out
       SCREAMING_SNAKE_CASE__     :				Optional[int]									=nn.GroupNorm(num_groups=32					,      epsilon=1e-5      )
       SCREAMING_SNAKE_CASE__     :				str									=nn.Conv(
           self.out_channels					,      kernel_size=(3, 3)					,      strides=(1, 1)					,      padding=((1, 1), (1, 1))					,      dtype=self.dtype					,      )




 def __call__(      self	:							Union[str, Any]					,      __lowercase	:							Optional[Any]					,      __lowercase	:							List[str]					,      __lowercase	:							str					,      __lowercase	:							Union[str, Any]=None					,      __lowercase	:							List[str]=None					,      __lowercase	:							bool = True					,      __lowercase	:							bool = False					,      )	->						Union[FlaxUNetaDConditionOutput, Tuple]:
       # 1. time
       if not isinstance(__lowercase					,      jnp.ndarray      ):
             SCREAMING_SNAKE_CASE__     :				List[Any]									=jnp.array([timesteps]					,      dtype=jnp.intaa      )
       elif isinstance(__lowercase					,      jnp.ndarray      ) and len(timesteps.shape      ) == 0:
             SCREAMING_SNAKE_CASE__     :				Optional[Any]									=timesteps.astype(dtype=jnp.floataa      )
             SCREAMING_SNAKE_CASE__     :				List[Any]									=jnp.expand_dims(__lowercase					,      0      )

       SCREAMING_SNAKE_CASE__     :				Optional[int]									=self.time_proj(__lowercase      )
       SCREAMING_SNAKE_CASE__     :				Tuple									=self.time_embedding(__lowercase      )

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

       # 3. down
       SCREAMING_SNAKE_CASE__     :				Tuple									=(sample,)
       for down_block in self.down_blocks:
             if isinstance(__lowercase					,      __lowercase      ):
                   SCREAMING_SNAKE_CASE__		,	SCREAMING_SNAKE_CASE__     :				str									=down_block(__lowercase					,      __lowercase					,      __lowercase					,      deterministic=not train      )
             else:
                   SCREAMING_SNAKE_CASE__		,	SCREAMING_SNAKE_CASE__     :				Any									=down_block(__lowercase					,      __lowercase					,      deterministic=not train      )
             down_block_res_samples += res_samples

       if down_block_additional_residuals is not None:
             SCREAMING_SNAKE_CASE__     :				int									=()

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

             SCREAMING_SNAKE_CASE__     :				Dict									=new_down_block_res_samples

       # 4. mid
       SCREAMING_SNAKE_CASE__     :				Union[str, Any]									=self.mid_block(__lowercase					,      __lowercase					,      __lowercase					,      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:
             SCREAMING_SNAKE_CASE__     :				List[str]									=down_block_res_samples[-(self.layers_per_block + 1) :]
             SCREAMING_SNAKE_CASE__     :				Optional[Any]									=down_block_res_samples[: -(self.layers_per_block + 1)]
             if isinstance(__lowercase					,      __lowercase      ):
                   SCREAMING_SNAKE_CASE__     :				Optional[int]									=up_block(
                       __lowercase					,      temb=__lowercase					,      encoder_hidden_states=__lowercase					,      res_hidden_states_tuple=__lowercase					,      deterministic=not train					,      )
             else:
                   SCREAMING_SNAKE_CASE__     :				int									=up_block(__lowercase					,      temb=__lowercase					,      res_hidden_states_tuple=__lowercase					,      deterministic=not train      )

        # 6. post-process
       SCREAMING_SNAKE_CASE__     :				List[Any]									=self.conv_norm_out(__lowercase      )
       SCREAMING_SNAKE_CASE__     :				Any									=nn.silu(__lowercase      )
       SCREAMING_SNAKE_CASE__     :				Union[str, Any]									=self.conv_out(__lowercase      )
       SCREAMING_SNAKE_CASE__     :				Any									=jnp.transpose(__lowercase					,      (0, 3, 1, 2)      )

       if not return_dict:
             return (sample,)

       return FlaxUNetaDConditionOutput(sample=__lowercase      )







'''simple docstring'''
import inspect
import os
import sys
import unittest

import accelerate
from accelerate.test_utils import execute_subprocess_async, require_tpu






class      __SCREAMING_SNAKE_CASE      (			unittest.TestCase				):




 def      __magic_name__	(      self	:							int      )	->						Tuple:
       SCREAMING_SNAKE_CASE__     :				Optional[int]									=inspect.getfile(accelerate.test_utils      )
       SCREAMING_SNAKE_CASE__     :				Optional[Any]									=os.path.sep.join(mod_file.split(os.path.sep      )[:-1] + ['''scripts''', '''test_script.py''']      )
       SCREAMING_SNAKE_CASE__     :				Dict									=os.path.sep.join(inspect.getfile(self.__class__      ).split(os.path.sep      )[:-1]      )




 @require_tpu
 def      __magic_name__	(      self	:							int      )	->						Tuple:
       SCREAMING_SNAKE_CASE__     :				Any									=F"\n            {self.test_dir}/xla_spawn.py\n            --num_cores 8\n            {self.test_file_path}\n        ".split()
       SCREAMING_SNAKE_CASE__     :				List[str]									=[sys.executable] + distributed_args
       execute_subprocess_async(__lowercase					,      env=os.environ.copy()      )