






'''simple docstring'''





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

from ...utils import logging


_UpperCamelCase		:		int								=			logging.get_logger(__name__)
def __UpperCAmelCase (			A			:	int	,   A			:	Tuple						)  -> Tuple:
			UpperCAmelCase_     :   Union[str, Any]       =   nn.functional.normalize(A						)
			UpperCAmelCase_     :   Tuple       =   nn.functional.normalize(A						)
			return torch.mm(A	,   normalized_text_embeds.t()						)







class 				snake_case__     (			UpperCamelCase):
					a_       =			CLIPConfig

					a_       =			["CLIPEncoderLayer"]




					def __init__(       self       :						Optional[Any]							,				_A       :						CLIPConfig     )  ->    int:
								super().__init__(_A     )

								UpperCAmelCase_     :   Union[str, Any]       =   CLIPVisionModel(config.vision_config     )
								UpperCAmelCase_     :   Dict       =   nn.Linear(config.vision_config.hidden_size							,				config.projection_dim							,				bias=_A     )

								UpperCAmelCase_     :   List[str]       =   nn.Parameter(torch.ones(17							,				config.projection_dim     )							,				requires_grad=_A     )
								UpperCAmelCase_     :   Optional[int]       =   nn.Parameter(torch.ones(3							,				config.projection_dim     )							,				requires_grad=_A     )

								UpperCAmelCase_     :   str       =   nn.Parameter(torch.ones(17     )							,				requires_grad=_A     )
								UpperCAmelCase_     :   Tuple       =   nn.Parameter(torch.ones(3     )							,				requires_grad=_A     )




					@torch.no_grad()
					def 						A     (       self       :						List[str]							,				_A       :						Optional[int]							,				_A       :						Tuple     )  ->    int:
								UpperCAmelCase_     :   Dict       =   self.vision_model(_A     )[1]  # pooled_output
								UpperCAmelCase_     :   Union[str, Any]       =   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_     :   Tuple       =   cosine_distance(_A							,				self.concept_embeds     ).cpu().float().numpy()

								UpperCAmelCase_     :   Optional[Any]       =   []
								UpperCAmelCase_     :   Any       =   image_embeds.shape[0]
								for i in range(_A     ):
											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_     :   Optional[int]       =   0.0

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

											for concept_idx in range(len(cos_dist[0]     )     ):
														UpperCAmelCase_     :   Dict       =   cos_dist[i][concept_idx]
														UpperCAmelCase_     :   List[str]       =   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(_A     )

											result.append(_A     )

								UpperCAmelCase_     :   Any       =   [len(res['''bad_concepts''']     ) > 0 for res in result]

								for idx, has_nsfw_concept in enumerate(_A     ):
											if has_nsfw_concept:
														if torch.is_tensor(_A     ) or torch.is_tensor(images[0]     ):
																	UpperCAmelCase_     :   Union[str, Any]       =   torch.zeros_like(images[idx]     )  # black image
														else:
																	UpperCAmelCase_     :   Optional[Any]       =   np.zeros(images[idx].shape     )  # black image

								if any(_A     ):
											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 						A     (       self       :						List[str]							,				_A       :						torch.FloatTensor							,				_A       :						torch.FloatTensor     )  ->    Any:
								UpperCAmelCase_     :   Optional[Any]       =   self.vision_model(_A     )[1]  # pooled_output
								UpperCAmelCase_     :   Dict       =   self.visual_projection(_A     )

								UpperCAmelCase_     :   Optional[int]       =   cosine_distance(_A							,				self.special_care_embeds     )
								UpperCAmelCase_     :   int       =   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_     :   Optional[int]       =   0.0

								UpperCAmelCase_     :   List[str]       =   special_cos_dist - self.special_care_embeds_weights + adjustment
								# special_scores = special_scores.round(decimals=3)
								UpperCAmelCase_     :   Dict       =   torch.any(special_scores > 0							,				dim=1     )
								UpperCAmelCase_     :   int       =   special_care * 0.01
								UpperCAmelCase_     :   str       =   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_     :   Any       =   torch.any(concept_scores > 0							,				dim=1     )

								UpperCAmelCase_     :   List[str]       =   0.0  # black image

								return images, has_nsfw_concepts










'''simple docstring'''





import unittest

import numpy as np
import PIL
import torch

from diffusers.image_processor import VaeImageProcessor


