


import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel

from ...utils import logging


_SCREAMING_SNAKE_CASE			:       str      =						logging.get_logger(__name__)







def     _lowercase							(  __lowerCamelCase      :       Union[str, Any]		,__lowerCamelCase      :       List[str]      )	-> List[Any]:





     '''simple docstring'''





     UpperCamelCase__				:	Union[str, Any]										=  nn.functional.normalize(UpperCamelCase__      )
     UpperCamelCase__				:	Tuple										=  nn.functional.normalize(UpperCamelCase__      )
     return torch.mm(UpperCamelCase__		,normalized_text_embeds.t()      )







class 			UpperCamelCase__			(			__lowerCamelCase      ):
      a__ :  List[Any]		    =    CLIPConfig

      a__ :  Tuple		    =    ['CLIPEncoderLayer']






      def __init__(  self       : Union[str, Any],		__lowerCamelCase       : CLIPConfig )	->			Any:
           super().__init__(__a )

           UpperCamelCase__				:	Dict										=  CLIPVisionModel(config.vision_config )
           UpperCamelCase__				:	Any										=  nn.Linear(config.vision_config.hidden_size,		config.projection_dim,		bias=__a )

           UpperCamelCase__				:	str										=  nn.Parameter(torch.ones(17,		config.projection_dim ),		requires_grad=__a )
           UpperCamelCase__				:	Any										=  nn.Parameter(torch.ones(3,		config.projection_dim ),		requires_grad=__a )

           UpperCamelCase__				:	List[Any]										=  nn.Parameter(torch.ones(17 ),		requires_grad=__a )
           UpperCamelCase__				:	Optional[int]										=  nn.Parameter(torch.ones(3 ),		requires_grad=__a )






      @torch.no_grad()
      def 						__lowercase(  self       : List[str],		__lowerCamelCase       : Any,		__lowerCamelCase       : str )	->			int:
           UpperCamelCase__				:	List[str]										=  self.vision_model(__a )[1]  # pooled_output
           UpperCamelCase__				:	Optional[int]										=  self.visual_projection(__a )

           # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
           UpperCamelCase__				:	List[str]										=  cosine_distance(__a,		self.special_care_embeds ).cpu().float().numpy()
           UpperCamelCase__				:	Dict										=  cosine_distance(__a,		self.concept_embeds ).cpu().float().numpy()

           UpperCamelCase__				:	Union[str, Any]										=  []
           UpperCamelCase__				:	Any										=  image_embeds.shape[0]
           for i in range(__a ):
                UpperCamelCase__				:	Any										=  {'''special_scores''': {}, '''special_care''': [], '''concept_scores''': {}, '''bad_concepts''': []}

                # increase this value to create a stronger `nfsw` filter
                # at the cost of increasing the possibility of filtering benign images
                UpperCamelCase__				:	Any										=  0.0

                for concept_idx in range(len(special_cos_dist[0] ) ):
                     UpperCamelCase__				:	Union[str, Any]										=  special_cos_dist[i][concept_idx]
                     UpperCamelCase__				:	List[str]										=  self.special_care_embeds_weights[concept_idx].item()
                     UpperCamelCase__				:	Union[str, Any]										=  round(concept_cos - concept_threshold + adjustment,		3 )
                     if result_img["special_scores"][concept_idx] > 0:
                          result_img["special_care"].append({concept_idx, result_img['''special_scores'''][concept_idx]} )
                          UpperCamelCase__				:	List[Any]										=  0.01

                for concept_idx in range(len(cos_dist[0] ) ):
                     UpperCamelCase__				:	Tuple										=  cos_dist[i][concept_idx]
                     UpperCamelCase__				:	Optional[int]										=  self.concept_embeds_weights[concept_idx].item()
                     UpperCamelCase__				:	Optional[Any]										=  round(concept_cos - concept_threshold + adjustment,		3 )
                     if result_img["concept_scores"][concept_idx] > 0:
                          result_img["bad_concepts"].append(__a )

                result.append(__a )

           UpperCamelCase__				:	Tuple										=  [len(res['''bad_concepts'''] ) > 0 for res in result]

           return images, has_nsfw_concepts






      @torch.no_grad()
      def 						__lowercase(  self       : str,		__lowerCamelCase       : torch.FloatTensor,		__lowerCamelCase       : torch.FloatTensor )	->			List[str]:
           UpperCamelCase__				:	str										=  self.vision_model(__a )[1]  # pooled_output
           UpperCamelCase__				:	Dict										=  self.visual_projection(__a )

           UpperCamelCase__				:	Dict										=  cosine_distance(__a,		self.special_care_embeds )
           UpperCamelCase__				:	Any										=  cosine_distance(__a,		self.concept_embeds )

           # increase this value to create a stronger `nsfw` filter
           # at the cost of increasing the possibility of filtering benign images
           UpperCamelCase__				:	Any										=  0.0

           UpperCamelCase__				:	Any										=  special_cos_dist - self.special_care_embeds_weights + adjustment
           # special_scores = special_scores.round(decimals=3)
           UpperCamelCase__				:	str										=  torch.any(special_scores > 0,		dim=1 )
           UpperCamelCase__				:	Any										=  special_care * 0.01
           UpperCamelCase__				:	Union[str, Any]										=  special_adjustment.unsqueeze(1 ).expand(-1,		cos_dist.shape[1] )

           UpperCamelCase__				:	Optional[int]										=  (cos_dist - self.concept_embeds_weights) + special_adjustment
           # concept_scores = concept_scores.round(decimals=3)
           UpperCamelCase__				:	List[str]										=  torch.any(concept_scores > 0,		dim=1 )

           return images, has_nsfw_concepts





