




"""simple docstring"""

import logging
import os
from dataclasses import dataclass, field
from typing import Dict, Optional

import numpy as np
from utils_multiple_choice import MultipleChoiceDataset, Split, processors

import transformers
from transformers import (
    AutoConfig,
    AutoModelForMultipleChoice,
    AutoTokenizer,
    DataCollatorWithPadding,
    EvalPrediction,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    set_seed,
)
from transformers.trainer_utils import is_main_process


_lowerCamelCase     :	Tuple         =				logging.getLogger(__name__)

def 						lowercase_					(						_UpperCAmelCase     ,					_UpperCAmelCase		):




	"""simple docstring"""






	return (preds == labels).mean()

@dataclass
class 				lowercase					:
				__lowerCAmelCase			:			str     				=      field(
				    metadata={"""help""": """Path to pretrained model or model identifier from huggingface.co/models"""})
				__lowerCAmelCase			:			Optional[str]     				=      field(
				    default=A__   ,		metadata={"""help""": """Pretrained config name or path if not the same as model_name"""})
				__lowerCAmelCase			:			Optional[str]     				=      field(
				    default=A__   ,		metadata={"""help""": """Pretrained tokenizer name or path if not the same as model_name"""})
				__lowerCAmelCase			:			Optional[str]     				=      field(
				    default=A__   ,		metadata={"""help""": """Where do you want to store the pretrained models downloaded from huggingface.co"""}   ,		)

@dataclass
class 				lowercase					:
				__lowerCAmelCase			:			str     				=      field(metadata={"""help""": """The name of the task to train on: """ + """, """.join(processors.keys())})
				__lowerCAmelCase			:			str     				=      field(metadata={"""help""": """Should contain the data files for the task."""})
				__lowerCAmelCase			:			int     				=      field(
				    default=128   ,		metadata={
				        """help""": (
				            """The maximum total input sequence length after tokenization. Sequences longer """
				            """than this will be truncated, sequences shorter will be padded."""
				        )
				    }   ,		)
				__lowerCAmelCase			:			bool     				=      field(
				    default=A__   ,		metadata={"""help""": """Overwrite the cached training and evaluation sets"""})

