



'''simple docstring'''






import datasets

from .evaluate import evaluate


UpperCamelCase__:     Tuple										=    "\\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"

UpperCamelCase__:     Union[str, Any]										=    "\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"

UpperCamelCase__:     Optional[Any]										=    "\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        SCREAMING_SNAKE_CASE(							datasets.Metric    ):







	"""simple docstring"""






	def   A			(		self    :  str	)    ->	List[str]:
								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   A			(		self    :  List[Any]	,    __snake_case    :  int	,    __snake_case    :  int	)    ->	Tuple:
								UpperCAmelCase				: Tuple											=	{prediction["""id"""]: prediction["""prediction_text"""] for prediction in predictions}
								UpperCAmelCase				: Tuple											=	[
								    {
								        """paragraphs""": [
								            {
								                """qas""": [
								                    {
								                        """answers""": [{"""text""": answer_text} for answer_text in ref["""answers"""]["""text"""]],
								                        """id""": ref["""id"""],
								                    }
								                    for ref in references
								                ]
								            }
								        ]
								    }
								]
								UpperCAmelCase				: Optional[int]											=	evaluate(dataset=a_	,    predictions=a_	)
								return score










'''simple docstring'''






import logging

import torch
from torch import nn
from torch.nn import CrossEntropyLoss, MSELoss

from transformers.file_utils import add_start_docstrings, add_start_docstrings_to_model_forward
from transformers.models.bert.modeling_bert import (
    BERT_INPUTS_DOCSTRING,
    BERT_START_DOCSTRING,
    BertEncoder,
    BertModel,
    BertPreTrainedModel,
)


UpperCamelCase__:     List[str]										=    logging.getLogger(__name__)

class        SCREAMING_SNAKE_CASE(							A__    ):







	"""simple docstring"""






	def   A			(		self    :  str	,    __snake_case    :  Tuple	,    __snake_case    :  int	,    __snake_case    :  List[Any]=None	,    __snake_case    :  List[Any]=None	)    ->	Optional[int]:
								UpperCAmelCase				: Union[str, Any]											=	self.layer[current_layer](__snake_case	,    __snake_case	,    head_mask[current_layer]	)

								UpperCAmelCase				: Optional[Any]											=	layer_outputs[0]

								return hidden_states

@add_start_docstrings(
    """The bare Bert Model transformer with PABEE outputting raw hidden-states without any specific head on top."""							,	A__							,	)
