



import sys
from collections.abc import Mapping

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter



class 		_lowercase		(							TensorFormatter[Mapping, np.ndarray, Mapping]						):

   def __init__(   self      :							Optional[Any]				, __lowerCAmelCase      :							List[str]=None				, **__lowerCAmelCase      :							List[str]				)							->					Tuple:


    """simple docstring"""






    super().__init__(features=__lowerCAmelCase				)
    a							=		np_array_kwargs

   def 		A				(   self      :							Optional[Any]				, __lowerCAmelCase      :							int				)							->					Optional[Any]:


    """simple docstring"""






    if isinstance(__lowerCAmelCase				, __lowerCAmelCase				):
     if column and all(
         isinstance(__lowerCAmelCase				, np.ndarray				) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column				):
      return np.stack(__lowerCAmelCase				)
     else:
      # don't use np.array(column, dtype=object)
      # since it fails in certain cases
      # see https://stackoverflow.com/q/51005699
      a							=		np.empty(len(__lowerCAmelCase				)				, dtype=__lowerCAmelCase				)
      a							=		column
      return out
    return column

   def 		A				(   self      :							Optional[int]				, __lowerCAmelCase      :							Dict				)							->					List[Any]:


    """simple docstring"""






    if isinstance(__lowerCAmelCase				, (str, bytes, type(__lowerCAmelCase				))				):
     return value
    elif isinstance(__lowerCAmelCase				, (np.character, np.ndarray)				) and np.issubdtype(value.dtype				, np.character				):
     return value
    elif isinstance(__lowerCAmelCase				, np.number				):
     return value

    a							=		{}

    if isinstance(__lowerCAmelCase				, np.ndarray				) and np.issubdtype(value.dtype				, np.integer				):
     a							=		{"dtype": np.intaa}
    elif isinstance(__lowerCAmelCase				, np.ndarray				) and np.issubdtype(value.dtype				, np.floating				):
     a							=		{"dtype": np.floataa}
    elif config.PIL_AVAILABLE and "PIL" in sys.modules:
     import PIL.Image

     if isinstance(__lowerCAmelCase				, PIL.Image.Image				):
      return np.asarray(__lowerCAmelCase				, **self.np_array_kwargs				)

    return np.asarray(__lowerCAmelCase				, **{**default_dtype, **self.np_array_kwargs}				)

   def 		A				(   self      :							Optional[int]				, __lowerCAmelCase      :							Union[str, Any]				)							->					Any:


    """simple docstring"""






    if config.TORCH_AVAILABLE and "torch" in sys.modules:
     import torch

     if isinstance(__lowerCAmelCase				, torch.Tensor				):
      return self._tensorize(data_struct.detach().cpu().numpy()[()]				)
    if hasattr(__lowerCAmelCase				, "__array__"				) and not isinstance(__lowerCAmelCase				, (np.ndarray, np.character, np.number)				):
     a							=		data_struct.__array__()
    # support for nested types like struct of list of struct
    if isinstance(__lowerCAmelCase				, np.ndarray				):
     if data_struct.dtype == object:
      return self._consolidate([self.recursive_tensorize(__lowerCAmelCase				) for substruct in data_struct]				)
    if isinstance(__lowerCAmelCase				, (list, tuple)				):
     return self._consolidate([self.recursive_tensorize(__lowerCAmelCase				) for substruct in data_struct]				)
    return self._tensorize(__lowerCAmelCase				)

   def 		A				(   self      :							Any				, __lowerCAmelCase      :							dict				)							->					Optional[int]:


    """simple docstring"""






    return map_nested(self._recursive_tensorize				, __lowerCAmelCase				, map_list=__lowerCAmelCase				)

   def 		A				(   self      :							Dict				, __lowerCAmelCase      :							pa.Table				)							->					Mapping:


    """simple docstring"""






    a							=		self.numpy_arrow_extractor().extract_row(__lowerCAmelCase				)
    a							=		self.python_features_decoder.decode_row(__lowerCAmelCase				)
    return self.recursive_tensorize(__lowerCAmelCase				)

   def 		A				(   self      :							Tuple				, __lowerCAmelCase      :							pa.Table				)							->					np.ndarray:


    """simple docstring"""






    a							=		self.numpy_arrow_extractor().extract_column(__lowerCAmelCase				)
    a							=		self.python_features_decoder.decode_column(__lowerCAmelCase				, pa_table.column_names[0]				)
    a							=		self.recursive_tensorize(__lowerCAmelCase				)
    a							=		self._consolidate(__lowerCAmelCase				)
    return column




   def 		A				(   self      :							Tuple				, __lowerCAmelCase      :							pa.Table				)							->					Mapping:


    """simple docstring"""






    a							=		self.numpy_arrow_extractor().extract_batch(__lowerCAmelCase				)
    a							=		self.python_features_decoder.decode_batch(__lowerCAmelCase				)
    a							=		self.recursive_tensorize(__lowerCAmelCase				)
    for column_name in batch:
     a							=		self._consolidate(batch[column_name]				)
    return batch





