


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

from ...utils import logging


__UpperCAmelCase			:			Tuple			      =					logging.get_logger(__name__)


def   a						(				SCREAMING_SNAKE_CASE_   :		List[str]				,      SCREAMING_SNAKE_CASE_   :		List[str]							):






       """simple docstring"""





       UpperCamelCase     :						int 						=		nn.functional.normalize(SCREAMING_SNAKE_CASE_							)
       UpperCamelCase     :						List[str] 						=		nn.functional.normalize(SCREAMING_SNAKE_CASE_							)
       return torch.mm(SCREAMING_SNAKE_CASE_				,      normalized_text_embeds.t()							)



class  UpperCAmelCase_						(				_a):
       '''simple docstring'''
       __UpperCamelCase   :  str				=       CLIPConfig

       __UpperCamelCase   :  Dict				=       ["CLIPEncoderLayer"]






       def __init__(  self			,       __SCREAMING_SNAKE_CASE    ):




              """simple docstring"""




              super().__init__(__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						Dict 						=		CLIPVisionModel(config.vision_config    )
              UpperCamelCase     :						Optional[int] 						=		nn.Linear(config.vision_config.hidden_size			,       config.projection_dim			,       bias=__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						Union[str, Any] 						=		nn.Parameter(torch.ones(17			,       config.projection_dim    )			,       requires_grad=__SCREAMING_SNAKE_CASE    )
              UpperCamelCase     :						Any 						=		nn.Parameter(torch.ones(3			,       config.projection_dim    )			,       requires_grad=__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						Optional[int] 						=		nn.Parameter(torch.ones(17    )			,       requires_grad=__SCREAMING_SNAKE_CASE    )
              UpperCamelCase     :						Any 						=		nn.Parameter(torch.ones(3    )			,       requires_grad=__SCREAMING_SNAKE_CASE    )






       @torch.no_grad()
       def 			_lowercase (  self			,       __SCREAMING_SNAKE_CASE			,       __SCREAMING_SNAKE_CASE    ):




              """simple docstring"""




              UpperCamelCase     :						Optional[int] 						=		self.vision_model(__SCREAMING_SNAKE_CASE    )[1]  # pooled_output
              UpperCamelCase     :						str 						=		self.visual_projection(__SCREAMING_SNAKE_CASE    )

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

              UpperCamelCase     :						List[str] 						=		[]
              UpperCamelCase     :						Any 						=		image_embeds.shape[0]
              for i in range(__SCREAMING_SNAKE_CASE    ):
                     UpperCamelCase     :						str 						=		{'''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     :						Dict 						=		0.0

                     for concept_idx in range(len(special_cos_dist[0]    )    ):
                            UpperCamelCase     :						List[Any] 						=		special_cos_dist[i][concept_idx]
                            UpperCamelCase     :						Dict 						=		self.special_care_embeds_weights[concept_idx].item()
                            UpperCamelCase     :						Optional[int] 						=		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     :						Union[str, Any] 						=		0.01

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

                     result.append(__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						int 						=		[len(res['''bad_concepts''']    ) > 0 for res in result]

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

              if any(__SCREAMING_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 			_lowercase (  self			,       __SCREAMING_SNAKE_CASE			,       __SCREAMING_SNAKE_CASE    ):




              """simple docstring"""




              UpperCamelCase     :						List[str] 						=		self.vision_model(__SCREAMING_SNAKE_CASE    )[1]  # pooled_output
              UpperCamelCase     :						str 						=		self.visual_projection(__SCREAMING_SNAKE_CASE    )

              UpperCamelCase     :						List[str] 						=		cosine_distance(__SCREAMING_SNAKE_CASE			,       self.special_care_embeds    )
              UpperCamelCase     :						Any 						=		cosine_distance(__SCREAMING_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
              UpperCamelCase     :						Dict 						=		0.0

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

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

              UpperCamelCase     :						List[str] 						=		0.0  # black image

              return images, has_nsfw_concepts





import os

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

from .logging import get_logger


__UpperCAmelCase			:			Optional[Any]			      =					2 << 20  # 2mb
__UpperCAmelCase			:			List[Any]			      =					get_logger(__name__)



class  UpperCAmelCase_						(				_a):
       '''simple docstring'''






       def 			_lowercase (  self    ):




              """simple docstring"""




              UpperCamelCase     :						Optional[Any] 						=		self._options.get_all_options().get('''runner'''    )
              return runner in [None, "DirectRunner", "PortableRunner"]




def   a						(				SCREAMING_SNAKE_CASE_   :		str				,      SCREAMING_SNAKE_CASE_   :		int				,      SCREAMING_SNAKE_CASE_   :		Optional[Any]=False							):






       """simple docstring"""





       UpperCamelCase     :						int 						=		FileSystems
       if fs.exists(SCREAMING_SNAKE_CASE_							):
              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(SCREAMING_SNAKE_CASE_							) as remote_file:
              with open(SCREAMING_SNAKE_CASE_				,      '''rb'''							) as local_file:
                     UpperCamelCase     :						List[str] 						=		local_file.read(SCREAMING_SNAKE_CASE_							)
                     while chunk:
                            remote_file.write(SCREAMING_SNAKE_CASE_							)
                            UpperCamelCase     :						Any 						=		local_file.read(SCREAMING_SNAKE_CASE_							)


def   a						(				SCREAMING_SNAKE_CASE_   :		str				,      SCREAMING_SNAKE_CASE_   :		Optional[Any]				,      SCREAMING_SNAKE_CASE_   :		List[str]=False							):






       """simple docstring"""





       UpperCamelCase     :						List[Any] 						=		FileSystems
       if os.path.exists(SCREAMING_SNAKE_CASE_							):
              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(SCREAMING_SNAKE_CASE_							) as remote_file:
              with open(SCREAMING_SNAKE_CASE_				,      '''wb'''							) as local_file:
                     UpperCamelCase     :						List[str] 						=		remote_file.read(SCREAMING_SNAKE_CASE_							)
                     while chunk:
                            local_file.write(SCREAMING_SNAKE_CASE_							)
                            UpperCamelCase     :						str 						=		remote_file.read(SCREAMING_SNAKE_CASE_							)