class 				snake_case__     (			unittest.TestCase):




					@property
					def 						A     (       self       :						Optional[Any]     )  ->    int:
								UpperCAmelCase_     :   Optional[int]       =   1
								UpperCAmelCase_     :   List[str]       =   3
								UpperCAmelCase_     :   Optional[int]       =   8
								UpperCAmelCase_     :   Optional[Any]       =   8

								UpperCAmelCase_     :   Any       =   torch.rand((batch_size, num_channels, height, width)     )

								return sample




					def 						A     (       self       :						Dict							,				_A       :						str     )  ->    List[Any]:
								if isinstance(image[0]							,				PIL.Image.Image     ):
											return np.stack([np.array(_A     ) for i in image]							,				axis=0     )
								elif isinstance(_A							,				torch.Tensor     ):
											return image.cpu().numpy().transpose(0							,				2							,				3							,				1     )
								return image




					def 						A     (       self       :						Tuple     )  ->    Optional[Any]:
								UpperCAmelCase_     :   int       =   VaeImageProcessor(do_resize=_A							,				do_normalize=_A     )

								UpperCAmelCase_     :   List[Any]       =   self.dummy_sample
								UpperCAmelCase_     :   Optional[int]       =   self.to_np(_A     )

								for output_type in ["pt", "np", "pil"]:
											UpperCAmelCase_     :   Any       =   image_processor.postprocess(
											    image_processor.preprocess(_A     )							,				output_type=_A							,				)
											UpperCAmelCase_     :   Union[str, Any]       =   self.to_np(_A     )
											UpperCAmelCase_     :   Union[str, Any]       =   (input_np * 2_55).round() if output_type == '''pil''' else input_np
											assert (
											    np.abs(in_np - out_np     ).max() < 1e-6
											), F"decoded output does not match input for output_type {output_type}"




					def 						A     (       self       :						Optional[Any]     )  ->    int:
								UpperCAmelCase_     :   Union[str, Any]       =   VaeImageProcessor(do_resize=_A							,				do_normalize=_A     )
								UpperCAmelCase_     :   List[Any]       =   self.dummy_sample.cpu().numpy().transpose(0							,				2							,				3							,				1     )

								for output_type in ["pt", "np", "pil"]:
											UpperCAmelCase_     :   Union[str, Any]       =   image_processor.postprocess(image_processor.preprocess(_A     )							,				output_type=_A     )

											UpperCAmelCase_     :   List[Any]       =   self.to_np(_A     )
											UpperCAmelCase_     :   Any       =   (input_np * 2_55).round() if output_type == '''pil''' else input_np
											assert (
											    np.abs(in_np - out_np     ).max() < 1e-6
											), F"decoded output does not match input for output_type {output_type}"




					def 						A     (       self       :						Any     )  ->    Optional[Any]:
								UpperCAmelCase_     :   Tuple       =   VaeImageProcessor(do_resize=_A							,				do_normalize=_A     )

								UpperCAmelCase_     :   Optional[Any]       =   self.dummy_sample.cpu().numpy().transpose(0							,				2							,				3							,				1     )
								UpperCAmelCase_     :   List[Any]       =   image_processor.numpy_to_pil(_A     )

								for output_type in ["pt", "np", "pil"]:
											UpperCAmelCase_     :   str       =   image_processor.postprocess(image_processor.preprocess(_A     )							,				output_type=_A     )
											for i, o in zip(_A							,				_A     ):
														UpperCAmelCase_     :   Optional[int]       =   np.array(_A     )
														UpperCAmelCase_     :   Dict       =   self.to_np(_A     ) if output_type == '''pil''' else (self.to_np(_A     ) * 2_55).round()
														assert (
														    np.abs(in_np - out_np     ).max() < 1e-6
														), F"decoded output does not match input for output_type {output_type}"




					def 						A     (       self       :						Optional[Any]     )  ->    Optional[Any]:
								UpperCAmelCase_     :   int       =   VaeImageProcessor(do_resize=_A							,				do_normalize=_A     )

								UpperCAmelCase_     :   Optional[Any]       =   self.dummy_sample
								UpperCAmelCase_     :   Optional[Any]       =   input_pt_ad.squeeze(0     )

								UpperCAmelCase_     :   Tuple       =   image_processor.postprocess(
								    image_processor.preprocess(_A     )							,				output_type='''np'''							,				)
								UpperCAmelCase_     :   Any       =   image_processor.postprocess(
								    image_processor.preprocess(_A     )							,				output_type='''np'''							,				)

								UpperCAmelCase_     :   str       =   self.to_np(self.dummy_sample     )
								UpperCAmelCase_     :   Tuple       =   input_np_ad.squeeze(0     )

								UpperCAmelCase_     :   Optional[int]       =   image_processor.postprocess(
								    image_processor.preprocess(_A     )							,				output_type='''np'''							,				)
								UpperCAmelCase_     :   Optional[Any]       =   image_processor.postprocess(
								    image_processor.preprocess(_A     )							,				output_type='''np'''							,				)

								assert np.abs(out_pt_ad - out_pt_ad     ).max() < 1e-6
								assert np.abs(out_np_ad - out_np_ad     ).max() < 1e-6





					def 						A     (       self       :						Dict     )  ->    Dict:
								UpperCAmelCase_     :   List[Any]       =   VaeImageProcessor(do_resize=_A							,				do_normalize=_A     )

								UpperCAmelCase_     :   Optional[int]       =   self.dummy_sample
								UpperCAmelCase_     :   Dict       =   list(_A     )

								UpperCAmelCase_     :   str       =   image_processor.postprocess(
								    image_processor.preprocess(_A     )							,				output_type='''np'''							,				)

								UpperCAmelCase_     :   Optional[Any]       =   image_processor.postprocess(
								    image_processor.preprocess(_A     )							,				output_type='''np'''							,				)

								UpperCAmelCase_     :   Dict       =   self.to_np(self.dummy_sample     )
								list(_A     )

								UpperCAmelCase_     :   str       =   image_processor.postprocess(
								    image_processor.preprocess(_A     )							,				output_type='''np'''							,				)

								UpperCAmelCase_     :   Optional[int]       =   image_processor.postprocess(
								    image_processor.preprocess(_A     )							,				output_type='''np'''							,				)

								assert np.abs(out_pt_ad - out_pt_list     ).max() < 1e-6
								assert np.abs(out_np_ad - out_np_list     ).max() < 1e-6


