

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,
)


__lowercase     : Any       =						logging.getLogger(__name__)



@dataclass(frozen=snake_case				)
class 						_A     :





  '''simple docstring'''



  __lowerCamelCase			:      str
  __lowerCamelCase			:      str
  __lowerCamelCase			:      Optional[str]					=   None
  __lowerCamelCase			:      Optional[str]					=   None
  __lowerCamelCase			:      Optional[str]					=   None







@dataclass(frozen=snake_case				)
class 						_A     :





  '''simple docstring'''



  __lowerCamelCase			:      List[int]
  __lowerCamelCase			:      Optional[List[int]]					=   None
  __lowerCamelCase			:      Optional[List[int]]					=   None
  __lowerCamelCase			:      Optional[Union[int, float]]					=   None
  __lowerCamelCase			:      Optional[int]					=   None


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



     class 						_A     (				snake_case				):





       '''simple docstring'''



       __lowerCamelCase			:      List[InputFeatures]





       def __init__(  self   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_ = None   ,SCREAMING_SNAKE_CASE_=False   ,SCREAMING_SNAKE_CASE_ = False   ,):

           '''simple docstring'''
           snake_case							:		str        =		hans_processors[task]()

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

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

                   snake_case							:		Union[str, Any]        =		(
                       processor.get_dev_examples(SCREAMING_SNAKE_CASE_    ) if evaluate else processor.get_train_examples(SCREAMING_SNAKE_CASE_    )
                   )

                   logger.info("""Training examples: %s"""   ,len(SCREAMING_SNAKE_CASE_    )    )
                   snake_case							:		Dict        =		hans_convert_examples_to_features(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_    )
                   logger.info("""Saving features into cached file %s"""   ,SCREAMING_SNAKE_CASE_    )
                   torch.save(self.features   ,SCREAMING_SNAKE_CASE_    )





       def __len__(  self    ):

           '''simple docstring'''
           return len(self.features    )





       def __getitem__(  self   ,SCREAMING_SNAKE_CASE_    ):

           '''simple docstring'''
           return self.features[i]





       def    snake_case_       (  self    ):

           '''simple docstring'''
           return self.label_list


