
from typing import Dict, Iterable, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_DEFAULT_MEAN,
    IMAGENET_DEFAULT_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, logging


__lowerCamelCase							=       logging.get_logger(__name__)






class   UpperCAmelCase		(						_lowerCamelCase ):
   A__					:		Any						     =     ["pixel_values"]



   def __init__(self		:					List[str]    ,						snake_case__		:					bool = True    ,						snake_case__		:					Dict[str, int] = None    ,						snake_case__		:					PILImageResampling = PILImageResampling.BICUBIC    ,						snake_case__		:					bool = True    ,						snake_case__		:					Dict[str, int] = None    ,						snake_case__		:					bool = True    ,						snake_case__		:					Union[int, float] = 1 / 2_55    ,						snake_case__		:					bool = True    ,						snake_case__		:					Optional[Union[float, Iterable[float]]] = IMAGENET_DEFAULT_MEAN    ,						snake_case__		:					Optional[Union[float, Iterable[float]]] = IMAGENET_DEFAULT_STD    ,						**snake_case__		:					Union[str, Any]    ,						)		->	Dict:


          '''simple docstring'''


          super().__init__(**lowercase_ )
          snake_case       :   List[str]							=      size if size is not None else {"shortest_edge": 2_24}
          snake_case       :   str							=      get_size_dict(lowercase_    ,						default_to_square=lowercase_ )
          snake_case       :   Optional[Any]							=      crop_size if crop_size is not None else {"height": 2_24, "width": 2_24}
          snake_case       :   Optional[Any]							=      get_size_dict(lowercase_    ,						param_name="crop_size" )

          snake_case       :   Optional[Any]							=      do_resize
          snake_case       :   List[str]							=      size
          snake_case       :   str							=      resample
          snake_case       :   Optional[int]							=      do_center_crop
          snake_case       :   List[str]							=      crop_size
          snake_case       :   Tuple							=      do_rescale
          snake_case       :   Dict							=      rescale_factor
          snake_case       :   int							=      do_normalize
          snake_case       :   Tuple							=      image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
          snake_case       :   Union[str, Any]							=      image_std if image_std is not None else IMAGENET_DEFAULT_STD



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[Any]    ,						snake_case__		:					np.ndarray    ,						snake_case__		:					Dict[str, int]    ,						snake_case__		:					PILImageResampling = PILImageResampling.BICUBIC    ,						snake_case__		:					Optional[Union[str, ChannelDimension]] = None    ,						**snake_case__		:					Any    ,						)		->	Any:


          '''simple docstring'''


          snake_case       :   Optional[Any]							=      get_size_dict(lowercase_    ,						default_to_square=lowercase_ )
          # size_dict is a dict with either keys "height" and "width" or "shortest_edge"
          if "shortest_edge" in size:
                 snake_case       :   Any							=      int((2_56 / 2_24) * size["shortest_edge"] )
                 snake_case       :   Tuple							=      get_resize_output_image_size(lowercase_    ,						size=lowercase_    ,						default_to_square=lowercase_ )
                 snake_case       :   Union[str, Any]							=      {"height": output_size[0], "width": output_size[1]}
          if "height" not in size_dict or "width" not in size_dict:
                 raise ValueError(
                     f"""Size dict must have keys 'height' and 'width' or 'shortest_edge'. Got {size_dict.keys()}""" )
          return resize(
              lowercase_    ,						size=(size_dict["height"], size_dict["width"])    ,						resample=lowercase_    ,						data_format=lowercase_    ,						**lowercase_ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Tuple    ,						snake_case__		:					np.ndarray    ,						snake_case__		:					Dict[str, int]    ,						snake_case__		:					Optional[Union[str, ChannelDimension]] = None    ,						**snake_case__		:					Dict    ,						)		->	Any:


          '''simple docstring'''


          snake_case       :   Optional[Any]							=      get_size_dict(lowercase_ )
          if "height" not in size or "width" not in size:
                 raise ValueError(f"""Size dict must have keys 'height' and 'width'. Got {size.keys()}""" )
          return center_crop(lowercase_    ,						size=(size["height"], size["width"])    ,						data_format=lowercase_    ,						**lowercase_ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[Any]    ,						snake_case__		:					np.ndarray    ,						snake_case__		:					Union[int, float]    ,						snake_case__		:					Optional[Union[str, ChannelDimension]] = None    ,						**snake_case__		:					Optional[int]    ,						)		->	Optional[int]:


          '''simple docstring'''


          return rescale(lowercase_    ,						scale=lowercase_    ,						data_format=lowercase_    ,						**lowercase_ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Union[str, Any]    ,						snake_case__		:					np.ndarray    ,						snake_case__		:					Union[float, List[float]]    ,						snake_case__		:					Union[float, List[float]]    ,						snake_case__		:					Optional[Union[str, ChannelDimension]] = None    ,						**snake_case__		:					int    ,						)		->	Tuple:


          '''simple docstring'''


          return normalize(lowercase_    ,						mean=lowercase_    ,						std=lowercase_    ,						data_format=lowercase_    ,						**lowercase_ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[int]    ,						snake_case__		:					ImageInput    ,						snake_case__		:					Optional[bool] = None    ,						snake_case__		:					Optional[Dict[str, int]] = None    ,						snake_case__		:					PILImageResampling = None    ,						snake_case__		:					Optional[bool] = None    ,						snake_case__		:					Optional[Dict[str, int]] = None    ,						snake_case__		:					Optional[bool] = None    ,						snake_case__		:					Optional[float] = None    ,						snake_case__		:					Optional[bool] = None    ,						snake_case__		:					Optional[Union[float, Iterable[float]]] = None    ,						snake_case__		:					Optional[Union[float, Iterable[float]]] = None    ,						snake_case__		:					Optional[TensorType] = None    ,						snake_case__		:					ChannelDimension = ChannelDimension.FIRST    ,						**snake_case__		:					int    ,						)		->	Dict:


          '''simple docstring'''


          snake_case       :   int							=      do_resize if do_resize is not None else self.do_resize
          snake_case       :   Optional[int]							=      resample if resample is not None else self.resample
          snake_case       :   Dict							=      do_center_crop if do_center_crop is not None else self.do_center_crop
          snake_case       :   Optional[int]							=      do_rescale if do_rescale is not None else self.do_rescale
          snake_case       :   Any							=      rescale_factor if rescale_factor is not None else self.rescale_factor
          snake_case       :   Union[str, Any]							=      do_normalize if do_normalize is not None else self.do_normalize
          snake_case       :   List[Any]							=      image_mean if image_mean is not None else self.image_mean
          snake_case       :   Tuple							=      image_std if image_std is not None else self.image_std

          snake_case       :   List[str]							=      size if size is not None else self.size
          snake_case       :   List[str]							=      get_size_dict(lowercase_    ,						default_to_square=lowercase_ )
          snake_case       :   List[str]							=      crop_size if crop_size is not None else self.crop_size
          snake_case       :   Optional[Any]							=      get_size_dict(lowercase_    ,						param_name="crop_size" )

          snake_case       :   Optional[int]							=      make_list_of_images(lowercase_ )

          if not valid_images(lowercase_ ):
                 raise ValueError(
                     "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
                     "torch.Tensor, tf.Tensor or jax.ndarray." )

          if do_resize and size is None:
                 raise ValueError("Size must be specified if do_resize is True." )

          if do_center_crop and crop_size is None:
                 raise ValueError("Crop size must be specified if do_center_crop is True." )

          if do_rescale and rescale_factor is None:
                 raise ValueError("Rescale factor must be specified if do_rescale is True." )

          if do_normalize and (image_mean is None or image_std is None):
                 raise ValueError("Image mean and std must be specified if do_normalize is True." )

          # All transformations expect numpy arrays.
          snake_case       :   Optional[Any]							=      [to_numpy_array(lowercase_ ) for image in images]

          if do_resize:
                 snake_case       :   List[str]							=      [self.resize(lowercase_    ,						lowercase_    ,						lowercase_ ) for image in images]

          if do_center_crop:
                 snake_case       :   Optional[Any]							=      [self.center_crop(lowercase_    ,						lowercase_ ) for image in images]

          if do_rescale:
                 snake_case       :   List[Any]							=      [self.rescale(lowercase_    ,						lowercase_ ) for image in images]

          if do_normalize:
                 snake_case       :   List[str]							=      [self.normalize(lowercase_    ,						lowercase_    ,						lowercase_ ) for image in images]

          snake_case       :   Any							=      [to_channel_dimension_format(lowercase_    ,						lowercase_ ) for image in images]

          snake_case       :   List[Any]							=      {"pixel_values": images}
          return BatchFeature(data=lowercase_    ,						tensor_type=lowercase_ )











