


import os
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ... import AutoTokenizer
from ...utils import logging


snake_case__									= 'block_records.npy'


snake_case__									= logging.get_logger(__name__)
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :str      ,						lowercase__  :int				)       ->  str:
						import tensorflow.compat.va as tf

						__magic_name__				:							Union[str, Any]				=      tf.data.TFRecordDataset(lowercase__      ,						buffer_size=5_1_2 * 1_0_2_4 * 1_0_2_4				)
						__magic_name__				:							Any				=      blocks_dataset.batch(lowercase__      ,						drop_remainder=lowercase__				)
						__magic_name__				:							Optional[int]				=      next(blocks_dataset.take(1				).as_numpy_iterator()				)

						return np_record



class     snake_case__  :
				def __init__(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase     ,      _UpperCAmelCase=2     ,      _UpperCAmelCase=1000     ,      _UpperCAmelCase=100     ,      _UpperCAmelCase=10_0000     ,      ):




										"""simple docstring"""




										from scann.scann_ops.py.scann_ops_pybind import builder as Builder

										__magic_name__				:							Any				=      Builder(db=_UpperCAmelCase     ,      num_neighbors=_UpperCAmelCase     ,      distance_measure='dot_product'      )
										__magic_name__				:							Any				=      builder.tree(
										    num_leaves=_UpperCAmelCase     ,      num_leaves_to_search=_UpperCAmelCase     ,      training_sample_size=_UpperCAmelCase      )
										__magic_name__				:							Dict				=      builder.score_ah(dimensions_per_block=_UpperCAmelCase      )

										__magic_name__				:							Union[str, Any]				=      builder.build()





				def  UpperCamelCase_						(  self     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										__magic_name__				:							int				=      self.searcher.search_batched(question_projection.detach().cpu()      )
										return retrieved_block_ids.astype('int64'      )






class     snake_case__  :
				def __init__(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										super().__init__()
										__magic_name__				:							Any				=      block_records
										__magic_name__				:							str				=      tokenizer
				def __call__(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase     ,      _UpperCAmelCase     ,      _UpperCAmelCase=None     ,      _UpperCAmelCase="pt"      ):




										"""simple docstring"""




										__magic_name__				:							Optional[int]				=      np.take(self.block_records     ,      indices=_UpperCAmelCase     ,      axis=0      )

										__magic_name__				:							Dict				=      self.tokenizer.decode(question_input_ids[0]     ,      skip_special_tokens=_UpperCAmelCase      )

										__magic_name__				:							List[str]				=      []
										__magic_name__				:							Optional[Any]				=      []
										for retrieved_block in retrieved_blocks:
																text.append(_UpperCAmelCase      )
																text_pair.append(retrieved_block.decode()      )

										__magic_name__				:							Union[str, Any]				=      self.tokenizer(
										    _UpperCAmelCase     ,      _UpperCAmelCase     ,      padding=_UpperCAmelCase     ,      truncation=_UpperCAmelCase     ,      return_special_tokens_mask=_UpperCAmelCase     ,      max_length=_UpperCAmelCase      )
										__magic_name__				:							Optional[int]				=      concat_inputs.convert_to_tensors(_UpperCAmelCase      )

										if answer_ids is not None:
																return self.block_has_answer(_UpperCAmelCase     ,      _UpperCAmelCase      ) + (concat_inputs_tensors,)
										else:
																return (None, None, None, concat_inputs_tensors)
				@classmethod
				def  UpperCamelCase_						(  cls     ,      _UpperCAmelCase     ,      *_UpperCAmelCase     ,      **_UpperCAmelCase      ):




										"""simple docstring"""




										if os.path.isdir(_UpperCAmelCase      ):
																__magic_name__				:							Optional[Any]				=      os.path.join(_UpperCAmelCase     ,      _REALM_BLOCK_RECORDS_FILENAME      )
										else:
																__magic_name__				:							str				=      hf_hub_download(
																    repo_id=_UpperCAmelCase     ,      filename=_REALM_BLOCK_RECORDS_FILENAME     ,      **_UpperCAmelCase      )
										__magic_name__				:							Optional[int]				=      np.load(_UpperCAmelCase     ,      allow_pickle=_UpperCAmelCase      )

										__magic_name__				:							Union[str, Any]				=      AutoTokenizer.from_pretrained(_UpperCAmelCase     ,      *_UpperCAmelCase     ,      **_UpperCAmelCase      )

										return cls(_UpperCAmelCase     ,      _UpperCAmelCase      )
				def  UpperCamelCase_						(  self     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										np.save(os.path.join(_UpperCAmelCase     ,      _REALM_BLOCK_RECORDS_FILENAME      )     ,      self.block_records      )
										# save tokenizer
										self.tokenizer.save_pretrained(_UpperCAmelCase      )





				def  UpperCamelCase_						(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										__magic_name__				:							Any				=      []
										__magic_name__				:							str				=      []
										__magic_name__				:							Any				=      []
										__magic_name__				:							List[Any]				=      0

										for input_id in concat_inputs.input_ids:
																__magic_name__				:							Union[str, Any]				=      input_id.tolist()
																# Check answers between two [SEP] tokens
																__magic_name__				:							Union[str, Any]				=      input_id_list.index(self.tokenizer.sep_token_id      )
																__magic_name__				:							Optional[int]				=      first_sep_idx + 1 + input_id_list[first_sep_idx + 1 :].index(self.tokenizer.sep_token_id      )

																start_pos.append([]      )
																end_pos.append([]      )
																for answer in answer_ids:
																						for idx in range(first_sep_idx + 1     ,      _UpperCAmelCase      ):
																												if answer[0] == input_id_list[idx]:
																																		if input_id_list[idx : idx + len(_UpperCAmelCase      )] == answer:
																																								start_pos[-1].append(_UpperCAmelCase      )
																																								end_pos[-1].append(idx + len(_UpperCAmelCase      ) - 1      )

																if len(start_pos[-1]      ) == 0:
																						has_answers.append(_UpperCAmelCase      )
																else:
																						has_answers.append(_UpperCAmelCase      )
																						if len(start_pos[-1]      ) > max_answers:
																												__magic_name__				:							List[str]				=      len(start_pos[-1]      )

        # Pad -1 to max_answers
										for start_pos_, end_pos_ in zip(_UpperCAmelCase     ,      _UpperCAmelCase      ):
																if len(start_pos_      ) < max_answers:
																						__magic_name__				:							List[str]				=      [-1] * (max_answers - len(start_pos_      ))
																						start_pos_ += padded
																						end_pos_ += padded
										return has_answers, start_pos, end_pos






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

from ...utils import logging


snake_case__     							= logging.get_logger(__name__)
def  __SCREAMING_SNAKE_CASE       (				lowercase__  :Optional[Any]      ,						lowercase__  :List[str]				)       ->  Tuple:
						__magic_name__				:							List[str]				=      nn.functional.normalize(lowercase__				)
						__magic_name__				:							Union[str, Any]				=      nn.functional.normalize(lowercase__				)
						return torch.mm(lowercase__      ,						normalized_text_embeds.t()				)



class     snake_case__  (      _A  ):
				__lowerCAmelCase		=      CLIPConfig

				__lowerCAmelCase		=      ['''CLIPEncoderLayer''']
				def __init__(  self     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										super().__init__(_UpperCAmelCase      )

										__magic_name__				:							List[Any]				=      CLIPVisionModel(config.vision_config      )
										__magic_name__				:							Tuple				=      nn.Linear(config.vision_config.hidden_size     ,      config.projection_dim     ,      bias=_UpperCAmelCase      )

										__magic_name__				:							str				=      nn.Parameter(torch.ones(17     ,      config.projection_dim      )     ,      requires_grad=_UpperCAmelCase      )
										__magic_name__				:							Optional[Any]				=      nn.Parameter(torch.ones(3     ,      config.projection_dim      )     ,      requires_grad=_UpperCAmelCase      )

										__magic_name__				:							Optional[Any]				=      nn.Parameter(torch.ones(17      )     ,      requires_grad=_UpperCAmelCase      )
										__magic_name__				:							str				=      nn.Parameter(torch.ones(3      )     ,      requires_grad=_UpperCAmelCase      )
				@torch.no_grad()
				def  UpperCamelCase_						(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										__magic_name__				:							Union[str, Any]				=      self.vision_model(_UpperCAmelCase      )[1]  # pooled_output
										__magic_name__				:							Any				=      self.visual_projection(_UpperCAmelCase      )

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

										__magic_name__				:							str				=      []
										__magic_name__				:							List[Any]				=      image_embeds.shape[0]
										for i in range(_UpperCAmelCase      ):
																__magic_name__				:							Optional[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
																__magic_name__				:							Optional[int]				=      0.0

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

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

																result.append(_UpperCAmelCase      )

										__magic_name__				:							Union[str, 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]      ):
																												__magic_name__				:							Union[str, Any]				=      torch.zeros_like(images[idx]      )  # black image
																						else:
																												__magic_name__				:							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  UpperCamelCase_						(  self     ,      _UpperCAmelCase     ,      _UpperCAmelCase      ):




										"""simple docstring"""




										__magic_name__				:							List[str]				=      self.vision_model(_UpperCAmelCase      )[1]  # pooled_output
										__magic_name__				:							List[str]				=      self.visual_projection(_UpperCAmelCase      )

										__magic_name__				:							Union[str, Any]				=      cosine_distance(_UpperCAmelCase     ,      self.special_care_embeds      )
										__magic_name__				:							Optional[int]				=      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
										__magic_name__				:							int				=      0.0

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

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

										__magic_name__				:							Dict				=      0.0  # black image

										return images, has_nsfw_concepts


