





import logging
import os
from dataclasses import dataclass
from typing import List, Optional, Union

import tqdm
from filelock import FileLock

from transformers import (
    BartTokenizer,
    BartTokenizerFast,
    DataProcessor,
    PreTrainedTokenizer,
    RobertaTokenizer,
    RobertaTokenizerFast,
    XLMRobertaTokenizer,
    is_tf_available,
    is_torch_available,
)


_UpperCAmelCase							:			str        =							logging.getLogger(__name__)

@dataclass(frozen=UpperCAmelCase_     )
class   lowerCAmelCase				:
 UpperCAmelCase__						=						42
 UpperCAmelCase__						=						42
 UpperCAmelCase__						=						None
 UpperCAmelCase__						=						None
 UpperCAmelCase__						=						None


@dataclass(frozen=UpperCAmelCase_     )
class   lowerCAmelCase				:
 UpperCAmelCase__						=						42
 UpperCAmelCase__						=						None
 UpperCAmelCase__						=						None
 UpperCAmelCase__						=						None
 UpperCAmelCase__						=						None


if is_torch_available():
    import torch
    from torch.utils.data import Dataset

    class   lowerCAmelCase				(   UpperCAmelCase_     ):
     UpperCAmelCase__						=						42
     def __init__( self    : Optional[Any]     ,	UpperCAmelCase    : str     ,	UpperCAmelCase    : PreTrainedTokenizer     ,	UpperCAmelCase    : str     ,	UpperCAmelCase    : Optional[int] = None     ,	UpperCAmelCase    : Tuple=False     ,	UpperCAmelCase    : bool = False     ,	)     ->      Tuple:
         lowerCamelCase__       : Optional[int]   					=						hans_processors[task]()

         lowerCamelCase__       : int   					=						os.path.join(
             _snake_case     ,	'cached_{}_{}_{}_{}'.format(
                 'dev' if evaluate else 'train'     ,	tokenizer.__class__.__name__     ,	str(_snake_case      )     ,	_snake_case     ,	)     ,	)
         lowerCamelCase__       : str   					=						processor.get_labels()
         if tokenizer.__class__ in (
             RobertaTokenizer,
             RobertaTokenizerFast,
             XLMRobertaTokenizer,
             BartTokenizer,
             BartTokenizerFast,
         ):
             # HACK(label indices are swapped in RoBERTa pretrained model)
             lowerCamelCase__					,				lowerCamelCase__       : int   					=						label_list[2], label_list[1]
         lowerCamelCase__       : Tuple   					=						label_list

         # Make sure only the first process in distributed training processes the dataset,
         # and the others will use the cache.
         lowerCamelCase__       : Optional[Any]   					=						cached_features_file + '.lock'
         with FileLock(_snake_case      ):
             if os.path.exists(_snake_case      ) and not overwrite_cache:
                 logger.info(F"""Loading features from cached file {cached_features_file}"""      )
                 lowerCamelCase__       : str   					=						torch.load(_snake_case      )
             else:
                 logger.info(F"""Creating features from dataset file at {data_dir}"""      )

                 lowerCamelCase__       : Tuple   					=						(
                     processor.get_dev_examples(_snake_case      ) if evaluate else processor.get_train_examples(_snake_case      )
                 )

                 logger.info('Training examples: %s'     ,	len(_snake_case      )      )
                 lowerCamelCase__       : List[Any]   					=						hans_convert_examples_to_features(_snake_case     ,	_snake_case     ,	_snake_case     ,	_snake_case      )
                 logger.info('Saving features into cached file %s'     ,	_snake_case      )
                 torch.save(self.features     ,	_snake_case      )
     def __len__( self    : Dict      )     ->      Tuple:
         return len(self.features      )
     def __getitem__( self    : Any     ,	UpperCAmelCase    : Dict      )     ->      Tuple:
         return self.features[i]







     def 	A_      ( self    : str      )     ->      str:
         return self.label_list


