






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,
)


lowerCamelCase     :							List[str]    ='hf-internal-testing/tiny-random-bert'
lowerCamelCase     :							str    =os.path.join(TRANSFORMERS_CACHE, '''models--hf-internal-testing--tiny-random-bert''')
lowerCamelCase     :							Optional[Any]    ='9b8c223d42b2188cb49d29af482996f9d0f3e5a6'





class       __a  (   unittest.TestCase	):







 def  __lowercase (		self :       Optional[Any]						):


      '''simple docstring'''






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

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

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







 def  __lowercase (		self :       List[str]						):


      '''simple docstring'''






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

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

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







 def  __lowercase (		self :       Tuple						):


      '''simple docstring'''






      with self.assertRaisesRegex(SCREAMING_SNAKE_CASE							,       "does not appear to have a file named"						):
           UpperCamelCase__      :       str	    =  cached_file(SCREAMING_SNAKE_CASE							,       "conf"						)

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

      UpperCamelCase__      :       Tuple	    =  cached_file(SCREAMING_SNAKE_CASE							,       "conf"							,       _raise_exceptions_for_missing_entries=SCREAMING_SNAKE_CASE						)
      self.assertIsNone(SCREAMING_SNAKE_CASE						)

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

      UpperCamelCase__      :       List[str]	    =  mock.Mock()
      UpperCamelCase__      :       int	    =  5_00
      UpperCamelCase__      :       Optional[Any]	    =  {}
      UpperCamelCase__      :       Optional[Any]	    =  HTTPError
      UpperCamelCase__      :       Optional[int]	    =  {}

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







 def  __lowercase (		self :       Optional[Any]						):


      '''simple docstring'''






      self.assertTrue(has_file("hf-internal-testing/tiny-bert-pt-only"							,       SCREAMING_SNAKE_CASE						)						)
      self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"							,       SCREAMING_SNAKE_CASE						)						)
      self.assertFalse(has_file("hf-internal-testing/tiny-bert-pt-only"							,       SCREAMING_SNAKE_CASE						)						)







 def  __lowercase (		self :       Optional[Any]						):


      '''simple docstring'''






      self.assertIsNone(get_file_from_repo("bert-base-cased"							,       "ahah.txt"						)						)

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

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

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







 def  __lowercase (		self :       List[Any]						):


      '''simple docstring'''






      with tempfile.TemporaryDirectory() as tmp_dir:
           UpperCamelCase__      :       List[Any]	    =  Path(SCREAMING_SNAKE_CASE						) / '''a.txt'''
           filename.touch()
           self.assertEqual(get_file_from_repo(SCREAMING_SNAKE_CASE							,       "a.txt"						)							,       str(SCREAMING_SNAKE_CASE						)						)

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








from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
 from ..tf_utils import stable_softmax

lowerCamelCase     :							str    =logging.get_logger(__name__)





@add_end_docstrings(A__	)
class       __a  (   A__	):







 def __init__(		self :       List[str]							,       **SCREAMING_SNAKE_CASE :       Tuple						):


      '''simple docstring'''






      super().__init__(**SCREAMING_SNAKE_CASE						)

      requires_backends(self							,       "vision"						)
      self.check_model_type(
          TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
          if self.framework == "tf"
          else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING						)







 def __call__(		self :       Dict							,       SCREAMING_SNAKE_CASE :       Union[str, List[str], "Image", List["Image"]]							,       **SCREAMING_SNAKE_CASE :       str						):


      '''simple docstring'''






      return super().__call__(SCREAMING_SNAKE_CASE							,       **SCREAMING_SNAKE_CASE						)







 def  __lowercase (		self :       List[str]							,       **SCREAMING_SNAKE_CASE :       List[str]						):


      '''simple docstring'''






      UpperCamelCase__      :       List[Any]	    =  {}
      if "candidate_labels" in kwargs:
           UpperCamelCase__      :       Optional[Any]	    =  kwargs["candidate_labels"]
      if "hypothesis_template" in kwargs:
           UpperCamelCase__      :       int	    =  kwargs["hypothesis_template"]

      return preprocess_params, {}, {}







 def  __lowercase (		self :       List[Any]							,       SCREAMING_SNAKE_CASE :       List[Any]							,       SCREAMING_SNAKE_CASE :       Tuple=None							,       SCREAMING_SNAKE_CASE :       Optional[int]="This is a photo of {}."						):


      '''simple docstring'''






      UpperCamelCase__      :       Dict	    =  load_image(SCREAMING_SNAKE_CASE						)
      UpperCamelCase__      :       List[Any]	    =  self.image_processor(images=[image]							,       return_tensors=self.framework						)
      UpperCamelCase__      :       Any	    =  candidate_labels
      UpperCamelCase__      :       Dict	    =  [hypothesis_template.format(SCREAMING_SNAKE_CASE						) for x in candidate_labels]
      UpperCamelCase__      :       Optional[Any]	    =  self.tokenizer(SCREAMING_SNAKE_CASE							,       return_tensors=self.framework							,       padding=SCREAMING_SNAKE_CASE						)
      UpperCamelCase__      :       List[Any]	    =  [text_inputs]
      return inputs







 def  __lowercase (		self :       int							,       SCREAMING_SNAKE_CASE :       Union[str, Any]						):


      '''simple docstring'''






      UpperCamelCase__      :       Tuple	    =  model_inputs.pop("candidate_labels"						)
      UpperCamelCase__      :       List[str]	    =  model_inputs.pop("text_inputs"						)
      if isinstance(text_inputs[0]							,       SCREAMING_SNAKE_CASE						):
           UpperCamelCase__      :       Dict	    =  text_inputs[0]
      else:
           # Batching case.
           UpperCamelCase__      :       Union[str, Any]	    =  text_inputs[0][0]

      UpperCamelCase__      :       Any	    =  self.model(**SCREAMING_SNAKE_CASE							,       **SCREAMING_SNAKE_CASE						)

      UpperCamelCase__      :       Dict	    =  {
          "candidate_labels": candidate_labels,
          "logits": outputs.logits_per_image,
      }
      return model_outputs







 def  __lowercase (		self :       Any							,       SCREAMING_SNAKE_CASE :       List[str]						):


      '''simple docstring'''






      UpperCamelCase__      :       Optional[int]	    =  model_outputs.pop("candidate_labels"						)
      UpperCamelCase__      :       int	    =  model_outputs["logits"][0]
      if self.framework == "pt":
           UpperCamelCase__      :       Dict	    =  logits.softmax(dim=-1						).squeeze(-1						)
           UpperCamelCase__      :       Optional[Any]	    =  probs.tolist()
           if not isinstance(SCREAMING_SNAKE_CASE							,       SCREAMING_SNAKE_CASE						):
                UpperCamelCase__      :       List[Any]	    =  [scores]
      elif self.framework == "tf":
           UpperCamelCase__      :       Optional[Any]	    =  stable_softmax(SCREAMING_SNAKE_CASE							,       axis=-1						)
           UpperCamelCase__      :       Optional[int]	    =  probs.numpy().tolist()
      else:
           raise ValueError(F'Unsupported framework: {self.framework}'						)

      UpperCamelCase__      :       Optional[int]	    =  [
          {"score": score, "label": candidate_label}
          for score, candidate_label in sorted(zip(SCREAMING_SNAKE_CASE							,       SCREAMING_SNAKE_CASE						)							,       key=lambda SCREAMING_SNAKE_CASE						: -x[0]						)
      ]
      return result