

"""simple docstring"""


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

from ...utils import logging


__A						   =			logging.get_logger(__name__)






def     a__    (    __SCREAMING_SNAKE_CASE					, __SCREAMING_SNAKE_CASE )	->	List[str]:
		__lowerCAmelCase:       str          =			nn.functional.normalize(SCREAMING_SNAKE_CASE__ )
		__lowerCAmelCase:       Dict          =			nn.functional.normalize(SCREAMING_SNAKE_CASE__ )
		return torch.mm(SCREAMING_SNAKE_CASE__					, normalized_text_embeds.t() )





class     snake_case	(				__lowerCamelCase  ):
				SCREAMING_SNAKE_CASE_   :      int         =  CLIPConfig

				SCREAMING_SNAKE_CASE_   :      List[str]         =  ["""CLIPEncoderLayer"""]

				def __init__(	self							:				Tuple     ,  UpperCamelCase__							:				CLIPConfig)->   Union[str, Any]:



						'''simple docstring'''


						super().__init__(UpperCamelCase__)

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

						__lowerCAmelCase:       int          =			nn.Parameter(torch.ones(1_7     ,  config.projection_dim)     ,  requires_grad=UpperCamelCase__)
						__lowerCAmelCase:       Dict          =			nn.Parameter(torch.ones(3     ,  config.projection_dim)     ,  requires_grad=UpperCamelCase__)

						__lowerCAmelCase:       Tuple          =			nn.Parameter(torch.ones(1_7)     ,  requires_grad=UpperCamelCase__)
						__lowerCAmelCase:       Union[str, Any]          =			nn.Parameter(torch.ones(3)     ,  requires_grad=UpperCamelCase__)

				@torch.no_grad()
				def 							lowercase_     (	self							:				List[str]     ,  UpperCamelCase__							:				Tuple     ,  UpperCamelCase__							:				List[str])->   Tuple:



						'''simple docstring'''


						__lowerCAmelCase:       Optional[Any]          =			self.vision_model(UpperCamelCase__)[1]  # pooled_output
						__lowerCAmelCase:       List[Any]          =			self.visual_projection(UpperCamelCase__)

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

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

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

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

								result.append(UpperCamelCase__)

						__lowerCAmelCase:       Optional[int]          =			[len(res["bad_concepts"]) > 0 for res in result]

						return images, has_nsfw_concepts




				@torch.no_grad()
				def 							lowercase_     (	self							:				Optional[Any]     ,  UpperCamelCase__							:				torch.FloatTensor     ,  UpperCamelCase__							:				torch.FloatTensor)->   Any:



						'''simple docstring'''


						__lowerCAmelCase:       Optional[Any]          =			self.vision_model(UpperCamelCase__)[1]  # pooled_output
						__lowerCAmelCase:       str          =			self.visual_projection(UpperCamelCase__)

						__lowerCAmelCase:       Tuple          =			cosine_distance(UpperCamelCase__     ,  self.special_care_embeds)
						__lowerCAmelCase:       Union[str, Any]          =			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
						__lowerCAmelCase:       Dict          =			0.0

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

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

						return images, has_nsfw_concepts









from __future__ import annotations

from sys import maxsize
from typing import Generic, TypeVar

__UpperCAmelCase       :   str				       =							TypeVar("T")







def 			A__							(    SCREAMING_SNAKE_CASE__)   ->    int:
  return (position - 1) // 2







def 			A__							(    SCREAMING_SNAKE_CASE__)   ->    int:
  return (2 * position) + 1







def 			A__							(    SCREAMING_SNAKE_CASE__)   ->    int:
  return (2 * position) + 2







