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


lowercase__						       =      logging.getLogger(__name__)






class 	snake_case__       (      __SCREAMING_SNAKE_CASE					):







		"""simple docstring"""

		lowerCamelCase     							=						"""sequence-classification"""



		def __init__(      self				:			Dict  ,	UpperCamelCase__				:			Dict      )						->			List[Any]:






				"""simple docstring"""







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

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



		def       lowerCAmelCase				(      self				:			Optional[Any]  ,	**UpperCamelCase__				:			Optional[Any]      )						->			Tuple:






				"""simple docstring"""







				return self.model(**UpperCamelCase__      )



		def       lowerCAmelCase				(      self				:			List[str]  ,	UpperCamelCase__				:			Tuple  ,	UpperCamelCase__				:			Dict      )						->			int:






				"""simple docstring"""







				snake_case  :       Dict     			=						{'''input_ids''': batch[0], '''attention_mask''': batch[1], '''labels''': batch[3]}

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

				snake_case  :       Tuple     			=						self(**UpperCamelCase__      )
				snake_case  :       Optional[Any]     			=						outputs[0]

				snake_case  :       List[Any]     			=						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       lowerCAmelCase				(      self				:			Optional[int]      )						->			Tuple:






				"""simple docstring"""







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

				for mode in ["train", "dev"]:
						snake_case  :       str     			=						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      )
								snake_case  :       Union[str, Any]     			=						(
								    processor.get_dev_examples(args.data_dir      )
								    if mode == '''dev'''
								    else processor.get_train_examples(args.data_dir      )
								)
								snake_case  :       List[str]     			=						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       lowerCAmelCase				(      self				:			Dict  ,	UpperCamelCase__				:			str  ,	UpperCamelCase__				:			int  ,	UpperCamelCase__				:			bool = False      )						->			DataLoader:






				"""simple docstring"""







				snake_case  :       List[Any]     			=						'''dev''' if mode == '''test''' else mode

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

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



		def       lowerCAmelCase				(      self				:			Tuple  ,	UpperCamelCase__				:			Optional[Any]  ,	UpperCamelCase__				:			int      )						->			int:






				"""simple docstring"""







				snake_case  :       Tuple     			=						{'''input_ids''': batch[0], '''attention_mask''': batch[1], '''labels''': batch[3]}

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

				snake_case  :       List[Any]     			=						self(**UpperCamelCase__      )
				snake_case			,snake_case  :       str     			=						outputs[:2]
				snake_case  :       Any     			=						logits.detach().cpu().numpy()
				snake_case  :       Dict     			=						inputs['''labels'''].detach().cpu().numpy()

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



		def       lowerCAmelCase				(      self				:			Dict  ,	UpperCamelCase__				:			Tuple      )						->			tuple:






				"""simple docstring"""







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

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

				snake_case  :       Tuple     			=						np.concatenate([x['''target'''] for x in outputs]  ,	axis=0      )
				snake_case  :       int     			=						[[] for _ in range(out_label_ids.shape[0]      )]
				snake_case  :       Any     			=						[[] for _ in range(out_label_ids.shape[0]      )]

				snake_case  :       int     			=						{**{'''val_loss''': val_loss_mean}, **compute_metrics(self.hparams.task  ,	UpperCamelCase__  ,	UpperCamelCase__      )}

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



		def       lowerCAmelCase				(      self				:			str  ,	UpperCamelCase__				:			list      )						->			dict:






				"""simple docstring"""







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



		def       lowerCAmelCase				(      self				:			List[str]  ,	UpperCamelCase__				:			Optional[Any]      )						->			dict:






				"""simple docstring"""







				snake_case			,snake_case			,snake_case  :       Union[str, Any]     			=						self._eval_end(UpperCamelCase__      )
				snake_case  :       Union[str, Any]     			=						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       lowerCAmelCase				(      UpperCamelCase__				:			List[Any]  ,	UpperCamelCase__				:			Union[str, Any]      )						->			Optional[int]:






				"""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    _UpperCamelCase						(       )  ->							int:


		'''simple docstring'''


		snake_case  :       Optional[int]     			=						argparse.ArgumentParser()
		add_generic_args(SCREAMING_SNAKE_CASE__ ,       os.getcwd()			)
		snake_case  :       Any     			=						GLUETransformer.add_model_specific_args(SCREAMING_SNAKE_CASE__ ,       os.getcwd()			)
		snake_case  :       Optional[int]     			=						parser.parse_args()

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

		snake_case  :       Any     			=						GLUETransformer(SCREAMING_SNAKE_CASE__			)
		snake_case  :       int     			=						generic_train(SCREAMING_SNAKE_CASE__ ,       SCREAMING_SNAKE_CASE__			)

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


if __name__ == "__main__":
						main()




'''simple docstring'''
import datasets

from .evaluate import evaluate


lowercase__						       =      "\\n@article{hendrycks2021cuad,\n      title={CUAD: An Expert-Annotated NLP Dataset for Legal Contract Review},\n      author={Dan Hendrycks and Collin Burns and Anya Chen and Spencer Ball},\n      journal={arXiv preprint arXiv:2103.06268},\n      year={2021}\n}\n"

lowercase__						       =      "\nThis metric wrap the official scoring script for version 1 of the Contract\nUnderstanding Atticus Dataset (CUAD).\nContract Understanding Atticus Dataset (CUAD) v1 is a corpus of more than 13,000 labels in 510\ncommercial legal contracts that have been manually labeled to identify 41 categories of important\nclauses that lawyers look for when reviewing contracts in connection with corporate transactions.\n"

lowercase__						       =      "\nComputes CUAD scores (EM, F1, AUPR, Precision@80%Recall, and Precision@90%Recall).\nArgs:\n    predictions: List of question-answers dictionaries with the following key-values:\n        - 'id': id of the question-answer pair as given in the references (see below)\n        - 'prediction_text': list of possible texts for the answer, as a list of strings\n        depending on a threshold on the confidence probability of each prediction.\n    references: List of question-answers dictionaries with the following key-values:\n        - 'id': id of the question-answer pair (see above),\n        - 'answers': a Dict in the CUAD dataset format\n            {\n                'text': list of possible texts for the answer, as a list of strings\n                'answer_start': list of start positions for the answer, as a list of ints\n            }\n            Note that answer_start values are not taken into account to compute the metric.\nReturns:\n    'exact_match': Exact match (the normalized answer exactly match the gold answer)\n    'f1': The F-score of predicted tokens versus the gold answer\n    'aupr': Area Under the Precision-Recall curve\n    'prec_at_80_recall': Precision at 80% recall\n    'prec_at_90_recall': Precision at 90% recall\nExamples:\n    >>> predictions = [{'prediction_text': ['The seller:', 'The buyer/End-User: Shenzhen LOHAS Supply Chain Management Co., Ltd.'], 'id': 'LohaCompanyltd_20191209_F-1_EX-10.16_11917878_EX-10.16_Supply Agreement__Parties'}]\n    >>> references = [{'answers': {'answer_start': [143, 49], 'text': ['The seller:', 'The buyer/End-User: Shenzhen LOHAS Supply Chain Management Co., Ltd.']}, 'id': 'LohaCompanyltd_20191209_F-1_EX-10.16_11917878_EX-10.16_Supply Agreement__Parties'}]\n    >>> cuad_metric = datasets.load_metric(\"cuad\")\n    >>> results = cuad_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {'exact_match': 100.0, 'f1': 100.0, 'aupr': 0.0, 'prec_at_80_recall': 1.0, 'prec_at_90_recall': 1.0}\n"






@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION    ,					_KWARGS_DESCRIPTION					)
class 	snake_case__       (      datasets.Metric					):







		"""simple docstring"""

		def       lowerCAmelCase				(      self				:			Optional[int]      )						->			Tuple:






				"""simple docstring"""







				return datasets.MetricInfo(
				    description=_DESCRIPTION  ,	citation=_CITATION  ,	inputs_description=_KWARGS_DESCRIPTION  ,	features=datasets.Features(
				        {
				            '''predictions''': {
				                '''id''': datasets.Value('''string'''      ),
				                '''prediction_text''': datasets.features.Sequence(datasets.Value('''string'''      )      ),
				            },
				            '''references''': {
				                '''id''': datasets.Value('''string'''      ),
				                '''answers''': datasets.features.Sequence(
				                    {
				                        '''text''': datasets.Value('''string'''      ),
				                        '''answer_start''': datasets.Value('''int32'''      ),
				                    }      ),
				            },
				        }      )  ,	codebase_urls=['''https://www.atticusprojectai.org/cuad''']  ,	reference_urls=['''https://www.atticusprojectai.org/cuad''']  ,	)



		def       lowerCAmelCase				(      self				:			str  ,	UpperCamelCase__				:			int  ,	UpperCamelCase__				:			str      )						->			Optional[Any]:






				"""simple docstring"""







				snake_case  :       Tuple     			=						{prediction['''id''']: prediction['''prediction_text'''] for prediction in predictions}
				snake_case  :       Dict     			=						[
				    {
				        '''paragraphs''': [
				            {
				                '''qas''': [
				                    {
				                        '''answers''': [{'''text''': answer_text} for answer_text in ref['''answers''']['''text''']],
				                        '''id''': ref['''id'''],
				                    }
				                    for ref in references
				                ]
				            }
				        ]
				    }
				]
				snake_case  :       int     			=						evaluate(dataset=UpperCamelCase__  ,	predictions=UpperCamelCase__      )
				return score


