





"""simple docstring"""


import logging
import os
from dataclasses import dataclass, field
from typing import Dict, List, Optional

import numpy as np
import torch
from utils_hans import HansDataset, InputFeatures, hans_processors, hans_tasks_num_labels

import transformers
from transformers import (
    AutoConfig,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    default_data_collator,
    set_seed,
)
from transformers.trainer_utils import is_main_process


a__					:     Tuple							=				logging.getLogger(__name__)



@dataclass
class      lowercase_				:
       snake_case__				:							str											=							field(
           metadata={"""help""": """Path to pretrained model or model identifier from huggingface.co/models"""}							)
       snake_case__				:							Optional[str]											=							field(
           default=lowercase__				,metadata={"""help""": """Pretrained config name or path if not the same as model_name"""}							)
       snake_case__				:							Optional[str]											=							field(
           default=lowercase__				,metadata={"""help""": """Pretrained tokenizer name or path if not the same as model_name"""}							)
       snake_case__				:							Optional[str]											=							field(
           default=lowercase__				,metadata={"""help""": """Where do you want to store the pretrained models downloaded from huggingface.co"""}				,)



@dataclass
class      lowercase_				:
       snake_case__				:							str											=							field(
           metadata={"""help""": """The name of the task to train selected in the list: """ + """, """.join(hans_processors.keys()							)}							)
       snake_case__				:							str											=							field(
           metadata={"""help""": """The input data dir. Should contain the .tsv files (or other data files) for the task."""}							)
       snake_case__				:							int											=							field(
           default=1_28				,metadata={
               """help""": (
                   """The maximum total input sequence length after tokenization. Sequences longer """
                   """than this will be truncated, sequences shorter will be padded."""
               )
           }				,)
       snake_case__				:							bool											=							field(
           default=lowercase__				,metadata={"""help""": """Overwrite the cached training and evaluation sets"""}							)





def 							a__			(  _UpperCAmelCase	)    ->  List[Any]:

 '''simple docstring'''






 snake_case__    							=				default_data_collator(_UpperCAmelCase	)
 snake_case__    							=				batch.pop("pairID"	,     _UpperCAmelCase	)
 return batch





def 							a__			(  )    ->  Tuple:

 '''simple docstring'''






 snake_case__    							=				HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)	)
 snake_case__    							=				parser.parse_args_into_dataclasses()

 if (
     os.path.exists(training_args.output_dir	)
     and os.listdir(training_args.output_dir	)
     and training_args.do_train
     and not training_args.overwrite_output_dir
 ):
  raise ValueError(
      f"Output directory ({training_args.output_dir}) already exists and is not empty. Use"
      " --overwrite_output_dir to overcome."	)

 # Setup logging
 logging.basicConfig(
     format="%(asctime)s - %(levelname)s - %(name)s - %(message)s"	,     datefmt="%m/%d/%Y %H:%M:%S"	,     level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN	,     )
 logger.warning(
     "Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s"	,     training_args.local_rank	,     training_args.device	,     training_args.n_gpu	,     bool(training_args.local_rank != -1	)	,     training_args.fpaa	,     )
 # Set the verbosity to info of the Transformers logger (on main process only):
 if is_main_process(training_args.local_rank	):
  transformers.utils.logging.set_verbosity_info()
  transformers.utils.logging.enable_default_handler()
  transformers.utils.logging.enable_explicit_format()
 logger.info("Training/evaluation parameters %s"	,     _UpperCAmelCase	)

 # Set seed
 set_seed(training_args.seed	)

 try:
  snake_case__    							=				hans_tasks_num_labels[data_args.task_name]
 except KeyError:
  raise ValueError("Task not found: %s" % (data_args.task_name)	)

 # Load pretrained model and tokenizer
 #
 # Distributed training:
 # The .from_pretrained methods guarantee that only one local process can concurrently
 # download model & vocab.

 snake_case__    							=				AutoConfig.from_pretrained(
     model_args.config_name if model_args.config_name else model_args.model_name_or_path	,     num_labels=_UpperCAmelCase	,     finetuning_task=data_args.task_name	,     cache_dir=model_args.cache_dir	,     )
 snake_case__    							=				AutoTokenizer.from_pretrained(
     model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path	,     cache_dir=model_args.cache_dir	,     )
 snake_case__    							=				AutoModelForSequenceClassification.from_pretrained(
     model_args.model_name_or_path	,     from_tf=bool(".ckpt" in model_args.model_name_or_path	)	,     config=_UpperCAmelCase	,     cache_dir=model_args.cache_dir	,     )

 # Get datasets
 snake_case__    							=				(
     HansDataset(
         data_dir=data_args.data_dir	,     tokenizer=_UpperCAmelCase	,     task=data_args.task_name	,     max_seq_length=data_args.max_seq_length	,     overwrite_cache=data_args.overwrite_cache	,     )
     if training_args.do_train
     else None
 )
 snake_case__    							=				(
     HansDataset(
         data_dir=data_args.data_dir	,     tokenizer=_UpperCAmelCase	,     task=data_args.task_name	,     max_seq_length=data_args.max_seq_length	,     overwrite_cache=data_args.overwrite_cache	,     evaluate=_UpperCAmelCase	,     )
     if training_args.do_eval
     else None
 )

 # Initialize our Trainer
 snake_case__    							=				Trainer(
     model=_UpperCAmelCase	,     args=_UpperCAmelCase	,     train_dataset=_UpperCAmelCase	,     eval_dataset=_UpperCAmelCase	,     data_collator=_UpperCAmelCase	,     )

 # Training
 if training_args.do_train:
  trainer.train(
      model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path	) else None	)
  trainer.save_model()
  # For convenience, we also re-save the tokenizer to the same directory,
  # so that you can share your model easily on huggingface.co/models =)
  if trainer.is_world_master():
   tokenizer.save_pretrained(training_args.output_dir	)

    # Evaluation
 if training_args.do_eval:
  logger.info("*** Evaluate ***"	)

  snake_case__    							=				trainer.predict(_UpperCAmelCase	)
  snake_case__    							=				output.predictions
  snake_case__    							=				np.argmax(_UpperCAmelCase	,     axis=1	)

  snake_case__    							=				[ex.pairID for ex in eval_dataset]
  snake_case__    							=				os.path.join(training_args.output_dir	,     "hans_predictions.txt"	)
  snake_case__    							=				eval_dataset.get_labels()
  if trainer.is_world_master():
   with open(_UpperCAmelCase	,     "w"	) as writer:
    writer.write("pairID,gold_label\n"	)
    for pid, pred in zip(_UpperCAmelCase	,     _UpperCAmelCase	):
     writer.write("ex" + str(_UpperCAmelCase	) + "," + label_list[int(_UpperCAmelCase	)] + "\n"	)

  trainer._log(output.metrics	)





