




"""simple docstring"""





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

from ...utils import logging


A   =       logging.get_logger(__name__)

def 					_UpperCamelCase    (			UpperCamelCase				,       UpperCamelCase							)			->			Union[str, Any]:


			"""simple docstring"""
			__UpperCAmelCase	: int		=					nn.functional.normalize(UpperCamelCase							)
			__UpperCAmelCase	: Any		=					nn.functional.normalize(UpperCamelCase							)
			return torch.mm(UpperCamelCase				,       normalized_text_embeds.t()							)





class      a__				(	__magic_name__   ):
			lowercase_    =	CLIPConfig

			lowercase_    =	["CLIPEncoderLayer"]


			def __init__(      self      :    List[Any]      ,  UpperCamelCase_      :    CLIPConfig):


						"""simple docstring"""




						super().__init__(UpperCamelCase_)

						__UpperCAmelCase	: str		=					CLIPVisionModel(config.vision_config)
						__UpperCAmelCase	: Union[str, Any]		=					nn.Linear(config.vision_config.hidden_size      ,  config.projection_dim      ,  bias=UpperCamelCase_)

						__UpperCAmelCase	: Tuple		=					nn.Parameter(torch.ones(17      ,  config.projection_dim)      ,  requires_grad=UpperCamelCase_)
						__UpperCAmelCase	: List[str]		=					nn.Parameter(torch.ones(3      ,  config.projection_dim)      ,  requires_grad=UpperCamelCase_)

						__UpperCAmelCase	: List[Any]		=					nn.Parameter(torch.ones(17)      ,  requires_grad=UpperCamelCase_)
						__UpperCAmelCase	: Optional[Any]		=					nn.Parameter(torch.ones(3)      ,  requires_grad=UpperCamelCase_)


			@torch.no_grad()
			def       a_     (      self      :    Union[str, Any]      ,  UpperCamelCase_      :    Dict      ,  UpperCamelCase_      :    int):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					self.vision_model(UpperCamelCase_)[1]  # pooled_output
						__UpperCAmelCase	: List[str]		=					self.visual_projection(UpperCamelCase_)

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

						__UpperCAmelCase	: List[Any]		=					[]
						__UpperCAmelCase	: Union[str, Any]		=					image_embeds.shape[0]
						for i in range(UpperCamelCase_):
									__UpperCAmelCase	: Optional[int]		=					{"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	: Union[str, Any]		=					0.0

									for concept_idx in range(len(special_cos_dist[0])):
												__UpperCAmelCase	: str		=					special_cos_dist[i][concept_idx]
												__UpperCAmelCase	: List[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	: int		=					0.01

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

									result.append(UpperCamelCase_)

						__UpperCAmelCase	: Tuple		=					[len(res["bad_concepts"]) > 0 for res in result]

						return images, has_nsfw_concepts


			@torch.no_grad()
			def       a_     (      self      :    Optional[Any]      ,  UpperCamelCase_      :    torch.FloatTensor      ,  UpperCamelCase_      :    torch.FloatTensor):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					self.vision_model(UpperCamelCase_)[1]  # pooled_output
						__UpperCAmelCase	: List[str]		=					self.visual_projection(UpperCamelCase_)

						__UpperCAmelCase	: Optional[int]		=					cosine_distance(UpperCamelCase_      ,  self.special_care_embeds)
						__UpperCAmelCase	: List[str]		=					cosine_distance(UpperCamelCase_      ,  self.concept_embeds)

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

						__UpperCAmelCase	: Optional[int]		=					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	: List[str]		=					special_care * 0.01
						__UpperCAmelCase	: Union[str, Any]		=					special_adjustment.unsqueeze(1).expand(-1      ,  cos_dist.shape[1])

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

						return images, has_nsfw_concepts








"""simple docstring"""





import shutil
import tempfile
import unittest

from transformers import ClapFeatureExtractor, ClapProcessor, RobertaTokenizer, RobertaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torchaudio

from .test_feature_extraction_clap import floats_list



@require_torchaudio
@require_sentencepiece
class      a__				(	unittest.TestCase   ):


			def       a_     (      self      :    Any):


						"""simple docstring"""




						__UpperCAmelCase	: Tuple		=					"laion/clap-htsat-unfused"
						__UpperCAmelCase	: List[str]		=					tempfile.mkdtemp()


			def       a_     (      self      :    Optional[Any]      ,  **UpperCamelCase_      :    Union[str, Any]):


						"""simple docstring"""




						return RobertaTokenizer.from_pretrained(self.checkpoint      ,  **UpperCamelCase_)


			def       a_     (      self      :    Union[str, Any]      ,  **UpperCamelCase_      :    Optional[int]):


						"""simple docstring"""




						return ClapFeatureExtractor.from_pretrained(self.checkpoint      ,  **UpperCamelCase_)


			def       a_     (      self      :    List[Any]):


						"""simple docstring"""




						shutil.rmtree(self.tmpdirname)


			def       a_     (      self      :    Optional[int]):


						"""simple docstring"""




						__UpperCAmelCase	: Union[str, Any]		=					self.get_tokenizer()
						__UpperCAmelCase	: str		=					self.get_feature_extractor()

						__UpperCAmelCase	: List[Any]		=					ClapProcessor(tokenizer=UpperCamelCase_      ,  feature_extractor=UpperCamelCase_)

						processor.save_pretrained(self.tmpdirname)
						__UpperCAmelCase	: Union[str, Any]		=					ClapProcessor.from_pretrained(self.tmpdirname)

						self.assertEqual(processor.tokenizer.get_vocab()      ,  tokenizer.get_vocab())
						self.assertIsInstance(processor.tokenizer      ,  UpperCamelCase_)

						self.assertEqual(processor.feature_extractor.to_json_string()      ,  feature_extractor.to_json_string())
						self.assertIsInstance(processor.feature_extractor      ,  UpperCamelCase_)


			def       a_     (      self      :    Any):


						"""simple docstring"""




						__UpperCAmelCase	: Union[str, Any]		=					ClapProcessor(tokenizer=self.get_tokenizer()      ,  feature_extractor=self.get_feature_extractor())
						processor.save_pretrained(self.tmpdirname)

						__UpperCAmelCase	: int		=					self.get_tokenizer(bos_token="(BOS)"      ,  eos_token="(EOS)")
						__UpperCAmelCase	: Any		=					self.get_feature_extractor(do_normalize=UpperCamelCase_      ,  padding_value=1.0)

						__UpperCAmelCase	: Tuple		=					ClapProcessor.from_pretrained(
						    self.tmpdirname      ,  bos_token="(BOS)"      ,  eos_token="(EOS)"      ,  do_normalize=UpperCamelCase_      ,  padding_value=1.0)

						self.assertEqual(processor.tokenizer.get_vocab()      ,  tokenizer_add_kwargs.get_vocab())
						self.assertIsInstance(processor.tokenizer      ,  UpperCamelCase_)

						self.assertEqual(processor.feature_extractor.to_json_string()      ,  feature_extractor_add_kwargs.to_json_string())
						self.assertIsInstance(processor.feature_extractor      ,  UpperCamelCase_)


			def       a_     (      self      :    Any):


						"""simple docstring"""




						__UpperCAmelCase	: Tuple		=					self.get_feature_extractor()
						__UpperCAmelCase	: int		=					self.get_tokenizer()

						__UpperCAmelCase	: Any		=					ClapProcessor(tokenizer=UpperCamelCase_      ,  feature_extractor=UpperCamelCase_)

						__UpperCAmelCase	: Optional[int]		=					floats_list((3, 1000))

						__UpperCAmelCase	: Tuple		=					feature_extractor(UpperCamelCase_      ,  return_tensors="np")
						__UpperCAmelCase	: Any		=					processor(audios=UpperCamelCase_      ,  return_tensors="np")

						for key in input_feat_extract.keys():
									self.assertAlmostEqual(input_feat_extract[key].sum()      ,  input_processor[key].sum()      ,  delta=1e-2)


			def       a_     (      self      :    Optional[Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Union[str, Any]		=					self.get_feature_extractor()
						__UpperCAmelCase	: Optional[int]		=					self.get_tokenizer()

						__UpperCAmelCase	: str		=					ClapProcessor(tokenizer=UpperCamelCase_      ,  feature_extractor=UpperCamelCase_)

						__UpperCAmelCase	: Any		=					"This is a test string"

						__UpperCAmelCase	: List[Any]		=					processor(text=UpperCamelCase_)

						__UpperCAmelCase	: Dict		=					tokenizer(UpperCamelCase_)

						for key in encoded_tok.keys():
									self.assertListEqual(encoded_tok[key]      ,  encoded_processor[key])


			def       a_     (      self      :    Tuple):


						"""simple docstring"""




						__UpperCAmelCase	: int		=					self.get_feature_extractor()
						__UpperCAmelCase	: List[str]		=					self.get_tokenizer()

						__UpperCAmelCase	: List[str]		=					ClapProcessor(tokenizer=UpperCamelCase_      ,  feature_extractor=UpperCamelCase_)

						__UpperCAmelCase	: Tuple		=					[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

						__UpperCAmelCase	: int		=					processor.batch_decode(UpperCamelCase_)
						__UpperCAmelCase	: Tuple		=					tokenizer.batch_decode(UpperCamelCase_)

						self.assertListEqual(UpperCamelCase_      ,  UpperCamelCase_)


			def       a_     (      self      :    Optional[int]):


						"""simple docstring"""




						__UpperCAmelCase	: Any		=					self.get_feature_extractor()
						__UpperCAmelCase	: int		=					self.get_tokenizer()

						__UpperCAmelCase	: int		=					ClapProcessor(tokenizer=UpperCamelCase_      ,  feature_extractor=UpperCamelCase_)

						self.assertListEqual(
						    processor.model_input_names[2:]      ,  feature_extractor.model_input_names      ,  msg="`processor` and `feature_extractor` model input names do not match"      ,  )


