


'''simple docstring'''



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

A__						:   Optional[int]				=    logging.get_logger(__name__)





@add_end_docstrings(_UpperCAmelCase    )
class   UpperCAmelCase_    (_UpperCAmelCase    ):





    """simple docstring"""






    def __init__(				self     ,	**SCREAMING_SNAKE_CASE_      )							->	List[str]:
         super().__init__(**SCREAMING_SNAKE_CASE_      )

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

         requires_backends(self     ,	'vision'      )
         self.check_model_type(SCREAMING_SNAKE_CASE_      )







    def __call__(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_ = None     ,	**SCREAMING_SNAKE_CASE_     ,	)							->	Optional[Any]:
         if "text_queries" in kwargs:
              __lowerCamelCase  :      List[Any]														=						kwargs.pop('text_queries'      )

         if isinstance(SCREAMING_SNAKE_CASE_     ,	(str, Image.Image)      ):
              __lowerCamelCase  :      str														=						{'image': image, 'candidate_labels': candidate_labels}
         else:
              __lowerCamelCase  :      List[str]														=						image
         __lowerCamelCase  :      Optional[int]														=						super().__call__(SCREAMING_SNAKE_CASE_     ,	**SCREAMING_SNAKE_CASE_      )
         return results







    def  lowercase_							(				self     ,	**SCREAMING_SNAKE_CASE_      )							->	List[Any]:
         __lowerCamelCase  :      Union[str, Any]														=						{}
         if "threshold" in kwargs:
              __lowerCamelCase  :      Any														=						kwargs['threshold']
         if "top_k" in kwargs:
              __lowerCamelCase  :      Tuple														=						kwargs['top_k']
         return {}, {}, postprocess_params







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_      )							->	List[str]:
         __lowerCamelCase  :      Union[str, Any]														=						load_image(inputs['image']      )
         __lowerCamelCase  :      int														=						inputs['candidate_labels']
         if isinstance(SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_      ):
              __lowerCamelCase  :      Optional[int]														=						candidate_labels.split(','      )

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







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_      )							->	Any:
         __lowerCamelCase  :      Optional[Any]														=						model_inputs.pop('target_size'      )
         __lowerCamelCase  :      List[Any]														=						model_inputs.pop('candidate_label'      )
         __lowerCamelCase  :      List[Any]														=						model_inputs.pop('is_last'      )

         __lowerCamelCase  :      str														=						self.model(**SCREAMING_SNAKE_CASE_      )

         __lowerCamelCase  :      str														=						{'target_size': target_size, 'candidate_label': candidate_label, 'is_last': is_last, **outputs}
         return model_outputs







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_     ,	SCREAMING_SNAKE_CASE_=0.1     ,	SCREAMING_SNAKE_CASE_=None      )							->	Tuple:
         __lowerCamelCase  :      Any														=						[]
         for model_output in model_outputs:
              __lowerCamelCase  :      Tuple														=						model_output['candidate_label']
              __lowerCamelCase  :      str														=						BaseModelOutput(SCREAMING_SNAKE_CASE_      )
              __lowerCamelCase  :      Dict														=						self.image_processor.post_process_object_detection(
                  outputs=SCREAMING_SNAKE_CASE_     ,	threshold=SCREAMING_SNAKE_CASE_     ,	target_sizes=model_output['target_size']      )[0]

              for index in outputs["scores"].nonzero():
                   __lowerCamelCase  :      Optional[int]														=						outputs['scores'][index].item()
                   __lowerCamelCase  :      List[Any]														=						self._get_bounding_box(outputs['boxes'][index][0]      )

                   __lowerCamelCase  :      Dict														=						{'score': score, 'label': label, 'box': box}
                   results.append(SCREAMING_SNAKE_CASE_      )

         __lowerCamelCase  :      str														=						sorted(SCREAMING_SNAKE_CASE_     ,	key=lambda SCREAMING_SNAKE_CASE_      : x["score"]     ,	reverse=SCREAMING_SNAKE_CASE_      )
         if top_k:
              __lowerCamelCase  :      List[str]														=						results[:top_k]

         return results







    def  lowercase_							(				self     ,	SCREAMING_SNAKE_CASE_      )							->	Dict[str, int]:
         if self.framework != "pt":
              raise ValueError('The ZeroShotObjectDetectionPipeline is only available in PyTorch.'      )
         __lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase							,			__lowerCamelCase  :      Union[str, Any]														=						box.int().tolist()
         __lowerCamelCase  :      Any														=						{
             'xmin': xmin,
             'ymin': ymin,
             'xmax': xmax,
             'ymax': ymax,
         }
         return bbox












import json
import logging
import os
import sys
from time import time
from unittest.mock import patch

from transformers.testing_utils import TestCasePlus, require_torch_tpu


logging.basicConfig(level=logging.DEBUG)

a__		     =     logging.getLogger()





def 				A__ (snake_case   :		Tuple				)     ->						Dict:
					__UpperCamelCase					:	Dict             =      {}
					__UpperCamelCase					:	Dict             =      os.path.join(snake_case		,	"""all_results.json"""				)
					if os.path.exists(snake_case				):
										with open(snake_case		,	"""r"""				) as f:
															__UpperCamelCase					:	List[Any]             =      json.load(snake_case				)
					else:
										raise ValueError(F'''can\'t find {path}'''				)
					return results


a__		     =     logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)







@require_torch_tpu
class    SCREAMING_SNAKE_CASE_				(       _UpperCamelCase   ):



	"""simple docstring"""


	def 					lowerCamelCase__     (					self		:						Dict	)	->       Dict:







						"""simple docstring"""







						import xla_spawn

						__UpperCamelCase					:	Dict             =      self.get_auto_remove_tmp_dir()
						__UpperCamelCase					:	str             =      F'''
            ./examples/pytorch/text-classification/run_glue.py
            --num_cores=8
            ./examples/pytorch/text-classification/run_glue.py
            --model_name_or_path distilbert-base-uncased
            --output_dir {tmp_dir}
            --overwrite_output_dir
            --train_file ./tests/fixtures/tests_samples/MRPC/train.csv
            --validation_file ./tests/fixtures/tests_samples/MRPC/dev.csv
            --do_train
            --do_eval
            --debug tpu_metrics_debug
            --per_device_train_batch_size=2
            --per_device_eval_batch_size=1
            --learning_rate=1e-4
            --max_steps=10
            --warmup_steps=2
            --seed=42
            --max_seq_length=128
            '''.split()

						with patch.object(lowerCAmelCase					,     """argv"""					,     lowerCAmelCase	):
											__UpperCamelCase					:	Union[str, Any]             =      time()
											xla_spawn.main()
											__UpperCamelCase					:	List[str]             =      time()

											__UpperCamelCase					:	Optional[int]             =      get_results(lowerCAmelCase	)
											self.assertGreaterEqual(result["""eval_accuracy"""]					,     0.75	)

											# Assert that the script takes less than 500 seconds to make sure it doesn't hang.
											self.assertLess(end - start					,     500	)






	def 					lowerCamelCase__     (					self		:						int	)	->       List[Any]:







						"""simple docstring"""







						import xla_spawn

						__UpperCamelCase					:	List[Any]             =      """
            ./tests/test_trainer_tpu.py
            --num_cores=8
            ./tests/test_trainer_tpu.py
            """.split()
						with patch.object(lowerCAmelCase					,     """argv"""					,     lowerCAmelCase	):
											xla_spawn.main()






