



"""simple docstring"""





import os
from typing import Optional

import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.utils import DEFAULT_BLOCK_SIZE



class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       ''''''
		_lowerCamelCase			  =       (
		    None  # protocol passed in prefix to the url. ex: "gzip", for gzip://file.txt::http://foo.bar/file.txt.gz
		)
		_lowerCamelCase			  =       None  # compression type in fsspec. ex: "gzip"
		_lowerCamelCase			  =       None  # extension of the filename to strip. ex: "".gz" to get file.txt from file.txt.gz





		def __init__(					self							,    _lowercase = ""							,    _lowercase = None							,    _lowercase = None							,    **_lowercase   ) ->						Union[str, Any]:







								'''simple docstring'''

								super().__init__(self							,    **_lowercase   )
								# always open as "rb" since fsspec can then use the TextIOWrapper to make it work for "r" mode
								snake_case_   : Dict						=			fsspec.open(
								    _lowercase							,    mode="""rb"""							,    protocol=_lowercase							,    compression=self.compression							,    client_kwargs={
								        """requote_redirect_url""": False,  # see https://github.com/huggingface/datasets/pull/5459
								        """trust_env""": True,  # Enable reading proxy env variables.
								        **(target_options or {}).pop("""client_kwargs"""							,    {}   ),  # To avoid issues if it was already passed.
								    }							,    **(target_options or {})							,    )
								snake_case_   : List[str]						=			os.path.basename(self.file.path.split("""::"""   )[0]   )
								snake_case_   : str						=			(
								    self.compressed_name[: self.compressed_name.rindex("""."""   )]
								    if """.""" in self.compressed_name
								    else self.compressed_name
								)
								snake_case_   : Any						=			None





		@classmethod
		def    UpperCAmelCase__							(					cls							,    _lowercase   ) ->						List[Any]:







								'''simple docstring'''

								return super()._strip_protocol(_lowercase   ).lstrip("""/"""   )





		def    UpperCAmelCase__							(					self   ) ->						Optional[int]:







								'''simple docstring'''

								if self.dir_cache is None:
														snake_case_   : Any						=			{**self.file.fs.info(self.file.path   ), """name""": self.uncompressed_name}
														snake_case_   : Dict						=			{f["""name"""]: f}





		def    UpperCAmelCase__							(					self							,    _lowercase   ) ->						Tuple:







								'''simple docstring'''

								return self.file.open().read()





		def    UpperCAmelCase__							(					self							,    _lowercase							,    _lowercase = "rb"							,    _lowercase=None							,    _lowercase=True							,    _lowercase=None							,    **_lowercase							,    ) ->						Any:







								'''simple docstring'''

								snake_case_   : str						=			self._strip_protocol(_lowercase   )
								if mode != "rb":
														raise ValueError(f'Tried to read with mode {mode} on file {self.file.path} opened with mode \'rb\''   )
								return self.file.open()




class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       '''bz2'''
		_lowerCamelCase			  =       '''bz2'''
		_lowerCamelCase			  =       '''.bz2'''




class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       '''gzip'''
		_lowerCamelCase			  =       '''gzip'''
		_lowerCamelCase			  =       '''.gz'''




class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       '''lz4'''
		_lowerCamelCase			  =       '''lz4'''
		_lowerCamelCase			  =       '''.lz4'''




class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       '''xz'''
		_lowerCamelCase			  =       '''xz'''
		_lowerCamelCase			  =       '''.xz'''




