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

from ...utils import logging


lowerCamelCase__           =				logging.get_logger(__name__)



def 							_lowerCamelCase(				__snake_case						,						__snake_case	)  ->   Optional[int]:
		__snake_case		       =     nn.functional.normalize(__snake_case	)
		__snake_case		       =     nn.functional.normalize(__snake_case	)
		return torch.mm(__snake_case						,						normalized_text_embeds.t()	)





class 					UpperCamelCase	(     snake_case__      ):
					__UpperCamelCase						=   CLIPConfig

					__UpperCamelCase						=   ["""CLIPEncoderLayer"""]




					def __init__( self :					Union[str, Any]      ,_lowerCAmelCase :					CLIPConfig						):


							"""simple docstring"""




							super().__init__(_lowerCAmelCase						)

							__snake_case		       =     CLIPVisionModel(config.vision_config						)
							__snake_case		       =     nn.Linear(config.vision_config.hidden_size      ,config.projection_dim      ,bias=_lowerCAmelCase						)

							__snake_case		       =     nn.Parameter(torch.ones(17      ,config.projection_dim						)      ,requires_grad=_lowerCAmelCase						)
							__snake_case		       =     nn.Parameter(torch.ones(3      ,config.projection_dim						)      ,requires_grad=_lowerCAmelCase						)

							__snake_case		       =     nn.Parameter(torch.ones(17						)      ,requires_grad=_lowerCAmelCase						)
							__snake_case		       =     nn.Parameter(torch.ones(3						)      ,requires_grad=_lowerCAmelCase						)




					@torch.no_grad()
					def 							UpperCamelCase_						( self :					List[Any]      ,_lowerCAmelCase :					List[Any]      ,_lowerCAmelCase :					Any						):


							"""simple docstring"""




							__snake_case		       =     self.vision_model(_lowerCAmelCase						)[1]  # pooled_output
							__snake_case		       =     self.visual_projection(_lowerCAmelCase						)

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

							__snake_case		       =     []
							__snake_case		       =     image_embeds.shape[0]
							for i in range(_lowerCAmelCase						):
									__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
									__snake_case		       =     0.0

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

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

									result.append(_lowerCAmelCase						)

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

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

							if any(_lowerCAmelCase						):
									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 							UpperCamelCase_						( self :					List[Any]      ,_lowerCAmelCase :					torch.FloatTensor      ,_lowerCAmelCase :					torch.FloatTensor						):


							"""simple docstring"""




							__snake_case		       =     self.vision_model(_lowerCAmelCase						)[1]  # pooled_output
							__snake_case		       =     self.visual_projection(_lowerCAmelCase						)

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

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

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

							__snake_case		       =     0.0  # black image

							return images, has_nsfw_concepts






from __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any




class 					UpperCamelCase	:
					__UpperCamelCase						=   ("""_front""", """_back""", """_len""")




					@dataclass
					class 					UpperCamelCase	:
										__UpperCamelCase						=   None
										__UpperCamelCase						=   None
										__UpperCamelCase						=   None




					class 					UpperCamelCase	:
										__UpperCamelCase						=   ("""_cur""",)




										def __init__( self :					Dict      ,_lowerCAmelCase :					Deque._Node | None						):


												"""simple docstring"""




												__snake_case		       =     cur




										def __iter__( self :					Dict						):


												"""simple docstring"""




												return self







										def 							UpperCamelCase_						( self :					Optional[Any]						):


												"""simple docstring"""




												if self._cur is None:
														# finished iterating
														raise StopIteration
												__snake_case		       =     self._cur.val
												__snake_case		       =     self._cur.next_node

												return val




					def __init__( self :					Optional[int]      ,_lowerCAmelCase :					Iterable[Any] | None = None						):


							"""simple docstring"""




							__snake_case		       =     None
							__snake_case		       =     None
							__snake_case		       =     0

							if iterable is not None:
									# append every value to the deque
									for val in iterable:
											self.append(_lowerCAmelCase						)




					def 							UpperCamelCase_						( self :					int      ,_lowerCAmelCase :					Any						):


							"""simple docstring"""




							__snake_case		       =     self._Node(_lowerCAmelCase      ,_lowerCAmelCase      ,_lowerCAmelCase						)
							if self.is_empty():
									# front = back
									__snake_case		       =     __snake_case		       =     node
									__snake_case		       =     1
							else:
									# connect nodes
									__snake_case		       =     node
									__snake_case		       =     self._back
									__snake_case		       =     node  # assign new back to the new node

									self._len += 1

									# make sure there were no errors
									assert not self.is_empty(), "Error on appending value."




					def 							UpperCamelCase_						( self :					Any      ,_lowerCAmelCase :					Any						):


							"""simple docstring"""




							__snake_case		       =     self._Node(_lowerCAmelCase      ,_lowerCAmelCase      ,_lowerCAmelCase						)
							if self.is_empty():
									# front = back
									__snake_case		       =     __snake_case		       =     node
									__snake_case		       =     1
							else:
									# connect nodes
									__snake_case		       =     self._front
									__snake_case		       =     node
									__snake_case		       =     node  # assign new front to the new node

									self._len += 1

									# make sure there were no errors
									assert not self.is_empty(), "Error on appending value."




					def 							UpperCamelCase_						( self :					int      ,_lowerCAmelCase :					Iterable[Any]						):


							"""simple docstring"""




							for val in iterable:
									self.append(_lowerCAmelCase						)




					def 							UpperCamelCase_						( self :					Optional[int]      ,_lowerCAmelCase :					Iterable[Any]						):


							"""simple docstring"""




							for val in iterable:
									self.appendleft(_lowerCAmelCase						)




					def 							UpperCamelCase_						( self :					List[Any]						):


							"""simple docstring"""




							assert not self.is_empty(), "Deque is empty."

							__snake_case		       =     self._back
							__snake_case		       =     self._back.prev_node  # set new back
							# drop the last node - python will deallocate memory automatically
							__snake_case		       =     None

							self._len -= 1

							return topop.val




					def 							UpperCamelCase_						( self :					List[str]						):


							"""simple docstring"""




							assert not self.is_empty(), "Deque is empty."

							__snake_case		       =     self._front
							__snake_case		       =     self._front.next_node  # set new front and drop the first node
							__snake_case		       =     None

							self._len -= 1

							return topop.val




					def 							UpperCamelCase_						( self :					int						):


							"""simple docstring"""




							return self._front is None




					def __len__( self :					Tuple						):


							"""simple docstring"""




							return self._len




					def __eq__( self :					Union[str, Any]      ,_lowerCAmelCase :					object						):


							"""simple docstring"""




							if not isinstance(_lowerCAmelCase      ,_lowerCAmelCase						):
									return NotImplemented

							__snake_case		       =     self._front
							__snake_case		       =     other._front

							# if the length of the dequeues are not the same, they are not equal
							if len(self						) != len(_lowerCAmelCase						):
									return False

							while me is not None and oth is not None:
									# compare every value
									if me.val != oth.val:
											return False
									__snake_case		       =     me.next_node
									__snake_case		       =     oth.next_node

							return True




					def __iter__( self :					Tuple						):


							"""simple docstring"""




							return Deque._Iterator(self._front						)







					def __repr__( self :					List[Any]						):


							"""simple docstring"""




							__snake_case		       =     []
							__snake_case		       =     self._front
							while aux is not None:
									# append the values in a list to display
									values_list.append(aux.val						)
									__snake_case		       =     aux.next_node

							return F"""[{", ".join(repr(_lowerCAmelCase						) for val in values_list						)}]"""


if __name__ == "__main__":
		import doctest

		doctest.testmod()