if is_tf_available():
    import tensorflow as tf

    class   lowerCAmelCase				:
     UpperCAmelCase__						=						42
     def __init__( self    : Union[str, Any]     ,	UpperCAmelCase    : str     ,	UpperCAmelCase    : PreTrainedTokenizer     ,	UpperCAmelCase    : str     ,	UpperCAmelCase    : Optional[int] = 128     ,	UpperCAmelCase    : List[Any]=False     ,	UpperCAmelCase    : bool = False     ,	)     ->      int:
         lowerCamelCase__       : List[Any]   					=						hans_processors[task]()
         lowerCamelCase__       : Union[str, Any]   					=						processor.get_labels()
         if tokenizer.__class__ in (
             RobertaTokenizer,
             RobertaTokenizerFast,
             XLMRobertaTokenizer,
             BartTokenizer,
             BartTokenizerFast,
         ):
             # HACK(label indices are swapped in RoBERTa pretrained model)
             lowerCamelCase__					,				lowerCamelCase__       : Any   					=						label_list[2], label_list[1]
         lowerCamelCase__       : Dict   					=						label_list

         lowerCamelCase__       : Union[str, Any]   					=						processor.get_dev_examples(_snake_case      ) if evaluate else processor.get_train_examples(_snake_case      )
         lowerCamelCase__       : Optional[int]   					=						hans_convert_examples_to_features(_snake_case     ,	_snake_case     ,	_snake_case     ,	_snake_case      )

         def gen():
             for ex_index, ex in tqdm.tqdm(enumerate(self.features      )     ,	desc='convert examples to features'      ):
                 if ex_index % 10000 == 0:
                     logger.info('Writing example %d of %d' % (ex_index, len(_snake_case      ))      )

                 yield (
                     {
                         "example_id": 0,
                         "input_ids": ex.input_ids,
                         "attention_mask": ex.attention_mask,
                         "token_type_ids": ex.token_type_ids,
                     },
                     ex.label,
                 )

         lowerCamelCase__       : Any   					=						tf.data.Dataset.from_generator(
             _snake_case     ,	(
                 {
                     'example_id': tf.intaa,
                     'input_ids': tf.intaa,
                     'attention_mask': tf.intaa,
                     'token_type_ids': tf.intaa,
                 },
                 tf.intaa,
             )     ,	(
                 {
                     'example_id': tf.TensorShape([]      ),
                     'input_ids': tf.TensorShape([None, None]      ),
                     'attention_mask': tf.TensorShape([None, None]      ),
                     'token_type_ids': tf.TensorShape([None, None]      ),
                 },
                 tf.TensorShape([]      ),
             )     ,	)
     def 	A_      ( self    : Tuple      )     ->      Optional[int]:
         return self.dataset
     def __len__( self    : Optional[int]      )     ->      Optional[Any]:
         return len(self.features      )
     def __getitem__( self    : str     ,	UpperCAmelCase    : Optional[int]      )     ->      Union[str, Any]:
         return self.features[i]







     def 	A_      ( self    : Any      )     ->      Any:
         return self.label_list

class   lowerCAmelCase				(   UpperCAmelCase_     ):
 def 	A_      ( self    : List[str]     ,	UpperCAmelCase    : int      )     ->      str:
     return self._create_examples(self._read_tsv(os.path.join(_snake_case     ,	'heuristics_train_set.txt'      )      )     ,	'train'      )
 def 	A_      ( self    : List[Any]     ,	UpperCAmelCase    : Union[str, Any]      )     ->      Optional[Any]:
     return self._create_examples(self._read_tsv(os.path.join(_snake_case     ,	'heuristics_evaluation_set.txt'      )      )     ,	'dev'      )
 def 	A_      ( self    : Optional[Any]      )     ->      List[str]:
     return ["contradiction", "entailment", "neutral"]







 def 	A_      ( self    : str     ,	UpperCAmelCase    : Tuple     ,	UpperCAmelCase    : Any      )     ->      Optional[int]:
     lowerCamelCase__       : Tuple   					=						[]
     for i, line in enumerate(_snake_case      ):
         if i == 0:
             continue
         lowerCamelCase__       : List[str]   					=						'%s-%s' % (set_type, line[0])
         lowerCamelCase__       : Tuple   					=						line[5]
         lowerCamelCase__       : int   					=						line[6]
         lowerCamelCase__       : int   					=						line[7][2:] if line[7].startswith('ex'      ) else line[7]
         lowerCamelCase__       : List[Any]   					=						line[0]
         examples.append(InputExample(guid=_snake_case     ,	text_a=_snake_case     ,	text_b=_snake_case     ,	label=_snake_case     ,	pairID=_snake_case      )      )
     return examples


def 			SCREAMING_SNAKE_CASE (			_UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      )     ->	str:
    lowerCamelCase__       : Optional[Any]   					=						{label: i for i, label in enumerate(__UpperCamelCase					)}

    lowerCamelCase__       : Dict   					=						[]
    for ex_index, example in tqdm.tqdm(enumerate(__UpperCamelCase					)		,      desc='convert examples to features'					):
        if ex_index % 1_0000 == 0:
            logger.info('Writing example %d' % (ex_index)					)

        lowerCamelCase__       : Any   					=						tokenizer(
            example.text_a		,      example.text_b		,      add_special_tokens=__UpperCamelCase		,      max_length=__UpperCamelCase		,      padding='max_length'		,      truncation=__UpperCamelCase		,      return_overflowing_tokens=__UpperCamelCase		,      )

        lowerCamelCase__       : Optional[int]   					=						label_map[example.label] if example.label in label_map else 0

        lowerCamelCase__       : Optional[Any]   					=						int(example.pairID					)

        features.append(InputFeatures(**__UpperCamelCase		,      label=__UpperCamelCase		,      pairID=__UpperCamelCase					)					)

    for i, example in enumerate(examples[:5]					):
        logger.info('*** Example ***'					)
        logger.info(F"""guid: {example}"""					)
        logger.info(F"""features: {features[i]}"""					)

    return features