class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       '''zstd'''
		_lowerCamelCase			  =       '''zstd'''
		_lowerCamelCase			  =       '''.zst'''





		def __init__(					self							,    _lowercase							,    _lowercase = "rb"							,    _lowercase = None							,    _lowercase = None							,    _lowercase = DEFAULT_BLOCK_SIZE							,    **_lowercase							,    ) ->						str:







								'''simple docstring'''

								super().__init__(
								    fo=_lowercase							,    mode=_lowercase							,    target_protocol=_lowercase							,    target_options=_lowercase							,    block_size=_lowercase							,    **_lowercase							,    )
								# We need to wrap the zstd decompressor to avoid this error in fsspec==2021.7.0 and zstandard==0.15.2:
								#
								# File "/Users/user/.virtualenvs/hf-datasets/lib/python3.7/site-packages/fsspec/core.py", line 145, in open
								#     out.close = close
								# AttributeError: 'zstd.ZstdDecompressionReader' object attribute 'close' is read-only
								#
								# see https://github.com/intake/filesystem_spec/issues/725
								snake_case_   : Union[str, Any]						=			self.file.__enter__



								class    _lowerCAmelCase    :





										"""simple docstring"""






										def __init__(					self							,    _lowercase   ) ->						Optional[int]:







																'''simple docstring'''

																snake_case_   : Optional[Any]						=			file_





										def __enter__(					self   ) ->						Optional[int]:







																'''simple docstring'''

																self._file.__enter__()
																return self





										def __exit__(					self							,    *_lowercase							,    **_lowercase   ) ->						List[Any]:







																'''simple docstring'''

																self._file.__exit__(*_lowercase							,    **_lowercase   )





										def __iter__(					self   ) ->						Dict:







																'''simple docstring'''

																return iter(self._file   )





										def    UpperCAmelCase__							(					self   ) ->						Tuple:







																'''simple docstring'''

																return next(self._file   )





										def __getattr__(					self							,    _lowercase   ) ->						Any:







																'''simple docstring'''

																return getattr(self._file							,    _lowercase   )

								def fixed_enter(*_lowercase							,    **_lowercase   ):
														return WrappedFile(_enter(*_lowercase							,    **_lowercase   )   )

								snake_case_   : List[str]						=			fixed_enter






"""simple docstring"""





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

from ...utils import logging


__lowerCAmelCase       :					Optional[int]      					=					logging.get_logger(__name__)
def 	__lowerCAmelCase						(		__UpperCamelCase      :			Tuple					, __UpperCamelCase      :			Optional[Any]    ):


						'''simple docstring'''






						snake_case_   : Union[str, Any]						=			nn.functional.normalize(__UpperCamelCase    )
						snake_case_   : Tuple						=			nn.functional.normalize(__UpperCamelCase    )
						return torch.mm(__UpperCamelCase					, normalized_text_embeds.t()    )






class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       CLIPConfig

		_lowerCamelCase			  =       ['''CLIPEncoderLayer''']





		def __init__(					self							,    _lowercase   ) ->						Union[str, Any]:







								'''simple docstring'''

								super().__init__(_lowercase   )

								snake_case_   : Tuple						=			CLIPVisionModel(config.vision_config   )
								snake_case_   : int						=			nn.Linear(config.vision_config.hidden_size							,    config.projection_dim							,    bias=_lowercase   )

								snake_case_   : Optional[Any]						=			nn.Parameter(torch.ones(1_7							,    config.projection_dim   )							,    requires_grad=_lowercase   )
								snake_case_   : Dict						=			nn.Parameter(torch.ones(3							,    config.projection_dim   )							,    requires_grad=_lowercase   )

								snake_case_   : Any						=			nn.Parameter(torch.ones(1_7   )							,    requires_grad=_lowercase   )
								snake_case_   : List[str]						=			nn.Parameter(torch.ones(3   )							,    requires_grad=_lowercase   )





		@torch.no_grad()
		def    UpperCAmelCase__							(					self							,    _lowercase							,    _lowercase   ) ->						Any:







								'''simple docstring'''

								snake_case_   : int						=			self.vision_model(_lowercase   )[1]  # pooled_output
								snake_case_   : str						=			self.visual_projection(_lowercase   )

								# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
								snake_case_   : Dict						=			cosine_distance(_lowercase							,    self.special_care_embeds   ).cpu().float().numpy()
								snake_case_   : List[str]						=			cosine_distance(_lowercase							,    self.concept_embeds   ).cpu().float().numpy()

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

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

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

														result.append(_lowercase   )

								snake_case_   : Union[str, Any]						=			[len(res["""bad_concepts"""]   ) > 0 for res in result]

								return images, has_nsfw_concepts





		@torch.no_grad()
		def    UpperCAmelCase__							(					self							,    _lowercase							,    _lowercase   ) ->						Tuple:







								'''simple docstring'''

								snake_case_   : Optional[Any]						=			self.vision_model(_lowercase   )[1]  # pooled_output
								snake_case_   : List[str]						=			self.visual_projection(_lowercase   )

								snake_case_   : str						=			cosine_distance(_lowercase							,    self.special_care_embeds   )
								snake_case_   : Optional[int]						=			cosine_distance(_lowercase							,    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_   : Tuple						=			0.0

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

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

								return images, has_nsfw_concepts

