


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

snake_case							:       str     			=      logging.get_logger(__name__)







@add_end_docstrings(lowerCamelCase_      )
class        snake_case_   (lowerCamelCase_      ):
					def __init__(      self   :List[str]			,**__snake_case   :Dict      )					->    List[str]:
												super().__init__(**__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(__snake_case      )
					def __call__(      self   :str			,__snake_case   :Union[str, "Image.Image", List[Dict[str, Any]]]			,__snake_case   :Union[str, List[str]] = None			,**__snake_case   :Dict			,)					->    Optional[int]:
												if "text_queries" in kwargs:
																			a__             =     kwargs.pop('text_queries'      )

												if isinstance(__snake_case			,(str, Image.Image)      ):
																			a__             =     {'image': image, 'candidate_labels': candidate_labels}
												else:
																			a__             =     image
												a__             =     super().__call__(__snake_case			,**__snake_case      )
												return results
					def  lowerCamelCase__(      self   :int			,**__snake_case   :List[Any]      )					->    str:
												a__             =     {}
												if "threshold" in kwargs:
																			a__             =     kwargs['threshold']
												if "top_k" in kwargs:
																			a__             =     kwargs['top_k']
												return {}, {}, postprocess_params
					def  lowerCamelCase__(      self   :Tuple			,__snake_case   :List[Any]      )					->    Optional[int]:
												a__             =     load_image(inputs['image']      )
												a__             =     inputs['candidate_labels']
												if isinstance(__snake_case			,__snake_case      ):
																			a__             =     candidate_labels.split(','      )

												a__             =     torch.tensor([[image.height, image.width]]			,dtype=torch.intaa      )
												for i, candidate_label in enumerate(__snake_case      ):
																			a__             =     self.tokenizer(__snake_case			,return_tensors=self.framework      )
																			a__             =     self.image_processor(__snake_case			,return_tensors=self.framework      )
																			yield {
																			    "is_last": i == len(__snake_case      ) - 1,
																			    "target_size": target_size,
																			    "candidate_label": candidate_label,
																			    **text_inputs,
																			    **image_features,
																			}
					def  lowerCamelCase__(      self   :Tuple			,__snake_case   :Optional[Any]      )					->    List[str]:
												a__             =     model_inputs.pop('target_size'      )
												a__             =     model_inputs.pop('candidate_label'      )
												a__             =     model_inputs.pop('is_last'      )

												a__             =     self.model(**__snake_case      )

												a__             =     {'target_size': target_size, 'candidate_label': candidate_label, 'is_last': is_last, **outputs}
												return model_outputs
					def  lowerCamelCase__(      self   :List[str]			,__snake_case   :Any			,__snake_case   :Tuple=0.1			,__snake_case   :int=None      )					->    Optional[Any]:
												a__             =     []
												for model_output in model_outputs:
																			a__             =     model_output['candidate_label']
																			a__             =     BaseModelOutput(__snake_case      )
																			a__             =     self.image_processor.post_process_object_detection(
																			    outputs=__snake_case			,threshold=__snake_case			,target_sizes=model_output['target_size']      )[0]

																			for index in outputs["scores"].nonzero():
																										a__             =     outputs['scores'][index].item()
																										a__             =     self._get_bounding_box(outputs['boxes'][index][0]      )

																										a__             =     {'score': score, 'label': label, 'box': box}
																										results.append(__snake_case      )

												a__             =     sorted(__snake_case			,key=lambda __snake_case      : x["score"]			,reverse=__snake_case      )
												if top_k:
																			a__             =     results[:top_k]

												return results



					def  lowerCamelCase__(      self   :Optional[int]			,__snake_case   :"torch.Tensor"      )					->    Dict[str, int]:
												if self.framework != "pt":
																			raise ValueError('The ZeroShotObjectDetectionPipeline is only available in PyTorch.'      )
												a__      ,					a__      ,					a__      ,					a__             =     box.int().tolist()
												a__             =     {
												    'xmin': xmin,
												    'ymin': ymin,
												    'xmax': xmax,
												    'ymax': ymax,
												}
												return bbox








from abc import ABC, abstractmethod
from argparse import ArgumentParser







class        snake_case_   (lowerCamelCase_      ):
					@staticmethod
					@abstractmethod
					def  lowerCamelCase__(      __snake_case   :ArgumentParser      )					->    Dict:
												raise NotImplementedError()



					@abstractmethod
					def  lowerCamelCase__(      self   :Union[str, Any]      )					->    Dict:
												raise NotImplementedError()




