






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

from ...utils import logging


A_       :List[Any]       =						logging.get_logger(__name__)






def 							A     (       a_							,a_					)     ->						List[Any]:
					__UpperCamelCase      :					int       		=nn.functional.normalize(a_					)
					__UpperCamelCase      :					List[Any]       		=nn.functional.normalize(a_					)
					return torch.mm(a_							,normalized_text_embeds.t()					)




class 				__A     (			a ):





				"""simple docstring"""







				UpperCamelCase__ :      Dict			 =CLIPConfig

				UpperCamelCase__ :      Union[str, Any]			 =["""CLIPEncoderLayer"""]


				def __init__( self							,  lowerCamelCase__    ):




									"""simple docstring"""

									super().__init__(lowerCamelCase__    )

									__UpperCamelCase      :					List[Any]       		=CLIPVisionModel(config.vision_config    )
									__UpperCamelCase      :					str       		=nn.Linear(config.vision_config.hidden_size							,  config.projection_dim							,  bias=lowerCamelCase__    )

									__UpperCamelCase      :					Dict       		=nn.Parameter(torch.ones(17							,  config.projection_dim    )							,  requires_grad=lowerCamelCase__    )
									__UpperCamelCase      :					Optional[int]       		=nn.Parameter(torch.ones(3							,  config.projection_dim    )							,  requires_grad=lowerCamelCase__    )

									__UpperCamelCase      :					Dict       		=nn.Parameter(torch.ones(17    )							,  requires_grad=lowerCamelCase__    )
									__UpperCamelCase      :					Dict       		=nn.Parameter(torch.ones(3    )							,  requires_grad=lowerCamelCase__    )


				@torch.no_grad()
				def       __lowercase							( self							,  lowerCamelCase__							,  lowerCamelCase__    ):




									"""simple docstring"""

									__UpperCamelCase      :					Optional[int]       		=self.vision_model(lowerCamelCase__    )[1]  # pooled_output
									__UpperCamelCase      :					Any       		=self.visual_projection(lowerCamelCase__    )

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

									__UpperCamelCase      :					Optional[Any]       		=[]
									__UpperCamelCase      :					Tuple       		=image_embeds.shape[0]
									for i in range(lowerCamelCase__    ):
														__UpperCamelCase      :					Dict       		={'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
														__UpperCamelCase      :					int       		=0.0

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

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

														result.append(lowerCamelCase__    )

									__UpperCamelCase      :					Optional[Any]       		=[len(res['bad_concepts']    ) > 0 for res in result]

									return images, has_nsfw_concepts







				@torch.no_grad()
				def       __lowercase							( self							,  lowerCamelCase__							,  lowerCamelCase__    ):




									"""simple docstring"""

									__UpperCamelCase      :					Any       		=self.vision_model(lowerCamelCase__    )[1]  # pooled_output
									__UpperCamelCase      :					List[str]       		=self.visual_projection(lowerCamelCase__    )

									__UpperCamelCase      :					List[Any]       		=cosine_distance(lowerCamelCase__							,  self.special_care_embeds    )
									__UpperCamelCase      :					List[Any]       		=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
									__UpperCamelCase      :					List[Any]       		=0.0

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

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

									return images, has_nsfw_concepts






"""simple docstring"""







import os
from argparse import ArgumentParser
from typing import List

import torch.utils.data

from datasets import Dataset, IterableDataset
from datasets.distributed import split_dataset_by_node


lowerCAmelCase_   						=   4
lowerCAmelCase_   						=   3
class        __A       (	A_    ):







		'''simple docstring'''





		pass





def      __UpperCAmelCase      (    __lowerCamelCase	)							->      Dict:
		for shard in shards:
				for i in range(__lowerCamelCase	):
						yield {"i": i, "shard": shard}



def      __UpperCAmelCase      (    )							->      Tuple:
		lowercase__   :    int				=			int(os.environ['''RANK''']	)
		lowercase__   :    str				=			int(os.environ['''WORLD_SIZE''']	)

		lowercase__   :    List[Any]				=			ArgumentParser()
		parser.add_argument('''--streaming'''						,     type=__lowerCamelCase	)
		parser.add_argument('''--local_rank'''						,     type=__lowerCamelCase	)
		parser.add_argument('''--num_workers'''						,     type=__lowerCamelCase						,     default=0	)
		lowercase__   :    int				=			parser.parse_args()
		lowercase__   :    Optional[Any]				=			args.streaming
		lowercase__   :    List[Any]				=			args.num_workers

		lowercase__   :    Optional[Any]				=			{'''shards''': [f"""shard_{shard_idx}""" for shard_idx in range(__lowerCamelCase	)]}
		lowercase__   :    Dict				=			IterableDataset.from_generator(__lowerCamelCase						,     gen_kwargs=__lowerCamelCase	)
		if not streaming:
				lowercase__   :    int				=			Dataset.from_list(list(__lowerCamelCase	)	)

		lowercase__   :    int				=			split_dataset_by_node(__lowerCamelCase						,     rank=__lowerCamelCase						,     world_size=__lowerCamelCase	)
		lowercase__   :    Optional[Any]				=			torch.utils.data.DataLoader(__lowerCamelCase						,     num_workers=__lowerCamelCase	)

		lowercase__   :    Optional[Any]				=			NUM_SHARDS * NUM_ITEMS_PER_SHARD
		lowercase__   :    str				=			full_size // world_size
		expected_local_size += int(rank < (full_size % world_size)	)

		lowercase__   :    str				=			sum(1 for _ in dataloader	)
		if local_size != expected_local_size:
				raise FailedTestError(f"""local_size {local_size} != expected_local_size {expected_local_size}"""	)


if __name__ == "__main__":
							main()
