


'''simple docstring'''


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

from ...utils import logging


lowerCamelCase__    	=     logging.get_logger(__name__)






def  __lowerCAmelCase (__lowerCAmelCase      , __lowerCAmelCase   ):
		_UpperCAmelCase		:       List[Any]								=      nn.functional.normalize(__lowerCAmelCase   )
		_UpperCAmelCase		:       List[str]								=      nn.functional.normalize(__lowerCAmelCase   )
		return torch.mm(__lowerCAmelCase      , normalized_text_embeds.t()   )






class 	lowerCAmelCase__			(			UpperCAmelCase__	):
			lowerCAmelCase :	Optional[Any]          =     CLIPConfig

			lowerCAmelCase :	Dict          =     ["CLIPEncoderLayer"]




			def __init__(		self  :     str	,  lowerCamelCase__  :     CLIPConfig )   ->Dict:



					'''simple docstring'''







					super().__init__(lowerCamelCase__ )

					_UpperCAmelCase		:       Optional[Any]								=      CLIPVisionModel(config.vision_config )
					_UpperCAmelCase		:       List[str]								=      nn.Linear(config.vision_config.hidden_size	,  config.projection_dim	,  bias=lowerCamelCase__ )

					_UpperCAmelCase		:       Any								=      nn.Parameter(torch.ones(17	,  config.projection_dim )	,  requires_grad=lowerCamelCase__ )
					_UpperCAmelCase		:       Dict								=      nn.Parameter(torch.ones(3	,  config.projection_dim )	,  requires_grad=lowerCamelCase__ )

					_UpperCAmelCase		:       str								=      nn.Parameter(torch.ones(17 )	,  requires_grad=lowerCamelCase__ )
					_UpperCAmelCase		:       List[Any]								=      nn.Parameter(torch.ones(3 )	,  requires_grad=lowerCamelCase__ )




			@torch.no_grad()
			def 							lowerCAmelCase__    (		self  :     Union[str, Any]	,  lowerCamelCase__  :     List[str]	,  lowerCamelCase__  :     Optional[Any] )   ->int:



					'''simple docstring'''







					_UpperCAmelCase		:       Union[str, Any]								=      self.vision_model(lowerCamelCase__ )[1]  # pooled_output
					_UpperCAmelCase		:       Union[str, Any]								=      self.visual_projection(lowerCamelCase__ )

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

					_UpperCAmelCase		:       str								=      []
					_UpperCAmelCase		:       Any								=      image_embeds.shape[0]
					for i in range(lowerCamelCase__ ):
							_UpperCAmelCase		:       List[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		:       List[Any]								=      0.0

							for concept_idx in range(len(special_cos_dist[0] ) ):
									_UpperCAmelCase		:       Dict								=      special_cos_dist[i][concept_idx]
									_UpperCAmelCase		:       str								=      self.special_care_embeds_weights[concept_idx].item()
									_UpperCAmelCase		:       List[str]								=      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		:       Tuple								=      0.0_1

							for concept_idx in range(len(cos_dist[0] ) ):
									_UpperCAmelCase		:       Union[str, Any]								=      cos_dist[i][concept_idx]
									_UpperCAmelCase		:       int								=      self.concept_embeds_weights[concept_idx].item()
									_UpperCAmelCase		:       List[Any]								=      round(concept_cos - concept_threshold + adjustment	,  3 )
									if result_img["concept_scores"][concept_idx] > 0:
											result_img["bad_concepts"].append(lowerCamelCase__ )

							result.append(lowerCamelCase__ )

					_UpperCAmelCase		:       Union[str, Any]								=      [len(res["bad_concepts"] ) > 0 for res in result]

					return images, has_nsfw_concepts






			@torch.no_grad()
			def 							lowerCAmelCase__    (		self  :     Optional[int]	,  lowerCamelCase__  :     torch.FloatTensor	,  lowerCamelCase__  :     torch.FloatTensor )   ->Optional[int]:



					'''simple docstring'''







					_UpperCAmelCase		:       Tuple								=      self.vision_model(lowerCamelCase__ )[1]  # pooled_output
					_UpperCAmelCase		:       Optional[Any]								=      self.visual_projection(lowerCamelCase__ )

					_UpperCAmelCase		:       Any								=      cosine_distance(lowerCamelCase__	,  self.special_care_embeds )
					_UpperCAmelCase		:       Optional[int]								=      cosine_distance(lowerCamelCase__	,  self.concept_embeds )

					# increase this value to create a stronger `nsfw` filter
					# at the cost of increasing the possibility of filtering benign images
					_UpperCAmelCase		:       Union[str, Any]								=      0.0

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

					_UpperCAmelCase		:       str								=      (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'''


# tests directory-specific settings - this file is run automatically
# by pytest before any tests are run

import sys
import warnings
from os.path import abspath, dirname, join


# allow having multiple repository checkouts and not needing to remember to rerun
# 'pip install -e .[dev]' when switching between checkouts and running tests.
lowerCamelCase__    	=     abspath(join(dirname(dirname(dirname(__file__))), 'src'))
sys.path.insert(1, git_repo_path)


# silence FutureWarning warnings in tests since often we can't act on them until
# they become normal warnings - i.e. the tests still need to test the current functionality
warnings.simplefilter(action='ignore', category=FutureWarning)






def  __lowerCAmelCase (__lowerCAmelCase   ):
		from transformers.testing_utils import pytest_addoption_shared

		pytest_addoption_shared(__lowerCAmelCase   )






def  __lowerCAmelCase (__lowerCAmelCase   ):
		from transformers.testing_utils import pytest_terminal_summary_main

		_UpperCAmelCase		:       Optional[int]								=      terminalreporter.config.getoption("--make-reports"   )
		if make_reports:
				pytest_terminal_summary_main(__lowerCAmelCase      , id=__lowerCAmelCase   )



