


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   __a			:
 def __init__(   self				, lowerCAmelCase__				, lowerCAmelCase__=12				, lowerCAmelCase__=7				, lowerCAmelCase__=True				, lowerCAmelCase__=True				, lowerCAmelCase__=True				, lowerCAmelCase__=99				, lowerCAmelCase__=32				, lowerCAmelCase__=32				, lowerCAmelCase__=2				, lowerCAmelCase__=4				, lowerCAmelCase__=37				, lowerCAmelCase__=0.1				, lowerCAmelCase__=0.1				, lowerCAmelCase__=512				, lowerCAmelCase__=0.0_2				, lowerCAmelCase__=0				, lowerCAmelCase__=None				, )					->				Dict:




        '''simple docstring'''
        lowercase__:   List[str]  			=   parent
        lowercase__:   str  			=   batch_size
        lowercase__:   Optional[Any]  			=   seq_length
        lowercase__:   List[str]  			=   is_training
        lowercase__:   List[str]  			=   use_input_mask
        lowercase__:   Optional[int]  			=   use_labels
        lowercase__:   Tuple  			=   vocab_size
        lowercase__:   int  			=   hidden_size
        lowercase__:   List[Any]  			=   projection_dim
        lowercase__:   int  			=   num_hidden_layers
        lowercase__:   Union[str, Any]  			=   num_attention_heads
        lowercase__:   Any  			=   intermediate_size
        lowercase__:   Dict  			=   dropout
        lowercase__:   int  			=   attention_dropout
        lowercase__:   Any  			=   max_position_embeddings
        lowercase__:   str  			=   initializer_range
        lowercase__:   List[str]  			=   scope
        lowercase__:   str  			=   bos_token_id
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Optional[Any]:




        '''simple docstring'''
        lowercase__:   Any  			=   ids_tensor([self.batch_size, self.seq_length]				, self.vocab_size )

        lowercase__:   int  			=   None
        if self.use_input_mask:
               lowercase__:   Any  			=   random_attention_mask([self.batch_size, self.seq_length] )

        if input_mask is not None:
               lowercase__:   int  			=   input_mask.numpy()
               lowercase__			,						lowercase__:   List[str]  			=   input_mask.shape
               lowercase__:   int  			=   np.random.randint(1				, seq_length - 1				, size=(batch_size,) )
               for batch_idx, start_index in enumerate(lowerCAmelCase__ ):
                      lowercase__:   List[str]  			=   1
                      lowercase__:   Tuple  			=   0

        lowercase__:   int  			=   self.get_config()

        return config, input_ids, tf.convert_to_tensor(lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Optional[Any]:




        '''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 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__ )					->				Dict:




        '''simple docstring'''
        lowercase__:   List[Any]  			=   TFBlipTextModel(config=lowerCAmelCase__ )
        lowercase__:   str  			=   model(lowerCAmelCase__				, attention_mask=lowerCAmelCase__				, training=lowerCAmelCase__ )
        lowercase__:   List[str]  			=   model(lowerCAmelCase__				, training=lowerCAmelCase__ )
        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 		SCREAMING_SNAKE_CASE__   (   self )					->				List[str]:




        '''simple docstring'''
        lowercase__:   Optional[Any]  			=   self.prepare_config_and_inputs()
        lowercase__			,						lowercase__			,						lowercase__:   Union[str, Any]  			=   config_and_inputs
        lowercase__:   List[Any]  			=   {'input_ids': input_ids, 'attention_mask': input_mask}
        return config, inputs_dict





@require_tf
class   __a			(    __UpperCamelCase       ,       unittest.TestCase      ):
 __lowercase :							Any        = (TFBlipTextModel,) if is_tf_available() else ()
 __lowercase :							Union[str, Any]        = False
 __lowercase :							str        = False
 __lowercase :							Union[str, Any]        = False
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Any:




        '''simple docstring'''
        lowercase__:   int  			=   BlipTextModelTester(self )
        lowercase__:   Union[str, Any]  			=   ConfigTester(self				, config_class=lowerCAmelCase__				, hidden_size=37 )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Union[str, Any]:




        '''simple docstring'''
        self.config_tester.run_common_tests()
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				List[Any]:




        '''simple docstring'''
        lowercase__:   List[str]  			=   self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Union[str, Any]:




        '''simple docstring'''
        pass
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				str:




        '''simple docstring'''
        pass
 @unittest.skip(reason='Blip does not use inputs_embeds' )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Tuple:




        '''simple docstring'''
        pass
 @unittest.skip(reason='BlipTextModel has no base class and is not available in MODEL_MAPPING' )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Tuple:




        '''simple docstring'''
        pass
 @unittest.skip(reason='BlipTextModel has no base class and is not available in MODEL_MAPPING' )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Union[str, Any]:




        '''simple docstring'''
        pass
 @slow
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Tuple:




        '''simple docstring'''
        for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               lowercase__:   Any  			=   TFBlipTextModel.from_pretrained(lowerCAmelCase__ )
               self.assertIsNotNone(lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__=True )					->				Dict:




        '''simple docstring'''
        super().test_pt_tf_model_equivalence(allow_missing_keys=lowerCAmelCase__ )






from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
 from ..tf_utils import stable_softmax

__lowerCAmelCase		       =							logging.get_logger(__name__)





@add_end_docstrings(__UpperCamelCase      )
class   __a			(    __UpperCamelCase      ):
 def __init__(   self				, **lowerCAmelCase__ )					->				List[str]:




        '''simple docstring'''
        super().__init__(**lowerCAmelCase__ )

        requires_backends(self				, 'vision' )
        self.check_model_type(
            TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
            if self.framework == 'tf'
            else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING )
 def __call__(   self				, lowerCAmelCase__				, **lowerCAmelCase__ )					->				Optional[Any]:




        '''simple docstring'''
        return super().__call__(lowerCAmelCase__				, **lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self				, **lowerCAmelCase__ )					->				Dict:




        '''simple docstring'''
        lowercase__:   int  			=   {}
        if "candidate_labels" in kwargs:
               lowercase__:   Dict  			=   kwargs['candidate_labels']
        if "hypothesis_template" in kwargs:
               lowercase__:   List[Any]  			=   kwargs['hypothesis_template']

        return preprocess_params, {}, {}
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__				, lowerCAmelCase__=None				, lowerCAmelCase__="This is a photo of {}." )					->				int:




        '''simple docstring'''
        lowercase__:   Optional[int]  			=   load_image(lowerCAmelCase__ )
        lowercase__:   Dict  			=   self.image_processor(images=[image]				, return_tensors=self.framework )
        lowercase__:   Tuple  			=   candidate_labels
        lowercase__:   List[str]  			=   [hypothesis_template.format(lowerCAmelCase__ ) for x in candidate_labels]
        lowercase__:   Optional[Any]  			=   self.tokenizer(lowerCAmelCase__				, return_tensors=self.framework				, padding=lowerCAmelCase__ )
        lowercase__:   str  			=   [text_inputs]
        return inputs
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__ )					->				int:




        '''simple docstring'''
        lowercase__:   str  			=   model_inputs.pop('candidate_labels' )
        lowercase__:   List[str]  			=   model_inputs.pop('text_inputs' )
        if isinstance(text_inputs[0]				, lowerCAmelCase__ ):
               lowercase__:   Any  			=   text_inputs[0]
        else:
               # Batching case.
               lowercase__:   Optional[int]  			=   text_inputs[0][0]

        lowercase__:   Tuple  			=   self.model(**lowerCAmelCase__				, **lowerCAmelCase__ )

        lowercase__:   Any  			=   {
            'candidate_labels': candidate_labels,
            'logits': outputs.logits_per_image,
        }
        return model_outputs
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__ )					->				Union[str, Any]:




        '''simple docstring'''
        lowercase__:   Tuple  			=   model_outputs.pop('candidate_labels' )
        lowercase__:   Dict  			=   model_outputs['logits'][0]
        if self.framework == "pt":
               lowercase__:   Optional[int]  			=   logits.softmax(dim=-1 ).squeeze(-1 )
               lowercase__:   Dict  			=   probs.tolist()
               if not isinstance(lowerCAmelCase__				, lowerCAmelCase__ ):
                      lowercase__:   Dict  			=   [scores]
        elif self.framework == "tf":
               lowercase__:   Optional[int]  			=   stable_softmax(lowerCAmelCase__				, axis=-1 )
               lowercase__:   Union[str, Any]  			=   probs.numpy().tolist()
        else:
               raise ValueError(F'Unsupported framework: {self.framework}' )

        lowercase__:   List[Any]  			=   [
            {'score': score, 'label': candidate_label}
            for score, candidate_label in sorted(zip(lowerCAmelCase__				, lowerCAmelCase__ )				, key=lambda lowerCAmelCase__ : -x[0] )
        ]
        return result


