
'''simple docstring'''



import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


UpperCamelCase__   :		Optional[Any]											=	get_tests_dir("fixtures")

class  _a					(unittest.TestCase):





       """simple docstring"""




       def 							UpperCamelCase					(				self   )     ->      Optional[int]:
          _SCREAMING_SNAKE_CASE  			=					mock.Mock()
          _SCREAMING_SNAKE_CASE  			=					5_00
          _SCREAMING_SNAKE_CASE  			=					{}
          _SCREAMING_SNAKE_CASE  			=					HTTPError
          _SCREAMING_SNAKE_CASE  			=					{}

          # Download this model to make sure it's in the cache.
          _SCREAMING_SNAKE_CASE  			=					WavaVecaFeatureExtractor.from_pretrained("""hf-internal-testing/tiny-random-wav2vec2"""   )
          # Under the mock environment we get a 500 error when trying to reach the model.
          with mock.patch("""requests.Session.request"""							,				return_value=lowerCAmelCase__   ) as mock_head:
             _SCREAMING_SNAKE_CASE  			=					WavaVecaFeatureExtractor.from_pretrained("""hf-internal-testing/tiny-random-wav2vec2"""   )
             # This check we did call the fake head request
             mock_head.assert_called()






       def 							UpperCamelCase					(				self   )     ->      int:
          _SCREAMING_SNAKE_CASE  			=					WavaVecaFeatureExtractor.from_pretrained(
              """https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"""   )






@is_staging_test
class  _a					(unittest.TestCase):





       """simple docstring"""




       @classmethod
       def 							UpperCamelCase					(				cls   )     ->      int:
          _SCREAMING_SNAKE_CASE  			=					TOKEN
          HfFolder.save_token(lowerCAmelCase__   )






       @classmethod
       def 							UpperCamelCase					(				cls   )     ->      str:
          try:
             delete_repo(token=cls._token							,				repo_id="""test-feature-extractor"""   )
          except HTTPError:
             pass

          try:
             delete_repo(token=cls._token							,				repo_id="""valid_org/test-feature-extractor-org"""   )
          except HTTPError:
             pass

          try:
             delete_repo(token=cls._token							,				repo_id="""test-dynamic-feature-extractor"""   )
          except HTTPError:
             pass






       def 							UpperCamelCase					(				self   )     ->      int:
          _SCREAMING_SNAKE_CASE  			=					WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase__   )
          feature_extractor.push_to_hub("""test-feature-extractor"""							,				use_auth_token=self._token   )

          _SCREAMING_SNAKE_CASE  			=					WavaVecaFeatureExtractor.from_pretrained(F"{USER}/test-feature-extractor"   )
          for k, v in feature_extractor.__dict__.items():
             self.assertEqual(lowerCAmelCase__							,				getattr(lowerCAmelCase__							,				lowerCAmelCase__   )   )

          # Reset repo
          delete_repo(token=self._token							,				repo_id="""test-feature-extractor"""   )

          # Push to hub via save_pretrained
          with tempfile.TemporaryDirectory() as tmp_dir:
             feature_extractor.save_pretrained(
                 lowerCAmelCase__							,				repo_id="""test-feature-extractor"""							,				push_to_hub=lowerCAmelCase__							,				use_auth_token=self._token   )

          _SCREAMING_SNAKE_CASE  			=					WavaVecaFeatureExtractor.from_pretrained(F"{USER}/test-feature-extractor"   )
          for k, v in feature_extractor.__dict__.items():
             self.assertEqual(lowerCAmelCase__							,				getattr(lowerCAmelCase__							,				lowerCAmelCase__   )   )






       def 							UpperCamelCase					(				self   )     ->      Tuple:
          _SCREAMING_SNAKE_CASE  			=					WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase__   )
          feature_extractor.push_to_hub("""valid_org/test-feature-extractor"""							,				use_auth_token=self._token   )

          _SCREAMING_SNAKE_CASE  			=					WavaVecaFeatureExtractor.from_pretrained("""valid_org/test-feature-extractor"""   )
          for k, v in feature_extractor.__dict__.items():
             self.assertEqual(lowerCAmelCase__							,				getattr(lowerCAmelCase__							,				lowerCAmelCase__   )   )

          # Reset repo
          delete_repo(token=self._token							,				repo_id="""valid_org/test-feature-extractor"""   )

          # Push to hub via save_pretrained
          with tempfile.TemporaryDirectory() as tmp_dir:
             feature_extractor.save_pretrained(
                 lowerCAmelCase__							,				repo_id="""valid_org/test-feature-extractor-org"""							,				push_to_hub=lowerCAmelCase__							,				use_auth_token=self._token   )

          _SCREAMING_SNAKE_CASE  			=					WavaVecaFeatureExtractor.from_pretrained("""valid_org/test-feature-extractor-org"""   )
          for k, v in feature_extractor.__dict__.items():
             self.assertEqual(lowerCAmelCase__							,				getattr(lowerCAmelCase__							,				lowerCAmelCase__   )   )






       def 							UpperCamelCase					(				self   )     ->      Optional[Any]:
          CustomFeatureExtractor.register_for_auto_class()
          _SCREAMING_SNAKE_CASE  			=					CustomFeatureExtractor.from_pretrained(lowerCAmelCase__   )

          feature_extractor.push_to_hub("""test-dynamic-feature-extractor"""							,				use_auth_token=self._token   )

          # This has added the proper auto_map field to the config
          self.assertDictEqual(
              feature_extractor.auto_map							,				{"""AutoFeatureExtractor""": """custom_feature_extraction.CustomFeatureExtractor"""}							,				)

          _SCREAMING_SNAKE_CASE  			=					AutoFeatureExtractor.from_pretrained(
              F"{USER}/test-dynamic-feature-extractor"							,				trust_remote_code=lowerCAmelCase__   )
          # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
          self.assertEqual(new_feature_extractor.__class__.__name__							,				"""CustomFeatureExtractor"""   )











