





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

from ...utils import logging


snake_case_           =      logging.get_logger(__name__)






def lowerCamelCase__   (							snake_case_     :  Any	,				snake_case_     :  str	)				->    Union[str, Any]:
			__snake_case								=       nn.functional.normalize(__UpperCAmelCase	)
			__snake_case								=       nn.functional.normalize(__UpperCAmelCase	)
			return torch.mm(__UpperCAmelCase	,				normalized_text_embeds.t()	)







class 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase  ):
						A_						:    int							=   CLIPConfig

						A_						:    Dict							=   ["CLIPEncoderLayer"]


						def __init__(self	:				Any    ,      a__	:				int				):





									"""simple docstring"""



									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        a			(self	:				Dict    ,      a__	:				Union[str, Any]    ,      a__	:				Dict				):





									"""simple docstring"""



									__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        a			(self	:				Optional[Any]    ,      a__	:				Union[str, Any]    ,      a__	:				Any				):





									"""simple docstring"""



									__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, Union

import numpy as np

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
    from PIL import Image

    from ..image_utils import load_image

if is_torch_available():
    import torch

    from ..models.auto.modeling_auto import MODEL_FOR_DEPTH_ESTIMATION_MAPPING

snake_case_           =      logging.get_logger(__name__)

@add_end_docstrings(_UpperCAmelCase  )
class 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase  ):


      def __init__(self	:				int    ,      *a__	:				List[Any]    ,      **a__	:				Dict				):





         """simple docstring"""



         super().__init__(*a__    ,      **a__				)
         requires_backends(self    ,      '''vision'''				)
         self.check_model_type(a__				)


      def __call__(self	:				Optional[Any]    ,      a__	:				Union[str, List[str], "Image.Image", List["Image.Image"]]    ,      **a__	:				List[str]				):





         """simple docstring"""



         return super().__call__(a__    ,      **a__				)


      def        a			(self	:				int    ,      **a__	:				int				):





         """simple docstring"""



         return {}, {}, {}


      def        a			(self	:				Optional[int]    ,      a__	:				Optional[int]				):





         """simple docstring"""



         __snake_case								=       load_image(a__				)
         __snake_case								=       image.size
         __snake_case								=       self.image_processor(images=a__    ,      return_tensors=self.framework				)
         return model_inputs


      def        a			(self	:				List[Any]    ,      a__	:				Union[str, Any]				):





         """simple docstring"""



         __snake_case								=       self.model(**a__				)
         return model_outputs


      def        a			(self	:				int    ,      a__	:				str				):





         """simple docstring"""



         __snake_case								=       model_outputs.predicted_depth
         __snake_case								=       torch.nn.functional.interpolate(
             predicted_depth.unsqueeze(1				)    ,      size=self.image_size[::-1]    ,      mode='''bicubic'''    ,      align_corners=a__				)
         __snake_case								=       prediction.squeeze().cpu().numpy()
         __snake_case								=       (output * 255 / np.max(a__				)).astype('''uint8'''				)
         __snake_case								=       Image.fromarray(a__				)
         __snake_case								=       {}
         __snake_case								=       predicted_depth
         __snake_case								=       depth
         return output_dict

