






"""simple docstring"""

from typing import Any

def 		__lowerCamelCase	(  __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       )						->    str:

		"""simple docstring"""







		_validation(
		    lowercase__      ,       lowercase__      ,       lowercase__      ,       lowercase__      ,       lowercase__      ,       )
		# Creates data structures and fill initial step
		lowerCAmelCase_   :							dict									=			{}
		lowerCAmelCase_   :							dict									=			{}
		for state in states_space:
				lowerCAmelCase_   :							List[Any]									=			observations_space[0]
				lowerCAmelCase_   :							str									=			(
				    initial_probabilities[state] * emission_probabilities[state][observation]
				)
				lowerCAmelCase_   :							List[Any]									=			None

		# Fills the data structure with the probabilities of
		# different transitions and pointers to previous states
		for o in range(1      ,       len(lowercase__       )       ):
				lowerCAmelCase_   :							List[Any]									=			observations_space[o]
				lowerCAmelCase_   :							Optional[Any]									=			observations_space[o - 1]
				for state in states_space:
						# Calculates the argmax for probability function
						lowerCAmelCase_   :							int									=			''''''
						lowerCAmelCase_   :							Any									=			-1
						for k_state in states_space:
								lowerCAmelCase_   :							int									=			(
								    probabilities[(k_state, prior_observation)]
								    * transition_probabilities[k_state][state]
								    * emission_probabilities[state][observation]
								)
								if probability > max_probability:
										lowerCAmelCase_   :							Tuple									=			probability
										lowerCAmelCase_   :							Union[str, Any]									=			k_state

            # Update probabilities and pointers dicts
						lowerCAmelCase_   :							str									=			(
						    probabilities[(arg_max, prior_observation)]
						    * transition_probabilities[arg_max][state]
						    * emission_probabilities[state][observation]
						)

						lowerCAmelCase_   :							Union[str, Any]									=			arg_max

    # The final observation
		lowerCAmelCase_   :							Union[str, Any]									=			observations_space[len(lowercase__       ) - 1]

		# argmax for given final observation
		lowerCAmelCase_   :							Tuple									=			''''''
		lowerCAmelCase_   :							Optional[int]									=			-1
		for k_state in states_space:
				lowerCAmelCase_   :							Dict									=			probabilities[(k_state, final_observation)]
				if probability > max_probability:
						lowerCAmelCase_   :							int									=			probability
						lowerCAmelCase_   :							Optional[int]									=			k_state
		lowerCAmelCase_   :							Optional[int]									=			arg_max

		# Process pointers backwards
		lowerCAmelCase_   :							List[Any]									=			last_state
		lowerCAmelCase_   :							Optional[Any]									=			[]
		for o in range(len(lowercase__       ) - 1      ,       -1      ,       -1       ):
				result.append(lowercase__       )
				lowerCAmelCase_   :							Tuple									=			pointers[previous, observations_space[o]]
		result.reverse()

		return result

def 		__lowerCamelCase	(  __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       )						->    int:

		"""simple docstring"""







		_validate_not_empty(
		    lowercase__      ,       lowercase__      ,       lowercase__      ,       lowercase__      ,       lowercase__      ,       )
		_validate_lists(lowercase__      ,       lowercase__       )
		_validate_dicts(
		    lowercase__      ,       lowercase__      ,       lowercase__       )

def 		__lowerCamelCase	(  __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       )						->    List[str]:

		"""simple docstring"""







		if not all(
		    [
		        observations_space,
		        states_space,
		        initial_probabilities,
		        transition_probabilities,
		        emission_probabilities,
		    ]       ):
				raise ValueError("There\'s an empty parameter"       )

def 		__lowerCamelCase	(  __UpperCamelCase      ,       __UpperCamelCase       )						->    Tuple:

		"""simple docstring"""







		_validate_list(lowercase__      ,       "observations_space"       )
		_validate_list(lowercase__      ,       "states_space"       )

def 		__lowerCamelCase	(  __UpperCamelCase      ,       __UpperCamelCase       )						->    int:

		"""simple docstring"""







		if not isinstance(_object      ,       lowercase__       ):
				lowerCAmelCase_   :							str									=			f'''{var_name} must be a list'''
				raise ValueError(lowercase__       )
		else:
				for x in _object:
						if not isinstance(lowercase__      ,       lowercase__       ):
								lowerCAmelCase_   :							str									=			f'''{var_name} must be a list of strings'''
								raise ValueError(lowercase__       )

def 		__lowerCamelCase	(  __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       )						->    str:

		"""simple docstring"""







		_validate_dict(lowercase__      ,       "initial_probabilities"      ,       lowercase__       )
		_validate_nested_dict(lowercase__      ,       "transition_probabilities"       )
		_validate_nested_dict(lowercase__      ,       "emission_probabilities"       )

