






import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


snake_case    				=		"""hf-internal-testing/tiny-random-bert"""
snake_case    				=		os.path.join(TRANSFORMERS_CACHE, """models--hf-internal-testing--tiny-random-bert""")
snake_case    				=		"""9b8c223d42b2188cb49d29af482996f9d0f3e5a6"""


class  SCREAMING_SNAKE_CASE			(			unittest.TestCase					):
     '''simple docstring'''


     def 				_A       (				self       :	int      ):
       SCREAMING_SNAKE_CASE       :      List[Any]							=    cached_file(UpperCAmelCase_					, UpperCAmelCase_      )
       # Should have downloaded the file in here
       self.assertTrue(os.path.isdir(UpperCAmelCase_      )      )
       # Cache should contain at least those three subfolders:
       for subfolder in ["blobs", "refs", "snapshots"]:
         self.assertTrue(os.path.isdir(os.path.join(UpperCAmelCase_					, UpperCAmelCase_      )      )      )
       with open(os.path.join(UpperCAmelCase_					, "refs"					, "main"      )      ) as f:
         SCREAMING_SNAKE_CASE       :      Union[str, Any]							=    f.read()
       self.assertEqual(UpperCAmelCase_					, os.path.join(UpperCAmelCase_					, "snapshots"					, UpperCAmelCase_					, UpperCAmelCase_      )      )
       self.assertTrue(os.path.isfile(UpperCAmelCase_      )      )

       # File is cached at the same place the second time.
       SCREAMING_SNAKE_CASE       :      Union[str, Any]							=    cached_file(UpperCAmelCase_					, UpperCAmelCase_      )
       self.assertEqual(UpperCAmelCase_					, UpperCAmelCase_      )

       # Using a specific revision to test the full commit hash.
       SCREAMING_SNAKE_CASE       :      List[str]							=    cached_file(UpperCAmelCase_					, UpperCAmelCase_					, revision="9b8c223"      )
       self.assertEqual(UpperCAmelCase_					, os.path.join(UpperCAmelCase_					, "snapshots"					, UpperCAmelCase_					, UpperCAmelCase_      )      )


     def 				_A       (				self       :	Optional[Any]      ):
       with self.assertRaisesRegex(UpperCAmelCase_					, "is not a valid model identifier"      ):
         SCREAMING_SNAKE_CASE       :      int							=    cached_file("tiny-random-bert"					, UpperCAmelCase_      )

       with self.assertRaisesRegex(UpperCAmelCase_					, "is not a valid git identifier"      ):
         SCREAMING_SNAKE_CASE       :      str							=    cached_file(UpperCAmelCase_					, UpperCAmelCase_					, revision="aaaa"      )

       with self.assertRaisesRegex(UpperCAmelCase_					, "does not appear to have a file named"      ):
         SCREAMING_SNAKE_CASE       :      Optional[Any]							=    cached_file(UpperCAmelCase_					, "conf"      )


     def 				_A       (				self       :	Tuple      ):
       with self.assertRaisesRegex(UpperCAmelCase_					, "does not appear to have a file named"      ):
         SCREAMING_SNAKE_CASE       :      Dict							=    cached_file(UpperCAmelCase_					, "conf"      )

       with open(os.path.join(UpperCAmelCase_					, "refs"					, "main"      )      ) as f:
         SCREAMING_SNAKE_CASE       :      Dict							=    f.read()
       self.assertTrue(os.path.isfile(os.path.join(UpperCAmelCase_					, ".no_exist"					, UpperCAmelCase_					, "conf"      )      )      )

       SCREAMING_SNAKE_CASE       :      Optional[int]							=    cached_file(UpperCAmelCase_					, "conf"					, _raise_exceptions_for_missing_entries=UpperCAmelCase_      )
       self.assertIsNone(UpperCAmelCase_      )

       SCREAMING_SNAKE_CASE       :      Union[str, Any]							=    cached_file(UpperCAmelCase_					, "conf"					, local_files_only=UpperCAmelCase_					, _raise_exceptions_for_missing_entries=UpperCAmelCase_      )
       self.assertIsNone(UpperCAmelCase_      )

       SCREAMING_SNAKE_CASE       :      Optional[Any]							=    mock.Mock()
       SCREAMING_SNAKE_CASE       :      List[Any]							=    500
       SCREAMING_SNAKE_CASE       :      Tuple							=    {}
       SCREAMING_SNAKE_CASE       :      Any							=    HTTPError
       SCREAMING_SNAKE_CASE       :      List[str]							=    {}

       # Under the mock environment we get a 500 error when trying to reach the tokenizer.
       with mock.patch("requests.Session.request"					, return_value=UpperCAmelCase_      ) as mock_head:
         SCREAMING_SNAKE_CASE       :      Any							=    cached_file(UpperCAmelCase_					, "conf"					, _raise_exceptions_for_connection_errors=UpperCAmelCase_      )
         self.assertIsNone(UpperCAmelCase_      )
         # This check we did call the fake head request
         mock_head.assert_called()


     def 				_A       (				self       :	List[Any]      ):
       self.assertTrue(has_file("hf-internal-testing/tiny-bert-pt-only"					, UpperCAmelCase_      )      )
       self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"					, UpperCAmelCase_      )      )
       self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"					, UpperCAmelCase_      )      )


     def 				_A       (				self       :	Union[str, Any]      ):
       # `get_file_from_repo` returns None if the file does not exist
       self.assertIsNone(get_file_from_repo("bert-base-cased"					, "ahah.txt"      )      )

       # The function raises if the repository does not exist.
       with self.assertRaisesRegex(UpperCAmelCase_					, "is not a valid model identifier"      ):
         get_file_from_repo("bert-base-case"					, UpperCAmelCase_      )

       # The function raises if the revision does not exist.
       with self.assertRaisesRegex(UpperCAmelCase_					, "is not a valid git identifier"      ):
         get_file_from_repo("bert-base-cased"					, UpperCAmelCase_					, revision="ahaha"      )

       SCREAMING_SNAKE_CASE       :      Tuple							=    get_file_from_repo("bert-base-cased"					, UpperCAmelCase_      )
       # The name is the cached name which is not very easy to test, so instead we load the content.
       SCREAMING_SNAKE_CASE       :      Optional[Any]							=    json.loads(open(UpperCAmelCase_					, "r"      ).read()      )
       self.assertEqual(config["hidden_size"]					, 768      )






     def 				_A       (				self       :	Any      ):
       with tempfile.TemporaryDirectory() as tmp_dir:
         SCREAMING_SNAKE_CASE       :      Union[str, Any]							=    Path(UpperCAmelCase_      ) / "a.txt"
         filename.touch()
         self.assertEqual(get_file_from_repo(UpperCAmelCase_					, "a.txt"      )					, str(UpperCAmelCase_      )      )

         self.assertIsNone(get_file_from_repo(UpperCAmelCase_					, "b.txt"      )      )














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

