





import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...models.auto.modeling_auto import MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
from ...utils import logging
from ..auto import CONFIG_MAPPING


__A									=				logging.get_logger(__name__)

__A									=				{
    "salesforce/blip2-opt-2.7b": "https://huggingface.co/salesforce/blip2-opt-2.7b/resolve/main/config.json",
}



class    _SCREAMING_SNAKE_CASE						(	__SCREAMING_SNAKE_CASE    ):







 '''simple docstring'''
 lowercase_		=			"blip_2_vision_model"

 def __init__(self				:   Union[str, Any]       , UpperCAmelCase_				:   Optional[Any]=1_408       , UpperCAmelCase_				:   int=6_144       , UpperCAmelCase_				:   Any=39       , UpperCAmelCase_				:   Optional[int]=16       , UpperCAmelCase_				:   Union[str, Any]=224       , UpperCAmelCase_				:   Union[str, Any]=14       , UpperCAmelCase_				:   Any="gelu"       , UpperCAmelCase_				:   Any=0.0_0001       , UpperCAmelCase_				:   Optional[Any]=0.0       , UpperCAmelCase_				:   Dict=1E-1_0       , UpperCAmelCase_				:   List[Any]=True       , **UpperCAmelCase_				:   int       , )       ->Union[str, Any]:



   '''simple docstring'''



   super().__init__(**UpperCAmelCase_)

   lowerCamelCase__:      Tuple			  =hidden_size
   lowerCamelCase__:      Tuple			  =intermediate_size
   lowerCamelCase__:      Optional[Any]			  =num_hidden_layers
   lowerCamelCase__:      int			  =num_attention_heads
   lowerCamelCase__:      Any			  =patch_size
   lowerCamelCase__:      Dict			  =image_size
   lowerCamelCase__:      int			  =initializer_range
   lowerCamelCase__:      Optional[Any]			  =attention_dropout
   lowerCamelCase__:      List[str]			  =layer_norm_eps
   lowerCamelCase__:      Union[str, Any]			  =hidden_act
   lowerCamelCase__:      str			  =qkv_bias







 @classmethod
 def 							SCREAMING_SNAKE_CASE_					(cls				:   List[Any]       , UpperCAmelCase_				:   Union[str, os.PathLike]       , **UpperCAmelCase_				:   List[Any])       ->"PretrainedConfig":



   '''simple docstring'''



   cls._set_token_in_kwargs(UpperCAmelCase_)

   lowerCamelCase__  ,					lowerCamelCase__:      Dict			  =cls.get_config_dict(UpperCAmelCase_       , **UpperCAmelCase_)

   # get the vision config dict if we are loading from Blip2Config
   if config_dict.get("model_type") == "blip-2":
     lowerCamelCase__:      Optional[int]			  =config_dict["vision_config"]

   if "model_type" in config_dict and hasattr(cls       , "model_type") and config_dict["model_type"] != cls.model_type:
     logger.warning(
         F"""You are using a model of type {config_dict["model_type"]} to instantiate a model of type """
         F"""{cls.model_type}. This is not supported for all configurations of models and can yield errors.""")

   return cls.from_dict(UpperCAmelCase_       , **UpperCAmelCase_)