if is_tf_available():
     import tensorflow as tf



     class 						_A     :





       '''simple docstring'''



       __lowerCamelCase			:      List[InputFeatures]





       def __init__(  self   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_ = 128   ,SCREAMING_SNAKE_CASE_=False   ,SCREAMING_SNAKE_CASE_ = False   ,):

           '''simple docstring'''
           snake_case							:		Any        =		hans_processors[task]()
           snake_case							:		List[str]        =		processor.get_labels()
           if tokenizer.__class__ in (
               RobertaTokenizer,
               RobertaTokenizerFast,
               XLMRobertaTokenizer,
               BartTokenizer,
               BartTokenizerFast,
           ):
               # HACK(label indices are swapped in RoBERTa pretrained model)
               snake_case      ,		snake_case							:		int        =		label_list[2], label_list[1]
           snake_case							:		List[str]        =		label_list

           snake_case							:		int        =		processor.get_dev_examples(SCREAMING_SNAKE_CASE_    ) if evaluate else processor.get_train_examples(SCREAMING_SNAKE_CASE_    )
           snake_case							:		Any        =		hans_convert_examples_to_features(SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_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(SCREAMING_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,
                   )

           snake_case							:		Any        =		tf.data.Dataset.from_generator(
               SCREAMING_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    snake_case_       (  self    ):

           '''simple docstring'''
           return self.dataset





       def __len__(  self    ):

           '''simple docstring'''
           return len(self.features    )





       def __getitem__(  self   ,SCREAMING_SNAKE_CASE_    ):

           '''simple docstring'''
           return self.features[i]





       def    snake_case_       (  self    ):

           '''simple docstring'''
           return self.label_list



class 						_A     (				snake_case				):





  '''simple docstring'''



  def    snake_case_       (  self   ,SCREAMING_SNAKE_CASE_    ):

      '''simple docstring'''
      return self._create_examples(self._read_tsv(os.path.join(SCREAMING_SNAKE_CASE_   ,"""heuristics_train_set.txt"""    )    )   ,"""train"""    )





  def    snake_case_       (  self   ,SCREAMING_SNAKE_CASE_    ):

      '''simple docstring'''
      return self._create_examples(self._read_tsv(os.path.join(SCREAMING_SNAKE_CASE_   ,"""heuristics_evaluation_set.txt"""    )    )   ,"""dev"""    )





  def    snake_case_       (  self    ):

      '''simple docstring'''
      return ["contradiction", "entailment", "neutral"]





  def    snake_case_       (  self   ,SCREAMING_SNAKE_CASE_   ,SCREAMING_SNAKE_CASE_    ):

      '''simple docstring'''
      snake_case							:		List[str]        =		[]
      for i, line in enumerate(SCREAMING_SNAKE_CASE_    ):
          if i == 0:
              continue
          snake_case							:		Any        =		"""%s-%s""" % (set_type, line[0])
          snake_case							:		Optional[int]        =		line[5]
          snake_case							:		Union[str, Any]        =		line[6]
          snake_case							:		Optional[Any]        =		line[7][2:] if line[7].startswith("""ex"""    ) else line[7]
          snake_case							:		Dict        =		line[0]
          examples.append(InputExample(guid=SCREAMING_SNAKE_CASE_   ,text_a=SCREAMING_SNAKE_CASE_   ,text_b=SCREAMING_SNAKE_CASE_   ,label=SCREAMING_SNAKE_CASE_   ,pairID=SCREAMING_SNAKE_CASE_    )    )
      return examples







def       lowercase					(	__A :					List[InputExample] ,    __A :					List[str] ,    __A :					int ,    __A :					PreTrainedTokenizer ,    )				->			Tuple:





    '''simple docstring'''




    snake_case							:		List[Any]        =		{label: i for i, label in enumerate(__A		)}

    snake_case							:		Union[str, Any]        =		[]
    for ex_index, example in tqdm.tqdm(enumerate(__A		) ,    desc="""convert examples to features"""		):
        if ex_index % 1_0000 == 0:
            logger.info("""Writing example %d""" % (ex_index)		)

        snake_case							:		Union[str, Any]        =		tokenizer(
            example.text_a ,    example.text_b ,    add_special_tokens=__A ,    max_length=__A ,    padding="""max_length""" ,    truncation=__A ,    return_overflowing_tokens=__A ,    )

        snake_case							:		Tuple        =		label_map[example.label] if example.label in label_map else 0

        snake_case							:		Tuple        =		int(example.pairID		)

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

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

    return features


__lowercase     : Dict       =						{
    '''hans''': 3,
}


__lowercase     : Union[str, Any]       =						{
    '''hans''': HansProcessor,
}










'''simple docstring'''

from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType
class 							__lowerCAmelCase	(       __magic_name__					):






       """simple docstring"""
       _snake_case      :	Dict									=       ['image_processor', 'tokenizer']
       _snake_case      :	int									=       'BlipImageProcessor'
       _snake_case      :	Tuple									=       ('BertTokenizer', 'BertTokenizerFast')





       def __init__(				self  :						Union[str, Any]			,					lowerCAmelCase__  :						List[Any]			,					lowerCAmelCase__  :						List[str]	)       ->	int:

            '''simple docstring'''




            _UpperCamelCase							       =			False
            super().__init__(lowerCAmelCase__			,					lowerCAmelCase__	)
            _UpperCamelCase							       =			self.image_processor





       def __call__(				self  :						int			,					lowerCAmelCase__  :						ImageInput = None			,					lowerCAmelCase__  :						Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None			,					lowerCAmelCase__  :						bool = True			,					lowerCAmelCase__  :						Union[bool, str, PaddingStrategy] = False			,					lowerCAmelCase__  :						Union[bool, str, TruncationStrategy] = None			,					lowerCAmelCase__  :						Optional[int] = None			,					lowerCAmelCase__  :						int = 0			,					lowerCAmelCase__  :						Optional[int] = None			,					lowerCAmelCase__  :						Optional[bool] = None			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = True			,					lowerCAmelCase__  :						Optional[Union[str, TensorType]] = None			,					**lowerCAmelCase__  :						Optional[Any]			,					)       ->	BatchEncoding:

            '''simple docstring'''




            if images is None and text is None:
                 raise ValueError('''You have to specify either images or text.'''	)

            # Get only text
            if images is None:
                 _UpperCamelCase							       =			self.tokenizer
                 _UpperCamelCase							       =			self.tokenizer(
                     text=lowerCAmelCase__			,					add_special_tokens=lowerCAmelCase__			,					padding=lowerCAmelCase__			,					truncation=lowerCAmelCase__			,					max_length=lowerCAmelCase__			,					stride=lowerCAmelCase__			,					pad_to_multiple_of=lowerCAmelCase__			,					return_attention_mask=lowerCAmelCase__			,					return_overflowing_tokens=lowerCAmelCase__			,					return_special_tokens_mask=lowerCAmelCase__			,					return_offsets_mapping=lowerCAmelCase__			,					return_token_type_ids=lowerCAmelCase__			,					return_length=lowerCAmelCase__			,					verbose=lowerCAmelCase__			,					return_tensors=lowerCAmelCase__			,					**lowerCAmelCase__			,					)
                 return text_encoding

            # add pixel_values
            _UpperCamelCase							       =			self.image_processor(lowerCAmelCase__			,					return_tensors=lowerCAmelCase__	)

            if text is not None:
                 _UpperCamelCase							       =			self.tokenizer(
                     text=lowerCAmelCase__			,					add_special_tokens=lowerCAmelCase__			,					padding=lowerCAmelCase__			,					truncation=lowerCAmelCase__			,					max_length=lowerCAmelCase__			,					stride=lowerCAmelCase__			,					pad_to_multiple_of=lowerCAmelCase__			,					return_attention_mask=lowerCAmelCase__			,					return_overflowing_tokens=lowerCAmelCase__			,					return_special_tokens_mask=lowerCAmelCase__			,					return_offsets_mapping=lowerCAmelCase__			,					return_token_type_ids=lowerCAmelCase__			,					return_length=lowerCAmelCase__			,					verbose=lowerCAmelCase__			,					return_tensors=lowerCAmelCase__			,					**lowerCAmelCase__			,					)
            else:
                 _UpperCamelCase							       =			None

            if text_encoding is not None:
                 encoding_image_processor.update(lowerCAmelCase__	)

            return encoding_image_processor





       def snake_case__		(				self  :						Any			,					*lowerCAmelCase__  :						Union[str, Any]			,					**lowerCAmelCase__  :						Dict	)       ->	List[str]:

            '''simple docstring'''




            return self.tokenizer.batch_decode(*lowerCAmelCase__			,					**lowerCAmelCase__	)





       def snake_case__		(				self  :						Tuple			,					*lowerCAmelCase__  :						int			,					**lowerCAmelCase__  :						Optional[int]	)       ->	Optional[int]:

            '''simple docstring'''




            return self.tokenizer.decode(*lowerCAmelCase__			,					**lowerCAmelCase__	)







       @property
       def snake_case__		(				self  :						Any	)       ->	str:

            '''simple docstring'''




            _UpperCamelCase							       =			self.tokenizer.model_input_names
            _UpperCamelCase							       =			self.image_processor.model_input_names
            return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names	)	)



