




from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block

@dataclass
class        _lowerCamelCase  (	_SCREAMING_SNAKE_CASE      ):

      """simple docstring"""

      lowerCAmelCase__	:       torch.FloatTensor







class        _lowerCamelCase  (	_SCREAMING_SNAKE_CASE							,	_SCREAMING_SNAKE_CASE      ):

      """simple docstring"""

      @register_to_config
      def __init__(   self    :       Optional[Any]       ,				snake_case    :       Optional[Union[int, Tuple[int, int]]] = None       ,				snake_case    :       int = 3       ,				snake_case    :       int = 3       ,				snake_case    :       bool = False       ,				snake_case    :       str = "positional"       ,				snake_case    :       int = 0       ,				snake_case    :       bool = True       ,				snake_case    :       Tuple[str] = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")       ,				snake_case    :       Tuple[str] = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")       ,				snake_case    :       Tuple[int] = (224, 448, 672, 896)       ,				snake_case    :       int = 2       ,				snake_case    :       float = 1       ,				snake_case    :       int = 1       ,				snake_case    :       str = "conv"       ,				snake_case    :       str = "conv"       ,				snake_case    :       str = "silu"       ,				snake_case    :       Optional[int] = 8       ,				snake_case    :       int = 32       ,				snake_case    :       float = 1E-5       ,				snake_case    :       str = "default"       ,				snake_case    :       bool = True       ,				snake_case    :       Optional[str] = None       ,				snake_case    :       Optional[int] = None       ,				):
          super().__init__()

          __UpperCamelCase     						=  sample_size
          __UpperCamelCase     						=  block_out_channels[0] * 4

          # Check inputs
          if len(snake_case ) != len(snake_case ):
              raise ValueError(
                  F"Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}." )

          if len(snake_case ) != len(snake_case ):
              raise ValueError(
                  F"Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}." )

          # input
          __UpperCamelCase     						=  nn.Convad(snake_case       ,				block_out_channels[0]       ,				kernel_size=3       ,				padding=(1, 1) )

          # time
          if time_embedding_type == "fourier":
              __UpperCamelCase     						=  GaussianFourierProjection(embedding_size=block_out_channels[0]       ,				scale=16 )
              __UpperCamelCase     						=  2 * block_out_channels[0]
          elif time_embedding_type == "positional":
              __UpperCamelCase     						=  Timesteps(block_out_channels[0]       ,				snake_case       ,				snake_case )
              __UpperCamelCase     						=  block_out_channels[0]

          __UpperCamelCase     						=  TimestepEmbedding(snake_case       ,				snake_case )

          # class embedding
          if class_embed_type is None and num_class_embeds is not None:
              __UpperCamelCase     						=  nn.Embedding(snake_case       ,				snake_case )
          elif class_embed_type == "timestep":
              __UpperCamelCase     						=  TimestepEmbedding(snake_case       ,				snake_case )
          elif class_embed_type == "identity":
              __UpperCamelCase     						=  nn.Identity(snake_case       ,				snake_case )
          else:
              __UpperCamelCase     						=  None

          __UpperCamelCase     						=  nn.ModuleList([] )
          __UpperCamelCase     						=  None
          __UpperCamelCase     						=  nn.ModuleList([] )

          # down
          __UpperCamelCase     						=  block_out_channels[0]
          for i, down_block_type in enumerate(snake_case ):
              __UpperCamelCase     						=  output_channel
              __UpperCamelCase     						=  block_out_channels[i]
              __UpperCamelCase     						=  i == len(snake_case ) - 1

              __UpperCamelCase     						=  get_down_block(
                  snake_case       ,				num_layers=snake_case       ,				in_channels=snake_case       ,				out_channels=snake_case       ,				temb_channels=snake_case       ,				add_downsample=not is_final_block       ,				resnet_eps=snake_case       ,				resnet_act_fn=snake_case       ,				resnet_groups=snake_case       ,				attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel       ,				downsample_padding=snake_case       ,				resnet_time_scale_shift=snake_case       ,				downsample_type=snake_case       ,				)
              self.down_blocks.append(snake_case )

          # mid
          __UpperCamelCase     						=  UNetMidBlockaD(
              in_channels=block_out_channels[-1]       ,				temb_channels=snake_case       ,				resnet_eps=snake_case       ,				resnet_act_fn=snake_case       ,				output_scale_factor=snake_case       ,				resnet_time_scale_shift=snake_case       ,				attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1]       ,				resnet_groups=snake_case       ,				add_attention=snake_case       ,				)

          # up
          __UpperCamelCase     						=  list(reversed(snake_case ) )
          __UpperCamelCase     						=  reversed_block_out_channels[0]
          for i, up_block_type in enumerate(snake_case ):
              __UpperCamelCase     						=  output_channel
              __UpperCamelCase     						=  reversed_block_out_channels[i]
              __UpperCamelCase     						=  reversed_block_out_channels[min(i + 1       ,				len(snake_case ) - 1 )]

              __UpperCamelCase     						=  i == len(snake_case ) - 1

              __UpperCamelCase     						=  get_up_block(
                  snake_case       ,				num_layers=layers_per_block + 1       ,				in_channels=snake_case       ,				out_channels=snake_case       ,				prev_output_channel=snake_case       ,				temb_channels=snake_case       ,				add_upsample=not is_final_block       ,				resnet_eps=snake_case       ,				resnet_act_fn=snake_case       ,				resnet_groups=snake_case       ,				attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel       ,				resnet_time_scale_shift=snake_case       ,				upsample_type=snake_case       ,				)
              self.up_blocks.append(snake_case )
              __UpperCamelCase     						=  output_channel

          # out
          __UpperCamelCase     						=  norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4       ,				32 )
          __UpperCamelCase     						=  nn.GroupNorm(num_channels=block_out_channels[0]       ,				num_groups=snake_case       ,				eps=snake_case )
          __UpperCamelCase     						=  nn.SiLU()
          __UpperCamelCase     						=  nn.Convad(block_out_channels[0]       ,				snake_case       ,				kernel_size=3       ,				padding=1 )







      def 					snake_case		(   self    :       Any       ,				snake_case    :       torch.FloatTensor       ,				snake_case    :       Union[torch.Tensor, float, int]       ,				snake_case    :       Optional[torch.Tensor] = None       ,				snake_case    :       bool = True       ,				):
          # 0. center input if necessary
          if self.config.center_input_sample:
              __UpperCamelCase     						=  2 * sample - 1.0

          # 1. time
          __UpperCamelCase     						=  timestep
          if not torch.is_tensor(snake_case ):
              __UpperCamelCase     						=  torch.tensor([timesteps]       ,				dtype=torch.long       ,				device=sample.device )
          elif torch.is_tensor(snake_case ) and len(timesteps.shape ) == 0:
              __UpperCamelCase     						=  timesteps[None].to(sample.device )

          # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
          __UpperCamelCase     						=  timesteps * torch.ones(sample.shape[0]       ,				dtype=timesteps.dtype       ,				device=timesteps.device )

          __UpperCamelCase     						=  self.time_proj(snake_case )

          # timesteps does not contain any weights and will always return f32 tensors
          # but time_embedding might actually be running in fp16. so we need to cast here.
          # there might be better ways to encapsulate this.
          __UpperCamelCase     						=  t_emb.to(dtype=self.dtype )
          __UpperCamelCase     						=  self.time_embedding(snake_case )

          if self.class_embedding is not None:
              if class_labels is None:
                  raise ValueError('''class_labels should be provided when doing class conditioning''' )

              if self.config.class_embed_type == "timestep":
                  __UpperCamelCase     						=  self.time_proj(snake_case )

              __UpperCamelCase     						=  self.class_embedding(snake_case ).to(dtype=self.dtype )
              __UpperCamelCase     						=  emb + class_emb

          # 2. pre-process
          __UpperCamelCase     						=  sample
          __UpperCamelCase     						=  self.conv_in(snake_case )

          # 3. down
          __UpperCamelCase     						=  (sample,)
          for downsample_block in self.down_blocks:
              if hasattr(snake_case       ,				'''skip_conv''' ):
                  __UpperCamelCase						,					__UpperCamelCase						,					__UpperCamelCase     						=  downsample_block(
                      hidden_states=snake_case       ,				temb=snake_case       ,				skip_sample=snake_case )
              else:
                  __UpperCamelCase						,					__UpperCamelCase     						=  downsample_block(hidden_states=snake_case       ,				temb=snake_case )

              down_block_res_samples += res_samples

          # 4. mid
          __UpperCamelCase     						=  self.mid_block(snake_case       ,				snake_case )

          # 5. up
          __UpperCamelCase     						=  None
          for upsample_block in self.up_blocks:
              __UpperCamelCase     						=  down_block_res_samples[-len(upsample_block.resnets ) :]
              __UpperCamelCase     						=  down_block_res_samples[: -len(upsample_block.resnets )]

              if hasattr(snake_case       ,				'''skip_conv''' ):
                  __UpperCamelCase						,					__UpperCamelCase     						=  upsample_block(snake_case       ,				snake_case       ,				snake_case       ,				snake_case )
              else:
                  __UpperCamelCase     						=  upsample_block(snake_case       ,				snake_case       ,				snake_case )

        # 6. post-process
          __UpperCamelCase     						=  self.conv_norm_out(snake_case )
          __UpperCamelCase     						=  self.conv_act(snake_case )
          __UpperCamelCase     						=  self.conv_out(snake_case )

          if skip_sample is not None:
              sample += skip_sample

          if self.config.time_embedding_type == "fourier":
              __UpperCamelCase     						=  timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:] ))) )
              __UpperCamelCase     						=  sample / timesteps

          if not return_dict:
              return (sample,)

          return UNetaDOutput(sample=snake_case )












