

from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..utils import maybe_allow_in_graph
from .activations import get_activation
from .attention_processor import Attention
from .embeddings import CombinedTimestepLabelEmbeddings


@maybe_allow_in_graph
class 				SCREAMING_SNAKE_CASE__ (nn.Module					):





    def __init__(				self						,       a						,       a						,       a						,       a=0.0						,       a = None						,       a = "geglu"						,       a = None						,       a = False						,       a = False						,       a = False						,       a = False						,       a = True						,       a = "layer_norm"						,       a = False						,       ):
         super().__init__()
         lowercase__   :       Union[str, Any]				=	only_cross_attention

         lowercase__   :       str				=	(num_embeds_ada_norm is not None) and norm_type == 'ada_norm_zero'
         lowercase__   :       Tuple				=	(num_embeds_ada_norm is not None) and norm_type == 'ada_norm'

         if norm_type in ("ada_norm", "ada_norm_zero") and num_embeds_ada_norm is None:
              raise ValueError(
                  f"""`norm_type` is set to {norm_type}, but `num_embeds_ada_norm` is not defined. Please make sure to"""
                  f""" define `num_embeds_ada_norm` if setting `norm_type` to {norm_type}.""")

         # Define 3 blocks. Each block has its own normalization layer.
         # 1. Self-Attn
         if self.use_ada_layer_norm:
              lowercase__   :       Optional[Any]				=	AdaLayerNorm(a						,       a)
         elif self.use_ada_layer_norm_zero:
              lowercase__   :       List[Any]				=	AdaLayerNormZero(a						,       a)
         else:
              lowercase__   :       Any				=	nn.LayerNorm(a						,       elementwise_affine=a)
         lowercase__   :       Optional[Any]				=	Attention(
             query_dim=a						,       heads=a						,       dim_head=a						,       dropout=a						,       bias=a						,       cross_attention_dim=cross_attention_dim if only_cross_attention else None						,       upcast_attention=a						,       )

         # 2. Cross-Attn
         if cross_attention_dim is not None or double_self_attention:
              # We currently only use AdaLayerNormZero for self attention where there will only be one attention block.
              # I.e. the number of returned modulation chunks from AdaLayerZero would not make sense if returned during
              # the second cross attention block.
              lowercase__   :       Optional[int]				=	(
                  AdaLayerNorm(a						,       a)
                  if self.use_ada_layer_norm
                  else nn.LayerNorm(a						,       elementwise_affine=a)
              )
              lowercase__   :       Tuple				=	Attention(
                  query_dim=a						,       cross_attention_dim=cross_attention_dim if not double_self_attention else None						,       heads=a						,       dim_head=a						,       dropout=a						,       bias=a						,       upcast_attention=a						,       )  # is self-attn if encoder_hidden_states is none
         else:
              lowercase__   :       Dict				=	None
              lowercase__   :       int				=	None

         # 3. Feed-forward
         lowercase__   :       Union[str, Any]				=	nn.LayerNorm(a						,       elementwise_affine=a)
         lowercase__   :       List[str]				=	FeedForward(a						,       dropout=a						,       activation_fn=a						,       final_dropout=a)

         # let chunk size default to None
         lowercase__   :       Dict				=	None
         lowercase__   :       List[str]				=	0





    def 				snake_case_    (				self						,       a						,       a):
         # Sets chunk feed-forward
         lowercase__   :       List[Any]				=	chunk_size
         lowercase__   :       Union[str, Any]				=	dim





    def 				snake_case_    (				self						,       a						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       ):
         # Notice that normalization is always applied before the real computation in the following blocks.
         # 1. Self-Attention
         if self.use_ada_layer_norm:
              lowercase__   :       List[Any]				=	self.norma(a						,       a)
         elif self.use_ada_layer_norm_zero:
              lowercase__   :       List[Any]				=	self.norma(
                  a						,       a						,       a						,       hidden_dtype=hidden_states.dtype)
         else:
              lowercase__   :       str				=	self.norma(a)

         lowercase__   :       int				=	cross_attention_kwargs if cross_attention_kwargs is not None else {}

         lowercase__   :       List[Any]				=	self.attna(
             a						,       encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None						,       attention_mask=a						,       **a						,       )
         if self.use_ada_layer_norm_zero:
              lowercase__   :       Tuple				=	gate_msa.unsqueeze(1) * attn_output
         lowercase__   :       int				=	attn_output + hidden_states

         # 2. Cross-Attention
         if self.attna is not None:
              lowercase__   :       Optional[int]				=	(
                  self.norma(a						,       a) if self.use_ada_layer_norm else self.norma(a)
              )

              lowercase__   :       Any				=	self.attna(
                  a						,       encoder_hidden_states=a						,       attention_mask=a						,       **a						,       )
              lowercase__   :       Tuple				=	attn_output + hidden_states

         # 3. Feed-forward
         lowercase__   :       Dict				=	self.norma(a)

         if self.use_ada_layer_norm_zero:
              lowercase__   :       Optional[Any]				=	norm_hidden_states * (1 + scale_mlp[:, None]) + shift_mlp[:, None]

         if self._chunk_size is not None:
              # "feed_forward_chunk_size" can be used to save memory
              if norm_hidden_states.shape[self._chunk_dim] % self._chunk_size != 0:
                   raise ValueError(
                       f"""`hidden_states` dimension to be chunked: {norm_hidden_states.shape[self._chunk_dim]} has to be divisible by chunk size: {self._chunk_size}. Make sure to set an appropriate `chunk_size` when calling `unet.enable_forward_chunking`.""")

              lowercase__   :       List[str]				=	norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
              lowercase__   :       Optional[int]				=	torch.cat(
                  [self.ff(a) for hid_slice in norm_hidden_states.chunk(a						,       dim=self._chunk_dim)]						,       dim=self._chunk_dim						,       )
         else:
              lowercase__   :       Optional[Any]				=	self.ff(a)

         if self.use_ada_layer_norm_zero:
              lowercase__   :       str				=	gate_mlp.unsqueeze(1) * ff_output

         lowercase__   :       Any				=	ff_output + hidden_states

         return hidden_states



