





from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
      from PIL import Image

      from ..image_utils import load_image

if is_torch_available():
      import torch

      from transformers.modeling_outputs import BaseModelOutput

      from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

UpperCAmelCase_					:   Any 							=						logging.get_logger(__name__)





@add_end_docstrings(__a  )
class      lowercase__    (							__a  ):







  def __init__(	self    ,						**_lowercase						):
       super().__init__(**lowerCAmelCase_						)

       if self.framework == "tf":
            raise ValueError(F'The {self.__class__} is only available in PyTorch.'						)

       requires_backends(self    ,						"""vision"""						)
       self.check_model_type(lowerCAmelCase_						)







  def __call__(	self    ,						_lowercase    ,						_lowercase = None    ,						**_lowercase    ,						):







       if "text_queries" in kwargs:
            lowerCAmelCase_					:       Optional[int]    =		kwargs.pop("""text_queries"""						)

       if isinstance(lowerCAmelCase_    ,						(str, Image.Image)						):
            lowerCAmelCase_					:       str    =		{"""image""": image, """candidate_labels""": candidate_labels}
       else:
            lowerCAmelCase_					:       Optional[int]    =		image
       lowerCAmelCase_					:       Union[str, Any]    =		super().__call__(lowerCAmelCase_    ,						**lowerCAmelCase_						)
       return results







  def 	UpperCAmelCase__		(	self    ,						**_lowercase						):
       lowerCAmelCase_					:       Optional[Any]    =		{}
       if "threshold" in kwargs:
            lowerCAmelCase_					:       Any    =		kwargs["""threshold"""]
       if "top_k" in kwargs:
            lowerCAmelCase_					:       Dict    =		kwargs["""top_k"""]
       return {}, {}, postprocess_params







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):
       lowerCAmelCase_					:       Union[str, Any]    =		load_image(inputs["""image"""]						)
       lowerCAmelCase_					:       List[str]    =		inputs["""candidate_labels"""]
       if isinstance(lowerCAmelCase_    ,						lowerCAmelCase_						):
            lowerCAmelCase_					:       List[Any]    =		candidate_labels.split(""","""						)

       lowerCAmelCase_					:       str    =		torch.tensor([[image.height, image.width]]    ,						dtype=torch.intaa						)
       for i, candidate_label in enumerate(lowerCAmelCase_						):
            lowerCAmelCase_					:       Optional[int]    =		self.tokenizer(lowerCAmelCase_    ,						return_tensors=self.framework						)
            lowerCAmelCase_					:       Tuple    =		self.image_processor(lowerCAmelCase_    ,						return_tensors=self.framework						)
            yield {
                "is_last": i == len(lowerCAmelCase_						) - 1,
                "target_size": target_size,
                "candidate_label": candidate_label,
                **text_inputs,
                **image_features,
            }







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):
       lowerCAmelCase_					:       Tuple    =		model_inputs.pop("""target_size"""						)
       lowerCAmelCase_					:       Optional[Any]    =		model_inputs.pop("""candidate_label"""						)
       lowerCAmelCase_					:       Any    =		model_inputs.pop("""is_last"""						)

       lowerCAmelCase_					:       List[str]    =		self.model(**lowerCAmelCase_						)

       lowerCAmelCase_					:       int    =		{"""target_size""": target_size, """candidate_label""": candidate_label, """is_last""": is_last, **outputs}
       return model_outputs







  def 	UpperCAmelCase__		(	self    ,						_lowercase    ,						_lowercase=0.1    ,						_lowercase=None						):
       lowerCAmelCase_					:       List[Any]    =		[]
       for model_output in model_outputs:
            lowerCAmelCase_					:       Union[str, Any]    =		model_output["""candidate_label"""]
            lowerCAmelCase_					:       Union[str, Any]    =		BaseModelOutput(lowerCAmelCase_						)
            lowerCAmelCase_					:       Optional[Any]    =		self.image_processor.post_process_object_detection(
                outputs=lowerCAmelCase_    ,						threshold=lowerCAmelCase_    ,						target_sizes=model_output["""target_size"""]						)[0]

            for index in outputs["scores"].nonzero():
                 lowerCAmelCase_					:       List[str]    =		outputs["""scores"""][index].item()
                 lowerCAmelCase_					:       Union[str, Any]    =		self._get_bounding_box(outputs["""boxes"""][index][0]						)

                 lowerCAmelCase_					:       Dict    =		{"""score""": score, """label""": label, """box""": box}
                 results.append(lowerCAmelCase_						)

       lowerCAmelCase_					:       int    =		sorted(lowerCAmelCase_    ,						key=lambda _lowercase						: x["score"]    ,						reverse=lowerCAmelCase_						)
       if top_k:
            lowerCAmelCase_					:       Optional[Any]    =		results[:top_k]

       return results







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):







       if self.framework != "pt":
            raise ValueError("""The ZeroShotObjectDetectionPipeline is only available in PyTorch."""						)
       lowerCAmelCase_   ,  lowerCAmelCase_   ,  lowerCAmelCase_   ,  lowerCAmelCase_					:       str    =		box.int().tolist()
       lowerCAmelCase_					:       Optional[int]    =		{
           """xmin""": xmin,
           """ymin""": ymin,
           """xmax""": xmax,
           """ymax""": ymax,
       }
       return bbox







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


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