class        SCREAMING_SNAKE_CASE(							A__    ):







	"""simple docstring"""






	def __init__(		self    :  int	,    __snake_case    :  Any	)    ->	Any:
								super().__init__(__snake_case	)

								UpperCAmelCase				: Any											=	BertEncoderWithPabee(__snake_case	)

								self.init_weights()
								UpperCAmelCase				: Optional[Any]											=	0
								UpperCAmelCase				: List[str]											=	0
								UpperCAmelCase				: Tuple											=	0

								UpperCAmelCase				: List[str]											=	0







	def   A			(		self    :  List[str]	,    __snake_case    :  List[str]	)    ->	List[Any]:
								UpperCAmelCase				: Union[str, Any]											=	threshold







	def   A			(		self    :  str	,    __snake_case    :  List[str]	)    ->	Optional[Any]:
								UpperCAmelCase				: List[str]											=	patience







	def   A			(		self    :  Dict	)    ->	str:
								UpperCAmelCase				: List[Any]											=	0
								UpperCAmelCase				: Dict											=	0







	def   A			(		self    :  Optional[Any]	)    ->	Optional[int]:
								UpperCAmelCase				: Dict											=	self.inference_layers_num / self.inference_instances_num
								UpperCAmelCase				: int											=	(
								    F"""*** Patience = {self.patience} Avg. Inference Layers = {avg_inf_layers:.2f} Speed Up ="""
								    F""" {1 - avg_inf_layers / self.config.num_hidden_layers:.2f} ***"""
								)
								print(__snake_case	)







	@add_start_docstrings_to_model_forward(__snake_case	)
	def   A			(		self    :  Union[str, Any]	,    __snake_case    :  Optional[Any]=None	,    __snake_case    :  str=None	,    __snake_case    :  int=None	,    __snake_case    :  Optional[Any]=None	,    __snake_case    :  Tuple=None	,    __snake_case    :  Tuple=None	,    __snake_case    :  Tuple=None	,    __snake_case    :  Optional[Any]=None	,    __snake_case    :  Union[str, Any]=None	,    __snake_case    :  Optional[int]=None	,    __snake_case    :  Optional[int]=False	,    )    ->	List[str]:

								if input_ids is not None and inputs_embeds is not None:
															raise ValueError('''You cannot specify both input_ids and inputs_embeds at the same time'''	)
								elif input_ids is not None:
															UpperCAmelCase				: Union[str, Any]											=	input_ids.size()
								elif inputs_embeds is not None:
															UpperCAmelCase				: List[Any]											=	inputs_embeds.size()[:-1]
								else:
															raise ValueError('''You have to specify either input_ids or inputs_embeds'''	)

								UpperCAmelCase				: Optional[Any]											=	input_ids.device if input_ids is not None else inputs_embeds.device

								if attention_mask is None:
															UpperCAmelCase				: Optional[Any]											=	torch.ones(__snake_case	,    device=__snake_case	)
								if token_type_ids is None:
															UpperCAmelCase				: Tuple											=	torch.zeros(__snake_case	,    dtype=torch.long	,    device=__snake_case	)

								# We can provide a self-attention mask of dimensions [batch_size, from_seq_length, to_seq_length]
								# ourselves in which case we just need to make it broadcastable to all heads.
								UpperCAmelCase				: torch.Tensor											=	self.get_extended_attention_mask(__snake_case	,    __snake_case	,    __snake_case	)

								# If a 2D ou 3D attention mask is provided for the cross-attention
								# we need to make broadcastable to [batch_size, num_heads, seq_length, seq_length]
								if self.config.is_decoder and encoder_hidden_states is not None:
															UpperCAmelCase							,	UpperCAmelCase							,	UpperCAmelCase				: Any											=	encoder_hidden_states.size()
															UpperCAmelCase				: List[Any]											=	(encoder_batch_size, encoder_sequence_length)
															if encoder_attention_mask is None:
																						UpperCAmelCase				: str											=	torch.ones(__snake_case	,    device=__snake_case	)
															UpperCAmelCase				: Tuple											=	self.invert_attention_mask(__snake_case	)
								else:
															UpperCAmelCase				: List[str]											=	None

								# Prepare head mask if needed
								# 1.0 in head_mask indicate we keep the head
								# attention_probs has shape bsz x n_heads x N x N
								# input head_mask has shape [num_heads] or [num_hidden_layers x num_heads]
								# and head_mask is converted to shape [num_hidden_layers x batch x num_heads x seq_length x seq_length]
								UpperCAmelCase				: List[Any]											=	self.get_head_mask(__snake_case	,    self.config.num_hidden_layers	)

								UpperCAmelCase				: Any											=	self.embeddings(
								    input_ids=__snake_case	,    position_ids=__snake_case	,    token_type_ids=__snake_case	,    inputs_embeds=__snake_case	)
								UpperCAmelCase				: Optional[int]											=	embedding_output

								if self.training:
															UpperCAmelCase				: str											=	[]
															for i in range(self.config.num_hidden_layers	):
																						UpperCAmelCase				: Optional[int]											=	self.encoder.adaptive_forward(
																						    __snake_case	,    current_layer=__snake_case	,    attention_mask=__snake_case	,    head_mask=__snake_case	)

																						UpperCAmelCase				: List[str]											=	self.pooler(__snake_case	)
																						UpperCAmelCase				: Dict											=	output_layers[i](output_dropout(__snake_case	)	)
																						res.append(__snake_case	)
								elif self.patience == 0:  # Use all layers for inference
															UpperCAmelCase				: Any											=	self.encoder(
															    __snake_case	,    attention_mask=__snake_case	,    head_mask=__snake_case	,    encoder_hidden_states=__snake_case	,    encoder_attention_mask=__snake_case	,    )
															UpperCAmelCase				: Dict											=	self.pooler(encoder_outputs[0]	)
															UpperCAmelCase				: List[Any]											=	[output_layers[self.config.num_hidden_layers - 1](__snake_case	)]
								else:
															UpperCAmelCase				: Tuple											=	0
															UpperCAmelCase				: List[str]											=	None
															UpperCAmelCase				: Optional[Any]											=	0
															for i in range(self.config.num_hidden_layers	):
																						calculated_layer_num += 1
																						UpperCAmelCase				: Optional[Any]											=	self.encoder.adaptive_forward(
																						    __snake_case	,    current_layer=__snake_case	,    attention_mask=__snake_case	,    head_mask=__snake_case	)

																						UpperCAmelCase				: List[str]											=	self.pooler(__snake_case	)
																						UpperCAmelCase				: Optional[int]											=	output_layers[i](__snake_case	)
																						if regression:
																													UpperCAmelCase				: Union[str, Any]											=	logits.detach()
																													if patient_result is not None:
																																				UpperCAmelCase				: List[str]											=	patient_result.detach()
																													if (patient_result is not None) and torch.abs(patient_result - labels	) < self.regression_threshold:
																																				patient_counter += 1
																													else:
																																				UpperCAmelCase				: Dict											=	0
																						else:
																													UpperCAmelCase				: Union[str, Any]											=	logits.detach().argmax(dim=1	)
																													if patient_result is not None:
																																				UpperCAmelCase				: str											=	patient_result.detach().argmax(dim=1	)
																													if (patient_result is not None) and torch.all(labels.eq(__snake_case	)	):
																																				patient_counter += 1
																													else:
																																				UpperCAmelCase				: Union[str, Any]											=	0

																						UpperCAmelCase				: List[str]											=	logits
																						if patient_counter == self.patience:
																													break
															UpperCAmelCase				: List[Any]											=	[patient_result]
															self.inference_layers_num += calculated_layer_num
															self.inference_instances_num += 1

								return res