class 				SCREAMING_SNAKE_CASE__ (nn.Module					):





    def __init__(				self						,       a						,       a = None						,       a = 4						,       a = 0.0						,       a = "geglu"						,       a = False						,       ):
         super().__init__()
         lowercase__   :       Optional[Any]				=	int(dim * mult)
         lowercase__   :       List[Any]				=	dim_out if dim_out is not None else dim

         if activation_fn == "gelu":
              lowercase__   :       str				=	GELU(a						,       a)
         if activation_fn == "gelu-approximate":
              lowercase__   :       Optional[int]				=	GELU(a						,       a						,       approximate='tanh')
         elif activation_fn == "geglu":
              lowercase__   :       Union[str, Any]				=	GEGLU(a						,       a)
         elif activation_fn == "geglu-approximate":
              lowercase__   :       Optional[int]				=	ApproximateGELU(a						,       a)

         lowercase__   :       str				=	nn.ModuleList([])
         # project in
         self.net.append(a)
         # project dropout
         self.net.append(nn.Dropout(a))
         # project out
         self.net.append(nn.Linear(a						,       a))
         # FF as used in Vision Transformer, MLP-Mixer, etc. have a final dropout
         if final_dropout:
              self.net.append(nn.Dropout(a))





    def 				snake_case_    (				self						,       a):
         for module in self.net:
              lowercase__   :       Any				=	module(a)
         return hidden_states



class 				SCREAMING_SNAKE_CASE__ (nn.Module					):





    def __init__(				self						,       a						,       a						,       a = "none"):
         super().__init__()
         lowercase__   :       Any				=	nn.Linear(a						,       a)
         lowercase__   :       int				=	approximate





    def 				snake_case_    (				self						,       a):
         if gate.device.type != "mps":
              return F.gelu(a						,       approximate=self.approximate)
         # mps: gelu is not implemented for float16
         return F.gelu(gate.to(dtype=torch.floataa)						,       approximate=self.approximate).to(dtype=gate.dtype)





    def 				snake_case_    (				self						,       a):
         lowercase__   :       int				=	self.proj(a)
         lowercase__   :       List[Any]				=	self.gelu(a)
         return hidden_states



