"""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      :			List[Any]							       =		logging.get_logger(__name__)







@add_end_docstrings(lowerCAmelCase_		)
class __snake_case    (						lowerCAmelCase_		):




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



     """simple docstring"""






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

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




   def      __a						(   self      :    Tuple	,     **_lowercase      :    Union[str, Any]				):



     """simple docstring"""






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




   def      __a						(   self      :    str	,     _lowercase      :    int				):



     """simple docstring"""






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

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



     """simple docstring"""






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

     SCREAMING_SNAKE_CASE__      							=				self.model(**__lowerCAmelCase				)

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




   def      __a						(   self      :    Union[str, Any]	,     _lowercase      :    List[str]	,     _lowercase      :    List[str]=0.1	,     _lowercase      :    Dict=None				):



     """simple docstring"""






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

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

         SCREAMING_SNAKE_CASE__      							=				{"""score""": score, """label""": label, """box""": box}
         results.append(__lowerCAmelCase				)

     SCREAMING_SNAKE_CASE__      							=				sorted(__lowerCAmelCase	,     key=lambda _lowercase				: x["score"]	,     reverse=__lowerCAmelCase				)
     if top_k:
       SCREAMING_SNAKE_CASE__      							=				results[:top_k]

     return results




   def      __a						(   self      :    List[str]	,     _lowercase      :    "torch.Tensor"				):



     """simple docstring"""






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

import PIL.Image
import PIL.ImageOps
from packaging import version
from PIL import Image


if version.parse(version.parse(PIL.__version__).base_version) >= version.parse('''9.1.0'''):
   __lowerCamelCase      :			Optional[Any]							       =		{
       '''linear''': PIL.Image.Resampling.BILINEAR,
       '''bilinear''': PIL.Image.Resampling.BILINEAR,
       '''bicubic''': PIL.Image.Resampling.BICUBIC,
       '''lanczos''': PIL.Image.Resampling.LANCZOS,
       '''nearest''': PIL.Image.Resampling.NEAREST,
   }
else:
   __lowerCamelCase      :			int							       =		{
       '''linear''': PIL.Image.LINEAR,
       '''bilinear''': PIL.Image.BILINEAR,
       '''bicubic''': PIL.Image.BICUBIC,
       '''lanczos''': PIL.Image.LANCZOS,
       '''nearest''': PIL.Image.NEAREST,
   }



def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	int     )							-> Optional[Any]:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__      							=				(images / 2 + 0.5).clamp(0			,					1     )
  SCREAMING_SNAKE_CASE__      							=				images.cpu().permute(0			,					2			,					3			,					1     ).float().numpy()
  SCREAMING_SNAKE_CASE__      							=				numpy_to_pil(__UpperCamelCase     )
  return images






def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	int     )							-> int:







  """simple docstring"""





  if images.ndim == 3:
    SCREAMING_SNAKE_CASE__      							=				images[None, ...]
  SCREAMING_SNAKE_CASE__      							=				(images * 2_55).round().astype("""uint8"""     )
  if images.shape[-1] == 1:
    # special case for grayscale (single channel) images
    SCREAMING_SNAKE_CASE__      							=				[Image.fromarray(image.squeeze()			,					mode="""L"""     ) for image in images]
  else:
    SCREAMING_SNAKE_CASE__      							=				[Image.fromarray(__UpperCamelCase     ) for image in images]

  return pil_images
