





"""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

_snake_case          =						logging.get_logger(__name__)
@add_end_docstrings(__UpperCAmelCase							)
class        _SCREAMING_SNAKE_CASE  (  __UpperCAmelCase							):






				'''simple docstring'''



				def __init__(					self	:			Tuple				,     **UpperCAmelCase_	:			Any  )		->  List[Any]:







											"""simple docstring"""




											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	:			int				,     UpperCAmelCase_	:			Union[str, "Image.Image", List[Dict[str, Any]]]				,     UpperCAmelCase_	:			Union[str, List[str]] = None				,     **UpperCAmelCase_	:			List[Any]				,     )		->  Optional[Any]:







											"""simple docstring"""




											if "text_queries" in kwargs:
																		_lowerCAmelCase  	=			kwargs.pop('text_queries'  )

											if isinstance(_lowerCamelCase				,     (str, Image.Image)  ):
																		_lowerCAmelCase  	=			{'image': image, 'candidate_labels': candidate_labels}
											else:
																		_lowerCAmelCase  	=			image
											_lowerCAmelCase  	=			super().__call__(_lowerCamelCase				,     **_lowerCamelCase  )
											return results


				def 				__lowerCamelCase	(					self	:			List[Any]				,     **UpperCAmelCase_	:			Dict  )		->  Dict:







											"""simple docstring"""




											_lowerCAmelCase  	=			{}
											if "threshold" in kwargs:
																		_lowerCAmelCase  	=			kwargs['threshold']
											if "top_k" in kwargs:
																		_lowerCAmelCase  	=			kwargs['top_k']
											return {}, {}, postprocess_params


				def 				__lowerCamelCase	(					self	:			str				,     UpperCAmelCase_	:			List[str]  )		->  int:







											"""simple docstring"""




											_lowerCAmelCase  	=			load_image(inputs['image']  )
											_lowerCAmelCase  	=			inputs['candidate_labels']
											if isinstance(_lowerCamelCase				,     _lowerCamelCase  ):
																		_lowerCAmelCase  	=			candidate_labels.split(','  )

											_lowerCAmelCase  	=			torch.tensor([[image.height, image.width]]				,     dtype=torch.intaa  )
											for i, candidate_label in enumerate(_lowerCamelCase  ):
																		_lowerCAmelCase  	=			self.tokenizer(_lowerCamelCase				,     return_tensors=self.framework  )
																		_lowerCAmelCase  	=			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 				__lowerCamelCase	(					self	:			str				,     UpperCAmelCase_	:			Optional[Any]  )		->  Dict:







											"""simple docstring"""




											_lowerCAmelCase  	=			model_inputs.pop('target_size'  )
											_lowerCAmelCase  	=			model_inputs.pop('candidate_label'  )
											_lowerCAmelCase  	=			model_inputs.pop('is_last'  )

											_lowerCAmelCase  	=			self.model(**_lowerCamelCase  )

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


				def 				__lowerCamelCase	(					self	:			Union[str, Any]				,     UpperCAmelCase_	:			Dict				,     UpperCAmelCase_	:			Dict=0.1				,     UpperCAmelCase_	:			List[Any]=None  )		->  str:







											"""simple docstring"""




											_lowerCAmelCase  	=			[]
											for model_output in model_outputs:
																		_lowerCAmelCase  	=			model_output['candidate_label']
																		_lowerCAmelCase  	=			BaseModelOutput(_lowerCamelCase  )
																		_lowerCAmelCase  	=			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  	=			outputs['scores'][index].item()
																									_lowerCAmelCase  	=			self._get_bounding_box(outputs['boxes'][index][0]  )

																									_lowerCAmelCase  	=			{'score': score, 'label': label, 'box': box}
																									results.append(_lowerCamelCase  )

											_lowerCAmelCase  	=			sorted(_lowerCamelCase				,     key=lambda UpperCAmelCase_  : x["score"]				,     reverse=_lowerCamelCase  )
											if top_k:
																		_lowerCAmelCase  	=			results[:top_k]

											return results







				def 				__lowerCamelCase	(					self	:			Optional[Any]				,     UpperCAmelCase_	:			"torch.Tensor"  )		->  List[Any]:







											"""simple docstring"""




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








"""simple docstring"""



import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import ClassLabel, Features, Image
from .base import TaskTemplate
@dataclass(frozen=UpperCAmelCase							)
class        _SCREAMING_SNAKE_CASE  (  UpperCAmelCase							):






				'''simple docstring'''



				SCREAMING_SNAKE_CASE_:							str							=					field(default="image-classification"      ,				metadata={"include_in_asdict_even_if_is_default": True}							)
				SCREAMING_SNAKE_CASE_:							ClassVar[Features]							=					Features({"image": Image()}							)
				SCREAMING_SNAKE_CASE_:							ClassVar[Features]							=					Features({"labels": ClassLabel}							)
				SCREAMING_SNAKE_CASE_:							str							=					"image"
				SCREAMING_SNAKE_CASE_:							str							=					"labels"


				def 				__lowerCamelCase	(					self	:			str				,     UpperCAmelCase_	:			Optional[Any]  )		->  Dict:







											"""simple docstring"""




											if self.label_column not in features:
																		raise ValueError(F"""Column {self.label_column} is not present in features."""  )
											if not isinstance(features[self.label_column]				,     UpperCAmelCase_  ):
																		raise ValueError(F"""Column {self.label_column} is not a ClassLabel."""  )
											_lowerCAmelCase  	=			copy.deepcopy(self  )
											_lowerCAmelCase  	=			self.label_schema.copy()
											_lowerCAmelCase  	=			features[self.label_column]
											_lowerCAmelCase  	=			label_schema
											return task_template







				@property
				def 				__lowerCamelCase	(					self	:			Union[str, Any]  )		->  Dict[str, str]:







											"""simple docstring"""




											return {
											    self.image_column: "image",
											    self.label_column: "labels",
											}






