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

from ...utils import logging


lowerCAmelCase__					:       List[str]		=       logging.get_logger(__name__)
def 			UpperCamelCase__						(    A__      ,			A__   )       -> Dict:
				snake_case__ :     Optional[int] 							=				nn.functional.normalize(A__   )
				snake_case__ :     int 							=				nn.functional.normalize(A__   )
				return torch.mm(A__      ,			normalized_text_embeds.t()   )





class 			__snake_case  (      A__     ):
					__lowerCamelCase	 =       CLIPConfig

					__lowerCamelCase	 =       ["""CLIPEncoderLayer"""]




					def __init__(       self   ,	__UpperCamelCase		)					->  Tuple:



									'''simple docstring'''







									super().__init__(_A		)

									snake_case__ :     Optional[Any] 							=				CLIPVisionModel(config.vision_config		)
									snake_case__ :     str 							=				nn.Linear(config.vision_config.hidden_size   ,	config.projection_dim   ,	bias=_A		)

									snake_case__ :     Union[str, Any] 							=				nn.Parameter(torch.ones(17   ,	config.projection_dim		)   ,	requires_grad=_A		)
									snake_case__ :     List[str] 							=				nn.Parameter(torch.ones(3   ,	config.projection_dim		)   ,	requires_grad=_A		)

									snake_case__ :     Dict 							=				nn.Parameter(torch.ones(17		)   ,	requires_grad=_A		)
									snake_case__ :     Any 							=				nn.Parameter(torch.ones(3		)   ,	requires_grad=_A		)




					@torch.no_grad()
					def       __a      (       self   ,	__UpperCamelCase   ,	__UpperCamelCase		)					->  Optional[int]:



									'''simple docstring'''







									snake_case__ :     str 							=				self.vision_model(_A		)[1]  # pooled_output
									snake_case__ :     Tuple 							=				self.visual_projection(_A		)

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

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

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

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

													result.append(_A		)

									snake_case__ :     Optional[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]		):
																					snake_case__ :     Any 							=				torch.zeros_like(images[idx]		)  # black image
																	else:
																					snake_case__ :     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   ,	__UpperCamelCase   ,	__UpperCamelCase		)					->  List[Any]:



									'''simple docstring'''







									snake_case__ :     Optional[Any] 							=				self.vision_model(_A		)[1]  # pooled_output
									snake_case__ :     Any 							=				self.visual_projection(_A		)

									snake_case__ :     int 							=				cosine_distance(_A   ,	self.special_care_embeds		)
									snake_case__ :     List[str] 							=				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
									snake_case__ :     List[Any] 							=				0.0

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

									snake_case__ :     Optional[Any] 							=				(cos_dist - self.concept_embeds_weights) + special_adjustment
									# concept_scores = concept_scores.round(decimals=3)
									snake_case__ :     List[str] 							=				torch.any(concept_scores > 0   ,	dim=1		)

									snake_case__ :     List[Any] 							=				0.0  # black image

									return images, has_nsfw_concepts



import inspect
import os
import unittest

import torch

import accelerate
from accelerate import debug_launcher
from accelerate.test_utils import (
    execute_subprocess_async,
    require_cpu,
    require_multi_gpu,
    require_single_gpu,
    test_sync,
)
from accelerate.utils import patch_environment





class 			__snake_case  (      unittest.TestCase     ):




					def       __a      (       self		)					->  Union[str, Any]:



									'''simple docstring'''







									snake_case__ :     List[Any] 							=				inspect.getfile(accelerate.test_utils		)
									snake_case__ :     Tuple 							=				os.path.sep.join(mod_file.split(os.path.sep		)[:-1] + ['scripts', 'test_sync.py']		)




					@require_cpu
					def       __a      (       self		)					->  int:



									'''simple docstring'''







									debug_launcher(test_sync.main   ,	num_processes=1		)




					@require_cpu
					def       __a      (       self		)					->  Dict:



									'''simple docstring'''







									debug_launcher(test_sync.main		)




					@require_single_gpu
					def       __a      (       self		)					->  int:



									'''simple docstring'''







									test_sync.main()




					@require_multi_gpu
					def       __a      (       self		)					->  str:



									'''simple docstring'''







									print(F"""Found {torch.cuda.device_count()} devices."""		)
									snake_case__ :     Any 							=				['torchrun', F"""--nproc_per_node={torch.cuda.device_count()}""", self.test_file_path]
									with patch_environment(omp_num_threads=1		):
													execute_subprocess_async(__UpperCamelCase   ,	env=os.environ.copy()		)


