





import argparse
import glob
import logging
import os
import time
from argparse import Namespace

import numpy as np
import torch
from lightning_base import BaseTransformer, add_generic_args, generic_train
from torch.utils.data import DataLoader, TensorDataset

from transformers import glue_compute_metrics as compute_metrics
from transformers import glue_convert_examples_to_features as convert_examples_to_features
from transformers import glue_output_modes, glue_tasks_num_labels
from transformers import glue_processors as processors


A__:    int  	=	logging.getLogger(__name__)




class        _a    (    _snake_case):

      """simple docstring"""


      UpperCamelCase__  		=       """sequence-classification"""






      def __init__(		self:			Any	,    __lowerCamelCase:			str					):
          '''simple docstring'''


          if type(UpperCamelCase__					) == dict:
              UpperCamelCase__:	Any					=     Namespace(**UpperCamelCase__					)
          UpperCamelCase__:	Optional[int]					=     glue_output_modes[hparams.task]
          UpperCamelCase__:	str					=     glue_tasks_num_labels[hparams.task]

          super().__init__(UpperCamelCase__	,    UpperCamelCase__	,    self.mode					)






      def 			UpperCAmelCase_       (		self:			Optional[Any]	,    **__lowerCamelCase:			Optional[int]					):
          '''simple docstring'''


          return self.model(**UpperCamelCase__					)






      def 			UpperCAmelCase_       (		self:			List[str]	,    __lowerCamelCase:			List[str]	,    __lowerCamelCase:			int					):
          '''simple docstring'''


          UpperCamelCase__:	Optional[Any]					=     {"input_ids": batch[0], "attention_mask": batch[1], "labels": batch[3]}

          if self.config.model_type not in ["distilbert", "bart"]:
              UpperCamelCase__:	Tuple					=     batch[2] if self.config.model_type in ["bert", "xlnet", "albert"] else None

          UpperCamelCase__:	List[str]					=     self(**UpperCamelCase__					)
          UpperCamelCase__:	int					=     outputs[0]

          UpperCamelCase__:	List[Any]					=     self.trainer.lr_schedulers[0]["scheduler"]
          UpperCamelCase__:	Dict					=     {"loss": loss, "rate": lr_scheduler.get_last_lr()[-1]}
          return {"loss": loss, "log": tensorboard_logs}






      def 			UpperCAmelCase_       (		self:			Any					):
          '''simple docstring'''


          UpperCamelCase__:	Any					=     self.hparams
          UpperCamelCase__:	int					=     processors[args.task]()
          UpperCamelCase__:	Tuple					=     processor.get_labels()

          for mode in ["train", "dev"]:
              UpperCamelCase__:	Optional[Any]					=     self._feature_file(UpperCamelCase__					)
              if os.path.exists(UpperCamelCase__					) and not args.overwrite_cache:
                  logger.info("Loading features from cached file %s"	,    UpperCamelCase__					)
              else:
                  logger.info("Creating features from dataset file at %s"	,    args.data_dir					)
                  UpperCamelCase__:	Union[str, Any]					=     (
                      processor.get_dev_examples(args.data_dir					)
                      if mode == "dev"
                      else processor.get_train_examples(args.data_dir					)
                  )
                  UpperCamelCase__:	Union[str, Any]					=     convert_examples_to_features(
                      UpperCamelCase__	,    self.tokenizer	,    max_length=args.max_seq_length	,    label_list=self.labels	,    output_mode=args.glue_output_mode	,    )
                  logger.info("Saving features into cached file %s"	,    UpperCamelCase__					)
                  torch.save(UpperCamelCase__	,    UpperCamelCase__					)






      def 			UpperCAmelCase_       (		self:			List[Any]	,    __lowerCamelCase:			Dict	,    __lowerCamelCase:			str	,    __lowerCamelCase:			List[Any] = False					):
          '''simple docstring'''


          UpperCamelCase__:	str					=     "dev" if mode == "test" else mode

          UpperCamelCase__:	str					=     self._feature_file(UpperCamelCase__					)
          logger.info("Loading features from cached file %s"	,    UpperCamelCase__					)
          UpperCamelCase__:	Union[str, Any]					=     torch.load(UpperCamelCase__					)
          UpperCamelCase__:	Any					=     torch.tensor([f.input_ids for f in features]	,    dtype=torch.long					)
          UpperCamelCase__:	Tuple					=     torch.tensor([f.attention_mask for f in features]	,    dtype=torch.long					)
          UpperCamelCase__:	int					=     torch.tensor([f.token_type_ids for f in features]	,    dtype=torch.long					)
          if self.hparams.glue_output_mode == "classification":
              UpperCamelCase__:	Union[str, Any]					=     torch.tensor([f.label for f in features]	,    dtype=torch.long					)
          elif self.hparams.glue_output_mode == "regression":
              UpperCamelCase__:	Optional[Any]					=     torch.tensor([f.label for f in features]	,    dtype=torch.float					)

          return DataLoader(
              TensorDataset(UpperCamelCase__	,    UpperCamelCase__	,    UpperCamelCase__	,    UpperCamelCase__					)	,    batch_size=UpperCamelCase__	,    shuffle=UpperCamelCase__	,    )






      def 			UpperCAmelCase_       (		self:			Union[str, Any]	,    __lowerCamelCase:			List[Any]	,    __lowerCamelCase:			Tuple					):
          '''simple docstring'''


          UpperCamelCase__:	Dict					=     {"input_ids": batch[0], "attention_mask": batch[1], "labels": batch[3]}

          if self.config.model_type not in ["distilbert", "bart"]:
              UpperCamelCase__:	Optional[int]					=     batch[2] if self.config.model_type in ["bert", "xlnet", "albert"] else None

          UpperCamelCase__:	Optional[Any]					=     self(**UpperCamelCase__					)
          UpperCamelCase__				, UpperCamelCase__:	Dict					=     outputs[:2]
          UpperCamelCase__:	Union[str, Any]					=     logits.detach().cpu().numpy()
          UpperCamelCase__:	Any					=     inputs["labels"].detach().cpu().numpy()

          return {"val_loss": tmp_eval_loss.detach().cpu(), "pred": preds, "target": out_label_ids}






      def 			UpperCAmelCase_       (		self:			str	,    __lowerCamelCase:			int					):
          '''simple docstring'''


          UpperCamelCase__:	Tuple					=     torch.stack([x["val_loss"] for x in outputs]					).mean().detach().cpu().item()
          UpperCamelCase__:	List[str]					=     np.concatenate([x["pred"] for x in outputs]	,    axis=0					)

          if self.hparams.glue_output_mode == "classification":
              UpperCamelCase__:	Any					=     np.argmax(UpperCamelCase__	,    axis=1					)
          elif self.hparams.glue_output_mode == "regression":
              UpperCamelCase__:	str					=     np.squeeze(UpperCamelCase__					)

          UpperCamelCase__:	List[str]					=     np.concatenate([x["target"] for x in outputs]	,    axis=0					)
          UpperCamelCase__:	int					=     [[] for _ in range(out_label_ids.shape[0]					)]
          UpperCamelCase__:	Optional[Any]					=     [[] for _ in range(out_label_ids.shape[0]					)]

          UpperCamelCase__:	Any					=     {**{"val_loss": val_loss_mean}, **compute_metrics(self.hparams.task	,    UpperCamelCase__	,    UpperCamelCase__					)}

          UpperCamelCase__:	Optional[Any]					=     dict(results.items()					)
          UpperCamelCase__:	List[Any]					=     results
          return ret, preds_list, out_label_list






      def 			UpperCAmelCase_       (		self:			Union[str, Any]	,    __lowerCamelCase:			Union[str, Any]					):
          '''simple docstring'''


          UpperCamelCase__				, UpperCamelCase__				, UpperCamelCase__:	str					=     self._eval_end(UpperCamelCase__					)
          UpperCamelCase__:	Tuple					=     ret["log"]
          return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}






      def 			UpperCAmelCase_       (		self:			Union[str, Any]	,    __lowerCamelCase:			Union[str, Any]					):
          '''simple docstring'''


          UpperCamelCase__				, UpperCamelCase__				, UpperCamelCase__:	Tuple					=     self._eval_end(UpperCamelCase__					)
          UpperCamelCase__:	int					=     ret["log"]
          # `val_loss` is the key returned by `self._eval_end()` but actually refers to `test_loss`
          return {"avg_test_loss": logs["val_loss"], "log": logs, "progress_bar": logs}






      @staticmethod
      def 			UpperCAmelCase_       (		__lowerCamelCase:			str	,    __lowerCamelCase:			Union[str, Any]					):
          '''simple docstring'''


          BaseTransformer.add_model_specific_args(UpperCamelCase__	,    UpperCamelCase__					)
          parser.add_argument(
              "--max_seq_length"	,    default=128	,    type=UpperCamelCase__	,    help=(
                  "The maximum total input sequence length after tokenization. Sequences longer "
                  "than this will be truncated, sequences shorter will be padded."
              )	,    )

          parser.add_argument(
              "--task"	,    default=""	,    type=UpperCamelCase__	,    required=UpperCamelCase__	,    help="The GLUE task to run"	,    )
          parser.add_argument(
              "--gpus"	,    default=0	,    type=UpperCamelCase__	,    help="The number of GPUs allocated for this, it is by default 0 meaning none"	,    )

          parser.add_argument(
              "--overwrite_cache"	,    action="store_true"	,    help="Overwrite the cached training and evaluation sets"					)

          return parser



