



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


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






class 		_UpperCamelCase    (				__SCREAMING_SNAKE_CASE       ):


			'''simple docstring'''

			_A :						List[str]            =						'''sequence-classification'''

			def __init__(				self	:					Dict				,	lowerCAmelCase__	:					Dict		):





					"""simple docstring"""






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

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

			def UpperCamelCase__   (				self	:					Optional[Any]				,	**lowerCAmelCase__	:					Optional[Any]		):





					"""simple docstring"""






					return self.model(**UpperCamelCase__		)

			def UpperCamelCase__   (				self	:					List[str]				,	lowerCAmelCase__	:					Tuple				,	lowerCAmelCase__	:					Dict		):





					"""simple docstring"""






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

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

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

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

			def UpperCamelCase__   (				self	:					Optional[int]		):





					"""simple docstring"""






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

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





					"""simple docstring"""






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

					__SCREAMING_SNAKE_CASE			:						List[Any]       			=						self._feature_file(UpperCamelCase__		)
					logger.info("""Loading features from cached file %s"""				,	UpperCamelCase__		)
					__SCREAMING_SNAKE_CASE			:						Tuple       			=						torch.load(UpperCamelCase__		)
					__SCREAMING_SNAKE_CASE			:						Union[str, Any]       			=						torch.tensor([f.input_ids for f in features]				,	dtype=torch.long		)
					__SCREAMING_SNAKE_CASE			:						List[Any]       			=						torch.tensor([f.attention_mask for f in features]				,	dtype=torch.long		)
					__SCREAMING_SNAKE_CASE			:						List[Any]       			=						torch.tensor([f.token_type_ids for f in features]				,	dtype=torch.long		)
					if self.hparams.glue_output_mode == "classification":
							__SCREAMING_SNAKE_CASE			:						Tuple       			=						torch.tensor([f.label for f in features]				,	dtype=torch.long		)
					elif self.hparams.glue_output_mode == "regression":
							__SCREAMING_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 UpperCamelCase__   (				self	:					Tuple				,	lowerCAmelCase__	:					Optional[Any]				,	lowerCAmelCase__	:					int		):





					"""simple docstring"""






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

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

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

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

			def UpperCamelCase__   (				self	:					Dict				,	lowerCAmelCase__	:					Tuple		):





					"""simple docstring"""






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

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

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

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

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

			def UpperCamelCase__   (				self	:					str				,	lowerCAmelCase__	:					list		):





					"""simple docstring"""






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

			def UpperCamelCase__   (				self	:					List[str]				,	lowerCAmelCase__	:					Optional[Any]		):





					"""simple docstring"""






					__SCREAMING_SNAKE_CASE			:						Union[str, Any]       			=						self._eval_end(UpperCamelCase__		)
					__SCREAMING_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 UpperCamelCase__   (				lowerCAmelCase__	:					List[Any]				,	lowerCAmelCase__	:					Union[str, Any]		):





					"""simple docstring"""






					BaseTransformer.add_model_specific_args(UpperCamelCase__				,	UpperCamelCase__		)
					parser.add_argument(
					    """--max_seq_length"""				,	default=1_2_8				,	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_  (      ):
		__SCREAMING_SNAKE_CASE			:						Optional[int]       			=						argparse.ArgumentParser()
		add_generic_args(SCREAMING_SNAKE_CASE__			, os.getcwd()   )
		__SCREAMING_SNAKE_CASE			:						Any       			=						GLUETransformer.add_model_specific_args(SCREAMING_SNAKE_CASE__			, os.getcwd()   )
		__SCREAMING_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:
				__SCREAMING_SNAKE_CASE			:						str       			=						os.path.join(
				    """./results"""			, F"{args.task}_{time.strftime('%Y%m%d_%H%M%S'   )}"			, )
				os.makedirs(args.output_dir   )

		__SCREAMING_SNAKE_CASE			:						Any       			=						GLUETransformer(SCREAMING_SNAKE_CASE__   )
		__SCREAMING_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:
				__SCREAMING_SNAKE_CASE			:						Tuple       			=						sorted(glob.glob(os.path.join(args.output_dir			, """checkpoint-epoch=*.ckpt"""   )			, recursive=SCREAMING_SNAKE_CASE__   )   )
				__SCREAMING_SNAKE_CASE			:						Any       			=						model.load_from_checkpoint(checkpoints[-1]   )
				return trainer.test(SCREAMING_SNAKE_CASE__   )


if __name__ == "__main__":
						main()

'''simple docstring'''
# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re

from ..utils import cached_file


# docstyle-ignore
lowercase__						       =      "\nHuman: <<task>>\n\nAssistant: "


lowercase__						       =      "huggingface-tools/default-prompts"
lowercase__						       =      {"chat": "chat_prompt_template.txt", "run": "run_prompt_template.txt"}




def    _UpperCamelCase						(       SCREAMING_SNAKE_CASE__ ,       SCREAMING_SNAKE_CASE__ ,       SCREAMING_SNAKE_CASE__="run"			)  ->							Dict:


		'''simple docstring'''


		if prompt_or_repo_id is None:
				snake_case  :       Union[str, Any]     			=						DEFAULT_PROMPTS_REPO

		# prompt is considered a repo ID when it does not contain any kind of space
		if re.search('''\\s''' ,       SCREAMING_SNAKE_CASE__			) is not None:
				return prompt_or_repo_id

		snake_case  :       List[Any]     			=						cached_file(
		    SCREAMING_SNAKE_CASE__ ,       PROMPT_FILES[mode] ,       repo_type='''dataset''' ,       user_agent={'''agent''': agent_name}			)
		with open(SCREAMING_SNAKE_CASE__ ,       '''r''' ,       encoding='''utf-8'''			) as f:
				return f.read()


