



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

from ...utils import logging


UpperCAmelCase :				Any							=logging.get_logger(__name__)



def  _lowerCAmelCase     (_lowerCAmelCase ,						_lowerCAmelCase):
					UpperCamelCase_			= nn.functional.normalize(_lowerCAmelCase)
					UpperCamelCase_			= nn.functional.normalize(_lowerCAmelCase)
					return torch.mm(_lowerCAmelCase ,						normalized_text_embeds.t())






class    _lowercase (a_   ):
				'''simple docstring'''







				lowercase__     =    CLIPConfig

				lowercase__     =    ["""CLIPEncoderLayer"""]

				def __init__(			self  ,			snake_case__     ):

									'''simple docstring'''


									super().__init__(snake_case__     )

									UpperCamelCase_			= CLIPVisionModel(config.vision_config     )
									UpperCamelCase_			= nn.Linear(config.vision_config.hidden_size  ,			config.projection_dim  ,			bias=snake_case__     )

									UpperCamelCase_			= nn.Parameter(torch.ones(17  ,			config.projection_dim     )  ,			requires_grad=snake_case__     )
									UpperCamelCase_			= nn.Parameter(torch.ones(3  ,			config.projection_dim     )  ,			requires_grad=snake_case__     )

									UpperCamelCase_			= nn.Parameter(torch.ones(17     )  ,			requires_grad=snake_case__     )
									UpperCamelCase_			= nn.Parameter(torch.ones(3     )  ,			requires_grad=snake_case__     )

				@torch.no_grad()
				def    _lowerCamelCase       (			self  ,			snake_case__  ,			snake_case__     ):

									'''simple docstring'''


									UpperCamelCase_			= self.vision_model(snake_case__     )[1]  # pooled_output
									UpperCamelCase_			= self.visual_projection(snake_case__     )

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

									UpperCamelCase_			= []
									UpperCamelCase_			= image_embeds.shape[0]
									for i in range(snake_case__     ):
														UpperCamelCase_			= {"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_			= 0.0

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

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

														result.append(snake_case__     )

									UpperCamelCase_			= [len(res["bad_concepts"]     ) > 0 for res in result]

									for idx, has_nsfw_concept in enumerate(snake_case__     ):
														if has_nsfw_concept:
																			if torch.is_tensor(snake_case__     ) or torch.is_tensor(images[0]     ):
																								UpperCamelCase_			= torch.zeros_like(images[idx]     )  # black image
																			else:
																								UpperCamelCase_			= np.zeros(images[idx].shape     )  # black image

									if any(snake_case__     ):
														logger.warning(
														    "Potential NSFW content was detected in one or more images. A black image will be returned instead."
														    " Try again with a different prompt and/or seed."     )

									return images, has_nsfw_concepts

				@torch.no_grad()
				def    _lowerCamelCase       (			self  ,			snake_case__  ,			snake_case__     ):

									'''simple docstring'''


									UpperCamelCase_			= self.vision_model(snake_case__     )[1]  # pooled_output
									UpperCamelCase_			= self.visual_projection(snake_case__     )

									UpperCamelCase_			= cosine_distance(snake_case__  ,			self.special_care_embeds     )
									UpperCamelCase_			= cosine_distance(snake_case__  ,			self.concept_embeds     )

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

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

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

									UpperCamelCase_			= 0.0  # black image

									return images, has_nsfw_concepts






from typing import List, Optional, Union

import numpy as np

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


UpperCamelCase__       				=	logging.get_logger(__name__)
class 							__lowercase   (			a__    ):
     _lowerCAmelCase		       =						["input_values", "attention_mask"]



     def __init__(  self   :							Union[str, Any]      ,		lowercase__   :							str=1      ,		lowercase__   :							int=1_6_0_0_0      ,		lowercase__   :							Optional[int]=0.0      ,		lowercase__   :							List[Any]=False      ,		lowercase__   :							Dict=True      ,		**lowercase__   :							Any      ,		):
       super().__init__(feature_size=lowercase__      ,		sampling_rate=lowercase__      ,		padding_value=lowercase__      ,		**lowercase__       )
       a_		    =  return_attention_mask
       a_		    =  do_normalize



     @staticmethod
     def 				__magic_name__		(  lowercase__   :							List[np.ndarray]      ,		lowercase__   :							List[np.ndarray]      ,		lowercase__   :							float = 0.0       ):
       if attention_mask is not None:
         a_		    =  np.array(lowercase__      ,		np.intaa       )
         a_		    =  []

         for vector, length in zip(lowercase__      ,		attention_mask.sum(-1       )       ):
           a_		    =  (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7       )
           if length < normed_slice.shape[0]:
             a_		    =  padding_value

           normed_input_values.append(lowercase__       )
       else:
         a_		    =  [(x - x.mean()) / np.sqrt(x.var() + 1e-7       ) for x in input_values]

       return normed_input_values



     def __call__(  self   :							Any      ,		lowercase__   :							Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]      ,		lowercase__   :							Union[bool, str, PaddingStrategy] = False      ,		lowercase__   :							Optional[int] = None      ,		lowercase__   :							bool = False      ,		lowercase__   :							Optional[int] = None      ,		lowercase__   :							Optional[bool] = None      ,		lowercase__   :							Optional[Union[str, TensorType]] = None      ,		lowercase__   :							Optional[int] = None      ,		**lowercase__   :							List[str]      ,		):

       if sampling_rate is not None:
         if sampling_rate != self.sampling_rate:
           raise ValueError(
               f"The model corresponding to this feature extractor: {self} was trained using a sampling rate of"
               f" {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with"
               f" {self.sampling_rate} and not {sampling_rate}."       )
       else:
         logger.warning(
             '''It is strongly recommended to pass the ``sampling_rate`` argument to this function. '''
             '''Failing to do so can result in silent errors that might be hard to debug.'''       )

       a_		    =  isinstance(lowercase__      ,		np.ndarray       ) and len(raw_speech.shape       ) > 1
       if is_batched_numpy and len(raw_speech.shape       ) > 2:
         raise ValueError(f"Only mono-channel audio is supported for input to {self}"       )
       a_		    =  is_batched_numpy or (
           isinstance(lowercase__      ,		(list, tuple)       ) and (isinstance(raw_speech[0]      ,		(np.ndarray, tuple, list)       ))
       )

       # always return batch
       if not is_batched:
         a_		    =  [raw_speech]

       # convert into correct format for padding
       a_		    =  BatchFeature({'''input_values''': raw_speech}       )

       a_		    =  self.pad(
           lowercase__      ,		padding=lowercase__      ,		max_length=lowercase__      ,		truncation=lowercase__      ,		pad_to_multiple_of=lowercase__      ,		return_attention_mask=lowercase__      ,		)

       # convert input values to correct format
       a_		    =  padded_inputs['''input_values''']
       if not isinstance(input_values[0]      ,		np.ndarray       ):
         a_		    =  [np.asarray(lowercase__      ,		dtype=np.floataa       ) for array in input_values]
       elif (
           not isinstance(lowercase__      ,		np.ndarray       )
           and isinstance(input_values[0]      ,		np.ndarray       )
           and input_values[0].dtype is np.dtype(np.floataa       )
       ):
         a_		    =  [array.astype(np.floataa       ) for array in input_values]
       elif isinstance(lowercase__      ,		np.ndarray       ) and input_values.dtype is np.dtype(np.floataa       ):
         a_		    =  input_values.astype(np.floataa       )

       # convert attention_mask to correct format
       a_		    =  padded_inputs.get('''attention_mask'''       )
       if attention_mask is not None:
         a_		    =  [np.asarray(lowercase__      ,		dtype=np.intaa       ) for array in attention_mask]

       # zero-mean and unit-variance normalization
       if self.do_normalize:
         a_		    =  (
             attention_mask
             if self._get_padding_strategies(lowercase__      ,		max_length=lowercase__       ) is not PaddingStrategy.DO_NOT_PAD
             else None
         )
         a_		    =  self.zero_mean_unit_var_norm(
             padded_inputs['''input_values''']      ,		attention_mask=lowercase__      ,		padding_value=self.padding_value       )

       if return_tensors is not None:
         a_		    =  padded_inputs.convert_to_tensors(lowercase__       )

       return padded_inputs