def 				lowerCAmelCase_ (							):
    UpperCamelCase__:	int					=     argparse.ArgumentParser()
    add_generic_args(UpperCAmelCase__    ,os.getcwd())
    UpperCamelCase__:	str					=     GLUETransformer.add_model_specific_args(UpperCAmelCase__    ,os.getcwd())
    UpperCamelCase__:	Any					=     parser.parse_args()

    # If output_dir not provided, a folder will be generated in pwd
    if args.output_dir is None:
        UpperCamelCase__:	int					=     os.path.join(
            "./results"    ,F"{args.task}_{time.strftime('%Y%m%d_%H%M%S')}"    ,)
        os.makedirs(args.output_dir)

    UpperCamelCase__:	int					=     GLUETransformer(UpperCAmelCase__)
    UpperCamelCase__:	Optional[Any]					=     generic_train(UpperCAmelCase__    ,UpperCAmelCase__)

    # Optionally, predict on dev set and write to output_dir
    if args.do_predict:
        UpperCamelCase__:	Tuple					=     sorted(glob.glob(os.path.join(args.output_dir    ,"checkpoint-epoch=*.ckpt")    ,recursive=UpperCAmelCase__))
        UpperCamelCase__:	Union[str, Any]					=     model.load_from_checkpoint(checkpoints[-1])
        return trainer.test(UpperCAmelCase__)