def 							a__			(  _UpperCAmelCase	)    ->  List[Any]:

 '''simple docstring'''






 main()


if __name__ == "__main__":
   main()








"""simple docstring"""


import json
import os
import random
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import BertTokenizer, BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
   from PIL import Image

   from transformers import FlavaImageProcessor, FlavaProcessor
   from transformers.models.flava.image_processing_flava import (
       FLAVA_CODEBOOK_MEAN,
       FLAVA_CODEBOOK_STD,
       FLAVA_IMAGE_MEAN,
       FLAVA_IMAGE_STD,
   )



@require_vision
class      lowercase_				(     unittest.TestCase							):

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        snake_case__    							=				tempfile.mkdtemp()

        # fmt: off
        snake_case__    							=				["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "want", "##want", "##ed", "wa", "un", "runn", "##ing", ",", "low", "lowest"]
        # fmt: on
        snake_case__    							=				os.path.join(self.tmpdirname   ,					VOCAB_FILES_NAMES["vocab_file"]     )

        with open(self.vocab_file   ,					"w"   ,					encoding="utf-8"     ) as fp:
         fp.write("".join([x + "\n" for x in vocab_tokens]     )     )

        snake_case__    							=				{
            "image_mean": FLAVA_IMAGE_MEAN,
            "image_std": FLAVA_IMAGE_STD,
            "do_normalize": True,
            "do_resize": True,
            "size": 224,
            "do_center_crop": True,
            "crop_size": 224,
            "input_size_patches": 14,
            "total_mask_patches": 75,
            "mask_group_max_patches": None,
            "mask_group_min_patches": 16,
            "mask_group_min_aspect_ratio": 0.3,
            "mask_group_max_aspect_ratio": None,
            "codebook_do_resize": True,
            "codebook_size": 112,
            "codebook_do_center_crop": True,
            "codebook_crop_size": 112,
            "codebook_do_map_pixels": True,
            "codebook_do_normalize": True,
            "codebook_image_mean": FLAVA_CODEBOOK_MEAN,
            "codebook_image_std": FLAVA_CODEBOOK_STD,
        }

        snake_case__    							=				os.path.join(self.tmpdirname   ,					__SCREAMING_SNAKE_CASE     )
        with open(self.image_processor_file   ,					"w"   ,					encoding="utf-8"     ) as fp:
         json.dump(__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE     )

       def 		_SCREAMING_SNAKE_CASE(			self   ,					**__SCREAMING_SNAKE_CASE     ):






        '''simple docstring'''

        return BertTokenizer.from_pretrained(self.tmpdirname   ,					**__SCREAMING_SNAKE_CASE     )

       def 		_SCREAMING_SNAKE_CASE(			self   ,					**__SCREAMING_SNAKE_CASE     ):






        '''simple docstring'''

        return BertTokenizerFast.from_pretrained(self.tmpdirname   ,					**__SCREAMING_SNAKE_CASE     )

       def 		_SCREAMING_SNAKE_CASE(			self   ,					**__SCREAMING_SNAKE_CASE     ):






        '''simple docstring'''

        return FlavaImageProcessor.from_pretrained(self.tmpdirname   ,					**__SCREAMING_SNAKE_CASE     )

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        shutil.rmtree(self.tmpdirname     )

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        snake_case__    							=				[np.random.randint(255   ,					size=(3, 30, 400)   ,					dtype=np.uinta     )]

        snake_case__    							=				[Image.fromarray(np.moveaxis(__SCREAMING_SNAKE_CASE   ,					0   ,					-1     )     ) for x in image_inputs]

        return image_inputs

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        snake_case__    							=				self.get_tokenizer()
        snake_case__    							=				self.get_rust_tokenizer()
        snake_case__    							=				self.get_image_processor()

        snake_case__    							=				FlavaProcessor(tokenizer=__SCREAMING_SNAKE_CASE   ,					image_processor=__SCREAMING_SNAKE_CASE     )
        processor_slow.save_pretrained(self.tmpdirname     )
        snake_case__    							=				FlavaProcessor.from_pretrained(self.tmpdirname   ,					use_fast=__SCREAMING_SNAKE_CASE     )

        snake_case__    							=				FlavaProcessor(tokenizer=__SCREAMING_SNAKE_CASE   ,					image_processor=__SCREAMING_SNAKE_CASE     )
        processor_fast.save_pretrained(self.tmpdirname     )
        snake_case__    							=				FlavaProcessor.from_pretrained(self.tmpdirname     )

        self.assertEqual(processor_slow.tokenizer.get_vocab()   ,					tokenizer_slow.get_vocab()     )
        self.assertEqual(processor_fast.tokenizer.get_vocab()   ,					tokenizer_fast.get_vocab()     )
        self.assertEqual(tokenizer_slow.get_vocab()   ,					tokenizer_fast.get_vocab()     )
        self.assertIsInstance(processor_slow.tokenizer   ,					__SCREAMING_SNAKE_CASE     )
        self.assertIsInstance(processor_fast.tokenizer   ,					__SCREAMING_SNAKE_CASE     )

        self.assertEqual(processor_slow.image_processor.to_json_string()   ,					image_processor.to_json_string()     )
        self.assertEqual(processor_fast.image_processor.to_json_string()   ,					image_processor.to_json_string()     )
        self.assertIsInstance(processor_slow.image_processor   ,					__SCREAMING_SNAKE_CASE     )
        self.assertIsInstance(processor_fast.image_processor   ,					__SCREAMING_SNAKE_CASE     )

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        snake_case__    							=				FlavaProcessor(tokenizer=self.get_tokenizer()   ,					image_processor=self.get_image_processor()     )
        processor.save_pretrained(self.tmpdirname     )

        snake_case__    							=				self.get_tokenizer(bos_token="(BOS)"   ,					eos_token="(EOS)"     )
        snake_case__    							=				self.get_image_processor(do_normalize=__SCREAMING_SNAKE_CASE   ,					padding_value=1.0     )

        snake_case__    							=				FlavaProcessor.from_pretrained(
            self.tmpdirname   ,					bos_token="(BOS)"   ,					eos_token="(EOS)"   ,					do_normalize=__SCREAMING_SNAKE_CASE   ,					padding_value=1.0     )

        self.assertEqual(processor.tokenizer.get_vocab()   ,					tokenizer_add_kwargs.get_vocab()     )
        self.assertIsInstance(processor.tokenizer   ,					__SCREAMING_SNAKE_CASE     )

        self.assertEqual(processor.image_processor.to_json_string()   ,					image_processor_add_kwargs.to_json_string()     )
        self.assertIsInstance(processor.image_processor   ,					__SCREAMING_SNAKE_CASE     )

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        snake_case__    							=				self.get_image_processor()
        snake_case__    							=				self.get_tokenizer()

        snake_case__    							=				FlavaProcessor(tokenizer=__SCREAMING_SNAKE_CASE   ,					image_processor=__SCREAMING_SNAKE_CASE     )

        snake_case__    							=				self.prepare_image_inputs()

        snake_case__    							=				image_processor(__SCREAMING_SNAKE_CASE   ,					return_tensors="np"     )
        snake_case__    							=				processor(images=__SCREAMING_SNAKE_CASE   ,					return_tensors="np"     )

        for key in input_feat_extract.keys():
         self.assertAlmostEqual(input_feat_extract[key].sum()   ,					input_processor[key].sum()   ,					delta=1e-2     )

        # With rest of the args
        random.seed(1234     )
        snake_case__    							=				image_processor(
            __SCREAMING_SNAKE_CASE   ,					return_image_mask=__SCREAMING_SNAKE_CASE   ,					return_codebook_pixels=__SCREAMING_SNAKE_CASE   ,					return_tensors="np"     )
        random.seed(1234     )
        snake_case__    							=				processor(
            images=__SCREAMING_SNAKE_CASE   ,					return_image_mask=__SCREAMING_SNAKE_CASE   ,					return_codebook_pixels=__SCREAMING_SNAKE_CASE   ,					return_tensors="np"     )

        for key in input_feat_extract.keys():
         self.assertAlmostEqual(input_feat_extract[key].sum()   ,					input_processor[key].sum()   ,					delta=1e-2     )

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        snake_case__    							=				self.get_image_processor()
        snake_case__    							=				self.get_tokenizer()

        snake_case__    							=				FlavaProcessor(tokenizer=__SCREAMING_SNAKE_CASE   ,					image_processor=__SCREAMING_SNAKE_CASE     )

        snake_case__    							=				"lower newer"

        snake_case__    							=				processor(text=__SCREAMING_SNAKE_CASE     )

        snake_case__    							=				tokenizer(__SCREAMING_SNAKE_CASE     )

        for key in encoded_tok.keys():
         self.assertListEqual(encoded_tok[key]   ,					encoded_processor[key]     )

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        snake_case__    							=				self.get_image_processor()
        snake_case__    							=				self.get_tokenizer()

        snake_case__    							=				FlavaProcessor(tokenizer=__SCREAMING_SNAKE_CASE   ,					image_processor=__SCREAMING_SNAKE_CASE     )

        snake_case__    							=				"lower newer"
        snake_case__    							=				self.prepare_image_inputs()

        snake_case__    							=				processor(text=__SCREAMING_SNAKE_CASE   ,					images=__SCREAMING_SNAKE_CASE     )

        self.assertListEqual(list(inputs.keys()     )   ,					["input_ids", "token_type_ids", "attention_mask", "pixel_values"]     )

        # add extra args
        snake_case__    							=				processor(text=__SCREAMING_SNAKE_CASE   ,					images=__SCREAMING_SNAKE_CASE   ,					return_codebook_pixels=__SCREAMING_SNAKE_CASE   ,					return_image_mask=__SCREAMING_SNAKE_CASE     )

        self.assertListEqual(
            list(inputs.keys()     )   ,					[
                "input_ids",
                "token_type_ids",
                "attention_mask",
                "pixel_values",
                "codebook_pixel_values",
                "bool_masked_pos",
            ]   ,					)

        # test if it raises when no input is passed
        with pytest.raises(__SCREAMING_SNAKE_CASE     ):
         processor()

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        snake_case__    							=				self.get_image_processor()
        snake_case__    							=				self.get_tokenizer()

        snake_case__    							=				FlavaProcessor(tokenizer=__SCREAMING_SNAKE_CASE   ,					image_processor=__SCREAMING_SNAKE_CASE     )

        snake_case__    							=				[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

        snake_case__    							=				processor.batch_decode(__SCREAMING_SNAKE_CASE     )
        snake_case__    							=				tokenizer.batch_decode(__SCREAMING_SNAKE_CASE     )

        self.assertListEqual(__SCREAMING_SNAKE_CASE   ,					__SCREAMING_SNAKE_CASE     )

       def 		_SCREAMING_SNAKE_CASE(			self     ):






        '''simple docstring'''

        snake_case__    							=				self.get_image_processor()
        snake_case__    							=				self.get_tokenizer()

        snake_case__    							=				FlavaProcessor(tokenizer=__SCREAMING_SNAKE_CASE   ,					image_processor=__SCREAMING_SNAKE_CASE     )

        snake_case__    							=				"lower newer"
        snake_case__    							=				self.prepare_image_inputs()

        snake_case__    							=				processor(text=__SCREAMING_SNAKE_CASE   ,					images=__SCREAMING_SNAKE_CASE     )

        self.assertListEqual(list(inputs.keys()     )   ,					processor.model_input_names     )

