

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

__lowerCamelCase										=			logging.get_logger(__name__)


@add_end_docstrings(__SCREAMING_SNAKE_CASE	)
class     A__					(					__SCREAMING_SNAKE_CASE	):



    def __init__(    self						, **UpperCamelCase__				)  ->					Any:





           '''simple docstring'''
           super().__init__(**UpperCAmelCase_				)

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

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



    def __call__(    self						, UpperCamelCase__						, UpperCamelCase__ = None						, **UpperCamelCase__						, )  ->					Union[str, Any]:





           '''simple docstring'''



           if "text_queries" in kwargs:
                  A_		    =					kwargs.pop("""text_queries"""				)

           if isinstance(UpperCAmelCase_						, (str, Image.Image)				):
                  A_		    =					{"image": image, "candidate_labels": candidate_labels}
           else:
                  A_		    =					image
           A_		    =					super().__call__(UpperCAmelCase_						, **UpperCAmelCase_				)
           return results



    def     snake_case_	(    self						, **UpperCamelCase__				)  ->					Dict:





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



    def     snake_case_	(    self						, UpperCamelCase__				)  ->					Union[str, Any]:





           '''simple docstring'''
           A_		    =					load_image(inputs["""image"""]				)
           A_		    =					inputs["candidate_labels"]
           if isinstance(UpperCAmelCase_						, UpperCAmelCase_				):
                  A_		    =					candidate_labels.split(""","""				)

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



    def     snake_case_	(    self						, UpperCamelCase__				)  ->					Optional[int]:





           '''simple docstring'''
           A_		    =					model_inputs.pop("""target_size"""				)
           A_		    =					model_inputs.pop("""candidate_label"""				)
           A_		    =					model_inputs.pop("""is_last"""				)

           A_		    =					self.model(**UpperCAmelCase_				)

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



    def     snake_case_	(    self						, UpperCamelCase__						, UpperCamelCase__=0.1						, UpperCamelCase__=None				)  ->					Tuple:





           '''simple docstring'''
           A_		    =					[]
           for model_output in model_outputs:
                  A_		    =					model_output["candidate_label"]
                  A_		    =					BaseModelOutput(UpperCAmelCase_				)
                  A_		    =					self.image_processor.post_process_object_detection(
                      outputs=UpperCAmelCase_						, threshold=UpperCAmelCase_						, 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(UpperCAmelCase_				)

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

           return results







    def     snake_case_	(    self						, UpperCamelCase__				)  ->					Dict[str, int]:





           '''simple docstring'''



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







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

from ..features import ClassLabel, Features, Image
from .base import TaskTemplate



@dataclass(frozen=__SCREAMING_SNAKE_CASE    )
class    _SCREAMING_SNAKE_CASE						(	__SCREAMING_SNAKE_CASE    ):







 '''simple docstring'''
 lowercase_		=			field(default="image-classification"    ,							metadata={"include_in_asdict_even_if_is_default": True}    )
 lowercase_		=			Features({"image": Image()}    )
 lowercase_		=			Features({"labels": ClassLabel}    )
 lowercase_		=			"image"
 lowercase_		=			"labels"

 def 							SCREAMING_SNAKE_CASE_					(self				:   Tuple       , UpperCAmelCase_				:   Union[str, Any])       ->Tuple:



   '''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__:      List[Any]			  =copy.deepcopy(self)
   lowerCamelCase__:      Optional[int]			  =self.label_schema.copy()
   lowerCamelCase__:      int			  =features[self.label_column]
   lowerCamelCase__:      int			  =label_schema
   return task_template







 @property
 def 							SCREAMING_SNAKE_CASE_					(self				:   Dict)       ->Dict[str, str]:



   '''simple docstring'''



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