class      lowercase__    (							unittest.TestCase  ):







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

       # File is cached at the same place the second time.
       lowerCAmelCase_					:       Dict    =		cached_file(_lowercase    ,						_lowercase						)
       self.assertEqual(_lowercase    ,						_lowercase						)

       # Using a specific revision to test the full commit hash.
       lowerCAmelCase_					:       Optional[int]    =		cached_file(_lowercase    ,						_lowercase    ,						revision="""9b8c223"""						)
       self.assertEqual(_lowercase    ,						os.path.join(_lowercase    ,						"""snapshots"""    ,						_lowercase    ,						_lowercase						)						)







  def 	UpperCAmelCase__		(	self						):
       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid model identifier"""						):
            lowerCAmelCase_					:       Dict    =		cached_file("""tiny-random-bert"""    ,						_lowercase						)

       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid git identifier"""						):
            lowerCAmelCase_					:       Tuple    =		cached_file(_lowercase    ,						_lowercase    ,						revision="""aaaa"""						)

       with self.assertRaisesRegex(_lowercase    ,						"""does not appear to have a file named"""						):
            lowerCAmelCase_					:       List[Any]    =		cached_file(_lowercase    ,						"""conf"""						)







  def 	UpperCAmelCase__		(	self						):
       with self.assertRaisesRegex(_lowercase    ,						"""does not appear to have a file named"""						):
            lowerCAmelCase_					:       Any    =		cached_file(_lowercase    ,						"""conf"""						)

       with open(os.path.join(_lowercase    ,						"""refs"""    ,						"""main"""						)						) as f:
            lowerCAmelCase_					:       str    =		f.read()
       self.assertTrue(os.path.isfile(os.path.join(_lowercase    ,						""".no_exist"""    ,						_lowercase    ,						"""conf"""						)						)						)

       lowerCAmelCase_					:       Union[str, Any]    =		cached_file(_lowercase    ,						"""conf"""    ,						_raise_exceptions_for_missing_entries=_lowercase						)
       self.assertIsNone(_lowercase						)

       lowerCAmelCase_					:       Any    =		cached_file(_lowercase    ,						"""conf"""    ,						local_files_only=_lowercase    ,						_raise_exceptions_for_missing_entries=_lowercase						)
       self.assertIsNone(_lowercase						)

       lowerCAmelCase_					:       Any    =		mock.Mock()
       lowerCAmelCase_					:       Optional[Any]    =		500
       lowerCAmelCase_					:       Any    =		{}
       lowerCAmelCase_					:       List[str]    =		HTTPError
       lowerCAmelCase_					:       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=_lowercase						) as mock_head:
            lowerCAmelCase_					:       Union[str, Any]    =		cached_file(_lowercase    ,						"""conf"""    ,						_raise_exceptions_for_connection_errors=_lowercase						)
            self.assertIsNone(_lowercase						)
            # This check we did call the fake head request
            mock_head.assert_called()







  def 	UpperCAmelCase__		(	self						):
       self.assertTrue(has_file("""hf-internal-testing/tiny-bert-pt-only"""    ,						_lowercase						)						)
       self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""    ,						_lowercase						)						)
       self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""    ,						_lowercase						)						)







  def 	UpperCAmelCase__		(	self						):
       # `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(_lowercase    ,						"""is not a valid model identifier"""						):
            get_file_from_repo("""bert-base-case"""    ,						_lowercase						)

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

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







  def 	UpperCAmelCase__		(	self						):
       with tempfile.TemporaryDirectory() as tmp_dir:
            lowerCAmelCase_					:       Optional[Any]    =		Path(_lowercase						) / """a.txt"""
            filename.touch()
            self.assertEqual(get_file_from_repo(_lowercase    ,						"""a.txt"""						)    ,						str(_lowercase						)						)

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