





"""simple docstring"""





from __future__ import annotations

import unittest

import numpy as np

from transformers import BlipTextConfig
from transformers.testing_utils import require_tf, slow
from transformers.utils import is_tf_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask


if is_tf_available():
  import tensorflow as tf

  from transformers import TFBlipTextModel
  from transformers.models.blip.modeling_tf_blip import TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST






class 					_lowerCAmelCase   :
       """simple docstring"""




       def __init__(    self    , __UpperCAmelCase    , __UpperCAmelCase=1_2    , __UpperCAmelCase=7    , __UpperCAmelCase=True    , __UpperCAmelCase=True    , __UpperCAmelCase=True    , __UpperCAmelCase=9_9    , __UpperCAmelCase=3_2    , __UpperCAmelCase=3_2    , __UpperCAmelCase=2    , __UpperCAmelCase=4    , __UpperCAmelCase=3_7    , __UpperCAmelCase=0.1    , __UpperCAmelCase=0.1    , __UpperCAmelCase=5_1_2    , __UpperCAmelCase=0.02    , __UpperCAmelCase=0    , __UpperCAmelCase=None    , ):



         '''simple docstring'''

         lowerCAmelCase__			:Dict											=							parent
         lowerCAmelCase__			:Union[str, Any]											=							batch_size
         lowerCAmelCase__			:List[Any]											=							seq_length
         lowerCAmelCase__			:Optional[int]											=							is_training
         lowerCAmelCase__			:Union[str, Any]											=							use_input_mask
         lowerCAmelCase__			:List[Any]											=							use_labels
         lowerCAmelCase__			:str											=							vocab_size
         lowerCAmelCase__			:Union[str, Any]											=							hidden_size
         lowerCAmelCase__			:List[Any]											=							projection_dim
         lowerCAmelCase__			:Optional[Any]											=							num_hidden_layers
         lowerCAmelCase__			:int											=							num_attention_heads
         lowerCAmelCase__			:Any											=							intermediate_size
         lowerCAmelCase__			:Any											=							dropout
         lowerCAmelCase__			:Any											=							attention_dropout
         lowerCAmelCase__			:List[Any]											=							max_position_embeddings
         lowerCAmelCase__			:Optional[int]											=							initializer_range
         lowerCAmelCase__			:int											=							scope
         lowerCAmelCase__			:Tuple											=							bos_token_id






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:str											=							ids_tensor([self.batch_size, self.seq_length]    , self.vocab_size      )

         lowerCAmelCase__			:str											=							None
         if self.use_input_mask:
           lowerCAmelCase__			:Tuple											=							random_attention_mask([self.batch_size, self.seq_length]      )

         if input_mask is not None:
           lowerCAmelCase__			:Union[str, Any]											=							input_mask.numpy()
           lowerCAmelCase__							,				lowerCAmelCase__			:Optional[Any]											=							input_mask.shape
           lowerCAmelCase__			:Optional[Any]											=							np.random.randint(1    , seq_length - 1    , size=(batch_size,)      )
           for batch_idx, start_index in enumerate(__UpperCAmelCase      ):
             lowerCAmelCase__			:Dict											=							1
             lowerCAmelCase__			:Any											=							0

         lowerCAmelCase__			:Optional[Any]											=							self.get_config()

         return config, input_ids, tf.convert_to_tensor(__UpperCAmelCase      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         return BlipTextConfig(
             vocab_size=self.vocab_size    , hidden_size=self.hidden_size    , projection_dim=self.projection_dim    , num_hidden_layers=self.num_hidden_layers    , num_attention_heads=self.num_attention_heads    , intermediate_size=self.intermediate_size    , dropout=self.dropout    , attention_dropout=self.attention_dropout    , max_position_embeddings=self.max_position_embeddings    , initializer_range=self.initializer_range    , bos_token_id=self.bos_token_id    , )






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      ):



         '''simple docstring'''

         lowerCAmelCase__			:Union[str, Any]											=							TFBlipTextModel(config=__UpperCAmelCase      )
         lowerCAmelCase__			:Union[str, Any]											=							model(__UpperCAmelCase    , attention_mask=__UpperCAmelCase    , training=__UpperCAmelCase      )
         lowerCAmelCase__			:Optional[int]											=							model(__UpperCAmelCase    , training=__UpperCAmelCase      )
         self.parent.assertEqual(result.last_hidden_state.shape    , (self.batch_size, self.seq_length, self.hidden_size)      )
         self.parent.assertEqual(result.pooler_output.shape    , (self.batch_size, self.hidden_size)      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:int											=							self.prepare_config_and_inputs()
         lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__			:Dict											=							config_and_inputs
         lowerCAmelCase__			:Union[str, Any]											=							{'input_ids': input_ids, 'attention_mask': input_mask}
         return config, inputs_dict






@require_tf
class 					_lowerCAmelCase   (     a  ,   unittest.TestCase       ):
       """simple docstring"""




       __magic_name__     :List[Any]		  =					(TFBlipTextModel,) if is_tf_available() else ()
       __magic_name__     :int		  =					False
       __magic_name__     :Dict		  =					False
       __magic_name__     :Optional[Any]		  =					False






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:Dict											=							BlipTextModelTester(self      )
         lowerCAmelCase__			:Optional[Any]											=							ConfigTester(self    , config_class=__UpperCAmelCase    , hidden_size=3_7      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         self.config_tester.run_common_tests()






       def      snake_case						(    self      ):



         '''simple docstring'''

         lowerCAmelCase__			:List[str]											=							self.model_tester.prepare_config_and_inputs()
         self.model_tester.create_and_check_model(*__UpperCAmelCase      )






       def      snake_case						(    self      ):



         '''simple docstring'''

         pass






       def      snake_case						(    self      ):



         '''simple docstring'''

         pass






       @unittest.skip(reason='Blip does not use inputs_embeds'      )
       def      snake_case						(    self      ):



         '''simple docstring'''

         pass






       @unittest.skip(reason='BlipTextModel has no base class and is not available in MODEL_MAPPING'      )
       def      snake_case						(    self      ):



         '''simple docstring'''

         pass






       @unittest.skip(reason='BlipTextModel has no base class and is not available in MODEL_MAPPING'      )
       def      snake_case						(    self      ):



         '''simple docstring'''

         pass






       @slow
       def      snake_case						(    self      ):



         '''simple docstring'''

         for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           lowerCAmelCase__			:List[Any]											=							TFBlipTextModel.from_pretrained(__UpperCAmelCase      )
           self.assertIsNotNone(__UpperCAmelCase      )






       def      snake_case						(    self    , __UpperCAmelCase=True      ):



         '''simple docstring'''

         super().test_pt_tf_model_equivalence(allow_missing_keys=__UpperCAmelCase      )







"""simple docstring"""





from collections import defaultdict
from typing import Optional

from ..image_utils import load_image
from ..utils import (
    add_end_docstrings,
    is_torch_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_torch_available():
  import torch

  from ..models.auto.modeling_auto import MODEL_FOR_MASK_GENERATION_MAPPING

__A											=		logging.get_logger(__name__)






@add_end_docstrings(a       )
class 					_lowerCAmelCase   (     a       ):
       """simple docstring"""




       def __init__(    self    , **__UpperCAmelCase      ):



         '''simple docstring'''

         super().__init__(**__UpperCAmelCase      )
         requires_backends(self    , 'vision'      )
         requires_backends(self    , 'torch'      )

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

         self.check_model_type(__UpperCAmelCase      )






       def      snake_case						(    self    , **__UpperCAmelCase      ):



         '''simple docstring'''

         lowerCAmelCase__			:List[str]											=							{}
         lowerCAmelCase__			:Tuple											=							{}
         lowerCAmelCase__			:Any											=							{}
         # preprocess args
         if "points_per_batch" in kwargs:
           lowerCAmelCase__			:Dict											=							kwargs['points_per_batch']
         if "points_per_crop" in kwargs:
           lowerCAmelCase__			:Union[str, Any]											=							kwargs['points_per_crop']
         if "crops_n_layers" in kwargs:
           lowerCAmelCase__			:Any											=							kwargs['crops_n_layers']
         if "crop_overlap_ratio" in kwargs:
           lowerCAmelCase__			:Any											=							kwargs['crop_overlap_ratio']
         if "crop_n_points_downscale_factor" in kwargs:
           lowerCAmelCase__			:Dict											=							kwargs['crop_n_points_downscale_factor']
         # postprocess args
         if "pred_iou_thresh" in kwargs:
           lowerCAmelCase__			:Tuple											=							kwargs['pred_iou_thresh']
         if "stability_score_offset" in kwargs:
           lowerCAmelCase__			:Optional[int]											=							kwargs['stability_score_offset']
         if "mask_threshold" in kwargs:
           lowerCAmelCase__			:List[Any]											=							kwargs['mask_threshold']
         if "stability_score_thresh" in kwargs:
           lowerCAmelCase__			:Optional[Any]											=							kwargs['stability_score_thresh']
         if "crops_nms_thresh" in kwargs:
           lowerCAmelCase__			:int											=							kwargs['crops_nms_thresh']
         if "output_rle_mask" in kwargs:
           lowerCAmelCase__			:Union[str, Any]											=							kwargs['output_rle_mask']
         if "output_bboxes_mask" in kwargs:
           lowerCAmelCase__			:Optional[Any]											=							kwargs['output_bboxes_mask']
         return preprocess_kwargs, forward_params, postprocess_kwargs






       def __call__(    self    , __UpperCAmelCase    , *__UpperCAmelCase    , __UpperCAmelCase=None    , __UpperCAmelCase=None    , **__UpperCAmelCase      ):



         '''simple docstring'''

         return super().__call__(__UpperCAmelCase    , *__UpperCAmelCase    , num_workers=__UpperCAmelCase    , batch_size=__UpperCAmelCase    , **__UpperCAmelCase      )






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase=6_4    , __UpperCAmelCase = 0    , __UpperCAmelCase = 5_1_2 / 1_5_0_0    , __UpperCAmelCase = 3_2    , __UpperCAmelCase = 1    , ):



         '''simple docstring'''

         lowerCAmelCase__			:Union[str, Any]											=							load_image(__UpperCAmelCase      )
         lowerCAmelCase__			:int											=							self.image_processor.size['longest_edge']
         lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__			:int											=							self.image_processor.generate_crop_boxes(
             __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      )
         lowerCAmelCase__			:Union[str, Any]											=							self.image_processor(images=__UpperCAmelCase    , return_tensors='pt'      )

         with self.device_placement():
           if self.framework == "pt":
             lowerCAmelCase__			:Optional[int]											=							self.get_inference_context()
             with inference_context():
               lowerCAmelCase__			:Any											=							self._ensure_tensor_on_device(__UpperCAmelCase    , device=self.device      )
               lowerCAmelCase__			:Tuple											=							self.model.get_image_embeddings(model_inputs.pop('pixel_values'      )      )
               lowerCAmelCase__			:Optional[int]											=							image_embeddings

         lowerCAmelCase__			:List[Any]											=							grid_points.shape[1]
         lowerCAmelCase__			:Union[str, Any]											=							points_per_batch if points_per_batch is not None else n_points

         if points_per_batch <= 0:
           raise ValueError(
               'Cannot have points_per_batch<=0. Must be >=1 to returned batched outputs. '
               'To return all points at once, set points_per_batch to None'      )

         for i in range(0    , __UpperCAmelCase    , __UpperCAmelCase      ):
           lowerCAmelCase__			:Optional[Any]											=							grid_points[:, i : i + points_per_batch, :, :]
           lowerCAmelCase__			:List[str]											=							input_labels[:, i : i + points_per_batch]
           lowerCAmelCase__			:List[Any]											=							i == n_points - points_per_batch
           yield {
               "input_points": batched_points,
               "input_labels": labels,
               "input_boxes": crop_boxes,
               "is_last": is_last,
               **model_inputs,
           }






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase=0.88    , __UpperCAmelCase=0.95    , __UpperCAmelCase=0    , __UpperCAmelCase=1    , ):



         '''simple docstring'''

         lowerCAmelCase__			:Any											=							model_inputs.pop('input_boxes'      )
         lowerCAmelCase__			:Optional[int]											=							model_inputs.pop('is_last'      )
         lowerCAmelCase__			:Dict											=							model_inputs.pop('original_sizes'      ).tolist()
         lowerCAmelCase__			:Dict											=							model_inputs.pop('reshaped_input_sizes'      ).tolist()

         lowerCAmelCase__			:Optional[int]											=							self.model(**__UpperCAmelCase      )

         # post processing happens here in order to avoid CPU GPU copies of ALL the masks
         lowerCAmelCase__			:int											=							model_outputs['pred_masks']
         lowerCAmelCase__			:Optional[Any]											=							self.image_processor.post_process_masks(
             __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , binarize=__UpperCAmelCase      )
         lowerCAmelCase__			:Any											=							model_outputs['iou_scores']
         lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__			:Tuple											=							self.image_processor.filter_masks(
             masks[0]    , iou_scores[0]    , original_sizes[0]    , input_boxes[0]    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , )
         return {
             "masks": masks,
             "is_last": is_last,
             "boxes": boxes,
             "iou_scores": iou_scores,
         }






       def      snake_case						(    self    , __UpperCAmelCase    , __UpperCAmelCase=False    , __UpperCAmelCase=False    , __UpperCAmelCase=0.7    , ):



         '''simple docstring'''

         lowerCAmelCase__			:Dict											=							[]
         lowerCAmelCase__			:Optional[Any]											=							[]
         lowerCAmelCase__			:int											=							[]
         for model_output in model_outputs:
           all_scores.append(model_output.pop('iou_scores'      )      )
           all_masks.extend(model_output.pop('masks'      )      )
           all_boxes.append(model_output.pop('boxes'      )      )

         lowerCAmelCase__			:Dict											=							torch.cat(__UpperCAmelCase      )
         lowerCAmelCase__			:Dict											=							torch.cat(__UpperCAmelCase      )
         lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__							,				lowerCAmelCase__			:Any											=							self.image_processor.post_process_for_mask_generation(
             __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase    , __UpperCAmelCase      )

         lowerCAmelCase__			:Tuple											=							defaultdict(__UpperCAmelCase      )
         for output in model_outputs:
           for k, v in output.items():
             extra[k].append(__UpperCAmelCase      )

         lowerCAmelCase__			:Optional[int]											=							{}
         if output_rle_mask:
           lowerCAmelCase__			:str											=							rle_mask

         if output_bboxes_mask:
           lowerCAmelCase__			:Optional[int]											=							bounding_boxes

         return {"masks": output_masks, "scores": iou_scores, **optional, **extra}
