






"""simple docstring"""



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

from ...utils import logging


_lowerCAmelCase     :     Union[str, Any]            =	logging.get_logger(__name__)


def      SCREAMING_SNAKE_CASE__    (   snake_case    :      List[str]       , snake_case    :      List[Any]							)->						Optional[int]:






     '''simple docstring'''



     UpperCAmelCase__		:  str				=				nn.functional.normalize(UpperCAmelCase__							)
     UpperCAmelCase__		:  Optional[Any]				=				nn.functional.normalize(UpperCAmelCase__							)
     return torch.mm(UpperCAmelCase__       , normalized_text_embeds.t()							)




class        lowerCAmelCase__       (  __magic_name__					):
   SCREAMING_SNAKE_CASE_					      =CLIPConfig

   SCREAMING_SNAKE_CASE_					      =["""CLIPEncoderLayer"""]
   def __init__(						self	:     Union[str, Any]				,       snake_case__	:     CLIPConfig				):


        '''simple docstring'''
        super().__init__(_UpperCamelCase				)

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

        UpperCAmelCase__		:  int				=				nn.Parameter(torch.ones(1_7				,       config.projection_dim				)				,       requires_grad=_UpperCamelCase				)
        UpperCAmelCase__		:  Tuple				=				nn.Parameter(torch.ones(3				,       config.projection_dim				)				,       requires_grad=_UpperCamelCase				)

        UpperCAmelCase__		:  Any				=				nn.Parameter(torch.ones(1_7				)				,       requires_grad=_UpperCamelCase				)
        UpperCAmelCase__		:  Dict				=				nn.Parameter(torch.ones(3				)				,       requires_grad=_UpperCamelCase				)
   @torch.no_grad()
   def 	__a				(						self	:     Optional[int]				,       snake_case__	:     List[Any]				,       snake_case__	:     Optional[Any]				):


        '''simple docstring'''
        UpperCAmelCase__		:  List[str]				=				self.vision_model(_UpperCamelCase				)[1]  # pooled_output
        UpperCAmelCase__		:  List[str]				=				self.visual_projection(_UpperCamelCase				)

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

        UpperCAmelCase__		:  Tuple				=				[]
        UpperCAmelCase__		:  List[str]				=				image_embeds.shape[0]
        for i in range(_UpperCamelCase				):
             UpperCAmelCase__		:  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
             UpperCAmelCase__		:  Tuple				=				0.0

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

             for concept_idx in range(len(cos_dist[0]				)				):
                  UpperCAmelCase__		:  Optional[Any]				=				cos_dist[i][concept_idx]
                  UpperCAmelCase__		:  Tuple				=				self.concept_embeds_weights[concept_idx].item()
                  UpperCAmelCase__		:  List[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				)

        UpperCAmelCase__		:  Any				=				[len(res["bad_concepts"]				) > 0 for res in result]

        for idx, has_nsfw_concept in enumerate(_UpperCamelCase				):
             if has_nsfw_concept:
                  if torch.is_tensor(_UpperCamelCase				) or torch.is_tensor(images[0]				):
                       UpperCAmelCase__		:  Dict				=				torch.zeros_like(images[idx]				)  # black image
                  else:
                       UpperCAmelCase__		:  Optional[int]				=				np.zeros(images[idx].shape				)  # black image

        if any(_UpperCamelCase				):
             logger.warning(
                 "Potential NSFW content was detected in one or more images. A black image will be returned instead."
                 " Try again with a different prompt and/or seed."				)

        return images, has_nsfw_concepts




   @torch.no_grad()
   def 	__a				(						self	:     List[Any]				,       snake_case__	:     torch.FloatTensor				,       snake_case__	:     torch.FloatTensor				):


        '''simple docstring'''
        UpperCAmelCase__		:  List[str]				=				self.vision_model(_UpperCamelCase				)[1]  # pooled_output
        UpperCAmelCase__		:  str				=				self.visual_projection(_UpperCamelCase				)

        UpperCAmelCase__		:  Tuple				=				cosine_distance(_UpperCamelCase				,       self.special_care_embeds				)
        UpperCAmelCase__		:  Tuple				=				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
        UpperCAmelCase__		:  str				=				0.0

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

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

        UpperCAmelCase__		:  Any				=				0.0  # black image

        return images, has_nsfw_concepts