class    _SCREAMING_SNAKE_CASE						(	__SCREAMING_SNAKE_CASE    ):







 '''simple docstring'''
 lowercase_		=			"blip_2_qformer"

 def __init__(self				:   Union[str, Any]       , UpperCAmelCase_				:   str=30_522       , UpperCAmelCase_				:   Tuple=768       , UpperCAmelCase_				:   str=12       , UpperCAmelCase_				:   Optional[Any]=12       , UpperCAmelCase_				:   Optional[Any]=3_072       , UpperCAmelCase_				:   Union[str, Any]="gelu"       , UpperCAmelCase_				:   List[Any]=0.1       , UpperCAmelCase_				:   Optional[int]=0.1       , UpperCAmelCase_				:   Tuple=512       , UpperCAmelCase_				:   Any=0.02       , UpperCAmelCase_				:   Optional[int]=1E-1_2       , UpperCAmelCase_				:   List[str]=0       , UpperCAmelCase_				:   Optional[int]="absolute"       , UpperCAmelCase_				:   str=2       , UpperCAmelCase_				:   Optional[Any]=1_408       , **UpperCAmelCase_				:   Any       , )       ->str:



   '''simple docstring'''



   super().__init__(pad_token_id=UpperCAmelCase_       , **UpperCAmelCase_)

   lowerCamelCase__:      Union[str, Any]			  =vocab_size
   lowerCamelCase__:      Optional[int]			  =hidden_size
   lowerCamelCase__:      Any			  =num_hidden_layers
   lowerCamelCase__:      Optional[int]			  =num_attention_heads
   lowerCamelCase__:      str			  =hidden_act
   lowerCamelCase__:      List[Any]			  =intermediate_size
   lowerCamelCase__:      List[str]			  =hidden_dropout_prob
   lowerCamelCase__:      int			  =attention_probs_dropout_prob
   lowerCamelCase__:      int			  =max_position_embeddings
   lowerCamelCase__:      Any			  =initializer_range
   lowerCamelCase__:      str			  =layer_norm_eps
   lowerCamelCase__:      Any			  =position_embedding_type
   lowerCamelCase__:      List[Any]			  =cross_attention_frequency
   lowerCamelCase__:      Union[str, Any]			  =encoder_hidden_size







 @classmethod
 def 							SCREAMING_SNAKE_CASE_					(cls				:   Optional[int]       , UpperCAmelCase_				:   Union[str, os.PathLike]       , **UpperCAmelCase_				:   Optional[Any])       ->"PretrainedConfig":



   '''simple docstring'''



   cls._set_token_in_kwargs(UpperCAmelCase_)

   lowerCamelCase__  ,					lowerCamelCase__:      Union[str, Any]			  =cls.get_config_dict(UpperCAmelCase_       , **UpperCAmelCase_)

   # get the qformer config dict if we are loading from Blip2Config
   if config_dict.get("model_type") == "blip-2":
     lowerCamelCase__:      Tuple			  =config_dict["qformer_config"]

   if "model_type" in config_dict and hasattr(cls       , "model_type") and config_dict["model_type"] != cls.model_type:
     logger.warning(
         F"""You are using a model of type {config_dict["model_type"]} to instantiate a model of type """
         F"""{cls.model_type}. This is not supported for all configurations of models and can yield errors.""")

   return cls.from_dict(UpperCAmelCase_       , **UpperCAmelCase_)



class    _SCREAMING_SNAKE_CASE						(	__SCREAMING_SNAKE_CASE    ):







 '''simple docstring'''
 lowercase_		=			"blip-2"
 lowercase_		=			True

 def __init__(self				:   Dict       , UpperCAmelCase_				:   Any=None       , UpperCAmelCase_				:   str=None       , UpperCAmelCase_				:   Optional[Any]=None       , UpperCAmelCase_				:   Any=32       , **UpperCAmelCase_				:   str)       ->int:



   '''simple docstring'''



   super().__init__(**UpperCAmelCase_)

   if vision_config is None:
     lowerCamelCase__:      Any			  ={}
     logger.info("vision_config is None. initializing the Blip2VisionConfig with default values.")

   if qformer_config is None:
     lowerCamelCase__:      str			  ={}
     logger.info("qformer_config is None. Initializing the Blip2QFormerConfig with default values.")

   if text_config is None:
     lowerCamelCase__:      Union[str, Any]			  ={}
     logger.info("text_config is None. Initializing the text config with default values (`OPTConfig`).")

   lowerCamelCase__:      Dict			  =BlipaVisionConfig(**UpperCAmelCase_)
   lowerCamelCase__:      List[str]			  =BlipaQFormerConfig(**UpperCAmelCase_)
   lowerCamelCase__:      Dict			  =text_config["model_type"] if "model_type" in text_config else "opt"
   lowerCamelCase__:      List[str]			  =CONFIG_MAPPING[text_model_type](**UpperCAmelCase_)

   lowerCamelCase__:      Optional[Any]			  =self.text_config.tie_word_embeddings
   lowerCamelCase__:      Optional[Any]			  =self.text_config.is_encoder_decoder

   lowerCamelCase__:      Any			  =num_query_tokens
   lowerCamelCase__:      List[Any]			  =self.vision_config.hidden_size
   lowerCamelCase__:      Optional[Any]			  =self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
   lowerCamelCase__:      List[Any]			  =1.0
   lowerCamelCase__:      Dict			  =0.02

 @classmethod
 def 							SCREAMING_SNAKE_CASE_					(cls				:   List[str]       , UpperCAmelCase_				:   BlipaVisionConfig       , UpperCAmelCase_				:   BlipaQFormerConfig       , UpperCAmelCase_				:   PretrainedConfig       , **UpperCAmelCase_				:   Dict       , )       ->int:



   '''simple docstring'''



   return cls(
       vision_config=vision_config.to_dict()       , qformer_config=qformer_config.to_dict()       , text_config=text_config.to_dict()       , **UpperCAmelCase_       , )







 def 							SCREAMING_SNAKE_CASE_					(self				:   List[Any])       ->List[str]:



   '''simple docstring'''



   lowerCamelCase__:      List[Any]			  =copy.deepcopy(self.__dict__)
   lowerCamelCase__:      Optional[int]			  =self.vision_config.to_dict()
   lowerCamelCase__:      List[Any]			  =self.qformer_config.to_dict()
   lowerCamelCase__:      Any			  =self.text_config.to_dict()
   lowerCamelCase__:      Union[str, Any]			  =self.__class__.model_type
   return output








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

