





'''simple docstring'''




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		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase=0.0		,      _UpperCAmelCase = None		,      _UpperCAmelCase = "geglu"		,      _UpperCAmelCase = None		,      _UpperCAmelCase = False		,      _UpperCAmelCase = False		,      _UpperCAmelCase = False		,      _UpperCAmelCase = False		,      _UpperCAmelCase = True		,      _UpperCAmelCase = "layer_norm"		,      _UpperCAmelCase = False		,      ):

          '''simple docstring'''






          super().__init__()
          __A		:					Optional[int]						=			only_cross_attention

          __A		:					int						=			(num_embeds_ada_norm is not None) and norm_type == 'ada_norm_zero'
          __A		:					Dict						=			(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:
               __A		:					Dict						=			AdaLayerNorm(__lowerCAmelCase		,      __lowerCAmelCase)
          elif self.use_ada_layer_norm_zero:
               __A		:					int						=			AdaLayerNormZero(__lowerCAmelCase		,      __lowerCAmelCase)
          else:
               __A		:					str						=			nn.LayerNorm(__lowerCAmelCase		,      elementwise_affine=__lowerCAmelCase)
          __A		:					List[str]						=			Attention(
              query_dim=__lowerCAmelCase		,      heads=__lowerCAmelCase		,      dim_head=__lowerCAmelCase		,      dropout=__lowerCAmelCase		,      bias=__lowerCAmelCase		,      cross_attention_dim=cross_attention_dim if only_cross_attention else None		,      upcast_attention=__lowerCAmelCase		,      )

          # 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.
               __A		:					Union[str, Any]						=			(
                   AdaLayerNorm(__lowerCAmelCase		,      __lowerCAmelCase)
                   if self.use_ada_layer_norm
                   else nn.LayerNorm(__lowerCAmelCase		,      elementwise_affine=__lowerCAmelCase)
               )
               __A		:					Optional[Any]						=			Attention(
                   query_dim=__lowerCAmelCase		,      cross_attention_dim=cross_attention_dim if not double_self_attention else None		,      heads=__lowerCAmelCase		,      dim_head=__lowerCAmelCase		,      dropout=__lowerCAmelCase		,      bias=__lowerCAmelCase		,      upcast_attention=__lowerCAmelCase		,      )  # is self-attn if encoder_hidden_states is none
          else:
               __A		:					int						=			None
               __A		:					Optional[int]						=			None

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

          # let chunk size default to None
          __A		:					Union[str, Any]						=			None
          __A		:					List[str]						=			0


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					Any						=			chunk_size
          __A		:					List[str]						=			dim


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase = None		,      _UpperCAmelCase = None		,      _UpperCAmelCase = None		,      _UpperCAmelCase = None		,      _UpperCAmelCase = None		,      _UpperCAmelCase = None		,      ):

          '''simple docstring'''






          if self.use_ada_layer_norm:
               __A		:					Dict						=			self.norma(__lowerCAmelCase		,      __lowerCAmelCase)
          elif self.use_ada_layer_norm_zero:
               __A					,__A					,__A					,__A					,__A		:					List[str]						=			self.norma(
                   __lowerCAmelCase		,      __lowerCAmelCase		,      __lowerCAmelCase		,      hidden_dtype=hidden_states.dtype)
          else:
               __A		:					Optional[int]						=			self.norma(__lowerCAmelCase)

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

          __A		:					int						=			self.attna(
              __lowerCAmelCase		,      encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None		,      attention_mask=__lowerCAmelCase		,      **__lowerCAmelCase		,      )
          if self.use_ada_layer_norm_zero:
               __A		:					Dict						=			gate_msa.unsqueeze(1) * attn_output
          __A		:					Any						=			attn_output + hidden_states

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

               __A		:					Tuple						=			self.attna(
                   __lowerCAmelCase		,      encoder_hidden_states=__lowerCAmelCase		,      attention_mask=__lowerCAmelCase		,      **__lowerCAmelCase		,      )
               __A		:					Optional[Any]						=			attn_output + hidden_states

          # 3. Feed-forward
          __A		:					Tuple						=			self.norma(__lowerCAmelCase)

          if self.use_ada_layer_norm_zero:
               __A		:					Tuple						=			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`.')

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

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

          __A		:					Any						=			ff_output + hidden_states

          return hidden_states




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase = None		,      _UpperCAmelCase = 4		,      _UpperCAmelCase = 0.0		,      _UpperCAmelCase = "geglu"		,      _UpperCAmelCase = False		,      ):

          '''simple docstring'''






          super().__init__()
          __A		:					Tuple						=			int(dim * mult)
          __A		:					Dict						=			dim_out if dim_out is not None else dim

          if activation_fn == "gelu":
               __A		:					str						=			GELU(__lowerCAmelCase		,      __lowerCAmelCase)
          if activation_fn == "gelu-approximate":
               __A		:					List[Any]						=			GELU(__lowerCAmelCase		,      __lowerCAmelCase		,      approximate='tanh')
          elif activation_fn == "geglu":
               __A		:					Any						=			GEGLU(__lowerCAmelCase		,      __lowerCAmelCase)
          elif activation_fn == "geglu-approximate":
               __A		:					Dict						=			ApproximateGELU(__lowerCAmelCase		,      __lowerCAmelCase)

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


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          for module in self.net:
               __A		:					Optional[int]						=			module(__lowerCAmelCase)
          return hidden_states




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase = "none"):

          '''simple docstring'''






          super().__init__()
          __A		:					Optional[Any]						=			nn.Linear(__lowerCAmelCase		,      __lowerCAmelCase)
          __A		:					List[str]						=			approximate


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          if gate.device.type != "mps":
               return F.gelu(__lowerCAmelCase		,      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 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					str						=			self.proj(__lowerCAmelCase)
          __A		:					str						=			self.gelu(__lowerCAmelCase)
          return hidden_states




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          super().__init__()
          __A		:					Optional[Any]						=			nn.Linear(__lowerCAmelCase		,      dim_out * 2)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






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


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          __A					,__A		:					List[str]						=			self.proj(__lowerCAmelCase).chunk(2		,      dim=-1)
          return hidden_states * self.gelu(__lowerCAmelCase)




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          super().__init__()
          __A		:					List[Any]						=			nn.Linear(__lowerCAmelCase		,      __lowerCAmelCase)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					List[Any]						=			self.proj(__lowerCAmelCase)
          return x * torch.sigmoid(1.702 * x)




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          super().__init__()
          __A		:					Union[str, Any]						=			nn.Embedding(__lowerCAmelCase		,      __lowerCAmelCase)
          __A		:					Dict						=			nn.SiLU()
          __A		:					List[Any]						=			nn.Linear(__lowerCAmelCase		,      embedding_dim * 2)
          __A		:					Optional[int]						=			nn.LayerNorm(__lowerCAmelCase		,      elementwise_affine=__lowerCAmelCase)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					str						=			self.linear(self.silu(self.emb(__lowerCAmelCase)))
          __A					,__A		:					str						=			torch.chunk(__lowerCAmelCase		,      2)
          __A		:					str						=			self.norm(__lowerCAmelCase) * (1 + scale) + shift
          return x




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          super().__init__()

          __A		:					Dict						=			CombinedTimestepLabelEmbeddings(__lowerCAmelCase		,      __lowerCAmelCase)

          __A		:					Dict						=			nn.SiLU()
          __A		:					Any						=			nn.Linear(__lowerCAmelCase		,      6 * embedding_dim		,      bias=__lowerCAmelCase)
          __A		:					Dict						=			nn.LayerNorm(__lowerCAmelCase		,      elementwise_affine=__lowerCAmelCase		,      eps=1e-6)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase=None):

          '''simple docstring'''






          __A		:					List[str]						=			self.linear(self.silu(self.emb(__lowerCAmelCase		,      __lowerCAmelCase		,      hidden_dtype=__lowerCAmelCase)))
          __A					,__A					,__A					,__A					,__A					,__A		:					int						=			emb.chunk(6		,      dim=1)
          __A		:					Any						=			self.norm(__lowerCAmelCase) * (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		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase = None		,      _UpperCAmelCase = 1e-5):

          '''simple docstring'''






          super().__init__()
          __A		:					Tuple						=			num_groups
          __A		:					Optional[int]						=			eps

          if act_fn is None:
               __A		:					Any						=			None
          else:
               __A		:					List[Any]						=			get_activation(__lowerCAmelCase)

          __A		:					Any						=			nn.Linear(__lowerCAmelCase		,      out_dim * 2)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          if self.act:
               __A		:					Tuple						=			self.act(__lowerCAmelCase)
          __A		:					int						=			self.linear(__lowerCAmelCase)
          __A		:					Union[str, Any]						=			emb[:, :, None, None]
          __A					,__A		:					str						=			emb.chunk(2		,      dim=1)

          __A		:					Union[str, Any]						=			F.group_norm(__lowerCAmelCase		,      self.num_groups		,      eps=self.eps)
          __A		:					int						=			x * (1 + scale) + shift
          return x




import unittest

from transformers import is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
   from PIL import Image
else:


   class 	lowerCamelCase__	:



    """simple docstring"""

    @staticmethod
    def     snake_case_      (    *__lowerCAmelCase					:					Optional[Any]			,						**__lowerCAmelCase					:					Any      )   ->	Any:
       pass


@is_pipeline_test
@require_vision
class 	lowerCamelCase__	(       unittest.TestCase):



 """simple docstring"""

 @require_torch
 def     snake_case_      (    self					:					Tuple      )   ->	Tuple:
    _A      =       pipeline(
        model='''hf-internal-testing/tiny-random-clip-zero-shot-image-classification'''			,						)
    _A      =       Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
    _A      =       image_classifier(__lowerCAmelCase			,						candidate_labels=['''a''', '''b''', '''c''']      )

    # The floating scores are so close, we enter floating error approximation and the order is not guaranteed across
    # python and torch versions.
    self.assertIn(
        nested_simplify(__lowerCAmelCase      )			,						[
            [{'''score''': 0.333, '''label''': '''a'''}, {'''score''': 0.333, '''label''': '''b'''}, {'''score''': 0.333, '''label''': '''c'''}],
            [{'''score''': 0.333, '''label''': '''a'''}, {'''score''': 0.333, '''label''': '''c'''}, {'''score''': 0.333, '''label''': '''b'''}],
        ]			,						)

    _A      =       image_classifier([image] * 5			,						candidate_labels=['''A''', '''B''', '''C''']			,						batch_size=2      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
        ]			,						)

 @require_tf
 def     snake_case_      (    self					:					int      )   ->	Optional[int]:
    _A      =       pipeline(
        model='''hf-internal-testing/tiny-random-clip-zero-shot-image-classification'''			,						framework='''tf'''      )
    _A      =       Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
    _A      =       image_classifier(__lowerCAmelCase			,						candidate_labels=['''a''', '''b''', '''c''']      )

    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[{'''score''': 0.333, '''label''': '''a'''}, {'''score''': 0.333, '''label''': '''b'''}, {'''score''': 0.333, '''label''': '''c'''}]			,						)

    _A      =       image_classifier([image] * 5			,						candidate_labels=['''A''', '''B''', '''C''']			,						batch_size=2      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
            [
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
                {'''score''': 0.333, '''label''': ANY(__lowerCAmelCase      )},
            ],
        ]			,						)

 @slow
 @require_torch
 def     snake_case_      (    self					:					Optional[int]      )   ->	int:
    _A      =       pipeline(
        task='''zero-shot-image-classification'''			,						model='''openai/clip-vit-base-patch32'''			,						)
    # This is an image of 2 cats with remotes and no planes
    _A      =       Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
    _A      =       image_classifier(__lowerCAmelCase			,						candidate_labels=['''cat''', '''plane''', '''remote''']      )

    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            {'''score''': 0.511, '''label''': '''remote'''},
            {'''score''': 0.485, '''label''': '''cat'''},
            {'''score''': 0.004, '''label''': '''plane'''},
        ]			,						)

    _A      =       image_classifier([image] * 5			,						candidate_labels=['''cat''', '''plane''', '''remote''']			,						batch_size=2      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            [
                {'''score''': 0.511, '''label''': '''remote'''},
                {'''score''': 0.485, '''label''': '''cat'''},
                {'''score''': 0.004, '''label''': '''plane'''},
            ],
        ]
        * 5			,						)

 @slow
 @require_tf
 def     snake_case_      (    self					:					Optional[int]      )   ->	Dict:
    _A      =       pipeline(
        task='''zero-shot-image-classification'''			,						model='''openai/clip-vit-base-patch32'''			,						framework='''tf'''      )
    # This is an image of 2 cats with remotes and no planes
    _A      =       Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
    _A      =       image_classifier(__lowerCAmelCase			,						candidate_labels=['''cat''', '''plane''', '''remote''']      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            {'''score''': 0.511, '''label''': '''remote'''},
            {'''score''': 0.485, '''label''': '''cat'''},
            {'''score''': 0.004, '''label''': '''plane'''},
        ]			,						)

    _A      =       image_classifier([image] * 5			,						candidate_labels=['''cat''', '''plane''', '''remote''']			,						batch_size=2      )
    self.assertEqual(
        nested_simplify(__lowerCAmelCase      )			,						[
            [
                {'''score''': 0.511, '''label''': '''remote'''},
                {'''score''': 0.485, '''label''': '''cat'''},
                {'''score''': 0.004, '''label''': '''plane'''},
            ],
        ]
        * 5			,						)



