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

from ...utils import logging


__snake_case								=  logging.get_logger(__name__)






def 							_lowercase							(     UpperCamelCase_	,	UpperCamelCase_		)		->		List[Any]:
					'''simple docstring'''


					SCREAMING_SNAKE_CASE__													= nn.functional.normalize(UpperCamelCase_		)
					SCREAMING_SNAKE_CASE__													= nn.functional.normalize(UpperCamelCase_		)
					return torch.mm(UpperCamelCase_	,	normalized_text_embeds.t()		)




class 						lowercase__  (						_UpperCAmelCase  ):
	A__    :   Dict				       =CLIPConfig

	A__    :   Optional[int]				       =["""CLIPEncoderLayer"""]
	def __init__(			self							:   Optional[Any]       ,   UpperCAmelCase_							:   CLIPConfig							):
						super().__init__(UpperCAmelCase_							)

						SCREAMING_SNAKE_CASE__													= CLIPVisionModel(config.vision_config							)
						SCREAMING_SNAKE_CASE__													= nn.Linear(config.vision_config.hidden_size       ,   config.projection_dim       ,   bias=UpperCAmelCase_							)

						SCREAMING_SNAKE_CASE__													= nn.Parameter(torch.ones(17       ,   config.projection_dim							)       ,   requires_grad=UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= nn.Parameter(torch.ones(3       ,   config.projection_dim							)       ,   requires_grad=UpperCAmelCase_							)

						SCREAMING_SNAKE_CASE__													= nn.Parameter(torch.ones(17							)       ,   requires_grad=UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= nn.Parameter(torch.ones(3							)       ,   requires_grad=UpperCAmelCase_							)
	@torch.no_grad()
	def      A_				(			self							:   str       ,   UpperCAmelCase_							:   Optional[Any]       ,   UpperCAmelCase_							:   str							):
						SCREAMING_SNAKE_CASE__													= self.vision_model(UpperCAmelCase_							)[1]  # pooled_output
						SCREAMING_SNAKE_CASE__													= self.visual_projection(UpperCAmelCase_							)

						# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
						SCREAMING_SNAKE_CASE__													= cosine_distance(UpperCAmelCase_       ,   self.special_care_embeds							).cpu().float().numpy()
						SCREAMING_SNAKE_CASE__													= cosine_distance(UpperCAmelCase_       ,   self.concept_embeds							).cpu().float().numpy()

						SCREAMING_SNAKE_CASE__													= []
						SCREAMING_SNAKE_CASE__													= image_embeds.shape[0]
						for i in range(UpperCAmelCase_							):
											SCREAMING_SNAKE_CASE__													= {'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
											SCREAMING_SNAKE_CASE__													= 0.0

											for concept_idx in range(len(special_cos_dist[0]							)							):
																SCREAMING_SNAKE_CASE__													= special_cos_dist[i][concept_idx]
																SCREAMING_SNAKE_CASE__													= self.special_care_embeds_weights[concept_idx].item()
																SCREAMING_SNAKE_CASE__													= 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]}							)
																					SCREAMING_SNAKE_CASE__													= 0.01

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

											result.append(UpperCAmelCase_							)

						SCREAMING_SNAKE_CASE__													= [len(res['bad_concepts']							) > 0 for res in result]

						return images, has_nsfw_concepts





	@torch.no_grad()
	def      A_				(			self							:   Tuple       ,   UpperCAmelCase_							:   torch.FloatTensor       ,   UpperCAmelCase_							:   torch.FloatTensor							):
						SCREAMING_SNAKE_CASE__													= self.vision_model(UpperCAmelCase_							)[1]  # pooled_output
						SCREAMING_SNAKE_CASE__													= self.visual_projection(UpperCAmelCase_							)

						SCREAMING_SNAKE_CASE__													= cosine_distance(UpperCAmelCase_       ,   self.special_care_embeds							)
						SCREAMING_SNAKE_CASE__													= 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
						SCREAMING_SNAKE_CASE__													= 0.0

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

						SCREAMING_SNAKE_CASE__													= (cos_dist - self.concept_embeds_weights) + special_adjustment
						# concept_scores = concept_scores.round(decimals=3)
						SCREAMING_SNAKE_CASE__													= torch.any(concept_scores > 0       ,   dim=1							)

						return images, has_nsfw_concepts







import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin
from .feature_extraction_wavaveca import WavaVecaFeatureExtractor
from .tokenization_wavaveca import WavaVecaCTCTokenizer