'''simple docstring'''



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

from ...utils import logging


lowerCAmelCase_  :							int             =    logging.get_logger(__name__)
def        UpperCAmelCase (    A	:					List[Any]       ,		A	:					Optional[int]					):
							SCREAMING_SNAKE_CASE       :   Any         =      nn.functional.normalize(A					)
							SCREAMING_SNAKE_CASE       :   List[str]         =      nn.functional.normalize(A					)
							return torch.mm(A       ,		normalized_text_embeds.t()					)






class 						lowerCamelCase_				(       snake_case_      ):
						_lowerCAmelCase  : Optional[Any]						  =					CLIPConfig

						_lowerCAmelCase  : List[str]						  =					['CLIPEncoderLayer']







						def __init__(	self :   int					,     lowerCAmelCase__ :   CLIPConfig					):

													"""simple docstring"""
													super().__init__(lowerCAmelCase__					)

													SCREAMING_SNAKE_CASE       :   Any         =      CLIPVisionModel(config.vision_config					)
													SCREAMING_SNAKE_CASE       :   List[str]         =      nn.Linear(config.vision_config.hidden_size					,     config.projection_dim					,     bias=lowerCAmelCase__					)

													SCREAMING_SNAKE_CASE       :   Any         =      nn.Parameter(torch.ones(17					,     config.projection_dim					)					,     requires_grad=lowerCAmelCase__					)
													SCREAMING_SNAKE_CASE       :   int         =      nn.Parameter(torch.ones(3					,     config.projection_dim					)					,     requires_grad=lowerCAmelCase__					)

													SCREAMING_SNAKE_CASE       :   Dict         =      nn.Parameter(torch.ones(17					)					,     requires_grad=lowerCAmelCase__					)
													SCREAMING_SNAKE_CASE       :   List[str]         =      nn.Parameter(torch.ones(3					)					,     requires_grad=lowerCAmelCase__					)







						@torch.no_grad()
						def        __lowercase     (	self :   List[str]					,     lowerCAmelCase__ :   List[str]					,     lowerCAmelCase__ :   Optional[Any]					):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE       :   Any         =      self.vision_model(lowerCAmelCase__					)[1]  # pooled_output
													SCREAMING_SNAKE_CASE       :   Any         =      self.visual_projection(lowerCAmelCase__					)

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

													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      []
													SCREAMING_SNAKE_CASE       :   int         =      image_embeds.shape[0]
													for i in range(lowerCAmelCase__					):
																				SCREAMING_SNAKE_CASE       :   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
																				SCREAMING_SNAKE_CASE       :   int         =      0.0

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

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

																				result.append(lowerCAmelCase__					)

													SCREAMING_SNAKE_CASE       :   Dict         =      [len(res['''bad_concepts''']					) > 0 for res in result]

													return images, has_nsfw_concepts







						@torch.no_grad()
						def        __lowercase     (	self :   List[str]					,     lowerCAmelCase__ :   torch.FloatTensor					,     lowerCAmelCase__ :   torch.FloatTensor					):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE       :   Union[str, Any]         =      self.vision_model(lowerCAmelCase__					)[1]  # pooled_output
													SCREAMING_SNAKE_CASE       :   List[Any]         =      self.visual_projection(lowerCAmelCase__					)

													SCREAMING_SNAKE_CASE       :   List[str]         =      cosine_distance(lowerCAmelCase__					,     self.special_care_embeds					)
													SCREAMING_SNAKE_CASE       :   List[str]         =      cosine_distance(lowerCAmelCase__					,     self.concept_embeds					)

													# increase this value to create a stronger `nsfw` filter
													# at the cost of increasing the possibility of filtering benign images
													SCREAMING_SNAKE_CASE       :   Optional[int]         =      0.0

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

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

													return images, has_nsfw_concepts