import json
import os
from dataclasses import dataclass

import numpy as np
import pyarrow as pa

import datasets
from utils import get_duration


_lowerCamelCase    :      int	 =						10_00_00_00_00_00
_lowerCamelCase    :      Union[str, Any]	 =						1_00_00

_lowerCamelCase ,  _lowerCamelCase    :      List[str]	 =						os.path.split(__file__)
_lowerCamelCase    :      Any	 =						os.path.join(RESULTS_BASEPATH, '''results''', RESULTS_FILENAME.replace('''.py''', '''.json'''))





def 		__lowerCamelCase							(UpperCAmelCase__   :							int					,       UpperCAmelCase__   :							int     ):
					SCREAMING_SNAKE_CASE				=				pa.Table.from_pydict({"col": [0] * chunk_size}     )
					SCREAMING_SNAKE_CASE				=				pa.concat_tables([table] * (num_examples // chunk_size)     )
					return datasets.Dataset(UpperCAmelCase__					,       fingerprint="table_100B"     )


@dataclass
class   lowercase			:
							lowercase__  :				int
							lowercase__  :				int
							lowercase__  :				int
							lowercase__  :				int




							def 		__snake_case(  self      :    Tuple					)    -> Dict:



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				np.random.default_rng(self.seed					)
												SCREAMING_SNAKE_CASE				=				rng.integers(low=self.low		,							high=self.high		,							size=self.size					).tolist()




							def __iter__(  self      :    Union[str, Any]					)    -> Dict:



												'''simple docstring'''

												return iter(self._sampled_values					)





							def __len__(  self      :    Any					)    -> int:



												'''simple docstring'''

												return self.size





@get_duration
def 		__lowerCamelCase							(UpperCAmelCase__   :							datasets.Dataset     ):
					SCREAMING_SNAKE_CASE				=				dataset[0]





@get_duration
def 		__lowerCamelCase							(UpperCAmelCase__   :							datasets.Dataset     ):
					SCREAMING_SNAKE_CASE				=				dataset[-1]





@get_duration
def 		__lowerCamelCase							(UpperCAmelCase__   :							datasets.Dataset     ):
					SCREAMING_SNAKE_CASE				=				dataset[range(len(UpperCAmelCase__     ) // 2					,       len(UpperCAmelCase__     ) // 2 + 1_0_2_4     )]





@get_duration
def 		__lowerCamelCase							(UpperCAmelCase__   :							datasets.Dataset     ):
					SCREAMING_SNAKE_CASE				=				dataset[RandIter(0					,       len(UpperCAmelCase__     )					,       1_0_2_4					,       seed=4_2     )]





def 		__lowerCamelCase							():
					SCREAMING_SNAKE_CASE				=				{"num examples": SPEED_TEST_N_EXAMPLES}
					SCREAMING_SNAKE_CASE				=				(get_first_row, get_last_row, get_batch_of_aaaa_rows, get_batch_of_aaaa_random_rows)
					print("generating dataset"     )
					SCREAMING_SNAKE_CASE				=				generate_100B_dataset(num_examples=UpperCAmelCase__					,       chunk_size=UpperCAmelCase__     )
					print("Functions"     )
					for func in functions:
										print(func.__name__     )
										SCREAMING_SNAKE_CASE				=				func(UpperCAmelCase__     )

					with open(UpperCAmelCase__					,       "wb"     ) as f:
										f.write(json.dumps(UpperCAmelCase__     ).encode("utf-8"     )     )


if __name__ == "__main__":  # useful to run the profiler
							benchmark_table_100B()