class 						lowercase__  (						_UpperCAmelCase  ):
	A__    :   Union[str, Any]				       ="""Wav2Vec2FeatureExtractor"""
	A__    :   Any				       ="""AutoTokenizer"""
	def __init__(			self							:   List[Any]       ,   UpperCAmelCase_							:   List[str]       ,   UpperCAmelCase_							:   str							):
						super().__init__(UpperCAmelCase_       ,   UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= self.feature_extractor
						SCREAMING_SNAKE_CASE__													= False
	@classmethod
	def      A_				(			cls							:   Union[str, Any]       ,   UpperCAmelCase_							:   Any       ,   **UpperCAmelCase_							:   Optional[Any]							):
						try:
											return super().from_pretrained(UpperCAmelCase_       ,   **UpperCAmelCase_							)
						except OSError:
											warnings.warn(
											    F'Loading a tokenizer inside {cls.__name__} from a config that does not'
											    ' include a `tokenizer_class` attribute is deprecated and will be '
											    'removed in v5. Please add `\'tokenizer_class\': \'Wav2Vec2CTCTokenizer\'`'
											    ' attribute to either your `config.json` or `tokenizer_config.json` '
											    'file to suppress this warning: '       ,   UpperCAmelCase_       ,   )

											SCREAMING_SNAKE_CASE__													= WavaVecaFeatureExtractor.from_pretrained(UpperCAmelCase_       ,   **UpperCAmelCase_							)
											SCREAMING_SNAKE_CASE__													= WavaVecaCTCTokenizer.from_pretrained(UpperCAmelCase_       ,   **UpperCAmelCase_							)

											return cls(feature_extractor=UpperCAmelCase_       ,   tokenizer=UpperCAmelCase_							)
	def __call__(			self							:   Any       ,   *UpperCAmelCase_							:   Dict       ,   **UpperCAmelCase_							:   Any							):
						# For backward compatibility
						if self._in_target_context_manager:
											return self.current_processor(*UpperCAmelCase_       ,   **UpperCAmelCase_							)

						if "raw_speech" in kwargs:
											warnings.warn('Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.'							)
											SCREAMING_SNAKE_CASE__													= kwargs.pop('raw_speech'							)
						else:
											SCREAMING_SNAKE_CASE__													= kwargs.pop('audio'       ,   UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= kwargs.pop('sampling_rate'       ,   UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= kwargs.pop('text'       ,   UpperCAmelCase_							)
						if len(UpperCAmelCase_							) > 0:
											SCREAMING_SNAKE_CASE__													= args[0]
											SCREAMING_SNAKE_CASE__													= args[1:]

						if audio is None and text is None:
											raise ValueError('You need to specify either an `audio` or `text` input to process.'							)

						if audio is not None:
											SCREAMING_SNAKE_CASE__													= self.feature_extractor(UpperCAmelCase_       ,   *UpperCAmelCase_       ,   sampling_rate=UpperCAmelCase_       ,   **UpperCAmelCase_							)
						if text is not None:
											SCREAMING_SNAKE_CASE__													= self.tokenizer(UpperCAmelCase_       ,   **UpperCAmelCase_							)

						if text is None:
											return inputs
						elif audio is None:
											return encodings
						else:
											SCREAMING_SNAKE_CASE__													= encodings['input_ids']
											return inputs
	def      A_				(			self							:   List[str]       ,   *UpperCAmelCase_							:   List[Any]       ,   **UpperCAmelCase_							:   Optional[int]							):
						# For backward compatibility
						if self._in_target_context_manager:
											return self.current_processor.pad(*UpperCAmelCase_       ,   **UpperCAmelCase_							)

						SCREAMING_SNAKE_CASE__													= kwargs.pop('input_features'       ,   UpperCAmelCase_							)
						SCREAMING_SNAKE_CASE__													= kwargs.pop('labels'       ,   UpperCAmelCase_							)
						if len(UpperCAmelCase_							) > 0:
											SCREAMING_SNAKE_CASE__													= args[0]
											SCREAMING_SNAKE_CASE__													= args[1:]

						if input_features is not None:
											SCREAMING_SNAKE_CASE__													= self.feature_extractor.pad(UpperCAmelCase_       ,   *UpperCAmelCase_       ,   **UpperCAmelCase_							)
						if labels is not None:
											SCREAMING_SNAKE_CASE__													= self.tokenizer.pad(UpperCAmelCase_       ,   **UpperCAmelCase_							)

						if labels is None:
											return input_features
						elif input_features is None:
											return labels
						else:
											SCREAMING_SNAKE_CASE__													= labels['input_ids']
											return input_features
	def      A_				(			self							:   List[Any]       ,   *UpperCAmelCase_							:   List[str]       ,   **UpperCAmelCase_							:   List[str]							):
						return self.tokenizer.batch_decode(*UpperCAmelCase_       ,   **UpperCAmelCase_							)
	def      A_				(			self							:   Tuple       ,   *UpperCAmelCase_							:   int       ,   **UpperCAmelCase_							:   List[str]							):
						return self.tokenizer.decode(*UpperCAmelCase_       ,   **UpperCAmelCase_							)





	@contextmanager
	def      A_				(			self							:   str							):
						warnings.warn(
						    '`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '
						    'labels by using the argument `text` of the regular `__call__` method (either in the same call as '
						    'your audio inputs, or in a separate call.'							)
						SCREAMING_SNAKE_CASE__													= True
						SCREAMING_SNAKE_CASE__													= self.tokenizer
						yield
						SCREAMING_SNAKE_CASE__													= self.feature_extractor
						SCREAMING_SNAKE_CASE__													= False





