

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   lowercase_   (  nn.Module					):



      def __init__(      self				:    str      , A__				:    int      , A__				:    int      , A__				:    int      , A__				:    List[str]=0.0      , A__				:    Optional[int] = None      , A__				:    str = "geglu"      , A__				:    Optional[int] = None      , A__				:    bool = False      , A__				:    bool = False      , A__				:    bool = False      , A__				:    bool = False      , A__				:    bool = True      , A__				:    str = "layer_norm"      , A__				:    bool = False      , )   ->					str:
        super().__init__()
        _snake_case						=							only_cross_attention

        _snake_case						=							(num_embeds_ada_norm is not None) and norm_type == '''ada_norm_zero'''
        _snake_case						=							(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:
          _snake_case						=							AdaLayerNorm(A__      , A__		)
        elif self.use_ada_layer_norm_zero:
          _snake_case						=							AdaLayerNormZero(A__      , A__		)
        else:
          _snake_case						=							nn.LayerNorm(A__      , elementwise_affine=A__		)
        _snake_case						=							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.
          _snake_case						=							(
              AdaLayerNorm(A__      , A__		)
              if self.use_ada_layer_norm
              else nn.LayerNorm(A__      , elementwise_affine=A__		)
          )
          _snake_case						=							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:
          _snake_case						=							None
          _snake_case						=							None

        # 3. Feed-forward
        _snake_case						=							nn.LayerNorm(A__      , elementwise_affine=A__		)
        _snake_case						=							FeedForward(A__      , dropout=A__      , activation_fn=A__      , final_dropout=A__		)

        # let chunk size default to None
        _snake_case						=							None
        _snake_case						=							0



      def 						UpperCamelCase_		(      self				:    Tuple      , A__				:    Optional[int]      , A__				:    int		)   ->					int:
        # Sets chunk feed-forward
        _snake_case						=							chunk_size
        _snake_case						=							dim







      def 						UpperCamelCase_		(      self				:    Any      , A__				:    torch.FloatTensor      , A__				:    Optional[torch.FloatTensor] = None      , A__				:    Optional[torch.FloatTensor] = None      , A__				:    Optional[torch.FloatTensor] = None      , A__				:    Optional[torch.LongTensor] = None      , A__				:    Dict[str, Any] = None      , A__				:    Optional[torch.LongTensor] = None      , )   ->					List[Any]:
        # Notice that normalization is always applied before the real computation in the following blocks.
        # 1. Self-Attention
        if self.use_ada_layer_norm:
          _snake_case						=							self.norma(A__      , A__		)
        elif self.use_ada_layer_norm_zero:
          _snake_case,					_snake_case,					_snake_case,					_snake_case,					_snake_case						=							self.norma(
              A__      , A__      , A__      , hidden_dtype=hidden_states.dtype		)
        else:
          _snake_case						=							self.norma(A__		)

        _snake_case						=							cross_attention_kwargs if cross_attention_kwargs is not None else {}

        _snake_case						=							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:
          _snake_case						=							gate_msa.unsqueeze(1		) * attn_output
        _snake_case						=							attn_output + hidden_states

        # 2. Cross-Attention
        if self.attna is not None:
          _snake_case						=							(
              self.norma(A__      , A__		) if self.use_ada_layer_norm else self.norma(A__		)
          )

          _snake_case						=							self.attna(
              A__      , encoder_hidden_states=A__      , attention_mask=A__      , **A__      , )
          _snake_case						=							attn_output + hidden_states

        # 3. Feed-forward
        _snake_case						=							self.norma(A__		)

        if self.use_ada_layer_norm_zero:
          _snake_case						=							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`."""		)

          _snake_case						=							norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
          _snake_case						=							torch.cat(
              [self.ff(A__		) for hid_slice in norm_hidden_states.chunk(A__      , dim=self._chunk_dim		)]      , dim=self._chunk_dim      , )
        else:
          _snake_case						=							self.ff(A__		)

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

        _snake_case						=							ff_output + hidden_states

        return hidden_states



class   lowercase_   (  nn.Module					):



      def __init__(      self				:    Tuple      , A__				:    int      , A__				:    Optional[int] = None      , A__				:    int = 4      , A__				:    float = 0.0      , A__				:    str = "geglu"      , A__				:    bool = False      , )   ->					Optional[Any]:
        super().__init__()
        _snake_case						=							int(dim * mult		)
        _snake_case						=							dim_out if dim_out is not None else dim

        if activation_fn == "gelu":
          _snake_case						=							GELU(A__      , A__		)
        if activation_fn == "gelu-approximate":
          _snake_case						=							GELU(A__      , A__      , approximate='''tanh'''		)
        elif activation_fn == "geglu":
          _snake_case						=							GEGLU(A__      , A__		)
        elif activation_fn == "geglu-approximate":
          _snake_case						=							ApproximateGELU(A__      , A__		)

        _snake_case						=							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 						UpperCamelCase_		(      self				:    Optional[int]      , A__				:    int		)   ->					Dict:
        for module in self.net:
          _snake_case						=							module(A__		)
        return hidden_states



class   lowercase_   (  nn.Module					):



      def __init__(      self				:    List[Any]      , A__				:    int      , A__				:    int      , A__				:    str = "none"		)   ->					Optional[int]:
        super().__init__()
        _snake_case						=							nn.Linear(A__      , A__		)
        _snake_case						=							approximate



      def 						UpperCamelCase_		(      self				:    Any      , A__				:    Optional[Any]		)   ->					Optional[int]:
        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 						UpperCamelCase_		(      self				:    Union[str, Any]      , A__				:    Tuple		)   ->					Optional[int]:
        _snake_case						=							self.proj(A__		)
        _snake_case						=							self.gelu(A__		)
        return hidden_states



class   lowercase_   (  nn.Module					):



      def __init__(      self				:    Tuple      , A__				:    int      , A__				:    int		)   ->					int:
        super().__init__()
        _snake_case						=							nn.Linear(A__      , dim_out * 2		)



      def 						UpperCamelCase_		(      self				:    int      , A__				:    int		)   ->					str:
        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 						UpperCamelCase_		(      self				:    Union[str, Any]      , A__				:    List[Any]		)   ->					Optional[int]:
        _snake_case,					_snake_case						=							self.proj(A__		).chunk(2      , dim=-1		)
        return hidden_states * self.gelu(A__		)



class   lowercase_   (  nn.Module					):



      def __init__(      self				:    List[str]      , A__				:    int      , A__				:    int		)   ->					Tuple:
        super().__init__()
        _snake_case						=							nn.Linear(A__      , A__		)







      def 						UpperCamelCase_		(      self				:    Dict      , A__				:    Dict		)   ->					int:
        _snake_case						=							self.proj(A__		)
        return x * torch.sigmoid(1.702 * x		)



class   lowercase_   (  nn.Module					):



      def __init__(      self				:    int      , A__				:    str      , A__				:    Union[str, Any]		)   ->					int:
        super().__init__()
        _snake_case						=							nn.Embedding(A__      , A__		)
        _snake_case						=							nn.SiLU()
        _snake_case						=							nn.Linear(A__      , embedding_dim * 2		)
        _snake_case						=							nn.LayerNorm(A__      , elementwise_affine=A__		)







      def 						UpperCamelCase_		(      self				:    List[Any]      , A__				:    Tuple      , A__				:    Optional[int]		)   ->					Optional[int]:
        _snake_case						=							self.linear(self.silu(self.emb(A__		)		)		)
        _snake_case,					_snake_case						=							torch.chunk(A__      , 2		)
        _snake_case						=							self.norm(A__		) * (1 + scale) + shift
        return x



class   lowercase_   (  nn.Module					):



      def __init__(      self				:    List[str]      , A__				:    Tuple      , A__				:    List[str]		)   ->					Any:
        super().__init__()

        _snake_case						=							CombinedTimestepLabelEmbeddings(A__      , A__		)

        _snake_case						=							nn.SiLU()
        _snake_case						=							nn.Linear(A__      , 6 * embedding_dim      , bias=A__		)
        _snake_case						=							nn.LayerNorm(A__      , elementwise_affine=A__      , eps=1e-6		)







      def 						UpperCamelCase_		(      self				:    List[Any]      , A__				:    List[Any]      , A__				:    Union[str, Any]      , A__				:    Any      , A__				:    Tuple=None		)   ->					Optional[int]:
        _snake_case						=							self.linear(self.silu(self.emb(A__      , A__      , hidden_dtype=A__		)		)		)
        _snake_case,					_snake_case,					_snake_case,					_snake_case,					_snake_case,					_snake_case						=							emb.chunk(6      , dim=1		)
        _snake_case						=							self.norm(A__		) * (1 + scale_msa[:, None]) + shift_msa[:, None]
        return x, gate_msa, shift_mlp, scale_mlp, gate_mlp



class   lowercase_   (  nn.Module					):



      def __init__(      self				:    Dict      , A__				:    int      , A__				:    int      , A__				:    int      , A__				:    Optional[str] = None      , A__				:    float = 1e-5		)   ->					List[str]:
        super().__init__()
        _snake_case						=							num_groups
        _snake_case						=							eps

        if act_fn is None:
          _snake_case						=							None
        else:
          _snake_case						=							get_activation(A__		)

        _snake_case						=							nn.Linear(A__      , out_dim * 2		)







      def 						UpperCamelCase_		(      self				:    str      , A__				:    Dict      , A__				:    List[Any]		)   ->					str:
        if self.act:
          _snake_case						=							self.act(A__		)
        _snake_case						=							self.linear(A__		)
        _snake_case						=							emb[:, :, None, None]
        _snake_case,					_snake_case						=							emb.chunk(2      , dim=1		)

        _snake_case						=							F.group_norm(A__      , self.num_groups      , eps=self.eps		)
        _snake_case						=							x * (1 + scale) + shift
        return x




import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_video_inputs


if is_torch_available():
 import torch

if is_vision_available():
 from PIL import Image

 from transformers import VivitImageProcessor



class   lowercase_   (  unittest.TestCase					):



      def __init__(      self				:    Union[str, Any]      , A__				:    int      , A__				:    List[str]=7      , A__				:    Tuple=3      , A__				:    List[str]=10      , A__				:    Optional[int]=18      , A__				:    int=30      , A__				:    Tuple=400      , A__				:    Dict=True      , A__				:    str=None      , A__				:    str=True      , A__				:    List[str]=[0.5, 0.5, 0.5]      , A__				:    int=[0.5, 0.5, 0.5]      , A__				:    List[Any]=None      , )   ->					int:
        _snake_case						=							size if size is not None else {'''shortest_edge''': 18}
        _snake_case						=							crop_size if crop_size is not None else {'''height''': 18, '''width''': 18}

        _snake_case						=							parent
        _snake_case						=							batch_size
        _snake_case						=							num_channels
        _snake_case						=							num_frames
        _snake_case						=							image_size
        _snake_case						=							min_resolution
        _snake_case						=							max_resolution
        _snake_case						=							do_resize
        _snake_case						=							size
        _snake_case						=							do_normalize
        _snake_case						=							image_mean
        _snake_case						=							image_std
        _snake_case						=							crop_size







      def 						UpperCamelCase_		(      self				:    List[str]		)   ->					Union[str, Any]:
        return {
            "image_mean": self.image_mean,
            "image_std": self.image_std,
            "do_normalize": self.do_normalize,
            "do_resize": self.do_resize,
            "size": self.size,
            "crop_size": self.crop_size,
        }



@require_torch
@require_vision
class   lowercase_   (  __lowercase       ,							unittest.TestCase					):
      UpperCamelCase_      :		Tuple										=    VivitImageProcessor if is_vision_available() else None



      def 						UpperCamelCase_		(      self				:    Union[str, Any]		)   ->					List[str]:
        _snake_case						=							VivitImageProcessingTester(self		)



      @property
      def 						UpperCamelCase_		(      self				:    Union[str, Any]		)   ->					Optional[Any]:
        return self.image_processor_tester.prepare_image_processor_dict()



      def 						UpperCamelCase_		(      self				:    Optional[int]		)   ->					Optional[Any]:
        _snake_case						=							self.image_processing_class(**self.image_processor_dict		)
        self.assertTrue(hasattr(A__      , '''image_mean'''		)		)
        self.assertTrue(hasattr(A__      , '''image_std'''		)		)
        self.assertTrue(hasattr(A__      , '''do_normalize'''		)		)
        self.assertTrue(hasattr(A__      , '''do_resize'''		)		)
        self.assertTrue(hasattr(A__      , '''do_center_crop'''		)		)
        self.assertTrue(hasattr(A__      , '''size'''		)		)



      def 						UpperCamelCase_		(      self				:    int		)   ->					List[Any]:
        _snake_case						=							self.image_processing_class.from_dict(self.image_processor_dict		)
        self.assertEqual(image_processor.size      , {'''shortest_edge''': 18}		)
        self.assertEqual(image_processor.crop_size      , {'''height''': 18, '''width''': 18}		)

        _snake_case						=							self.image_processing_class.from_dict(self.image_processor_dict      , size=42      , crop_size=84		)
        self.assertEqual(image_processor.size      , {'''shortest_edge''': 42}		)
        self.assertEqual(image_processor.crop_size      , {'''height''': 84, '''width''': 84}		)



      def 						UpperCamelCase_		(      self				:    List[Any]		)   ->					Optional[Any]:
        # Initialize image_processing
        _snake_case						=							self.image_processing_class(**self.image_processor_dict		)
        # create random PIL videos
        _snake_case						=							prepare_video_inputs(self.image_processor_tester      , equal_resolution=A__		)
        for video in video_inputs:
          self.assertIsInstance(A__      , A__		)
          self.assertIsInstance(video[0]      , Image.Image		)

        # Test not batched input
        _snake_case						=							image_processing(video_inputs[0]      , return_tensors='''pt'''		).pixel_values
        self.assertEqual(
            encoded_videos.shape      , (
                1,
                self.image_processor_tester.num_frames,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size['''height'''],
                self.image_processor_tester.crop_size['''width'''],
            )      , )

        # Test batched
        _snake_case						=							image_processing(A__      , return_tensors='''pt'''		).pixel_values
        self.assertEqual(
            encoded_videos.shape      , (
                self.image_processor_tester.batch_size,
                self.image_processor_tester.num_frames,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size['''height'''],
                self.image_processor_tester.crop_size['''width'''],
            )      , )



      def 						UpperCamelCase_		(      self				:    Any		)   ->					List[str]:
        # Initialize image_processing
        _snake_case						=							self.image_processing_class(**self.image_processor_dict		)
        # create random numpy tensors
        _snake_case						=							prepare_video_inputs(self.image_processor_tester      , equal_resolution=A__      , numpify=A__		)
        for video in video_inputs:
          self.assertIsInstance(A__      , A__		)
          self.assertIsInstance(video[0]      , np.ndarray		)

        # Test not batched input
        _snake_case						=							image_processing(video_inputs[0]      , return_tensors='''pt'''		).pixel_values
        self.assertEqual(
            encoded_videos.shape      , (
                1,
                self.image_processor_tester.num_frames,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size['''height'''],
                self.image_processor_tester.crop_size['''width'''],
            )      , )

        # Test batched
        _snake_case						=							image_processing(A__      , return_tensors='''pt'''		).pixel_values
        self.assertEqual(
            encoded_videos.shape      , (
                self.image_processor_tester.batch_size,
                self.image_processor_tester.num_frames,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size['''height'''],
                self.image_processor_tester.crop_size['''width'''],
            )      , )







      def 						UpperCamelCase_		(      self				:    Optional[Any]		)   ->					int:
        # Initialize image_processing
        _snake_case						=							self.image_processing_class(**self.image_processor_dict		)
        # create random PyTorch tensors
        _snake_case						=							prepare_video_inputs(self.image_processor_tester      , equal_resolution=A__      , torchify=A__		)
        for video in video_inputs:
          self.assertIsInstance(A__      , A__		)
          self.assertIsInstance(video[0]      , torch.Tensor		)

        # Test not batched input
        _snake_case						=							image_processing(video_inputs[0]      , return_tensors='''pt'''		).pixel_values
        self.assertEqual(
            encoded_videos.shape      , (
                1,
                self.image_processor_tester.num_frames,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size['''height'''],
                self.image_processor_tester.crop_size['''width'''],
            )      , )

        # Test batched
        _snake_case						=							image_processing(A__      , return_tensors='''pt'''		).pixel_values
        self.assertEqual(
            encoded_videos.shape      , (
                self.image_processor_tester.batch_size,
                self.image_processor_tester.num_frames,
                self.image_processor_tester.num_channels,
                self.image_processor_tester.crop_size['''height'''],
                self.image_processor_tester.crop_size['''width'''],
            )      , )

