






import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available







class  _lowercase   (   _A						):
      _a							:						Optional[int]										=							['image_processor', 'tokenizer']
      _a							:						Tuple										=							'OwlViTImageProcessor'
      _a							:						Optional[int]										=							('CLIPTokenizer', 'CLIPTokenizerFast')
      def __init__(     self				,			a=None				,			a=None				,			**a   ):
       snake_case__  :    Dict							       =None
       if "feature_extractor" in kwargs:
        warnings.warn(
            """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
            """ instead."""				,			a				,			)
        snake_case__  :    List[Any]							       =kwargs.pop("""feature_extractor"""   )

       snake_case__  :    List[str]							       =image_processor if image_processor is not None else feature_extractor
       if image_processor is None:
        raise ValueError("""You need to specify an `image_processor`."""   )
       if tokenizer is None:
        raise ValueError("""You need to specify a `tokenizer`."""   )

       super().__init__(a				,			a   )
      def __call__(     self				,			a=None				,			a=None				,			a=None				,			a="max_length"				,			a="np"				,			**a   ):

       if text is None and query_images is None and images is None:
        raise ValueError(
            """You have to specify at least one text or query image or image. All three cannot be none."""   )

       if text is not None:
        if isinstance(a				,			a   ) or (isinstance(a				,			a   ) and not isinstance(text[0]				,			a   )):
         snake_case__  :    List[Any]							       =[self.tokenizer(a				,			padding=a				,			return_tensors=a				,			**a   )]

        elif isinstance(a				,			a   ) and isinstance(text[0]				,			a   ):
         snake_case__  :    Any							       =[]

         # Maximum number of queries across batch
         snake_case__  :    List[str]							       =max([len(a   ) for t in text]   )

         # Pad all batch samples to max number of text queries
         for t in text:
          if len(a   ) != max_num_queries:
           snake_case__  :    str							       =t + [""" """] * (max_num_queries - len(a   ))

          snake_case__  :    Optional[Any]							       =self.tokenizer(a				,			padding=a				,			return_tensors=a				,			**a   )
          encodings.append(a   )
        else:
         raise TypeError("""Input text should be a string, a list of strings or a nested list of strings"""   )

        if return_tensors == "np":
         snake_case__  :    Union[str, Any]							       =np.concatenate([encoding["""input_ids"""] for encoding in encodings]				,			axis=0   )
         snake_case__  :    Any							       =np.concatenate([encoding["""attention_mask"""] for encoding in encodings]				,			axis=0   )

        elif return_tensors == "jax" and is_flax_available():
         import jax.numpy as jnp

         snake_case__  :    int							       =jnp.concatenate([encoding["""input_ids"""] for encoding in encodings]				,			axis=0   )
         snake_case__  :    Any							       =jnp.concatenate([encoding["""attention_mask"""] for encoding in encodings]				,			axis=0   )

        elif return_tensors == "pt" and is_torch_available():
         import torch

         snake_case__  :    int							       =torch.cat([encoding["""input_ids"""] for encoding in encodings]				,			dim=0   )
         snake_case__  :    Dict							       =torch.cat([encoding["""attention_mask"""] for encoding in encodings]				,			dim=0   )

        elif return_tensors == "tf" and is_tf_available():
         import tensorflow as tf

         snake_case__  :    Dict							       =tf.stack([encoding["""input_ids"""] for encoding in encodings]				,			axis=0   )
         snake_case__  :    Optional[Any]							       =tf.stack([encoding["""attention_mask"""] for encoding in encodings]				,			axis=0   )

        else:
         raise ValueError("""Target return tensor type could not be returned"""   )

        snake_case__  :    List[str]							       =BatchEncoding()
        snake_case__  :    Union[str, Any]							       =input_ids
        snake_case__  :    Optional[int]							       =attention_mask

       if query_images is not None:
        snake_case__  :    str							       =BatchEncoding()
        snake_case__  :    Any							       =self.image_processor(
            a				,			return_tensors=a				,			**a   ).pixel_values
        snake_case__  :    int							       =query_pixel_values

       if images is not None:
        snake_case__  :    int							       =self.image_processor(a				,			return_tensors=a				,			**a   )

       if text is not None and images is not None:
        snake_case__  :    int							       =image_features.pixel_values
        return encoding
       elif query_images is not None and images is not None:
        snake_case__  :    List[Any]							       =image_features.pixel_values
        return encoding
       elif text is not None or query_images is not None:
        return encoding
       else:
        return BatchEncoding(data=dict(**a   )				,			tensor_type=a   )
      def 						lowercase__					(     self				,			*a				,			**a   ):
       return self.image_processor.post_process(*a				,			**a   )
      def 						lowercase__					(     self				,			*a				,			**a   ):
       return self.image_processor.post_process_object_detection(*a				,			**a   )
      def 						lowercase__					(     self				,			*a				,			**a   ):
       return self.image_processor.post_process_image_guided_detection(*a				,			**a   )
      def 						lowercase__					(     self				,			*a				,			**a   ):
       return self.tokenizer.batch_decode(*a				,			**a   )
      def 						lowercase__					(     self				,			*a				,			**a   ):
       return self.tokenizer.decode(*a				,			**a   )
      @property
      def 						lowercase__					(     self   ):
       warnings.warn(
           """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""				,			a				,			)
       return self.image_processor_class



      @property
      def 						lowercase__					(     self   ):
       warnings.warn(
           """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""				,			a				,			)
       return self.image_processor













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







@add_end_docstrings(_A						)
class  _lowercase   (   _A						):
      def __init__(     self				,			**a   ):
       super().__init__(**a   )

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

       requires_backends(self				,			"""vision"""   )
       self.check_model_type(a   )
      def __call__(     self				,			a				,			a = None				,			**a				,			):
       if "text_queries" in kwargs:
        snake_case__  :    Any							       =kwargs.pop("""text_queries"""   )

       if isinstance(a				,			(str, Image.Image)   ):
        snake_case__  :    Union[str, Any]							       ={"""image""": image, """candidate_labels""": candidate_labels}
       else:
        snake_case__  :    str							       =image
       snake_case__  :    Union[str, Any]							       =super().__call__(a				,			**a   )
       return results
      def 						lowercase__					(     self				,			**a   ):
       snake_case__  :    Union[str, Any]							       ={}
       if "threshold" in kwargs:
        snake_case__  :    Optional[Any]							       =kwargs["""threshold"""]
       if "top_k" in kwargs:
        snake_case__  :    Dict							       =kwargs["""top_k"""]
       return {}, {}, postprocess_params
      def 						lowercase__					(     self				,			a   ):
       snake_case__  :    List[Any]							       =load_image(inputs["""image"""]   )
       snake_case__  :    Dict							       =inputs["""candidate_labels"""]
       if isinstance(a				,			a   ):
        snake_case__  :    str							       =candidate_labels.split(""","""   )

       snake_case__  :    Optional[Any]							       =torch.tensor([[image.height, image.width]]				,			dtype=torch.intaa   )
       for i, candidate_label in enumerate(a   ):
        snake_case__  :    Tuple							       =self.tokenizer(a				,			return_tensors=self.framework   )
        snake_case__  :    int							       =self.image_processor(a				,			return_tensors=self.framework   )
        yield {
            "is_last": i == len(a   ) - 1,
            "target_size": target_size,
            "candidate_label": candidate_label,
            **text_inputs,
            **image_features,
        }
      def 						lowercase__					(     self				,			a   ):
       snake_case__  :    Tuple							       =model_inputs.pop("""target_size"""   )
       snake_case__  :    Tuple							       =model_inputs.pop("""candidate_label"""   )
       snake_case__  :    Dict							       =model_inputs.pop("""is_last"""   )

       snake_case__  :    Any							       =self.model(**a   )

       snake_case__  :    List[Any]							       ={"""target_size""": target_size, """candidate_label""": candidate_label, """is_last""": is_last, **outputs}
       return model_outputs
      def 						lowercase__					(     self				,			a				,			a=0.1				,			a=None   ):
       snake_case__  :    Optional[int]							       =[]
       for model_output in model_outputs:
        snake_case__  :    Tuple							       =model_output["""candidate_label"""]
        snake_case__  :    Optional[int]							       =BaseModelOutput(a   )
        snake_case__  :    int							       =self.image_processor.post_process_object_detection(
            outputs=a				,			threshold=a				,			target_sizes=model_output["""target_size"""]   )[0]

        for index in outputs["scores"].nonzero():
         snake_case__  :    int							       =outputs["""scores"""][index].item()
         snake_case__  :    Tuple							       =self._get_bounding_box(outputs["""boxes"""][index][0]   )

         snake_case__  :    Tuple							       ={"""score""": score, """label""": label, """box""": box}
         results.append(a   )

       snake_case__  :    Tuple							       =sorted(a				,			key=lambda a   : x["score"]				,			reverse=a   )
       if top_k:
        snake_case__  :    Union[str, Any]							       =results[:top_k]

       return results



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





