
"""simple docstring"""



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

from ...utils import logging


UpperCAmelCase__          = logging.get_logger(__name__)
def 					_UpperCAmelCase      ( __lowerCamelCase   :       Dict ,					__lowerCamelCase   :       Optional[Any]		)     ->					List[Any]:
  _snake_case					     =       nn.functional.normalize(UpperCamelCase_		)
  _snake_case					     =       nn.functional.normalize(UpperCamelCase_		)
  return torch.mm(UpperCamelCase_ ,					normalized_text_embeds.t()		)

class       lowerCAmelCase__				(						_UpperCAmelCase	):
     __a						=					CLIPConfig

     __a						=					["""CLIPEncoderLayer"""]







     def __init__(							self			:			Optional[Any]    ,      _lowerCamelCase			:			CLIPConfig			):
       super().__init__(UpperCAmelCase_			)

       _snake_case					     =       CLIPVisionModel(config.vision_config			)
       _snake_case					     =       nn.Linear(config.vision_config.hidden_size    ,      config.projection_dim    ,      bias=UpperCAmelCase_			)

       _snake_case					     =       nn.Parameter(torch.ones(17    ,      config.projection_dim			)    ,      requires_grad=UpperCAmelCase_			)
       _snake_case					     =       nn.Parameter(torch.ones(3    ,      config.projection_dim			)    ,      requires_grad=UpperCAmelCase_			)

       _snake_case					     =       nn.Parameter(torch.ones(17			)    ,      requires_grad=UpperCAmelCase_			)
       _snake_case					     =       nn.Parameter(torch.ones(3			)    ,      requires_grad=UpperCAmelCase_			)







     @torch.no_grad()
     def      lowercase	(							self			:			str    ,      _lowerCamelCase			:			Optional[Any]    ,      _lowerCamelCase			:			str			):
       _snake_case					     =       self.vision_model(UpperCAmelCase_			)[1]  # pooled_output
       _snake_case					     =       self.visual_projection(UpperCAmelCase_			)

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

       _snake_case					     =       []
       _snake_case					     =       image_embeds.shape[0]
       for i in range(UpperCAmelCase_			):
         _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(UpperCAmelCase_			)

         result.append(UpperCAmelCase_			)

       _snake_case					     =       [len(res['''bad_concepts''']			) > 0 for res in result]

       return images, has_nsfw_concepts







     @torch.no_grad()
     def      lowercase	(							self			:			Tuple    ,      _lowerCamelCase			:			torch.FloatTensor    ,      _lowerCamelCase			:			torch.FloatTensor			):
       _snake_case					     =       self.vision_model(UpperCAmelCase_			)[1]  # pooled_output
       _snake_case					     =       self.visual_projection(UpperCAmelCase_			)

       _snake_case					     =       cosine_distance(UpperCAmelCase_    ,      self.special_care_embeds			)
       _snake_case					     =       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
       _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			)

       return images, has_nsfw_concepts





from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


__snake_case								=  logging.get_logger(__name__)


__snake_case								=  TypeVar("""DatasetType""", Dataset, IterableDataset)






def 							_lowercase							(     UpperCamelCase_	,	UpperCamelCase_ = None	,	UpperCamelCase_ = None	,	UpperCamelCase_ = None	,	UpperCamelCase_ = None	,	UpperCamelCase_ = "first_exhausted"	,	)		->		DatasetType:
					'''simple docstring'''


					from .arrow_dataset import Dataset
					from .iterable_dataset import IterableDataset

					if not datasets:
										raise ValueError('Unable to interleave an empty list of datasets.'		)
					for i, dataset in enumerate(UpperCamelCase_		):
										if not isinstance(UpperCamelCase_	,	(Dataset, IterableDataset)		):
															if isinstance(UpperCamelCase_	,	(DatasetDict, IterableDatasetDict)		):
																				if not dataset:
																									raise ValueError(
																									    F'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '
																									    'is an empty dataset dictionary.'		)
																				raise ValueError(
																				    F'Dataset at position {i} has at least one split: {list(UpperCamelCase_		)}\n'
																				    F'Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(UpperCamelCase_		)		)}\']'		)
															raise ValueError(
															    F'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(UpperCamelCase_		).__name__}.'		)
										if i == 0:
															SCREAMING_SNAKE_CASE__						, SCREAMING_SNAKE_CASE__													= (
															    (Dataset, IterableDataset) if isinstance(UpperCamelCase_	,	UpperCamelCase_		) else (IterableDataset, Dataset)
															)
										elif not isinstance(UpperCamelCase_	,	UpperCamelCase_		):
															raise ValueError(
															    F'Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'		)
					if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
										raise ValueError(F'{stopping_strategy} is not supported. Please enter a valid stopping_strategy.'		)
					if dataset_type is Dataset:
										return _interleave_map_style_datasets(
										    UpperCamelCase_	,	UpperCamelCase_	,	UpperCamelCase_	,	info=UpperCamelCase_	,	split=UpperCamelCase_	,	stopping_strategy=UpperCamelCase_		)
					else:
										return _interleave_iterable_datasets(
										    UpperCamelCase_	,	UpperCamelCase_	,	UpperCamelCase_	,	info=UpperCamelCase_	,	split=UpperCamelCase_	,	stopping_strategy=UpperCamelCase_		)






def 							_lowercase							(     UpperCamelCase_	,	UpperCamelCase_ = None	,	UpperCamelCase_ = None	,	UpperCamelCase_ = 0	,	)		->		DatasetType:
					'''simple docstring'''


					if not dsets:
										raise ValueError('Unable to concatenate an empty list of datasets.'		)
					for i, dataset in enumerate(UpperCamelCase_		):
										if not isinstance(UpperCamelCase_	,	(Dataset, IterableDataset)		):
															if isinstance(UpperCamelCase_	,	(DatasetDict, IterableDatasetDict)		):
																				if not dataset:
																									raise ValueError(
																									    F'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '
																									    'is an empty dataset dictionary.'		)
																				raise ValueError(
																				    F'Dataset at position {i} has at least one split: {list(UpperCamelCase_		)}\n'
																				    F'Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(UpperCamelCase_		)		)}\']'		)
															raise ValueError(
															    F'Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(UpperCamelCase_		).__name__}.'		)
										if i == 0:
															SCREAMING_SNAKE_CASE__						, SCREAMING_SNAKE_CASE__													= (
															    (Dataset, IterableDataset) if isinstance(UpperCamelCase_	,	UpperCamelCase_		) else (IterableDataset, Dataset)
															)
										elif not isinstance(UpperCamelCase_	,	UpperCamelCase_		):
															raise ValueError(
															    F'Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'		)
					if dataset_type is Dataset:
										return _concatenate_map_style_datasets(UpperCamelCase_	,	info=UpperCamelCase_	,	split=UpperCamelCase_	,	axis=UpperCamelCase_		)
					else:
										return _concatenate_iterable_datasets(UpperCamelCase_	,	info=UpperCamelCase_	,	split=UpperCamelCase_	,	axis=UpperCamelCase_		)





