



"""simple docstring"""
from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import rescale, resize, to_channel_dimension_format
from ...image_utils import (
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
       import PIL


UpperCAmelCase						:		Optional[Any]			= logging.get_logger(__name__)


def 			_SCREAMING_SNAKE_CASE						(__lowerCAmelCase ,  __lowerCAmelCase							) ->						Any:







  '''simple docstring'''


  lowercase_								= b.T
  lowercase_								= np.sum(np.square(__snake_case							) ,  axis=1							)
  lowercase_								= np.sum(np.square(__snake_case							) ,  axis=0							)
  lowercase_								= np.matmul(__snake_case ,  __snake_case							)
  lowercase_								= aa[:, None] - 2 * ab + ba[None, :]
  return d


def 			_SCREAMING_SNAKE_CASE						(__lowerCAmelCase ,  __lowerCAmelCase							) ->						Any:







  '''simple docstring'''


  lowercase_								= x.reshape(-1 ,  3							)
  lowercase_								= squared_euclidean_distance(__snake_case ,  __snake_case							)
  return np.argmin(__snake_case ,  axis=1							)


class    SCREAMING_SNAKE_CASE__ (							_A   ):
    lowercase__      =   ["pixel_values"]

    def __init__(				self			:				int ,  lowerCAmelCase_			:				Union[str, Any] = None ,  lowerCAmelCase_			:				List[Any] = True ,  lowerCAmelCase_			:				Tuple = None ,  lowerCAmelCase_			:				Tuple = PILImageResampling.BILINEAR ,  lowerCAmelCase_			:				List[str] = True ,  lowerCAmelCase_			:				Tuple = True ,  **lowerCAmelCase_			:				Union[str, Any] ,  ):
      super().__init__(**lowerCAmelCase_)
      lowercase_								= size if size is not None else {"height": 2_5_6, "width": 2_5_6}
      lowercase_								= get_size_dict(lowerCAmelCase_)
      lowercase_								= np.array(lowerCAmelCase_) if clusters is not None else None
      lowercase_								= do_resize
      lowercase_								= size
      lowercase_								= resample
      lowercase_								= do_normalize
      lowercase_								= do_color_quantize

    def 					_UpperCAmelCase	(				self			:				Tuple ,  lowerCAmelCase_			:				List[Any] ,  lowerCAmelCase_			:				Dict ,  lowerCAmelCase_			:				Union[str, Any] = PILImageResampling.BILINEAR ,  lowerCAmelCase_			:				List[str] = None ,  **lowerCAmelCase_			:				List[str] ,  ):
      lowercase_								= get_size_dict(lowerCAmelCase_)
      if "height" not in size or "width" not in size:
        raise ValueError(F'''Size dictionary must contain both height and width keys. Got {size.keys()}''')
      return resize(
          lowerCAmelCase_ ,  size=(size["""height"""], size["""width"""]) ,  resample=lowerCAmelCase_ ,  data_format=lowerCAmelCase_ ,  **lowerCAmelCase_)

    def 					_UpperCAmelCase	(				self			:				Tuple ,  lowerCAmelCase_			:				Dict ,  lowerCAmelCase_			:				List[Any] = None ,  ):
      lowercase_								= rescale(image=lowerCAmelCase_ ,  scale=1 / 127.5 ,  data_format=lowerCAmelCase_)
      lowercase_								= image - 1
      return image




    def 					_UpperCAmelCase	(				self			:				List[str] ,  lowerCAmelCase_			:				Any ,  lowerCAmelCase_			:				Any = None ,  lowerCAmelCase_			:				Optional[int] = None ,  lowerCAmelCase_			:				Dict = None ,  lowerCAmelCase_			:				str = None ,  lowerCAmelCase_			:				int = None ,  lowerCAmelCase_			:				Tuple = None ,  lowerCAmelCase_			:				int = None ,  lowerCAmelCase_			:				Optional[int] = ChannelDimension.FIRST ,  **lowerCAmelCase_			:				Any ,  ):
      lowercase_								= do_resize if do_resize is not None else self.do_resize
      lowercase_								= size if size is not None else self.size
      lowercase_								= get_size_dict(lowerCAmelCase_)
      lowercase_								= resample if resample is not None else self.resample
      lowercase_								= do_normalize if do_normalize is not None else self.do_normalize
      lowercase_								= do_color_quantize if do_color_quantize is not None else self.do_color_quantize
      lowercase_								= clusters if clusters is not None else self.clusters
      lowercase_								= np.array(lowerCAmelCase_)

      lowercase_								= make_list_of_images(lowerCAmelCase_)

      if not valid_images(lowerCAmelCase_):
        raise ValueError(
            """Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, """
            """torch.Tensor, tf.Tensor or jax.ndarray.""")

      if do_resize and size is None or resample is None:
        raise ValueError("""Size and resample must be specified if do_resize is True.""")

      if do_color_quantize and clusters is None:
        raise ValueError("""Clusters must be specified if do_color_quantize is True.""")

      # All transformations expect numpy arrays.
      lowercase_								= [to_numpy_array(lowerCAmelCase_) for image in images]

      if do_resize:
        lowercase_								= [self.resize(image=lowerCAmelCase_ ,  size=lowerCAmelCase_ ,  resample=lowerCAmelCase_) for image in images]

      if do_normalize:
        lowercase_								= [self.normalize(image=lowerCAmelCase_) for image in images]

      if do_color_quantize:
        lowercase_								= [to_channel_dimension_format(lowerCAmelCase_ ,  ChannelDimension.LAST) for image in images]
        # color quantize from (batch_size, height, width, 3) to (batch_size, height, width)
        lowercase_								= np.array(lowerCAmelCase_)
        lowercase_								= color_quantize(lowerCAmelCase_ ,  lowerCAmelCase_).reshape(images.shape[:-1])

        # flatten to (batch_size, height*width)
        lowercase_								= images.shape[0]
        lowercase_								= images.reshape(lowerCAmelCase_ ,  -1)

        # We need to convert back to a list of images to keep consistent behaviour across processors.
        lowercase_								= list(lowerCAmelCase_)
      else:
        lowercase_								= [to_channel_dimension_format(lowerCAmelCase_ ,  lowerCAmelCase_) for image in images]

      lowercase_								= {"input_ids": images}
      return BatchFeature(data=lowerCAmelCase_ ,  tensor_type=lowerCAmelCase_)






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

UpperCAmelCase						:		Optional[Any]			= logging.get_logger(__name__)


@add_end_docstrings(__UpperCAmelCase   )
class    SCREAMING_SNAKE_CASE__ (							__UpperCAmelCase   ):

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







      """simple docstring"""







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

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

    def 					_UpperCAmelCase	(				self			:				Optional[Any] ,  **lowerCAmelCase_			:				Optional[Any]):







      """simple docstring"""







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

    def 					_UpperCAmelCase	(				self			:				int ,  lowerCAmelCase_			:				List[Any]):







      """simple docstring"""







      lowercase_								= load_image(inputs["""image"""])
      lowercase_								= inputs["""candidate_labels"""]
      if isinstance(lowerCAmelCase_ ,  lowerCAmelCase_):
        lowercase_								= candidate_labels.split(""",""")

      lowercase_								= torch.tensor([[image.height, image.width]] ,  dtype=torch.intaa)
      for i, candidate_label in enumerate(lowerCAmelCase_):
        lowercase_								= self.tokenizer(lowerCAmelCase_ ,  return_tensors=self.framework)
        lowercase_								= 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			:				List[Any] ,  lowerCAmelCase_			:				List[Any]):







      """simple docstring"""







      lowercase_								= model_inputs.pop("""target_size""")
      lowercase_								= model_inputs.pop("""candidate_label""")
      lowercase_								= model_inputs.pop("""is_last""")

      lowercase_								= self.model(**lowerCAmelCase_)

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

    def 					_UpperCAmelCase	(				self			:				int ,  lowerCAmelCase_			:				Optional[int] ,  lowerCAmelCase_			:				Tuple=0.1 ,  lowerCAmelCase_			:				List[str]=None):







      """simple docstring"""







      lowercase_								= []
      for model_output in model_outputs:
        lowercase_								= model_output["""candidate_label"""]
        lowercase_								= BaseModelOutput(lowerCAmelCase_)
        lowercase_								= self.image_processor.post_process_object_detection(
            outputs=lowerCAmelCase_ ,  threshold=lowerCAmelCase_ ,  target_sizes=model_output["""target_size"""])[0]

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

          lowercase_								= {"""score""": score, """label""": label, """box""": box}
          results.append(lowerCAmelCase_)

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

      return results




    def 					_UpperCAmelCase	(				self			:				Union[str, Any] ,  lowerCAmelCase_			:				"torch.Tensor"):







      """simple docstring"""







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