def 						lowercase_					(						):




	"""simple docstring"""






	A_     : Optional[Any]		      =						HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)		)
	A_     : int		      =						parser.parse_args_into_dataclasses()

	if (
	    os.path.exists(training_args.output_dir		)
	    and os.listdir(training_args.output_dir		)
	    and training_args.do_train
	    and not training_args.overwrite_output_dir
	):
		raise ValueError(
		    f"""Output directory ({training_args.output_dir}) already exists and is not empty. Use"""
		    ''' --overwrite_output_dir to overcome.'''		)

	# Setup logging
	logging.basicConfig(
	    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s'''     ,					datefmt='''%m/%d/%Y %H:%M:%S'''     ,					level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN     ,					)
	logger.warning(
	    '''Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s'''     ,					training_args.local_rank     ,					training_args.device     ,					training_args.n_gpu     ,					bool(training_args.local_rank != -1		)     ,					training_args.fpaa     ,					)
	# Set the verbosity to info of the Transformers logger (on main process only):
	if is_main_process(training_args.local_rank		):
		transformers.utils.logging.set_verbosity_info()
		transformers.utils.logging.enable_default_handler()
		transformers.utils.logging.enable_explicit_format()
	logger.info('''Training/evaluation parameters %s'''     ,					__lowerCAmelCase		)

	# Set seed
	set_seed(training_args.seed		)

	try:
		A_     : Any		      =						processors[data_args.task_name]()
		A_     : Any		      =						processor.get_labels()
		A_     : Dict		      =						len(__lowerCAmelCase		)
	except KeyError:
		raise ValueError('''Task not found: %s''' % (data_args.task_name)		)

	# Load pretrained model and tokenizer
	#
	# Distributed training:
	# The .from_pretrained methods guarantee that only one local process can concurrently
	# download model & vocab.

	A_     : Union[str, Any]		      =						AutoConfig.from_pretrained(
	    model_args.config_name if model_args.config_name else model_args.model_name_or_path     ,					num_labels=__lowerCAmelCase     ,					finetuning_task=data_args.task_name     ,					cache_dir=model_args.cache_dir     ,					)
	A_     : Dict		      =						AutoTokenizer.from_pretrained(
	    model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path     ,					cache_dir=model_args.cache_dir     ,					)
	A_     : str		      =						AutoModelForMultipleChoice.from_pretrained(
	    model_args.model_name_or_path     ,					from_tf=bool('''.ckpt''' in model_args.model_name_or_path		)     ,					config=__lowerCAmelCase     ,					cache_dir=model_args.cache_dir     ,					)

	# Get datasets
	A_     : str		      =						(
	    MultipleChoiceDataset(
	        data_dir=data_args.data_dir     ,					tokenizer=__lowerCAmelCase     ,					task=data_args.task_name     ,					max_seq_length=data_args.max_seq_length     ,					overwrite_cache=data_args.overwrite_cache     ,					mode=Split.train     ,					)
	    if training_args.do_train
	    else None
	)
	A_     : List[Any]		      =						(
	    MultipleChoiceDataset(
	        data_dir=data_args.data_dir     ,					tokenizer=__lowerCAmelCase     ,					task=data_args.task_name     ,					max_seq_length=data_args.max_seq_length     ,					overwrite_cache=data_args.overwrite_cache     ,					mode=Split.dev     ,					)
	    if training_args.do_eval
	    else None
	)

	def compute_metrics(_UpperCAmelCase		) -> Dict:
		A_     : Any		      =						np.argmax(p.predictions     ,					axis=1		)
		return {"acc": simple_accuracy(__lowerCAmelCase     ,					p.label_ids		)}

	# Data collator
	A_     : Optional[Any]		      =						DataCollatorWithPadding(__lowerCAmelCase     ,					pad_to_multiple_of=8		) if training_args.fpaa else None

	# Initialize our Trainer
	A_     : Tuple		      =						Trainer(
	    model=__lowerCAmelCase     ,					args=__lowerCAmelCase     ,					train_dataset=__lowerCAmelCase     ,					eval_dataset=__lowerCAmelCase     ,					compute_metrics=__lowerCAmelCase     ,					data_collator=__lowerCAmelCase     ,					)

	# Training
	if training_args.do_train:
		trainer.train(
		    model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path		) else None		)
		trainer.save_model()
		# For convenience, we also re-save the tokenizer to the same directory,
		# so that you can share your model easily on huggingface.co/models =)
		if trainer.is_world_master():
			tokenizer.save_pretrained(training_args.output_dir		)

    # Evaluation
	A_     : str		      =						{}
	if training_args.do_eval:
		logger.info('''*** Evaluate ***'''		)

		A_     : int		      =						trainer.evaluate()

		A_     : Union[str, Any]		      =						os.path.join(training_args.output_dir     ,					'''eval_results.txt'''		)
		if trainer.is_world_master():
			with open(__lowerCAmelCase     ,					'''w'''		) as writer:
				logger.info('''***** Eval results *****'''		)
				for key, value in result.items():
					logger.info('''  %s = %s'''     ,					__lowerCAmelCase     ,					__lowerCAmelCase		)
					writer.write('''%s = %s\n''' % (key, value)		)

				results.update(__lowerCAmelCase		)

	return results

def 						lowercase_					(						_UpperCAmelCase		):




	"""simple docstring"""






	main()


if __name__ == "__main__":
		main()













# 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 torch

from ..models.auto import AutoModelForSequenceClassification, AutoTokenizer
from .base import PipelineTool





class 			__a  (    A__       ):
 _lowerCAmelCase       :       str        = '''facebook/bart-large-mnli'''
 _lowerCAmelCase       :       Tuple        = (
     '''This is a tool that classifies an English text using provided labels. It takes two inputs: `text`, which '''
     '''should be the text to classify, and `labels`, which should be the list of labels to use for classification. '''
     '''It returns the most likely label in the list of provided `labels` for the input text.'''
 )
 _lowerCAmelCase       :       Any        = '''text_classifier'''
 _lowerCAmelCase       :       int        = AutoTokenizer
 _lowerCAmelCase       :       Union[str, Any]        = AutoModelForSequenceClassification

 _lowerCAmelCase       :       Union[str, Any]        = ['''text''', ['''text''']]
 _lowerCAmelCase       :       Dict        = ['''text''']







 def 					__lowercase					(		self					:   int						):


      '''simple docstring'''






      super().setup()
      UpperCamelCase__			:   Dict       				=  self.model.config
      UpperCamelCase__			:   Union[str, Any]       				=  -1
      for idx, label in config.idalabel.items():
           if label.lower().startswith("entail"						):
                UpperCamelCase__			:   List[str]       				=  int(SCREAMING_SNAKE_CASE						)
      if self.entailment_id == -1:
           raise ValueError("Could not determine the entailment ID from the model config, please pass it at init."						)







 def 					__lowercase					(		self					:   Tuple							,   SCREAMING_SNAKE_CASE					:   Tuple							,   SCREAMING_SNAKE_CASE					:   List[Any]						):


      '''simple docstring'''






      UpperCamelCase__			:   Any       				=  labels
      return self.pre_processor(
          [text] * len(SCREAMING_SNAKE_CASE						)							,   [F'This example is {label}' for label in labels]							,   return_tensors="pt"							,   padding="max_length"							,   )







 def 					__lowercase					(		self					:   Optional[Any]							,   SCREAMING_SNAKE_CASE					:   Dict						):


      '''simple docstring'''






      UpperCamelCase__			:   List[Any]       				=  outputs.logits
      UpperCamelCase__			:   Any       				=  torch.argmax(logits[:, 2]						).item()
      return self._labels[label_id]