class 				SCREAMING_SNAKE_CASE__ (nn.Module					):





    def __init__(				self						,       a						,       a):
         super().__init__()
         lowercase__   :       Optional[Any]				=	nn.Linear(a						,       dim_out * 2)





    def 				snake_case_    (				self						,       a):
         if gate.device.type != "mps":
              return F.gelu(a)
         # mps: gelu is not implemented for float16
         return F.gelu(gate.to(dtype=torch.floataa)).to(dtype=gate.dtype)





    def 				snake_case_    (				self						,       a):
         lowercase__   :       int				=	self.proj(a).chunk(2						,       dim=-1)
         return hidden_states * self.gelu(a)



class 				SCREAMING_SNAKE_CASE__ (nn.Module					):





    def __init__(				self						,       a						,       a):
         super().__init__()
         lowercase__   :       Dict				=	nn.Linear(a						,       a)





    def 				snake_case_    (				self						,       a):
         lowercase__   :       int				=	self.proj(a)
         return x * torch.sigmoid(1.702 * x)



class 				SCREAMING_SNAKE_CASE__ (nn.Module					):





    def __init__(				self						,       a						,       a):
         super().__init__()
         lowercase__   :       Optional[int]				=	nn.Embedding(a						,       a)
         lowercase__   :       Dict				=	nn.SiLU()
         lowercase__   :       Tuple				=	nn.Linear(a						,       embedding_dim * 2)
         lowercase__   :       List[Any]				=	nn.LayerNorm(a						,       elementwise_affine=a)





    def 				snake_case_    (				self						,       a						,       a):
         lowercase__   :       Union[str, Any]				=	self.linear(self.silu(self.emb(a)))
         lowercase__   :       str				=	torch.chunk(a						,       2)
         lowercase__   :       Any				=	self.norm(a) * (1 + scale) + shift
         return x



class 				SCREAMING_SNAKE_CASE__ (nn.Module					):





    def __init__(				self						,       a						,       a):
         super().__init__()

         lowercase__   :       List[Any]				=	CombinedTimestepLabelEmbeddings(a						,       a)

         lowercase__   :       Any				=	nn.SiLU()
         lowercase__   :       Any				=	nn.Linear(a						,       6 * embedding_dim						,       bias=a)
         lowercase__   :       List[Any]				=	nn.LayerNorm(a						,       elementwise_affine=a						,       eps=1e-6)





    def 				snake_case_    (				self						,       a						,       a						,       a						,       a=None):
         lowercase__   :       Tuple				=	self.linear(self.silu(self.emb(a						,       a						,       hidden_dtype=a)))
         lowercase__   :       Optional[Any]				=	emb.chunk(6						,       dim=1)
         lowercase__   :       Optional[Any]				=	self.norm(a) * (1 + scale_msa[:, None]) + shift_msa[:, None]
         return x, gate_msa, shift_mlp, scale_mlp, gate_mlp



class 				SCREAMING_SNAKE_CASE__ (nn.Module					):





    def __init__(				self						,       a						,       a						,       a						,       a = None						,       a = 1e-5):
         super().__init__()
         lowercase__   :       Optional[int]				=	num_groups
         lowercase__   :       Dict				=	eps

         if act_fn is None:
              lowercase__   :       List[str]				=	None
         else:
              lowercase__   :       Dict				=	get_activation(a)

         lowercase__   :       Optional[Any]				=	nn.Linear(a						,       out_dim * 2)





    def 				snake_case_    (				self						,       a						,       a):
         if self.act:
              lowercase__   :       Optional[Any]				=	self.act(a)
         lowercase__   :       Optional[Any]				=	self.linear(a)
         lowercase__   :       Dict				=	emb[:, :, None, None]
         lowercase__   :       Union[str, Any]				=	emb.chunk(2						,       dim=1)

         lowercase__   :       List[str]				=	F.group_norm(a						,       self.num_groups						,       eps=self.eps)
         lowercase__   :       int				=	x * (1 + scale) + shift
         return x




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,
    convert_to_rgb,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    OPENAI_CLIP_MEAN,
    OPENAI_CLIP_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


snake_case_        =			logging.get_logger(__name__)


if is_vision_available():
       import PIL