_UpperCAmelCase							:			List[str]        =							{
    """hans""": 3,
}


_UpperCAmelCase							:			List[str]        =							{
    """hans""": HansProcessor,
}










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       :							Dict					,       _snake_case       :							Union[str, Any]					,       _snake_case       :							Optional[Any]=12					,       _snake_case       :							Any=7					,       _snake_case       :							List[str]=True					,       _snake_case       :							int=True					,       _snake_case       :							int=True					,       _snake_case       :							Tuple=99					,       _snake_case       :							List[Any]=32					,       _snake_case       :							Optional[int]=32					,       _snake_case       :							List[str]=2					,       _snake_case       :							List[str]=4					,       _snake_case       :							List[Any]=37					,       _snake_case       :							Union[str, Any]=0.1					,       _snake_case       :							Tuple=0.1					,       _snake_case       :							Dict=5_12					,       _snake_case       :							Union[str, Any]=0.02					,       _snake_case       :							Any=0					,       _snake_case       :							Optional[Any]=None					,       ):






								"""simple docstring"""


								A__	  =     parent
								A__	  =     batch_size
								A__	  =     seq_length
								A__	  =     is_training
								A__	  =     use_input_mask
								A__	  =     use_labels
								A__	  =     vocab_size
								A__	  =     hidden_size
								A__	  =     projection_dim
								A__	  =     num_hidden_layers
								A__	  =     num_attention_heads
								A__	  =     intermediate_size
								A__	  =     dropout
								A__	  =     attention_dropout
								A__	  =     max_position_embeddings
								A__	  =     initializer_range
								A__	  =     scope
								A__	  =     bos_token_id



					def      _a				(  self       :							Optional[Any]       ):






								"""simple docstring"""


								A__	  =     ids_tensor([self.batch_size, self.seq_length]					,       self.vocab_size       )

								A__	  =     None
								if self.use_input_mask:
											A__	  =     random_attention_mask([self.batch_size, self.seq_length]       )

								if input_mask is not None:
											A__	  =     input_mask.numpy()
											A__  ,			A__	  =     input_mask.shape
											A__	  =     np.random.randint(1					,       seq_length - 1					,       size=(batch_size,)       )
											for batch_idx, start_index in enumerate(_snake_case       ):
														A__	  =     1
														A__	  =     0

								A__	  =     self.get_config()

								return config, input_ids, tf.convert_to_tensor(_snake_case       )



					def      _a				(  self       :							Tuple       ):






								"""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      _a				(  self       :							int					,       _snake_case       :							Union[str, Any]					,       _snake_case       :							Any					,       _snake_case       :							List[str]       ):






								"""simple docstring"""


								A__	  =     TFBlipTextModel(config=_snake_case       )
								A__	  =     model(_snake_case					,       attention_mask=_snake_case					,       training=_snake_case       )
								A__	  =     model(_snake_case					,       training=_snake_case       )
								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      _a				(  self       :							str       ):






								"""simple docstring"""


								A__	  =     self.prepare_config_and_inputs()
								A__  ,			A__  ,			A__	  =     config_and_inputs
								A__	  =     {'input_ids': input_ids, 'attention_mask': input_mask}
								return config, inputs_dict







@require_tf
class   __lowerCAmelCase  (		UpperCAmelCase_				,  unittest.TestCase   ):
					"""simple docstring"""






					A__					:		Tuple      							=     (TFBlipTextModel,) if is_tf_available() else ()
					A__					:		Optional[int]      							=     False
					A__					:		Union[str, Any]      							=     False
					A__					:		Union[str, Any]      							=     False



					def      _a				(  self       :							Any       ):






								"""simple docstring"""


								A__	  =     BlipTextModelTester(self       )
								A__	  =     ConfigTester(self					,       config_class=_snake_case					,       hidden_size=37       )



					def      _a				(  self       :							List[str]       ):






								"""simple docstring"""


								self.config_tester.run_common_tests()



					def      _a				(  self       :							Union[str, Any]       ):






								"""simple docstring"""


								A__	  =     self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*_snake_case       )



					def      _a				(  self       :							Tuple       ):






								"""simple docstring"""


								pass



					def      _a				(  self       :							int       ):






								"""simple docstring"""


								pass



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






								"""simple docstring"""


								pass



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






								"""simple docstring"""


								pass



					@unittest.skip(reason='BlipTextModel has no base class and is not available in MODEL_MAPPING'       )
					def      _a				(  self       :							Optional[Any]       ):






								"""simple docstring"""


								pass



					@slow
					def      _a				(  self       :							Union[str, Any]       ):






								"""simple docstring"""


								for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
											A__	  =     TFBlipTextModel.from_pretrained(_snake_case       )
											self.assertIsNotNone(_snake_case       )



					def      _a				(  self       :							int					,       _snake_case       :							int=True       ):






								"""simple docstring"""


								super().test_pt_tf_model_equivalence(allow_missing_keys=_snake_case       )