'''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 						a				(   _lowerCamelCase     ):
				snake_case_      =		42







@flax_register_to_config
class 						a				(   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, 1_280, 1_280)
				snake_case_      =		2
				snake_case_      =		8
				snake_case_      =		None
				snake_case_      =		1_280
				snake_case_      =		0.0
				snake_case_      =		False
				snake_case_      =		jnp.floataa
				snake_case_      =		True
				snake_case_      =		0
				snake_case_      =		False


				def   A_ (		self    :					Optional[int]       ,				lowercase_    :					jax.random.KeyArray					):
									# init input tensors
									snake_case_									=				(1, self.in_channels, self.sample_size, self.sample_size)
									snake_case_									=				jnp.zeros(lowercase_       ,				dtype=jnp.floataa					)
									snake_case_									=				jnp.ones((1,)       ,				dtype=jnp.intaa					)
									snake_case_									=				jnp.zeros((1, 1, self.cross_attention_dim)       ,				dtype=jnp.floataa					)

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

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


				def   A_ (		self    :					List[str]					):
									snake_case_									=				self.block_out_channels
									snake_case_									=				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.
									snake_case_									=				self.num_attention_heads or self.attention_head_dim

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

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

									snake_case_									=				self.only_cross_attention
									if isinstance(lowercase_       ,				lowercase_					):
														snake_case_									=				(only_cross_attention,) * len(self.down_block_types					)

									if isinstance(lowercase_       ,				lowercase_					):
														snake_case_									=				(num_attention_heads,) * len(self.down_block_types					)

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

														if down_block_type == "CrossAttnDownBlock2D":
																			snake_case_									=				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:
																			snake_case_									=				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_					)
									snake_case_									=				down_blocks

									# mid
									snake_case_									=				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
									snake_case_									=				[]
									snake_case_									=				list(reversed(lowercase_					)					)
									snake_case_									=				list(reversed(lowercase_					)					)
									snake_case_									=				list(reversed(lowercase_					)					)
									snake_case_									=				reversed_block_out_channels[0]
									for i, up_block_type in enumerate(self.up_block_types					):
														snake_case_									=				output_channel
														snake_case_									=				reversed_block_out_channels[i]
														snake_case_									=				reversed_block_out_channels[min(i + 1       ,				len(lowercase_					) - 1					)]

														snake_case_									=				i == len(lowercase_					) - 1

														if up_block_type == "CrossAttnUpBlock2D":
																			snake_case_									=				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:
																			snake_case_									=				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_					)
														snake_case_									=				output_channel
									snake_case_									=				up_blocks

									# out
									snake_case_									=				nn.GroupNorm(num_groups=32       ,				epsilon=1e-5					)
									snake_case_									=				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_    :					Union[str, Any]       ,				lowercase_    :					int       ,				lowercase_    :					Any       ,				lowercase_    :					int=None       ,				lowercase_    :					Any=None       ,				lowercase_    :					bool = True       ,				lowercase_    :					bool = False       ,				):
									# 1. time
									if not isinstance(lowercase_       ,				jnp.ndarray					):
														snake_case_									=				jnp.array([timesteps]       ,				dtype=jnp.intaa					)
									elif isinstance(lowercase_       ,				jnp.ndarray					) and len(timesteps.shape					) == 0:
														snake_case_									=				timesteps.astype(dtype=jnp.floataa					)
														snake_case_									=				jnp.expand_dims(lowercase_       ,				0					)

									snake_case_									=				self.time_proj(lowercase_					)
									snake_case_									=				self.time_embedding(lowercase_					)

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

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

									if down_block_additional_residuals is not None:
														snake_case_									=				()

														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,)

														snake_case_									=				new_down_block_res_samples

									# 4. mid
									snake_case_									=				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:
														snake_case_									=				down_block_res_samples[-(self.layers_per_block + 1) :]
														snake_case_									=				down_block_res_samples[: -(self.layers_per_block + 1)]
														if isinstance(lowercase_       ,				lowercase_					):
																			snake_case_									=				up_block(
																			    lowercase_       ,				temb=lowercase_       ,				encoder_hidden_states=lowercase_       ,				res_hidden_states_tuple=lowercase_       ,				deterministic=not train       ,				)
														else:
																			snake_case_									=				up_block(lowercase_       ,				temb=lowercase_       ,				res_hidden_states_tuple=lowercase_       ,				deterministic=not train					)

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

									if not return_dict:
														return (sample,)

									return FlaxUNetaDConditionOutput(sample=lowercase_					)