__A									=				logging.get_logger(__name__)



@add_end_docstrings(__SCREAMING_SNAKE_CASE    )
class    _SCREAMING_SNAKE_CASE						(	__SCREAMING_SNAKE_CASE    ):







 '''simple docstring'''

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



   '''simple docstring'''



   if "text_queries" in kwargs:
     lowerCamelCase__:      Any			  =kwargs.pop("text_queries")

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

 def 							SCREAMING_SNAKE_CASE_					(self				:   Optional[int]       , **UpperCAmelCase_				:   Union[str, Any])       ->Dict:



   '''simple docstring'''



   lowerCamelCase__:      List[str]			  ={}
   if "threshold" in kwargs:
     lowerCamelCase__:      List[Any]			  =kwargs["threshold"]
   if "top_k" in kwargs:
     lowerCamelCase__:      Any			  =kwargs["top_k"]
   return {}, {}, postprocess_params

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



   '''simple docstring'''



   lowerCamelCase__:      List[Any]			  =load_image(inputs["image"])
   lowerCamelCase__:      Dict			  =inputs["candidate_labels"]
   if isinstance(UpperCAmelCase_       , UpperCAmelCase_):
     lowerCamelCase__:      Any			  =candidate_labels.split(",")

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



   '''simple docstring'''



   lowerCamelCase__:      Dict			  =model_inputs.pop("target_size")
   lowerCamelCase__:      Dict			  =model_inputs.pop("candidate_label")
   lowerCamelCase__:      Dict			  =model_inputs.pop("is_last")

   lowerCamelCase__:      Union[str, Any]			  =self.model(**UpperCAmelCase_)

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

 def 							SCREAMING_SNAKE_CASE_					(self				:   Optional[int]       , UpperCAmelCase_				:   int       , UpperCAmelCase_				:   Any=0.1       , UpperCAmelCase_				:   str=None)       ->Tuple:



   '''simple docstring'''



   lowerCamelCase__:      Union[str, Any]			  =[]
   for model_output in model_outputs:
     lowerCamelCase__:      Optional[Any]			  =model_output["candidate_label"]
     lowerCamelCase__:      Tuple			  =BaseModelOutput(UpperCAmelCase_)
     lowerCamelCase__:      Dict			  =self.image_processor.post_process_object_detection(
         outputs=UpperCAmelCase_       , threshold=UpperCAmelCase_       , target_sizes=model_output["target_size"])[0]

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

       lowerCamelCase__:      Optional[Any]			  ={"score": score, "label": label, "box": box}
       results.append(UpperCAmelCase_)

   lowerCamelCase__:      List[str]			  =sorted(UpperCAmelCase_       , key=lambda UpperCAmelCase_: x["score"]       , reverse=UpperCAmelCase_)
   if top_k:
     lowerCamelCase__:      Dict			  =results[:top_k]

   return results







 def 							SCREAMING_SNAKE_CASE_					(self				:   str       , UpperCAmelCase_				:   "torch.Tensor")       ->Dict[str, int]:



   '''simple docstring'''



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