from ...utils import logging


snake_case    				=		logging.get_logger(__name__)


class  SCREAMING_SNAKE_CASE			(			lowerCAmelCase					):
     '''simple docstring'''


     UpperCamelCase_							:		Union[str, Any]			=	CLIPConfig

     UpperCamelCase_							:		Optional[int]			=	['''CLIPEncoderLayer''']


     def __init__(				self       :	List[Any]					, UpperCAmelCase_       :	CLIPConfig      ):
       super().__init__(UpperCAmelCase_      )

       SCREAMING_SNAKE_CASE       :      Any							=    CLIPVisionModelWithProjection(config.vision_config      )

       SCREAMING_SNAKE_CASE       :      List[Any]							=    nn.Linear(config.vision_config.projection_dim					, 1      )
       SCREAMING_SNAKE_CASE       :      List[str]							=    nn.Linear(config.vision_config.projection_dim					, 1      )






     @torch.no_grad()
     def 				_A       (				self       :	Union[str, Any]					, UpperCAmelCase_       :	int					, UpperCAmelCase_       :	Dict					, UpperCAmelCase_       :	List[Any]=0.5					, UpperCAmelCase_       :	Optional[Any]=0.5      ):
       SCREAMING_SNAKE_CASE       :      str							=    self.vision_model(UpperCAmelCase_      )[0]

       SCREAMING_SNAKE_CASE       :      Union[str, Any]							=    self.p_head(UpperCAmelCase_      )
       SCREAMING_SNAKE_CASE       :      List[str]							=    nsfw_detected.flatten()
       SCREAMING_SNAKE_CASE       :      List[str]							=    nsfw_detected > p_threshold
       SCREAMING_SNAKE_CASE       :      List[Any]							=    nsfw_detected.tolist()

       if any(UpperCAmelCase_      ):
         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."      )

       for idx, nsfw_detected_ in enumerate(UpperCAmelCase_      ):
         if nsfw_detected_:
           SCREAMING_SNAKE_CASE       :      Dict							=    np.zeros(images[idx].shape      )

       SCREAMING_SNAKE_CASE       :      Any							=    self.w_head(UpperCAmelCase_      )
       SCREAMING_SNAKE_CASE       :      List[Any]							=    watermark_detected.flatten()
       SCREAMING_SNAKE_CASE       :      Union[str, Any]							=    watermark_detected > w_threshold
       SCREAMING_SNAKE_CASE       :      Optional[int]							=    watermark_detected.tolist()

       if any(UpperCAmelCase_      ):
         logger.warning(
             "Potential watermarked content was detected in one or more images. A black image will be returned instead."
             " Try again with a different prompt and/or seed."      )

       for idx, watermark_detected_ in enumerate(UpperCAmelCase_      ):
         if watermark_detected_:
           SCREAMING_SNAKE_CASE       :      Optional[int]							=    np.zeros(images[idx].shape      )

       return images, nsfw_detected, watermark_detected






