



"""simple docstring"""





import flax.linen as nn
import jax.numpy as jnp

from .attention_flax import FlaxTransformeraDModel
from .resnet_flax import FlaxDownsampleaD, FlaxResnetBlockaD, FlaxUpsampleaD



class  _lowerCAmelCase     (					nn.Module		):





		"""simple docstring"""






		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							0.0
		_lowerCamelCase      	=							1
		_lowerCamelCase      	=							1
		_lowerCamelCase      	=							True
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							jnp.floataa





		def  UpperCAmelCase__				(       self )			->    Tuple:







								'''simple docstring'''

								snake_case_ :			Optional[int]            =      []
								snake_case_ :			Optional[int]            =      []

								for i in range(self.num_layers ):
														snake_case_ :			List[str]            =      self.in_channels if i == 0 else self.out_channels

														snake_case_ :			Optional[int]            =      FlaxResnetBlockaD(
														    in_channels=_lowercase				,     out_channels=self.out_channels				,     dropout_prob=self.dropout				,     dtype=self.dtype				,     )
														resnets.append(_lowercase )

														snake_case_ :			Any            =      FlaxTransformeraDModel(
														    in_channels=self.out_channels				,     n_heads=self.num_attention_heads				,     d_head=self.out_channels // self.num_attention_heads				,     depth=1				,     use_linear_projection=self.use_linear_projection				,     only_cross_attention=self.only_cross_attention				,     use_memory_efficient_attention=self.use_memory_efficient_attention				,     dtype=self.dtype				,     )
														attentions.append(_lowercase )

								snake_case_ :			Any            =      resnets
								snake_case_ :			Any            =      attentions

								if self.add_downsample:
														snake_case_ :			Any            =      FlaxDownsampleaD(self.out_channels				,     dtype=self.dtype )





		def __call__(       self				,     _lowercase				,     _lowercase				,     _lowercase				,     _lowercase=True )			->    Optional[int]:







								'''simple docstring'''

								snake_case_ :			Union[str, Any]            =      ()

								for resnet, attn in zip(self.resnets				,     self.attentions ):
														snake_case_ :			List[Any]            =      resnet(_lowercase				,     _lowercase				,     deterministic=_lowercase )
														snake_case_ :			Optional[int]            =      attn(_lowercase				,     _lowercase				,     deterministic=_lowercase )
														output_states += (hidden_states,)

								if self.add_downsample:
														snake_case_ :			List[Any]            =      self.downsamplers_a(_lowercase )
														output_states += (hidden_states,)

								return hidden_states, output_states




class  _lowerCAmelCase     (					nn.Module		):





		"""simple docstring"""






		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							0.0
		_lowerCamelCase      	=							1
		_lowerCamelCase      	=							True
		_lowerCamelCase      	=							jnp.floataa





		def  UpperCAmelCase__				(       self )			->    Optional[Any]:







								'''simple docstring'''

								snake_case_ :			int            =      []

								for i in range(self.num_layers ):
														snake_case_ :			int            =      self.in_channels if i == 0 else self.out_channels

														snake_case_ :			Optional[Any]            =      FlaxResnetBlockaD(
														    in_channels=_lowercase				,     out_channels=self.out_channels				,     dropout_prob=self.dropout				,     dtype=self.dtype				,     )
														resnets.append(_lowercase )
								snake_case_ :			Dict            =      resnets

								if self.add_downsample:
														snake_case_ :			Any            =      FlaxDownsampleaD(self.out_channels				,     dtype=self.dtype )





		def __call__(       self				,     _lowercase				,     _lowercase				,     _lowercase=True )			->    List[Any]:







								'''simple docstring'''

								snake_case_ :			Dict            =      ()

								for resnet in self.resnets:
														snake_case_ :			List[Any]            =      resnet(_lowercase				,     _lowercase				,     deterministic=_lowercase )
														output_states += (hidden_states,)

								if self.add_downsample:
														snake_case_ :			Union[str, Any]            =      self.downsamplers_a(_lowercase )
														output_states += (hidden_states,)

								return hidden_states, output_states




class  _lowerCAmelCase     (					nn.Module		):





		"""simple docstring"""






		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							0.0
		_lowerCamelCase      	=							1
		_lowerCamelCase      	=							1
		_lowerCamelCase      	=							True
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							jnp.floataa





		def  UpperCAmelCase__				(       self )			->    Optional[int]:







								'''simple docstring'''

								snake_case_ :			List[Any]            =      []
								snake_case_ :			Tuple            =      []

								for i in range(self.num_layers ):
														snake_case_ :			Dict            =      self.in_channels if (i == self.num_layers - 1) else self.out_channels
														snake_case_ :			Tuple            =      self.prev_output_channel if i == 0 else self.out_channels

														snake_case_ :			str            =      FlaxResnetBlockaD(
														    in_channels=resnet_in_channels + res_skip_channels				,     out_channels=self.out_channels				,     dropout_prob=self.dropout				,     dtype=self.dtype				,     )
														resnets.append(_lowercase )

														snake_case_ :			Optional[int]            =      FlaxTransformeraDModel(
														    in_channels=self.out_channels				,     n_heads=self.num_attention_heads				,     d_head=self.out_channels // self.num_attention_heads				,     depth=1				,     use_linear_projection=self.use_linear_projection				,     only_cross_attention=self.only_cross_attention				,     use_memory_efficient_attention=self.use_memory_efficient_attention				,     dtype=self.dtype				,     )
														attentions.append(_lowercase )

								snake_case_ :			Union[str, Any]            =      resnets
								snake_case_ :			Any            =      attentions

								if self.add_upsample:
														snake_case_ :			Tuple            =      FlaxUpsampleaD(self.out_channels				,     dtype=self.dtype )





		def __call__(       self				,     _lowercase				,     _lowercase				,     _lowercase				,     _lowercase				,     _lowercase=True )			->    Union[str, Any]:







								'''simple docstring'''

								for resnet, attn in zip(self.resnets				,     self.attentions ):
														# pop res hidden states
														snake_case_ :			Union[str, Any]            =      res_hidden_states_tuple[-1]
														snake_case_ :			Optional[Any]            =      res_hidden_states_tuple[:-1]
														snake_case_ :			List[Any]            =      jnp.concatenate((hidden_states, res_hidden_states)				,     axis=-1 )

														snake_case_ :			List[str]            =      resnet(_lowercase				,     _lowercase				,     deterministic=_lowercase )
														snake_case_ :			int            =      attn(_lowercase				,     _lowercase				,     deterministic=_lowercase )

								if self.add_upsample:
														snake_case_ :			List[Any]            =      self.upsamplers_a(_lowercase )

								return hidden_states




class  _lowerCAmelCase     (					nn.Module		):





		"""simple docstring"""






		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							0.0
		_lowerCamelCase      	=							1
		_lowerCamelCase      	=							True
		_lowerCamelCase      	=							jnp.floataa





		def  UpperCAmelCase__				(       self )			->    Any:







								'''simple docstring'''

								snake_case_ :			Union[str, Any]            =      []

								for i in range(self.num_layers ):
														snake_case_ :			int            =      self.in_channels if (i == self.num_layers - 1) else self.out_channels
														snake_case_ :			List[str]            =      self.prev_output_channel if i == 0 else self.out_channels

														snake_case_ :			Optional[int]            =      FlaxResnetBlockaD(
														    in_channels=resnet_in_channels + res_skip_channels				,     out_channels=self.out_channels				,     dropout_prob=self.dropout				,     dtype=self.dtype				,     )
														resnets.append(_lowercase )

								snake_case_ :			Optional[int]            =      resnets

								if self.add_upsample:
														snake_case_ :			str            =      FlaxUpsampleaD(self.out_channels				,     dtype=self.dtype )





		def __call__(       self				,     _lowercase				,     _lowercase				,     _lowercase				,     _lowercase=True )			->    Optional[int]:







								'''simple docstring'''

								for resnet in self.resnets:
														# pop res hidden states
														snake_case_ :			Optional[Any]            =      res_hidden_states_tuple[-1]
														snake_case_ :			List[str]            =      res_hidden_states_tuple[:-1]
														snake_case_ :			str            =      jnp.concatenate((hidden_states, res_hidden_states)				,     axis=-1 )

														snake_case_ :			Optional[int]            =      resnet(_lowercase				,     _lowercase				,     deterministic=_lowercase )

								if self.add_upsample:
														snake_case_ :			int            =      self.upsamplers_a(_lowercase )

								return hidden_states




class  _lowerCAmelCase     (					nn.Module		):





		"""simple docstring"""






		_lowerCamelCase      	=							42
		_lowerCamelCase      	=							0.0
		_lowerCamelCase      	=							1
		_lowerCamelCase      	=							1
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							False
		_lowerCamelCase      	=							jnp.floataa





		def  UpperCAmelCase__				(       self )			->    Tuple:







								'''simple docstring'''

								snake_case_ :			Any            =      [
								    FlaxResnetBlockaD(
								        in_channels=self.in_channels				,     out_channels=self.in_channels				,     dropout_prob=self.dropout				,     dtype=self.dtype				,     )
								]

								snake_case_ :			Tuple            =      []

								for _ in range(self.num_layers ):
														snake_case_ :			List[Any]            =      FlaxTransformeraDModel(
														    in_channels=self.in_channels				,     n_heads=self.num_attention_heads				,     d_head=self.in_channels // self.num_attention_heads				,     depth=1				,     use_linear_projection=self.use_linear_projection				,     use_memory_efficient_attention=self.use_memory_efficient_attention				,     dtype=self.dtype				,     )
														attentions.append(_lowercase )

														snake_case_ :			List[str]            =      FlaxResnetBlockaD(
														    in_channels=self.in_channels				,     out_channels=self.in_channels				,     dropout_prob=self.dropout				,     dtype=self.dtype				,     )
														resnets.append(_lowercase )

								snake_case_ :			Optional[Any]            =      resnets
								snake_case_ :			List[Any]            =      attentions





		def __call__(       self				,     _lowercase				,     _lowercase				,     _lowercase				,     _lowercase=True )			->    Union[str, Any]:







								'''simple docstring'''

								snake_case_ :			Any            =      self.resnets[0](_lowercase				,     _lowercase )
								for attn, resnet in zip(self.attentions				,     self.resnets[1:] ):
														snake_case_ :			Dict            =      attn(_lowercase				,     _lowercase				,     deterministic=_lowercase )
														snake_case_ :			Optional[int]            =      resnet(_lowercase				,     _lowercase				,     deterministic=_lowercase )

								return hidden_states






'''simple docstring'''



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

from ...utils import logging


lowerCAmelCase_  :							int             =    logging.get_logger(__name__)
def        UpperCAmelCase (    A	:					List[Any]       ,		A	:					Optional[int]					):
							SCREAMING_SNAKE_CASE       :   Any         =      nn.functional.normalize(A					)
							SCREAMING_SNAKE_CASE       :   List[str]         =      nn.functional.normalize(A					)
							return torch.mm(A       ,		normalized_text_embeds.t()					)






class 						lowerCamelCase_				(       snake_case_      ):
						_lowerCAmelCase  : Optional[Any]						  =					CLIPConfig

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







						def __init__(	self :   int					,     lowerCAmelCase__ :   CLIPConfig					):

													"""simple docstring"""
													super().__init__(lowerCAmelCase__					)

													SCREAMING_SNAKE_CASE       :   Any         =      CLIPVisionModel(config.vision_config					)
													SCREAMING_SNAKE_CASE       :   List[str]         =      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       :   int         =      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       :   List[str]         =      nn.Parameter(torch.ones(3					)					,     requires_grad=lowerCAmelCase__					)







						@torch.no_grad()
						def        __lowercase     (	self :   List[str]					,     lowerCAmelCase__ :   List[str]					,     lowerCAmelCase__ :   Optional[Any]					):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE       :   Any         =      self.vision_model(lowerCAmelCase__					)[1]  # pooled_output
													SCREAMING_SNAKE_CASE       :   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       :   Union[str, Any]         =      cosine_distance(lowerCAmelCase__					,     self.special_care_embeds					).cpu().float().numpy()
													SCREAMING_SNAKE_CASE       :   List[str]         =      cosine_distance(lowerCAmelCase__					,     self.concept_embeds					).cpu().float().numpy()

													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      []
													SCREAMING_SNAKE_CASE       :   int         =      image_embeds.shape[0]
													for i in range(lowerCAmelCase__					):
																				SCREAMING_SNAKE_CASE       :   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
																				SCREAMING_SNAKE_CASE       :   int         =      0.0

																				for concept_idx in range(len(special_cos_dist[0]					)					):
																											SCREAMING_SNAKE_CASE       :   Optional[int]         =      special_cos_dist[i][concept_idx]
																											SCREAMING_SNAKE_CASE       :   int         =      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       :   Tuple         =      0.01

																				for concept_idx in range(len(cos_dist[0]					)					):
																											SCREAMING_SNAKE_CASE       :   Optional[Any]         =      cos_dist[i][concept_idx]
																											SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      self.concept_embeds_weights[concept_idx].item()
																											SCREAMING_SNAKE_CASE       :   Dict         =      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       :   Dict         =      [len(res['''bad_concepts''']					) > 0 for res in result]

													return images, has_nsfw_concepts







						@torch.no_grad()
						def        __lowercase     (	self :   List[str]					,     lowerCAmelCase__ :   torch.FloatTensor					,     lowerCAmelCase__ :   torch.FloatTensor					):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      self.vision_model(lowerCAmelCase__					)[1]  # pooled_output
													SCREAMING_SNAKE_CASE       :   List[Any]         =      self.visual_projection(lowerCAmelCase__					)

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

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




