


"""simple docstring"""





import argparse
import glob
import logging
import os
from argparse import Namespace
from importlib import import_module

import numpy as np
import torch
from lightning_base import BaseTransformer, add_generic_args, generic_train
from seqeval.metrics import accuracy_score, fa_score, precision_score, recall_score
from torch.nn import CrossEntropyLoss
from torch.utils.data import DataLoader, TensorDataset
from utils_ner import TokenClassificationTask


snake_case__						:   int			 = logging.getLogger(__name__)



class 						snake_case_(				a__	):
     __UpperCamelCase     =    '''token-classification'''






     def __init__(		self     :    int    , UpperCamelCase_     :    Tuple ):
           if type(UpperCamelCase_ ) == dict:
                 lowerCAmelCase    :		Union[str, Any]				=      Namespace(**UpperCamelCase_ )
           lowerCAmelCase    :		Dict				=      import_module('''tasks''' )
           try:
                 lowerCAmelCase    :		str				=      getattr(UpperCamelCase_    , hparams.task_type )
                 lowerCAmelCase    :		TokenClassificationTask				=      token_classification_task_clazz()
           except AttributeError:
                 raise ValueError(
                     F'''Task {hparams.task_type} needs to be defined as a TokenClassificationTask subclass in {module}. '''
                     F'''Available tasks classes are: {TokenClassificationTask.__subclasses__()}''' )
           lowerCAmelCase    :		Any				=      self.token_classification_task.get_labels(hparams.labels )
           lowerCAmelCase    :		Optional[Any]				=      CrossEntropyLoss().ignore_index
           super().__init__(UpperCamelCase_    , len(self.labels )    , self.mode )






     def    lowerCamelCase__   (		self     :    int    , **UpperCamelCase_     :    List[Any] ):
           return self.model(**UpperCamelCase_ )






     def    lowerCamelCase__   (		self     :    Optional[Any]    , UpperCamelCase_     :    Any    , UpperCamelCase_     :    int ):
           lowerCAmelCase    :		List[str]				=      {'''input_ids''': batch[0], '''attention_mask''': batch[1], '''labels''': batch[3]}
           if self.config.model_type != "distilbert":
                 lowerCAmelCase    :		Dict				=      (
                     batch[2] if self.config.model_type in ['''bert''', '''xlnet'''] else None
                 )  # XLM and RoBERTa don"t use token_type_ids

           lowerCAmelCase    :		Optional[int]				=      self(**UpperCamelCase_ )
           lowerCAmelCase    :		str				=      outputs[0]
           # tensorboard_logs = {"loss": loss, "rate": self.lr_scheduler.get_last_lr()[-1]}
           return {"loss": loss}






     def    lowerCamelCase__   (		self     :    Tuple ):
           lowerCAmelCase    :		List[str]				=      self.hparams
           for mode in ["train", "dev", "test"]:
                 lowerCAmelCase    :		Dict				=      self._feature_file(UpperCamelCase_ )
                 if os.path.exists(UpperCamelCase_ ) and not args.overwrite_cache:
                       logger.info('''Loading features from cached file %s'''    , UpperCamelCase_ )
                       lowerCAmelCase    :		int				=      torch.load(UpperCamelCase_ )
                 else:
                       logger.info('''Creating features from dataset file at %s'''    , args.data_dir )
                       lowerCAmelCase    :		Union[str, Any]				=      self.token_classification_task.read_examples_from_file(args.data_dir    , UpperCamelCase_ )
                       lowerCAmelCase    :		Optional[Any]				=      self.token_classification_task.convert_examples_to_features(
                           UpperCamelCase_    , self.labels    , args.max_seq_length    , self.tokenizer    , cls_token_at_end=bool(self.config.model_type in ['''xlnet'''] )    , cls_token=self.tokenizer.cls_token    , cls_token_segment_id=2 if self.config.model_type in ['''xlnet'''] else 0    , sep_token=self.tokenizer.sep_token    , sep_token_extra=UpperCamelCase_    , pad_on_left=bool(self.config.model_type in ['''xlnet'''] )    , pad_token=self.tokenizer.pad_token_id    , pad_token_segment_id=self.tokenizer.pad_token_type_id    , pad_token_label_id=self.pad_token_label_id    , )
                       logger.info('''Saving features into cached file %s'''    , UpperCamelCase_ )
                       torch.save(UpperCamelCase_    , UpperCamelCase_ )






     def    lowerCamelCase__   (		self     :    Union[str, Any]    , UpperCamelCase_     :    int    , UpperCamelCase_     :    int    , UpperCamelCase_     :    bool = False ):
           lowerCAmelCase    :		Optional[int]				=      self._feature_file(UpperCamelCase_ )
           logger.info('''Loading features from cached file %s'''    , UpperCamelCase_ )
           lowerCAmelCase    :		str				=      torch.load(UpperCamelCase_ )
           lowerCAmelCase    :		int				=      torch.tensor([f.input_ids for f in features]    , dtype=torch.long )
           lowerCAmelCase    :		Tuple				=      torch.tensor([f.attention_mask for f in features]    , dtype=torch.long )
           if features[0].token_type_ids is not None:
                 lowerCAmelCase    :		int				=      torch.tensor([f.token_type_ids for f in features]    , dtype=torch.long )
           else:
                 lowerCAmelCase    :		Optional[Any]				=      torch.tensor([0 for f in features]    , dtype=torch.long )
                 # HACK(we will not use this anymore soon)
           lowerCAmelCase    :		int				=      torch.tensor([f.label_ids for f in features]    , dtype=torch.long )
           return DataLoader(
               TensorDataset(UpperCamelCase_    , UpperCamelCase_    , UpperCamelCase_    , UpperCamelCase_ )    , batch_size=UpperCamelCase_ )






     def    lowerCamelCase__   (		self     :    Any    , UpperCamelCase_     :    Union[str, Any]    , UpperCamelCase_     :    List[Any] ):
           """Compute validation""" ""
           lowerCAmelCase    :		str				=      {'''input_ids''': batch[0], '''attention_mask''': batch[1], '''labels''': batch[3]}
           if self.config.model_type != "distilbert":
                 lowerCAmelCase    :		Optional[int]				=      (
                     batch[2] if self.config.model_type in ['''bert''', '''xlnet'''] else None
                 )  # XLM and RoBERTa don"t use token_type_ids
           lowerCAmelCase    :		Dict				=      self(**UpperCamelCase_ )
           lowerCAmelCase,					lowerCAmelCase    :		Union[str, Any]				=      outputs[:2]
           lowerCAmelCase    :		Optional[int]				=      logits.detach().cpu().numpy()
           lowerCAmelCase    :		Any				=      inputs['''labels'''].detach().cpu().numpy()
           return {"val_loss": tmp_eval_loss.detach().cpu(), "pred": preds, "target": out_label_ids}






     def    lowerCamelCase__   (		self     :    Union[str, Any]    , UpperCamelCase_     :    Any ):
           lowerCAmelCase    :		str				=      torch.stack([x['''val_loss'''] for x in outputs] ).mean()
           lowerCAmelCase    :		List[str]				=      np.concatenate([x['''pred'''] for x in outputs]    , axis=0 )
           lowerCAmelCase    :		Optional[Any]				=      np.argmax(UpperCamelCase_    , axis=2 )
           lowerCAmelCase    :		Optional[Any]				=      np.concatenate([x['''target'''] for x in outputs]    , axis=0 )

           lowerCAmelCase    :		Dict				=      dict(enumerate(self.labels ) )
           lowerCAmelCase    :		Tuple				=      [[] for _ in range(out_label_ids.shape[0] )]
           lowerCAmelCase    :		str				=      [[] for _ in range(out_label_ids.shape[0] )]

           for i in range(out_label_ids.shape[0] ):
                 for j in range(out_label_ids.shape[1] ):
                       if out_label_ids[i, j] != self.pad_token_label_id:
                             out_label_list[i].append(label_map[out_label_ids[i][j]] )
                             preds_list[i].append(label_map[preds[i][j]] )

           lowerCAmelCase    :		List[Any]				=      {
               '''val_loss''': val_loss_mean,
               '''accuracy_score''': accuracy_score(UpperCamelCase_    , UpperCamelCase_ ),
               '''precision''': precision_score(UpperCamelCase_    , UpperCamelCase_ ),
               '''recall''': recall_score(UpperCamelCase_    , UpperCamelCase_ ),
               '''f1''': fa_score(UpperCamelCase_    , UpperCamelCase_ ),
           }

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






     def    lowerCamelCase__   (		self     :    Union[str, Any]    , UpperCamelCase_     :    List[Any] ):
           # when stable
           lowerCAmelCase,					lowerCAmelCase,					lowerCAmelCase    :		Union[str, Any]				=      self._eval_end(UpperCamelCase_ )
           lowerCAmelCase    :		Any				=      ret['''log''']
           return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}






     def    lowerCamelCase__   (		self     :    Tuple    , UpperCamelCase_     :    Dict ):
           # updating to test_epoch_end instead of deprecated test_end
           lowerCAmelCase,					lowerCAmelCase,					lowerCAmelCase    :		Optional[Any]				=      self._eval_end(UpperCamelCase_ )

           # Converting to the dict required by pl
           # https://github.com/PyTorchLightning/pytorch-lightning/blob/master/\
           # pytorch_lightning/trainer/logging.py#L139
           lowerCAmelCase    :		Optional[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[str]    , UpperCamelCase_     :    List[str] ):
           # Add NER specific options
           BaseTransformer.add_model_specific_args(UpperCamelCase_    , UpperCamelCase_ )
           parser.add_argument(
               '''--task_type'''    , default='''NER'''    , type=UpperCamelCase_    , help='''Task type to fine tune in training (e.g. NER, POS, etc)''' )
           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(
               '''--labels'''    , default=''''''    , type=UpperCamelCase_    , help='''Path to a file containing all labels. If not specified, CoNLL-2003 labels are used.'''    , )
           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


