




"""simple docstring"""


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_      			=       logging.getLogger(__name__)






class       lowercase(  __a				):







 '''simple docstring'''





 lowercase__				      =      "sequence-classification"


 def __init__(       self:	Union[str, Any],   a_:	Any	):




   '''simple docstring'''



   if type(a_	) == dict:
     _snake_case							:	Dict			=						Namespace(**a_	)
   _snake_case							:	List[Any]			=						glue_output_modes[hparams.task]
   _snake_case							:	Optional[Any]			=						glue_tasks_num_labels[hparams.task]

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


 def 				UpperCamelCase_     (       self:	List[Any],   **a_:	Optional[Any]	):




   '''simple docstring'''



   return self.model(**a_	)


 def 				UpperCamelCase_     (       self:	Optional[int],   a_:	List[Any],   a_:	Union[str, Any]	):




   '''simple docstring'''



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

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

   _snake_case							:	Union[str, Any]			=						self(**a_	)
   _snake_case							:	List[str]			=						outputs[0]

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


 def 				UpperCamelCase_     (       self:	Optional[Any]	):




   '''simple docstring'''



   _snake_case							:	Dict			=						self.hparams
   _snake_case							:	List[str]			=						processors[args.task]()
   _snake_case							:	Optional[Any]			=						processor.get_labels()

   for mode in ["train", "dev"]:
     _snake_case							:	Union[str, Any]			=						self._feature_file(a_	)
     if os.path.exists(a_	) and not args.overwrite_cache:
       logger.info("""Loading features from cached file %s""",   a_	)
     else:
       logger.info("""Creating features from dataset file at %s""",   args.data_dir	)
       _snake_case							:	Any			=						(
           processor.get_dev_examples(args.data_dir	)
           if mode == """dev"""
           else processor.get_train_examples(args.data_dir	)
       )
       _snake_case							:	str			=						convert_examples_to_features(
           a_,   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""",   a_	)
       torch.save(a_,   a_	)


 def 				UpperCamelCase_     (       self:	str,   a_:	str,   a_:	int,   a_:	bool = False	):




   '''simple docstring'''



   _snake_case							:	Optional[int]			=						"""dev""" if mode == """test""" else mode

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

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


 def 				UpperCamelCase_     (       self:	Tuple,   a_:	Dict,   a_:	List[str]	):




   '''simple docstring'''



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

   if self.config.model_type not in ["distilbert", "bart"]:
     _snake_case							:	Union[str, Any]			=						batch[2] if self.config.model_type in ["""bert""", """xlnet""", """albert"""] else None

   _snake_case							:	Optional[int]			=						self(**a_	)
   _snake_case	,      _snake_case							:	Union[str, Any]			=						outputs[:2]
   _snake_case							:	List[str]			=						logits.detach().cpu().numpy()
   _snake_case							:	Union[str, Any]			=						inputs["""labels"""].detach().cpu().numpy()

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


 def 				UpperCamelCase_     (       self:	Tuple,   a_:	int	):




   '''simple docstring'''



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

   if self.hparams.glue_output_mode == "classification":
     _snake_case							:	Tuple			=						np.argmax(a_,   axis=1	)
   elif self.hparams.glue_output_mode == "regression":
     _snake_case							:	Optional[int]			=						np.squeeze(a_	)

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

   _snake_case							:	Tuple			=						{**{"""val_loss""": val_loss_mean}, **compute_metrics(self.hparams.task,   a_,   a_	)}

   _snake_case							:	Optional[int]			=						dict(results.items()	)
   _snake_case							:	str			=						results
   return ret, preds_list, out_label_list


 def 				UpperCamelCase_     (       self:	str,   a_:	list	):




   '''simple docstring'''



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


 def 				UpperCamelCase_     (       self:	str,   a_:	Any	):




   '''simple docstring'''



   _snake_case	,      _snake_case	,      _snake_case							:	str			=						self._eval_end(a_	)
   _snake_case							:	List[str]			=						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_     (       a_:	Optional[Any],   a_:	Tuple	):




   '''simple docstring'''



   BaseTransformer.add_model_specific_args(a_,   a_	)
   parser.add_argument(
       """--max_seq_length""",   default=128,   type=a_,   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=a_,   required=a_,   help="""The GLUE task to run""",   )
   parser.add_argument(
       """--gpus""",   default=0,   type=a_,   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       UpperCAmelCase__ ():
  """simple docstring"""

  _snake_case							:	Dict			=						argparse.ArgumentParser()
  add_generic_args(snake_case__	,     os.getcwd()					)
  _snake_case							:	Optional[Any]			=						GLUETransformer.add_model_specific_args(snake_case__	,     os.getcwd()					)
  _snake_case							:	Tuple			=						parser.parse_args()

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

  _snake_case							:	Optional[Any]			=						GLUETransformer(snake_case__					)
  _snake_case							:	Union[str, Any]			=						generic_train(snake_case__	,     snake_case__					)

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


if __name__ == "__main__":
     main()








"""simple docstring"""


import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


A_      			=       logging.get_logger(__name__)

A_      			=       {
    '''BridgeTower/bridgetower-base''': '''https://huggingface.co/BridgeTower/bridgetower-base/blob/main/config.json''',
    '''BridgeTower/bridgetower-base-itm-mlm''': (
        '''https://huggingface.co/BridgeTower/bridgetower-base-itm-mlm/blob/main/config.json'''
    ),
}






class       lowercase(  __a				):







 '''simple docstring'''





 lowercase__				      =      "bridgetower_vision_model"


 def __init__(       self:	Tuple,   a_:	str=768,   a_:	Union[str, Any]=12,   a_:	List[str]=3,   a_:	Optional[int]=16,   a_:	List[Any]=288,   a_:	Optional[Any]=1,   a_:	Any=1E-05,   a_:	Dict=False,   a_:	Any=True,   a_:	int=False,   **a_:	int,   ):




   '''simple docstring'''



   super().__init__(**a_	)
   _snake_case							:	str			=						hidden_size
   _snake_case							:	int			=						num_hidden_layers
   _snake_case							:	Any			=						num_channels
   _snake_case							:	Union[str, Any]			=						patch_size
   _snake_case							:	Dict			=						image_size
   _snake_case							:	Optional[Any]			=						initializer_factor
   _snake_case							:	Any			=						layer_norm_eps
   _snake_case							:	int			=						stop_gradient
   _snake_case							:	Any			=						share_layernorm
   _snake_case							:	List[Any]			=						remove_last_layer


 @classmethod
 def 				UpperCamelCase_     (       cls:	Union[str, Any],   a_:	Union[str, os.PathLike],   **a_:	Optional[Any]	):




   '''simple docstring'''



   _snake_case	,      _snake_case							:	List[Any]			=						cls.get_config_dict(a_,   **a_	)

   if config_dict.get("""model_type"""	) == "bridgetower":
     _snake_case							:	str			=						config_dict["""text_config"""]

   if "model_type" in config_dict and hasattr(cls,   """model_type"""	) and config_dict["model_type"] != cls.model_type:
     logger.warning(
         f"You are using a model of type {config_dict['model_type']} to instantiate a model of type "
         f"{cls.model_type}. This is not supported for all configurations of models and can yield errors."	)

   return cls.from_dict(a_,   **a_	)







class       lowercase(  __a				):







 '''simple docstring'''





 lowercase__				      =      "bridgetower_text_model"


 def __init__(       self:	str,   a_:	Dict=50_265,   a_:	List[Any]=768,   a_:	Union[str, Any]=12,   a_:	List[str]=12,   a_:	str=1,   a_:	Optional[Any]=3_072,   a_:	int="gelu",   a_:	int=0.1,   a_:	int=0.1,   a_:	Optional[int]=514,   a_:	Tuple=1,   a_:	Tuple=1E-05,   a_:	Optional[int]=1,   a_:	Union[str, Any]=0,   a_:	str=2,   a_:	Any="absolute",   a_:	List[Any]=True,   **a_:	Union[str, Any],   ):




   '''simple docstring'''



   super().__init__(**a_	)

   _snake_case							:	str			=						vocab_size
   _snake_case							:	Optional[int]			=						hidden_size
   _snake_case							:	Dict			=						num_hidden_layers
   _snake_case							:	Optional[int]			=						num_attention_heads
   _snake_case							:	Optional[int]			=						hidden_act
   _snake_case							:	List[Any]			=						initializer_factor
   _snake_case							:	Optional[int]			=						intermediate_size
   _snake_case							:	int			=						hidden_dropout_prob
   _snake_case							:	Tuple			=						attention_probs_dropout_prob
   _snake_case							:	List[str]			=						max_position_embeddings
   _snake_case							:	Optional[int]			=						type_vocab_size
   _snake_case							:	List[Any]			=						layer_norm_eps
   _snake_case							:	Dict			=						position_embedding_type
   _snake_case							:	Dict			=						use_cache
   _snake_case							:	int			=						pad_token_id
   _snake_case							:	Union[str, Any]			=						bos_token_id
   _snake_case							:	Union[str, Any]			=						eos_token_id


 @classmethod
 def 				UpperCamelCase_     (       cls:	str,   a_:	Union[str, os.PathLike],   **a_:	int	):




   '''simple docstring'''



   _snake_case	,      _snake_case							:	Optional[int]			=						cls.get_config_dict(a_,   **a_	)

   if config_dict.get("""model_type"""	) == "bridgetower":
     _snake_case							:	Union[str, Any]			=						config_dict["""text_config"""]

   if "model_type" in config_dict and hasattr(cls,   """model_type"""	) and config_dict["model_type"] != cls.model_type:
     logger.warning(
         f"You are using a model of type {config_dict['model_type']} to instantiate a model of type "
         f"{cls.model_type}. This is not supported for all configurations of models and can yield errors."	)

   return cls.from_dict(a_,   **a_	)







class       lowercase(  __a				):







 '''simple docstring'''





 lowercase__				      =      "bridgetower"


 def __init__(       self:	int,   a_:	List[str]=True,   a_:	Any="gelu",   a_:	List[Any]=768,   a_:	int=1,   a_:	Optional[int]=1E-05,   a_:	Tuple=False,   a_:	Optional[Any]="add",   a_:	List[str]=12,   a_:	Union[str, Any]=6,   a_:	int=False,   a_:	Any=False,   a_:	Dict=None,   a_:	Any=None,   **a_:	str,   ):




   '''simple docstring'''



   _snake_case							:	str			=						kwargs.pop("""text_config_dict""",   a_	)
   _snake_case							:	Optional[Any]			=						kwargs.pop("""vision_config_dict""",   a_	)

   super().__init__(**a_	)
   _snake_case							:	str			=						share_cross_modal_transformer_layers
   _snake_case							:	Any			=						hidden_act
   _snake_case							:	Union[str, Any]			=						hidden_size
   _snake_case							:	Union[str, Any]			=						initializer_factor
   _snake_case							:	Dict			=						layer_norm_eps
   _snake_case							:	Dict			=						share_link_tower_layers
   _snake_case							:	Optional[int]			=						link_tower_type
   _snake_case							:	Any			=						num_attention_heads
   _snake_case							:	int			=						num_hidden_layers
   _snake_case							:	int			=						tie_word_embeddings
   _snake_case							:	Optional[Any]			=						init_layernorm_from_vision_encoder

   if text_config is None:
     _snake_case							:	Optional[Any]			=						{}
     logger.info("""`text_config` is `None`. Initializing the `BridgeTowerTextConfig` with default values."""	)

   if vision_config is None:
     _snake_case							:	str			=						{}
     logger.info("""`vision_config` is `None`. Initializing the `BridgeTowerVisionConfig` with default values."""	)

   _snake_case							:	Any			=						BridgeTowerTextConfig(**a_	)
   _snake_case							:	List[Any]			=						BridgeTowerVisionConfig(**a_	)


 @classmethod
 def 				UpperCamelCase_     (       cls:	Union[str, Any],   a_:	BridgeTowerTextConfig,   a_:	BridgeTowerVisionConfig,   **a_:	Optional[Any]	):




   '''simple docstring'''



   return cls(text_config=text_config.to_dict(),   vision_config=vision_config.to_dict(),   **a_	)


 def 				UpperCamelCase_     (       self:	int	):




   '''simple docstring'''



   _snake_case							:	Optional[int]			=						copy.deepcopy(self.__dict__	)
   _snake_case							:	str			=						self.text_config.to_dict()
   _snake_case							:	List[str]			=						self.vision_config.to_dict()
   _snake_case							:	Tuple			=						self.__class__.model_type
   return output