@add_start_docstrings(
    """Bert Model transformer with PABEE and a sequence classification/regression head on top (a linear layer on top of
    the pooled output) e.g. for GLUE tasks. """							,	A__							,	)
class        SCREAMING_SNAKE_CASE(							A__    ):







	"""simple docstring"""






	def __init__(		self    :  str	,    __snake_case    :  str	)    ->	Optional[Any]:
								super().__init__(__snake_case	)
								UpperCAmelCase				: Optional[int]											=	config.num_labels

								UpperCAmelCase				: Optional[Any]											=	BertModelWithPabee(__snake_case	)
								UpperCAmelCase				: Tuple											=	nn.Dropout(config.hidden_dropout_prob	)
								UpperCAmelCase				: Dict											=	nn.ModuleList(
								    [nn.Linear(config.hidden_size	,    self.config.num_labels	) for _ in range(config.num_hidden_layers	)]	)

								self.init_weights()







	@add_start_docstrings_to_model_forward(__snake_case	)
	def   A			(		self    :  Any	,    __snake_case    :  Dict=None	,    __snake_case    :  str=None	,    __snake_case    :  List[Any]=None	,    __snake_case    :  List[Any]=None	,    __snake_case    :  Union[str, Any]=None	,    __snake_case    :  int=None	,    __snake_case    :  str=None	,    )    ->	Union[str, Any]:
								UpperCAmelCase				: Optional[int]											=	self.bert(
								    input_ids=__snake_case	,    attention_mask=__snake_case	,    token_type_ids=__snake_case	,    position_ids=__snake_case	,    head_mask=__snake_case	,    inputs_embeds=__snake_case	,    output_dropout=self.dropout	,    output_layers=self.classifiers	,    regression=self.num_labels == 1	,    )

								UpperCAmelCase				: List[str]											=	(logits[-1],)

								if labels is not None:
															UpperCAmelCase				: Tuple											=	None
															UpperCAmelCase				: Any											=	0
															for ix, logits_item in enumerate(__snake_case	):
																						if self.num_labels == 1:
																													#  We are doing regression
																													UpperCAmelCase				: Optional[Any]											=	MSELoss()
																													UpperCAmelCase				: Optional[Any]											=	loss_fct(logits_item.view(-1	)	,    labels.view(-1	)	)
																						else:
																													UpperCAmelCase				: Dict											=	CrossEntropyLoss()
																													UpperCAmelCase				: Dict											=	loss_fct(logits_item.view(-1	,    self.num_labels	)	,    labels.view(-1	)	)
																						if total_loss is None:
																													UpperCAmelCase				: Dict											=	loss
																						else:
																													total_loss += loss * (ix + 1)
																						total_weights += ix + 1
															UpperCAmelCase				: Any											=	(total_loss / total_weights,) + outputs

								return outputs





