





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


__lowerCamelCase :  Union[str, Any]										=  logging.getLogger(__name__)



class       A__     ( __snake_case    ):
      _UpperCAmelCase			:Optional[int]							=     'token-classification'





      def __init__(				self ,							A_     ):
       '''simple docstring'''




       if type(A_     ) == dict:
        UpperCamelCase			:      Any 			=     Namespace(**A_     )
       UpperCamelCase			:      Tuple 			=     import_module("tasks"     )
       try:
        UpperCamelCase			:      Dict 			=     getattr(A_ ,							hparams.task_type     )
        UpperCamelCase			:      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__()}"""     )
       UpperCamelCase			:      List[Any] 			=     self.token_classification_task.get_labels(hparams.labels     )
       UpperCamelCase			:      Any 			=     CrossEntropyLoss().ignore_index
       super().__init__(A_ ,							len(self.labels     ) ,							self.mode     )





      def  __UpperCamelCase(				self ,							**A_     ):
       '''simple docstring'''




       return self.model(**A_     )





      def  __UpperCamelCase(				self ,							A_ ,							A_     ):
       '''simple docstring'''




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

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





      def  __UpperCamelCase(				self     ):
       '''simple docstring'''




       UpperCamelCase			:      List[str] 			=     self.hparams
       for mode in ["train", "dev", "test"]:
        UpperCamelCase			:      str 			=     self._feature_file(A_     )
        if os.path.exists(A_     ) and not args.overwrite_cache:
         logger.info("Loading features from cached file %s" ,							A_     )
         UpperCamelCase			:      Union[str, Any] 			=     torch.load(A_     )
        else:
         logger.info("Creating features from dataset file at %s" ,							args.data_dir     )
         UpperCamelCase			:      int 			=     self.token_classification_task.read_examples_from_file(args.data_dir ,							A_     )
         UpperCamelCase			:      Optional[Any] 			=     self.token_classification_task.convert_examples_to_features(
             A_ ,							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=A_ ,							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" ,							A_     )
         torch.save(A_ ,							A_     )





      def  __UpperCamelCase(				self ,							A_ ,							A_ ,							A_ = False     ):
       '''simple docstring'''




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





      def  __UpperCamelCase(				self ,							A_ ,							A_     ):
       '''simple docstring'''




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





      def  __UpperCamelCase(				self ,							A_     ):
       '''simple docstring'''




       UpperCamelCase			:      Optional[int] 			=     torch.stack([x["val_loss"] for x in outputs]     ).mean()
       UpperCamelCase			:      List[str] 			=     np.concatenate([x["pred"] for x in outputs] ,							axis=0     )
       UpperCamelCase			:      List[str] 			=     np.argmax(A_ ,							axis=2     )
       UpperCamelCase			:      Union[str, Any] 			=     np.concatenate([x["target"] for x in outputs] ,							axis=0     )

       UpperCamelCase			:      List[str] 			=     dict(enumerate(self.labels     )     )
       UpperCamelCase			:      Tuple 			=     [[] for _ in range(out_label_ids.shape[0]     )]
       UpperCamelCase			:      List[Any] 			=     [[] 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]]     )

       UpperCamelCase			:      Union[str, Any] 			=     {
           "val_loss": val_loss_mean,
           "accuracy_score": accuracy_score(A_ ,							A_     ),
           "precision": precision_score(A_ ,							A_     ),
           "recall": recall_score(A_ ,							A_     ),
           "f1": fa_score(A_ ,							A_     ),
       }

       UpperCamelCase			:      Tuple 			=     dict(results.items()     )
       UpperCamelCase			:      int 			=     results
       return ret, preds_list, out_label_list





      def  __UpperCamelCase(				self ,							A_     ):
       '''simple docstring'''




       UpperCamelCase			,					UpperCamelCase			,					UpperCamelCase			:      str 			=     self._eval_end(A_     )
       UpperCamelCase			:      Dict 			=     ret["log"]
       return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}





      def  __UpperCamelCase(				self ,							A_     ):
       '''simple docstring'''




       UpperCamelCase			,					UpperCamelCase			,					UpperCamelCase			:      List[str] 			=     self._eval_end(A_     )

       # Converting to the dict required by pl
       # https://github.com/PyTorchLightning/pytorch-lightning/blob/master/\
       # pytorch_lightning/trainer/logging.py#L139
       UpperCamelCase			:      Tuple 			=     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(				A_ ,							A_     ):
       '''simple docstring'''




       BaseTransformer.add_model_specific_args(A_ ,							A_     )
       parser.add_argument(
           "--task_type" ,							default="NER" ,							type=A_ ,							help="Task type to fine tune in training (e.g. NER, POS, etc)"     )
       parser.add_argument(
           "--max_seq_length" ,							default=128 ,							type=A_ ,							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=A_ ,							help="Path to a file containing all labels. If not specified, CoNLL-2003 labels are used." ,							)
       parser.add_argument(
           "--gpus" ,							default=0 ,							type=A_ ,							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__":
    __lowerCamelCase :  Tuple										=  argparse.ArgumentParser()
    add_generic_args(parser, os.getcwd())
    __lowerCamelCase :  Union[str, Any]										=  NERTransformer.add_model_specific_args(parser, os.getcwd())
    __lowerCamelCase :  Tuple										=  parser.parse_args()
    __lowerCamelCase :  Any										=  NERTransformer(args)
    __lowerCamelCase :  Tuple										=  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
        __lowerCamelCase :  List[Any]										=  sorted(glob.glob(os.path.join(args.output_dir, """checkpoint-epoch=*.ckpt"""), recursive=True))
        __lowerCamelCase :  Union[str, Any]										=  model.load_from_checkpoint(checkpoints[-1])
        trainer.test(model)







def 	A_       (  _lowerCAmelCase							,     _lowerCAmelCase  )		->				int:
 return int(input_a == input_a == 0  )



def 	A_       (  )		->				None:
 print("Truth Table of NOR Gate:"  )
 print("|   Input 1   |  Input 2  | Output |"  )
 print(F"""|      0      |    0      |  {nor_gate(0							,     0  )}     |"""  )
 print(F"""|      0      |    1      |  {nor_gate(0							,     1  )}     |"""  )
 print(F"""|      1      |    0      |  {nor_gate(1							,     0  )}     |"""  )
 print(F"""|      1      |    1      |  {nor_gate(1							,     1  )}     |"""  )


if __name__ == "__main__":
    import doctest

    doctest.testmod()
    main()