'''simple docstring'''


from tempfile import TemporaryDirectory
from unittest import TestCase
from unittest.mock import MagicMock, patch

from transformers import AutoModel, TFAutoModel
from transformers.onnx import FeaturesManager
from transformers.testing_utils import SMALL_MODEL_IDENTIFIER, require_tf, require_torch






@require_torch
@require_tf
class     A ( UpperCAmelCase ):





							def    snake_case__    (   self					:				str						)				->						Any:
								__UpperCAmelCase      					=      SMALL_MODEL_IDENTIFIER
								__UpperCAmelCase      					=      '''pt'''
								__UpperCAmelCase      					=      '''tf'''





							def    snake_case__    (   self					:				int  ,		__a					:				int						)				->						Union[str, Any]:
								__UpperCAmelCase      					=      AutoModel.from_pretrained(self.test_model						)
								model_pt.save_pretrained(__a						)





							def    snake_case__    (   self					:				Any  ,		__a					:				Optional[int]						)				->						Tuple:
								__UpperCAmelCase      					=      TFAutoModel.from_pretrained(self.test_model  ,		from_pt=__a						)
								model_tf.save_pretrained(__a						)





							def    snake_case__    (   self					:				Dict						)				->						Dict:
								__UpperCAmelCase      					=      '''mock_framework'''

								# Framework provided - return whatever the user provides
								__UpperCAmelCase      					=      FeaturesManager.determine_framework(self.test_model  ,		__a						)
								self.assertEqual(__a  ,		__a						)

								# Local checkpoint and framework provided - return provided framework
								# PyTorch checkpoint
								with TemporaryDirectory() as local_pt_ckpt:
									self._setup_pt_ckpt(__a						)
									__UpperCAmelCase      					=      FeaturesManager.determine_framework(__a  ,		__a						)
									self.assertEqual(__a  ,		__a						)

								# TensorFlow checkpoint
								with TemporaryDirectory() as local_tf_ckpt:
									self._setup_tf_ckpt(__a						)
									__UpperCAmelCase      					=      FeaturesManager.determine_framework(__a  ,		__a						)
									self.assertEqual(__a  ,		__a						)





							def    snake_case__    (   self					:				Dict						)				->						List[str]:
								# PyTorch checkpoint
								with TemporaryDirectory() as local_pt_ckpt:
									self._setup_pt_ckpt(__a						)
									__UpperCAmelCase      					=      FeaturesManager.determine_framework(__a						)
									self.assertEqual(__a  ,		self.framework_pt						)

								# TensorFlow checkpoint
								with TemporaryDirectory() as local_tf_ckpt:
									self._setup_tf_ckpt(__a						)
									__UpperCAmelCase      					=      FeaturesManager.determine_framework(__a						)
									self.assertEqual(__a  ,		self.framework_tf						)

								# Invalid local checkpoint
								with TemporaryDirectory() as local_invalid_ckpt:
									with self.assertRaises(__a						):
										__UpperCAmelCase      					=      FeaturesManager.determine_framework(__a						)






							def    snake_case__    (   self					:				str						)				->						Union[str, Any]:
								__UpperCAmelCase      					=      MagicMock(return_value=__a						)
								with patch('''transformers.onnx.features.is_tf_available'''  ,		__a						):
									__UpperCAmelCase      					=      FeaturesManager.determine_framework(self.test_model						)
									self.assertEqual(__a  ,		self.framework_pt						)

								# PyTorch not in environment -> use TensorFlow
								__UpperCAmelCase      					=      MagicMock(return_value=__a						)
								with patch('''transformers.onnx.features.is_torch_available'''  ,		__a						):
									__UpperCAmelCase      					=      FeaturesManager.determine_framework(self.test_model						)
									self.assertEqual(__a  ,		self.framework_tf						)

								# Both in environment -> use PyTorch
								__UpperCAmelCase      					=      MagicMock(return_value=__a						)
								__UpperCAmelCase      					=      MagicMock(return_value=__a						)
								with patch('''transformers.onnx.features.is_tf_available'''  ,		__a						), patch(
								    '''transformers.onnx.features.is_torch_available'''  ,		__a						):
									__UpperCAmelCase      					=      FeaturesManager.determine_framework(self.test_model						)
									self.assertEqual(__a  ,		self.framework_pt						)

								# Both not in environment -> raise error
								__UpperCAmelCase      					=      MagicMock(return_value=__a						)
								__UpperCAmelCase      					=      MagicMock(return_value=__a						)
								with patch('''transformers.onnx.features.is_tf_available'''  ,		__a						), patch(
								    '''transformers.onnx.features.is_torch_available'''  ,		__a						):
									with self.assertRaises(__a						):
										__UpperCAmelCase      					=      FeaturesManager.determine_framework(self.test_model						)