class     __snake_case       (							Generic[T]			):





       '''simple docstring'''







       def __init__(  self :    Tuple		):
         __snake_case:						list[tuple[T, int]]   =						[]
         __snake_case:						dict[T, int]   =						{}
         __snake_case:						int   =						0




       def __len__(  self :    Tuple		):
         return self.elements




       def __repr__(  self :    Optional[int]		):
         return str(self.heap		)




       def 							UpperCAmelCase__ (  self :    List[Any]		):
         # Check if the priority queue is empty
         return self.elements == 0




       def 							UpperCAmelCase__ (  self :    List[Any]							,				A :    T							,				A :    int		):
         # Add an element with given priority to the queue
         self.heap.append((elem, weight)		)
         __snake_case:						Optional[Any]   =						self.elements
         self.elements += 1
         self._bubble_up(A		)




       def 							UpperCAmelCase__ (  self :    List[Any]		):
         # Remove and return the element with lowest weight (highest priority)
         if self.elements > 1:
           self._swap_nodes(0							,				self.elements - 1		)
         __snake_case  ,   __snake_case:						Union[str, Any]   =						self.heap.pop()
         del self.position_map[elem]
         self.elements -= 1
         if self.elements > 0:
           __snake_case  ,   __snake_case:						int   =						self.heap[0]
           self._bubble_down(A		)
         return elem




       def 							UpperCAmelCase__ (  self :    Any							,				A :    T							,				A :    int		):
         # Update the weight of the given key
         __snake_case:						Dict   =						self.position_map[elem]
         __snake_case:						Optional[Any]   =						(elem, weight)
         if position > 0:
           __snake_case:						Optional[int]   =						get_parent_position(A		)
           __snake_case  ,   __snake_case:						Dict   =						self.heap[parent_position]
           if parent_weight > weight:
             self._bubble_up(A		)
           else:
             self._bubble_down(A		)
         else:
           self._bubble_down(A		)




       def 							UpperCAmelCase__ (  self :    int							,				A :    T		):
         # Place a node at the proper position (upward movement) [to be used internally
         # only]
         __snake_case:						int   =						self.position_map[elem]
         if curr_pos == 0:
           return None
         __snake_case:						Optional[Any]   =						get_parent_position(A		)
         __snake_case  ,   __snake_case:						Tuple   =						self.heap[curr_pos]
         __snake_case  ,   __snake_case:						Optional[Any]   =						self.heap[parent_position]
         if parent_weight > weight:
           self._swap_nodes(A							,				A		)
           return self._bubble_up(A		)
         return None




       def 							UpperCAmelCase__ (  self :    List[str]							,				A :    T		):
         # Place a node at the proper position (downward movement) [to be used
         # internally only]
         __snake_case:						Any   =						self.position_map[elem]
         __snake_case  ,   __snake_case:						int   =						self.heap[curr_pos]
         __snake_case:						Union[str, Any]   =						get_child_left_position(A		)
         __snake_case:						Dict   =						get_child_right_position(A		)
         if child_left_position < self.elements and child_right_position < self.elements:
           __snake_case  ,   __snake_case:						Any   =						self.heap[child_left_position]
           __snake_case  ,   __snake_case:						List[str]   =						self.heap[child_right_position]
           if child_right_weight < child_left_weight and child_right_weight < weight:
             self._swap_nodes(A							,				A		)
             return self._bubble_down(A		)
         if child_left_position < self.elements:
           __snake_case  ,   __snake_case:						List[Any]   =						self.heap[child_left_position]
           if child_left_weight < weight:
             self._swap_nodes(A							,				A		)
             return self._bubble_down(A		)
         else:
           return None
         if child_right_position < self.elements:
           __snake_case  ,   __snake_case:						List[Any]   =						self.heap[child_right_position]
           if child_right_weight < weight:
             self._swap_nodes(A							,				A		)
             return self._bubble_down(A		)
         return None






       def 							UpperCAmelCase__ (  self :    int							,				A :    int							,				A :    int		):
         # Swap the nodes at the given positions
         __snake_case:						Optional[int]   =						self.heap[nodea_pos][0]
         __snake_case:						Optional[Any]   =						self.heap[nodea_pos][0]
         __snake_case  ,   __snake_case:						Tuple   =						(
             self.heap[nodea_pos],
             self.heap[nodea_pos],
         )
         __snake_case:						Any   =						nodea_pos
         __snake_case:						str   =						nodea_pos







class     __snake_case       (							Generic[T]			):





       '''simple docstring'''







       def __init__(  self :    Any		):
         __snake_case:						dict[T, dict[T, int]]   =						{}
         __snake_case:						int   =						0




       def __repr__(  self :    Dict		):
         return str(self.connections		)




       def __len__(  self :    List[str]		):
         return self.nodes




       def 							UpperCAmelCase__ (  self :    Optional[Any]							,				A :    T		):
         # Add a node in the graph if it is not in the graph
         if node not in self.connections:
           __snake_case:						List[str]   =						{}
           self.nodes += 1






       def 							UpperCAmelCase__ (  self :    Optional[Any]							,				A :    T							,				A :    T							,				A :    int		):
         # Add an edge between 2 nodes in the graph
         self.add_node(A		)
         self.add_node(A		)
         __snake_case:						Optional[Any]   =						weight
         __snake_case:						str   =						weight







def 			A__							(    SCREAMING_SNAKE_CASE__					,					)   ->    tuple[dict[T, int], dict[T, T | None]]:
  __snake_case:						dict[T, int]   =						{node: maxsize for node in graph.connections}
  __snake_case:						dict[T, T | None]   =						{node: None for node in graph.connections}

  __snake_case:						MinPriorityQueue[T]   =						MinPriorityQueue()
  for node, weight in dist.items():
    priority_queue.push(SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)

  if priority_queue.is_empty():
    return dist, parent

  # initialization
  __snake_case:						Optional[int]   =						priority_queue.extract_min()
  __snake_case:						Any   =						0
  for neighbour in graph.connections[node]:
    if dist[neighbour] > dist[node] + graph.connections[node][neighbour]:
      __snake_case:						Tuple   =						dist[node] + graph.connections[node][neighbour]
      priority_queue.update_key(SCREAMING_SNAKE_CASE__					,					dist[neighbour])
      __snake_case:						Optional[int]   =						node

    # running prim's algorithm
  while not priority_queue.is_empty():
    __snake_case:						Optional[int]   =						priority_queue.extract_min()
    for neighbour in graph.connections[node]:
      if dist[neighbour] > dist[node] + graph.connections[node][neighbour]:
        __snake_case:						Tuple   =						dist[node] + graph.connections[node][neighbour]
        priority_queue.update_key(SCREAMING_SNAKE_CASE__					,					dist[neighbour])
        __snake_case:						List[Any]   =						node
  return dist, parent