from typing import Dict, List, Optional, Union

import numpy as np

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


if is_vision_available():
 import PIL


a_      						=   logging.get_logger(__name__)

class        _lowerCamelCase  (	_SCREAMING_SNAKE_CASE      ):

      """simple docstring"""

      lowerCAmelCase__	:       Union[str, Any]			=    ["pixel_values"]







      def __init__(   self    :       Union[str, Any]       ,				snake_case    :       bool = True       ,				snake_case    :       Dict[str, int] = None       ,				snake_case    :       bool = True       ,				snake_case    :       Dict[str, int] = None       ,				snake_case    :       PILImageResampling = PILImageResampling.BICUBIC       ,				snake_case    :       bool = True       ,				snake_case    :       Union[int, float] = 1 / 255       ,				snake_case    :       bool = True       ,				snake_case    :       Optional[Union[float, List[float]]] = None       ,				snake_case    :       Optional[Union[float, List[float]]] = None       ,				**snake_case    :       Union[str, Any]       ,				):
          super().__init__(**snake_case )
          __UpperCamelCase     						=  crop_size if crop_size is not None else {'''height''': 256, '''width''': 256}
          __UpperCamelCase     						=  get_size_dict(snake_case       ,				param_name='''crop_size''' )
          __UpperCamelCase     						=  size if size is not None else {'''height''': 224, '''width''': 224}
          __UpperCamelCase     						=  get_size_dict(snake_case )

          __UpperCamelCase     						=  do_center_crop
          __UpperCamelCase     						=  crop_size
          __UpperCamelCase     						=  do_resize
          __UpperCamelCase     						=  size
          __UpperCamelCase     						=  resample
          __UpperCamelCase     						=  do_rescale
          __UpperCamelCase     						=  rescale_factor
          __UpperCamelCase     						=  do_normalize
          __UpperCamelCase     						=  image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
          __UpperCamelCase     						=  image_std if image_std is not None else IMAGENET_DEFAULT_STD







      def 					snake_case		(   self    :       Optional[Any]       ,				snake_case    :       np.ndarray       ,				snake_case    :       Dict[str, int]       ,				snake_case    :       Optional[int] = None       ,				snake_case    :       Optional[Union[str, ChannelDimension]] = None       ,				**snake_case    :       List[Any]       ,				):
          __UpperCamelCase     						=  self.size if size is None else size
          __UpperCamelCase     						=  get_size_dict(snake_case )
          __UpperCamelCase     						=  get_size_dict(snake_case       ,				param_name='''crop_size''' )

          __UpperCamelCase						,					__UpperCamelCase     						=  get_image_size(snake_case )
          __UpperCamelCase     						=  min(snake_case       ,				snake_case )
          __UpperCamelCase     						=  (size['''height'''] / crop_size['''height''']) * min_dim
          __UpperCamelCase     						=  (size['''width'''] / crop_size['''width''']) * min_dim
          return center_crop(snake_case       ,				size=(cropped_height, cropped_width)       ,				data_format=snake_case       ,				**snake_case )







      def 					snake_case		(   self    :       Union[str, Any]       ,				snake_case    :       np.ndarray       ,				snake_case    :       Dict[str, int]       ,				snake_case    :       PILImageResampling = PIL.Image.BICUBIC       ,				snake_case    :       Optional[Union[str, ChannelDimension]] = None       ,				**snake_case    :       List[str]       ,				):
          __UpperCamelCase     						=  get_size_dict(snake_case )
          if "height" not in size or "width" not in size:
              raise ValueError(F"The size dictionary must contain the keys 'height' and 'width'. Got {size.keys()}" )
          return resize(
              snake_case       ,				size=(size['''height'''], size['''width'''])       ,				resample=snake_case       ,				data_format=snake_case       ,				**snake_case )







      def 					snake_case		(   self    :       Union[str, Any]       ,				snake_case    :       np.ndarray       ,				snake_case    :       Union[int, float]       ,				snake_case    :       Optional[Union[str, ChannelDimension]] = None       ,				**snake_case    :       Optional[int]       ,				):
          return rescale(snake_case       ,				scale=snake_case       ,				data_format=snake_case       ,				**snake_case )







      def 					snake_case		(   self    :       Optional[int]       ,				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    :       Any       ,				):
          return normalize(snake_case       ,				mean=snake_case       ,				std=snake_case       ,				data_format=snake_case       ,				**snake_case )







      def 					snake_case		(   self    :       List[Any]       ,				snake_case    :       ImageInput       ,				snake_case    :       Optional[bool] = None       ,				snake_case    :       Optional[Dict[str, int]] = None       ,				snake_case    :       Optional[bool] = None       ,				snake_case    :       Optional[Dict[str, int]] = None       ,				snake_case    :       PILImageResampling = None       ,				snake_case    :       Optional[bool] = None       ,				snake_case    :       Optional[float] = None       ,				snake_case    :       Optional[bool] = None       ,				snake_case    :       Optional[Union[float, List[float]]] = None       ,				snake_case    :       Optional[Union[float, List[float]]] = None       ,				snake_case    :       Optional[Union[str, TensorType]] = None       ,				snake_case    :       ChannelDimension = ChannelDimension.FIRST       ,				**snake_case    :       Tuple       ,				):
          __UpperCamelCase     						=  do_center_crop if do_center_crop is not None else self.do_center_crop
          __UpperCamelCase     						=  crop_size if crop_size is not None else self.crop_size
          __UpperCamelCase     						=  get_size_dict(snake_case       ,				param_name='''crop_size''' )
          __UpperCamelCase     						=  do_resize if do_resize is not None else self.do_resize
          __UpperCamelCase     						=  size if size is not None else self.size
          __UpperCamelCase     						=  get_size_dict(snake_case )
          __UpperCamelCase     						=  resample if resample is not None else self.resample
          __UpperCamelCase     						=  do_rescale if do_rescale is not None else self.do_rescale
          __UpperCamelCase     						=  rescale_factor if rescale_factor is not None else self.rescale_factor
          __UpperCamelCase     						=  do_normalize if do_normalize is not None else self.do_normalize
          __UpperCamelCase     						=  image_mean if image_mean is not None else self.image_mean
          __UpperCamelCase     						=  image_std if image_std is not None else self.image_std

          __UpperCamelCase     						=  make_list_of_images(snake_case )

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

          if do_center_crop and crop_size is None:
              raise ValueError('''If `do_center_crop` is set to `True`, `crop_size` must be provided.''' )

          if do_resize and size is None:
              raise ValueError('''Size must be specified if do_resize 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 image standard deviation must be specified if do_normalize is True.''' )

          # All transformations expect numpy arrays.
          __UpperCamelCase     						=  [to_numpy_array(snake_case ) for image in images]

          if do_center_crop:
              __UpperCamelCase     						=  [self.center_crop(snake_case       ,				snake_case       ,				size=snake_case ) for image in images]

          if do_resize:
              __UpperCamelCase     						=  [self.resize(image=snake_case       ,				size=snake_case       ,				resample=snake_case ) for image in images]

          if do_rescale:
              __UpperCamelCase     						=  [self.rescale(image=snake_case       ,				scale=snake_case ) for image in images]

          if do_normalize:
              __UpperCamelCase     						=  [self.normalize(image=snake_case       ,				mean=snake_case       ,				std=snake_case ) for image in images]

          __UpperCamelCase     						=  [to_channel_dimension_format(snake_case       ,				snake_case ) for image in images]

          __UpperCamelCase     						=  {'''pixel_values''': images}
          return BatchFeature(data=snake_case       ,				tensor_type=snake_case )