if __name__ == "__main__":
 snake_case__						:   Optional[Any]			 = argparse.ArgumentParser()
 add_generic_args(parser, os.getcwd())
 snake_case__						:   Dict			 = NERTransformer.add_model_specific_args(parser, os.getcwd())
 snake_case__						:   str			 = parser.parse_args()
 snake_case__						:   Optional[Any]			 = NERTransformer(args)
 snake_case__						:   str			 = generic_train(model, args)

 if args.do_predict:
  # See https://github.com/huggingface/transformers/issues/3159
  # pl use this default format to create a checkpoint:
  # https://github.com/PyTorchLightning/pytorch-lightning/blob/master\
  # /pytorch_lightning/callbacks/model_checkpoint.py#L322
  snake_case__						:   Tuple			 = sorted(glob.glob(os.path.join(args.output_dir, '''checkpoint-epoch=*.ckpt'''), recursive=True))
  snake_case__						:   str			 = model.load_from_checkpoint(checkpoints[-1])
  trainer.test(model)










"""simple docstring"""



def 						lowercase    (				__snake_case				: int	):
		if not isinstance(__snake_case		,					__snake_case	):
				raise ValueError('''Input must be an integer'''	)
		if input_num <= 0:
				raise ValueError('''Input must be positive'''	)
		return sum(
		    divisor for divisor in range(1		,					input_num // 2 + 1	) if input_num % divisor == 0	)


if __name__ == "__main__":
						import doctest

						doctest.testmod()