"""simple docstring"""







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

from ...utils import logging


lowerCamelCase_  :  Tuple     		=    logging.get_logger(__name__)
def _A		(		lowercase			,       lowercase						):







    """simple docstring"""
    a											=nn.functional.normalize(lowercase						)
    a											=nn.functional.normalize(lowercase						)
    return torch.mm(lowercase			,       normalized_text_embeds.t()						)






class  __A							(			_SCREAMING_SNAKE_CASE     ):






 """simple docstring"""




 __lowerCAmelCase					   =    CLIPConfig

 __lowerCAmelCase					   =    ["CLIPEncoderLayer"]


 def __init__(						self					,       __A		)   ->	int:
     super().__init__(__A		)

     a											=CLIPVisionModel(config.vision_config		)
     a											=nn.Linear(config.vision_config.hidden_size					,       config.projection_dim					,       bias=__A		)

     a											=nn.Parameter(torch.ones(17					,       config.projection_dim		)					,       requires_grad=__A		)
     a											=nn.Parameter(torch.ones(3					,       config.projection_dim		)					,       requires_grad=__A		)

     a											=nn.Parameter(torch.ones(17		)					,       requires_grad=__A		)
     a											=nn.Parameter(torch.ones(3		)					,       requires_grad=__A		)


 @torch.no_grad()
 def  SCREAMING_SNAKE_CASE				(						self					,       __A					,       __A		)   ->	Union[str, Any]:
     a											=self.vision_model(__A		)[1]  # pooled_output
     a											=self.visual_projection(__A		)

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

     a											=[]
     a											=image_embeds.shape[0]
     for i in range(__A		):
         a											={'''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
         a											=0.0

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

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

         result.append(__A		)

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

     for idx, has_nsfw_concept in enumerate(__A		):
         if has_nsfw_concept:
             if torch.is_tensor(__A		) or torch.is_tensor(images[0]		):
                 a											=torch.zeros_like(images[idx]		)  # black image
             else:
                 a											=np.zeros(images[idx].shape		)  # black image

     if any(__A		):
         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  SCREAMING_SNAKE_CASE				(						self					,       __A					,       __A		)   ->	str:
     a											=self.vision_model(__A		)[1]  # pooled_output
     a											=self.visual_projection(__A		)

     a											=cosine_distance(__A					,       self.special_care_embeds		)
     a											=cosine_distance(__A					,       self.concept_embeds		)

     # increase this value to create a stronger `nsfw` filter
     # at the cost of increasing the possibility of filtering benign images
     a											=0.0

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

     a											=(cos_dist - self.concept_embeds_weights) + special_adjustment
     # concept_scores = concept_scores.round(decimals=3)
     a											=torch.any(concept_scores > 0					,       dim=1		)

     a											=0.0  # black image

     return images, has_nsfw_concepts