class 				SCREAMING_SNAKE_CASE__ (__snake_case					):
    __lowerCamelCase						:	Optional[Any]														=					["""pixel_values"""]





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

         lowercase__   :       List[str]				=	do_resize
         lowercase__   :       List[Any]				=	size
         lowercase__   :       Tuple				=	resample
         lowercase__   :       int				=	do_center_crop
         lowercase__   :       Union[str, Any]				=	crop_size
         lowercase__   :       int				=	do_rescale
         lowercase__   :       List[str]				=	rescale_factor
         lowercase__   :       Tuple				=	do_normalize
         lowercase__   :       str				=	image_mean if image_mean is not None else OPENAI_CLIP_MEAN
         lowercase__   :       Optional[Any]				=	image_std if image_std is not None else OPENAI_CLIP_STD
         lowercase__   :       List[Any]				=	do_convert_rgb





    def 				snake_case_    (				self						,       a						,       a						,       a = PILImageResampling.BICUBIC						,       a = None						,       **a						,       ):
         lowercase__   :       str				=	get_size_dict(a						,       default_to_square=a)
         if "shortest_edge" not in size:
              raise ValueError(f"""The `size` parameter must contain the key `shortest_edge`. Got {size.keys()}""")
         lowercase__   :       str				=	get_resize_output_image_size(a						,       size=size['shortest_edge']						,       default_to_square=a)
         return resize(a						,       size=a						,       resample=a						,       data_format=a						,       **a)





    def 				snake_case_    (				self						,       a						,       a						,       a = None						,       **a						,       ):
         lowercase__   :       List[str]				=	get_size_dict(a)
         if "height" not in size or "width" not in size:
              raise ValueError(f"""The `size` parameter must contain the keys (height, width). Got {size.keys()}""")
         return center_crop(a						,       size=(size['height'], size['width'])						,       data_format=a						,       **a)





    def 				snake_case_    (				self						,       a						,       a						,       a = None						,       **a						,       ):
         return rescale(a						,       scale=a						,       data_format=a						,       **a)





    def 				snake_case_    (				self						,       a						,       a						,       a						,       a = None						,       **a						,       ):
         return normalize(a						,       mean=a						,       std=a						,       data_format=a						,       **a)





    def 				snake_case_    (				self						,       a						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       a = None						,       a = ChannelDimension.FIRST						,       **a						,       ):
         lowercase__   :       int				=	do_resize if do_resize is not None else self.do_resize
         lowercase__   :       Tuple				=	size if size is not None else self.size
         lowercase__   :       Union[str, Any]				=	get_size_dict(a						,       param_name='size'						,       default_to_square=a)
         lowercase__   :       Optional[Any]				=	resample if resample is not None else self.resample
         lowercase__   :       int				=	do_center_crop if do_center_crop is not None else self.do_center_crop
         lowercase__   :       List[Any]				=	crop_size if crop_size is not None else self.crop_size
         lowercase__   :       Union[str, Any]				=	get_size_dict(a						,       param_name='crop_size'						,       default_to_square=a)
         lowercase__   :       Dict				=	do_rescale if do_rescale is not None else self.do_rescale
         lowercase__   :       int				=	rescale_factor if rescale_factor is not None else self.rescale_factor
         lowercase__   :       List[str]				=	do_normalize if do_normalize is not None else self.do_normalize
         lowercase__   :       Optional[Any]				=	image_mean if image_mean is not None else self.image_mean
         lowercase__   :       List[str]				=	image_std if image_std is not None else self.image_std
         lowercase__   :       Any				=	do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

         lowercase__   :       str				=	make_list_of_images(a)

         if not valid_images(a):
              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.')

         # PIL RGBA images are converted to RGB
         if do_convert_rgb:
              lowercase__   :       str				=	[convert_to_rgb(a) for image in images]

         # All transformations expect numpy arrays.
         lowercase__   :       Dict				=	[to_numpy_array(a) for image in images]

         if do_resize:
              lowercase__   :       Tuple				=	[self.resize(image=a						,       size=a						,       resample=a) for image in images]

         if do_center_crop:
              lowercase__   :       List[str]				=	[self.center_crop(image=a						,       size=a) for image in images]

         if do_rescale:
              lowercase__   :       str				=	[self.rescale(image=a						,       scale=a) for image in images]

         if do_normalize:
              lowercase__   :       Tuple				=	[self.normalize(image=a						,       mean=a						,       std=a) for image in images]

         lowercase__   :       Optional[int]				=	[to_channel_dimension_format(a						,       a) for image in images]

         lowercase__   :       Dict				=	{'pixel_values': images}
         return BatchFeature(data=a						,       tensor_type=a)