def 		__lowerCamelCase	(  __UpperCamelCase      ,       __UpperCamelCase       )						->    int:

		"""simple docstring"""







		_validate_dict(_object      ,       lowercase__      ,       lowercase__       )
		for x in _object.values():
				_validate_dict(lowercase__      ,       lowercase__      ,       lowercase__      ,       lowercase__       )

def 		__lowerCamelCase	(  __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase      ,       __UpperCamelCase = False       )						->    List[str]:

		"""simple docstring"""







		if not isinstance(_object      ,       lowercase__       ):
				lowerCAmelCase_   :							str									=			f'''{var_name} must be a dict'''
				raise ValueError(lowercase__       )
		if not all(isinstance(lowercase__      ,       lowercase__       ) for x in _object       ):
				lowerCAmelCase_   :							Dict									=			f'''{var_name} all keys must be strings'''
				raise ValueError(lowercase__       )
		if not all(isinstance(lowercase__      ,       lowercase__       ) for x in _object.values()       ):
				lowerCAmelCase_   :							Tuple									=			'''nested dictionary ''' if nested else ''''''
				lowerCAmelCase_   :							Optional[Any]									=			f'''{var_name} {nested_text}all values must be {value_type.__name__}'''
				raise ValueError(lowercase__       )


if __name__ == "__main__":
					from doctest import testmod

					testmod()




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

from ...utils import logging


__lowerCAmelCase	: Dict  				=logging.get_logger(__name__)


def        _UpperCamelCase							(      lowercase__    ,       lowercase__      ):
						__SCREAMING_SNAKE_CASE						:     List[Any]					=		nn.functional.normalize(lowercase__      )
						__SCREAMING_SNAKE_CASE						:     Tuple					=		nn.functional.normalize(lowercase__      )
						return torch.mm(lowercase__    ,       normalized_text_embeds.t()      )
class        _lowercase      (			A__       ):






				'''simple docstring'''



				SCREAMING_SNAKE_CASE__				:			Optional[Any]						=		CLIPConfig

				SCREAMING_SNAKE_CASE__				:			List[str]						=		['''CLIPEncoderLayer''']


				def __init__(       self			:str			,			lowerCAmelCase__			:CLIPConfig	)		->			Tuple:
										super().__init__(lowerCAmelCase__	)

										__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		CLIPVisionModel(config.vision_config	)
										__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		nn.Linear(config.vision_config.hidden_size			,			config.projection_dim			,			bias=lowerCAmelCase__	)

										__SCREAMING_SNAKE_CASE						:     Any					=		nn.Parameter(torch.ones(17			,			config.projection_dim	)			,			requires_grad=lowerCAmelCase__	)
										__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		nn.Parameter(torch.ones(3			,			config.projection_dim	)			,			requires_grad=lowerCAmelCase__	)

										__SCREAMING_SNAKE_CASE						:     Dict					=		nn.Parameter(torch.ones(17	)			,			requires_grad=lowerCAmelCase__	)
										__SCREAMING_SNAKE_CASE						:     str					=		nn.Parameter(torch.ones(3	)			,			requires_grad=lowerCAmelCase__	)


				@torch.no_grad()
				def 		__magic_name__(       self			:Optional[int]			,			lowerCAmelCase__			:int			,			lowerCAmelCase__			:Optional[int]	)		->			Dict:
										__SCREAMING_SNAKE_CASE						:     int					=		self.vision_model(lowerCAmelCase__	)[1]  # pooled_output
										__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		self.visual_projection(lowerCAmelCase__	)

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

										__SCREAMING_SNAKE_CASE						:     List[Any]					=		[]
										__SCREAMING_SNAKE_CASE						:     List[Any]					=		image_embeds.shape[0]
										for i in range(lowerCAmelCase__	):
																__SCREAMING_SNAKE_CASE						:     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
																__SCREAMING_SNAKE_CASE						:     List[Any]					=		0.0

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

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

																result.append(lowerCAmelCase__	)

										__SCREAMING_SNAKE_CASE						:     List[str]					=		[len(res['''bad_concepts''']	) > 0 for res in result]

										return images, has_nsfw_concepts



				@torch.no_grad()
				def 		__magic_name__(       self			:Optional[int]			,			lowerCAmelCase__			:torch.FloatTensor			,			lowerCAmelCase__			:torch.FloatTensor	)		->			Any:
										__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		self.vision_model(lowerCAmelCase__	)[1]  # pooled_output
										__SCREAMING_SNAKE_CASE						:     List[str]					=		self.visual_projection(lowerCAmelCase__	)

										__SCREAMING_SNAKE_CASE						:     Dict					=		cosine_distance(lowerCAmelCase__			,			self.special_care_embeds	)
										__SCREAMING_SNAKE_CASE						:     Optional[int]					=		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
										__SCREAMING_SNAKE_CASE						:     List[Any]					=		0.0

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

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

										return images, has_nsfw_concepts
