





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             =							logging.get_logger(__name__)






@add_end_docstrings(__lowercase						)
class        snake_case_    (		__lowercase						):






  def __init__(							self			:					Any       ,  **_snake_case			:					List[str]      )->List[str]:






   '''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			:					Tuple       ,  _snake_case			:					List[str]       ,  _snake_case			:					int = None       ,  **_snake_case			:					Union[str, Any]       ,  )->int:






   '''simple docstring'''

   if "text_queries" in kwargs:
    __lowerCAmelCase					:	List[Any]      =       kwargs.pop("""text_queries"""      )

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






  def 				UpperCAmelCase__						(							self			:					Any       ,  **_snake_case			:					Union[str, Any]      )->Any:






   '''simple docstring'''

   __lowerCAmelCase					:	Tuple      =       {}
   if "threshold" in kwargs:
    __lowerCAmelCase					:	str      =       kwargs["""threshold"""]
   if "top_k" in kwargs:
    __lowerCAmelCase					:	int      =       kwargs["""top_k"""]
   return {}, {}, postprocess_params






  def 				UpperCAmelCase__						(							self			:					Union[str, Any]       ,  _snake_case			:					List[Any]      )->int:






   '''simple docstring'''

   __lowerCAmelCase					:	Tuple      =       load_image(inputs["""image"""]      )
   __lowerCAmelCase					:	Tuple      =       inputs["""candidate_labels"""]
   if isinstance(lowerCamelCase_       ,  lowerCamelCase_      ):
    __lowerCAmelCase					:	Union[str, Any]      =       candidate_labels.split(""","""      )

   __lowerCAmelCase					:	Union[str, Any]      =       torch.tensor([[image.height, image.width]]       ,  dtype=torch.intaa      )
   for i, candidate_label in enumerate(lowerCamelCase_      ):
    __lowerCAmelCase					:	Dict      =       self.tokenizer(lowerCamelCase_       ,  return_tensors=self.framework      )
    __lowerCAmelCase					:	str      =       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			:					Dict       ,  _snake_case			:					Union[str, Any]      )->Tuple:






   '''simple docstring'''

   __lowerCAmelCase					:	Dict      =       model_inputs.pop("""target_size"""      )
   __lowerCAmelCase					:	Union[str, Any]      =       model_inputs.pop("""candidate_label"""      )
   __lowerCAmelCase					:	Any      =       model_inputs.pop("""is_last"""      )

   __lowerCAmelCase					:	Any      =       self.model(**lowerCamelCase_      )

   __lowerCAmelCase					:	List[str]      =       {"""target_size""": target_size, """candidate_label""": candidate_label, """is_last""": is_last, **outputs}
   return model_outputs






  def 				UpperCAmelCase__						(							self			:					Union[str, Any]       ,  _snake_case			:					int       ,  _snake_case			:					str=0.1       ,  _snake_case			:					Tuple=None      )->str:






   '''simple docstring'''

   __lowerCAmelCase					:	Dict      =       []
   for model_output in model_outputs:
    __lowerCAmelCase					:	Union[str, Any]      =       model_output["""candidate_label"""]
    __lowerCAmelCase					:	List[str]      =       BaseModelOutput(lowerCamelCase_      )
    __lowerCAmelCase					:	List[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					:	str      =       outputs["""scores"""][index].item()
     __lowerCAmelCase					:	Union[str, Any]      =       self._get_bounding_box(outputs["""boxes"""][index][0]      )

     __lowerCAmelCase					:	List[str]      =       {"""score""": score, """label""": label, """box""": box}
     results.append(lowerCamelCase_      )

   __lowerCAmelCase					:	str      =       sorted(lowerCamelCase_       ,  key=lambda _snake_case      : x["score"]       ,  reverse=lowerCamelCase_      )
   if top_k:
    __lowerCAmelCase					:	Union[str, Any]      =       results[:top_k]

   return results






  def 				UpperCAmelCase__						(							self			:					List[str]       ,  _snake_case			:					Optional[int]      )->Dict[str, int]:






   '''simple docstring'''

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






def 	_SCREAMING_SNAKE_CASE		(      SCREAMING_SNAKE_CASE		:list[int]   )      ->   int:
 if not numbers:
  return 0

 if not isinstance(SCREAMING_SNAKE_CASE   ,   (list, tuple)   ) or not all(
     isinstance(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   ) for number in numbers   ):
  raise ValueError("""numbers must be an iterable of integers"""   )

 __lowerCAmelCase					:	int      =       numbers[0]

 for i in range(1   ,   len(SCREAMING_SNAKE_CASE   )   ):
  # update the maximum and minimum subarray products
  __lowerCAmelCase					:	List[str]      =       numbers[i]
  if number < 0:
   __lowerCAmelCase ,					__lowerCAmelCase					:	Union[str, Any]      =       min_till_now, max_till_now
  __lowerCAmelCase					:	Optional[int]      =       max(SCREAMING_SNAKE_CASE   ,   max_till_now * number   )
  __lowerCAmelCase					:	List[Any]      =       min(SCREAMING_SNAKE_CASE   ,   min_till_now * number   )

  # update the maximum product found till now
  __lowerCAmelCase					:	List[str]      =       max(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   )

 return max_prod