if __name__ == "__main__":
   main()




'''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        A__   :



    def __init__(       self		,					UpperCamelCase__		,					UpperCamelCase__=12		,					UpperCamelCase__=7		,					UpperCamelCase__=True		,					UpperCamelCase__=True		,					UpperCamelCase__=True		,					UpperCamelCase__=99		,					UpperCamelCase__=32		,					UpperCamelCase__=32		,					UpperCamelCase__=2		,					UpperCamelCase__=4		,					UpperCamelCase__=37		,					UpperCamelCase__=0.1		,					UpperCamelCase__=0.1		,					UpperCamelCase__=512		,					UpperCamelCase__=0.02		,					UpperCamelCase__=0		,					UpperCamelCase__=None		,					)    ->   Union[str, Any]:





           '''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        snake_case_      (       self							)    ->   Dict:





           '''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(UpperCamelCase__							):
                         A_						       =   1
                         A_						       =   0

           A_						       =   self.get_config()

           return config, input_ids, tf.convert_to_tensor(UpperCamelCase__							)



    def        snake_case_      (       self							)    ->   Optional[int]:





           '''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__							)    ->   int:





           '''simple docstring'''
           A_						       =   TFBlipTextModel(config=UpperCamelCase__							)
           A_						       =   model(UpperCamelCase__		,					attention_mask=UpperCamelCase__		,					training=UpperCamelCase__							)
           A_						       =   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							)    ->   List[Any]:





           '''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        A__   (   _snake_case    ,    unittest.TestCase      ):
    lowercase						       =		(TFBlipTextModel,) if is_tf_available() else ()
    lowercase						       =		False
    lowercase						       =		False
    lowercase						       =		False



    def        snake_case_      (       self							)    ->   Optional[int]:





           '''simple docstring'''
           A_						       =   BlipTextModelTester(self							)
           A_						       =   ConfigTester(self		,					config_class=UpperCamelCase__		,					hidden_size=37							)



    def        snake_case_      (       self							)    ->   Tuple:





           '''simple docstring'''
           self.config_tester.run_common_tests()



    def        snake_case_      (       self							)    ->   str:





           '''simple docstring'''
           A_						       =   self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_model(*UpperCamelCase__							)



    def        snake_case_      (       self							)    ->   List[Any]:





           '''simple docstring'''
           pass



    def        snake_case_      (       self							)    ->   Union[str, Any]:





           '''simple docstring'''
           pass



    @unittest.skip(reason="""Blip does not use inputs_embeds"""							)
    def        snake_case_      (       self							)    ->   Union[str, Any]:





           '''simple docstring'''
           pass



    @unittest.skip(reason="""BlipTextModel has no base class and is not available in MODEL_MAPPING"""							)
    def        snake_case_      (       self							)    ->   List[Any]:





           '''simple docstring'''
           pass



    @unittest.skip(reason="""BlipTextModel has no base class and is not available in MODEL_MAPPING"""							)
    def        snake_case_      (       self							)    ->   Any:





           '''simple docstring'''
           pass



    @slow
    def        snake_case_      (       self							)    ->   List[Any]:





           '''simple docstring'''
           for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                  A_						       =   TFBlipTextModel.from_pretrained(UpperCamelCase__							)
                  self.assertIsNotNone(UpperCamelCase__							)







    def        snake_case_      (       self		,					UpperCamelCase__=True							)    ->   str:





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