




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

from ...utils import logging


A				: Tuple					      =			logging.get_logger(__name__)
def 							snake_case_    (  a__						:      str       ,a__						:      Tuple     ):



    """simple docstring"""




    __lowercase     							=		nn.functional.normalize(lowercase_     )
    __lowercase     							=		nn.functional.normalize(lowercase_     )
    return torch.mm(lowercase_       ,normalized_text_embeds.t()     )






class 			SCREAMING_SNAKE_CASE(							A_	):
       snake_case_					:	Tuple       =   CLIPConfig

       snake_case_					:	Optional[Any]       =   ["""CLIPEncoderLayer"""]

       def __init__(       self   ,						lowerCamelCase__	)       ->   str:

           """simple docstring"""



           super().__init__(snake_case_	)

           __lowercase     							=		CLIPVisionModel(config.vision_config	)
           __lowercase     							=		nn.Linear(config.vision_config.hidden_size   ,						config.projection_dim   ,						bias=snake_case_	)

           __lowercase     							=		nn.Parameter(torch.ones(17   ,						config.projection_dim	)   ,						requires_grad=snake_case_	)
           __lowercase     							=		nn.Parameter(torch.ones(3   ,						config.projection_dim	)   ,						requires_grad=snake_case_	)

           __lowercase     							=		nn.Parameter(torch.ones(17	)   ,						requires_grad=snake_case_	)
           __lowercase     							=		nn.Parameter(torch.ones(3	)   ,						requires_grad=snake_case_	)

       @torch.no_grad()
       def        snake_case__     (       self   ,						lowerCamelCase__   ,						lowerCamelCase__	)       ->   Tuple:

           """simple docstring"""



           __lowercase     							=		self.vision_model(snake_case_	)[1]  # pooled_output
           __lowercase     							=		self.visual_projection(snake_case_	)

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

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

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

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

               result.append(snake_case_	)

           __lowercase     							=		[len(res["""bad_concepts"""]	) > 0 for res in result]

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

           if any(snake_case_	):
               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        snake_case__     (       self   ,						lowerCamelCase__   ,						lowerCamelCase__	)       ->   Dict:

           """simple docstring"""



           __lowercase     							=		self.vision_model(snake_case_	)[1]  # pooled_output
           __lowercase     							=		self.visual_projection(snake_case_	)

           __lowercase     							=		cosine_distance(snake_case_   ,						self.special_care_embeds	)
           __lowercase     							=		cosine_distance(snake_case_   ,						self.concept_embeds	)

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

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

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

           __lowercase     							=		0.0  # black image

           return images, has_nsfw_concepts










"""simple docstring"""



import os

from apache_beam.io.filesystems import FileSystems
from apache_beam.pipeline import Pipeline

from .logging import get_logger


SCREAMING_SNAKE_CASE							=			2 << 20  # 2mb
SCREAMING_SNAKE_CASE							=			get_logger(__name__)




class       UpperCAmelCase_    (		A_				):




    def 				__magic_name__				(       self		:			Tuple ) ->				Optional[int]:




     '''simple docstring'''

     A__			=							self._options.get_all_options().get("runner" )
     return runner in [None, "DirectRunner", "PortableRunner"]







def     _SCREAMING_SNAKE_CASE  (				lowercase_     ,					lowercase_     ,					lowercase_=False		)		->						Dict:
 A__			=							FileSystems
 if fs.exists(lowercase_		):
  if force_upload:
   logger.info(f"""Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True."""		)
  else:
   logger.info(f"""Remote path already exist: {remote_file_path}. Skipping it as force_upload=False."""		)
   return
 with fs.create(lowercase_		) as remote_file:
  with open(lowercase_     ,					"rb"		) as local_file:
   A__			=							local_file.read(lowercase_		)
   while chunk:
    remote_file.write(lowercase_		)
    A__			=							local_file.read(lowercase_		)







def     _SCREAMING_SNAKE_CASE  (				lowercase_     ,					lowercase_     ,					lowercase_=False		)		->						int:
 A__			=							FileSystems
 if os.path.exists(lowercase_		):
  if force_download:
   logger.info(f"""Local path already exist: {remote_file_path}. Overwriting it as force_upload=True."""		)
  else:
   logger.info(f"""Local path already exist: {remote_file_path}. Skipping it as force_upload=False."""		)
   return
 with fs.open(lowercase_		) as remote_file:
  with open(lowercase_     ,					"wb"		) as local_file:
   A__			=							remote_file.read(lowercase_		)
   while chunk:
    local_file.write(lowercase_		)
    A__			=							remote_file.read(lowercase_		)





