





import logging
import os
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Union

from filelock import FileLock

from transformers import PreTrainedTokenizer, is_tf_available, is_torch_available


__UpperCAmelCase       :   str				       =							logging.getLogger(__name__)







@dataclass
class     __snake_case       :





       '''simple docstring'''







       lowerCAmelCase__      		=     42
       lowerCAmelCase__      		=     42
       lowerCAmelCase__      		=     42







@dataclass
class     __snake_case       :





       '''simple docstring'''







       lowerCAmelCase__      		=     42
       lowerCAmelCase__      		=     42
       lowerCAmelCase__      		=     None
       lowerCAmelCase__      		=     None







class     __snake_case       (							__lowerCamelCase			):





       '''simple docstring'''







       lowerCAmelCase__      		=     """train"""
       lowerCAmelCase__      		=     """dev"""
       lowerCAmelCase__      		=     """test"""







class     __snake_case       :





       '''simple docstring'''







       @staticmethod
       def 							UpperCAmelCase__ (  A :    int							,				A :    Union[Split, str]		):
         raise NotImplementedError




       @staticmethod
       def 							UpperCAmelCase__ (  A :    str		):
         raise NotImplementedError






       @staticmethod
       def 							UpperCAmelCase__ (  A :    List[InputExample]							,				A :    List[str]							,				A :    int							,				A :    PreTrainedTokenizer							,				A :    Dict=False							,				A :    Optional[Any]="[CLS]"							,				A :    int=1							,				A :    str="[SEP]"							,				A :    Optional[int]=False							,				A :    List[Any]=False							,				A :    Optional[Any]=0							,				A :    Tuple=0							,				A :    int=-100							,				A :    Optional[Any]=0							,				A :    Union[str, Any]=True							,				):
         __snake_case:						Optional[Any]   =						{label: i for i, label in enumerate(A		)}

         __snake_case:						str   =						[]
         for ex_index, example in enumerate(A		):
           if ex_index % 10_000 == 0:
             logger.info("""Writing example %d of %d"""							,				A							,				len(A		)		)

           __snake_case:						Union[str, Any]   =						[]
           __snake_case:						str   =						[]
           for word, label in zip(example.words							,				example.labels		):
             __snake_case:						Optional[Any]   =						tokenizer.tokenize(A		)

             # bert-base-multilingual-cased sometimes output "nothing ([]) when calling tokenize with just a space.
             if len(A		) > 0:
               tokens.extend(A		)
               # Use the real label id for the first token of the word, and padding ids for the remaining tokens
               label_ids.extend([label_map[label]] + [pad_token_label_id] * (len(A		) - 1)		)

            # Account for [CLS] and [SEP] with "- 2" and with "- 3" for RoBERTa.
           __snake_case:						Tuple   =						tokenizer.num_special_tokens_to_add()
           if len(A		) > max_seq_length - special_tokens_count:
             __snake_case:						str   =						tokens[: (max_seq_length - special_tokens_count)]
             __snake_case:						List[str]   =						label_ids[: (max_seq_length - special_tokens_count)]

           # The convention in BERT is:
           # (a) For sequence pairs:
           #  tokens:   [CLS] is this jack ##son ##ville ? [SEP] no it is not . [SEP]
           #  type_ids:   0   0  0    0    0     0       0   0   1  1  1  1   1   1
           # (b) For single sequences:
           #  tokens:   [CLS] the dog is hairy . [SEP]
           #  type_ids:   0   0   0   0  0     0   0
           #
           # Where "type_ids" are used to indicate whether this is the first
           # sequence or the second sequence. The embedding vectors for `type=0` and
           # `type=1` were learned during pre-training and are added to the wordpiece
           # embedding vector (and position vector). This is not *strictly* necessary
           # since the [SEP] token unambiguously separates the sequences, but it makes
           # it easier for the model to learn the concept of sequences.
           #
           # For classification tasks, the first vector (corresponding to [CLS]) is
           # used as the "sentence vector". Note that this only makes sense because
           # the entire model is fine-tuned.
           tokens += [sep_token]
           label_ids += [pad_token_label_id]
           if sep_token_extra:
             # roberta uses an extra separator b/w pairs of sentences
             tokens += [sep_token]
             label_ids += [pad_token_label_id]
           __snake_case:						List[str]   =						[sequence_a_segment_id] * len(A		)

           if cls_token_at_end:
             tokens += [cls_token]
             label_ids += [pad_token_label_id]
             segment_ids += [cls_token_segment_id]
           else:
             __snake_case:						List[str]   =						[cls_token] + tokens
             __snake_case:						str   =						[pad_token_label_id] + label_ids
             __snake_case:						int   =						[cls_token_segment_id] + segment_ids

           __snake_case:						Union[str, Any]   =						tokenizer.convert_tokens_to_ids(A		)

           # The mask has 1 for real tokens and 0 for padding tokens. Only real
           # tokens are attended to.
           __snake_case:						str   =						[1 if mask_padding_with_zero else 0] * len(A		)

           # Zero-pad up to the sequence length.
           __snake_case:						Optional[Any]   =						max_seq_length - len(A		)
           if pad_on_left:
             __snake_case:						Optional[int]   =						([pad_token] * padding_length) + input_ids
             __snake_case:						str   =						([0 if mask_padding_with_zero else 1] * padding_length) + input_mask
             __snake_case:						List[str]   =						([pad_token_segment_id] * padding_length) + segment_ids
             __snake_case:						int   =						([pad_token_label_id] * padding_length) + label_ids
           else:
             input_ids += [pad_token] * padding_length
             input_mask += [0 if mask_padding_with_zero else 1] * padding_length
             segment_ids += [pad_token_segment_id] * padding_length
             label_ids += [pad_token_label_id] * padding_length

           assert len(A		) == max_seq_length
           assert len(A		) == max_seq_length
           assert len(A		) == max_seq_length
           assert len(A		) == max_seq_length

           if ex_index < 5:
             logger.info("""*** Example ***"""		)
             logger.info("""guid: %s"""							,				example.guid		)
             logger.info("""tokens: %s"""							,				""" """.join([str(A		) for x in tokens]		)		)
             logger.info("""input_ids: %s"""							,				""" """.join([str(A		) for x in input_ids]		)		)
             logger.info("""input_mask: %s"""							,				""" """.join([str(A		) for x in input_mask]		)		)
             logger.info("""segment_ids: %s"""							,				""" """.join([str(A		) for x in segment_ids]		)		)
             logger.info("""label_ids: %s"""							,				""" """.join([str(A		) for x in label_ids]		)		)

           if "token_type_ids" not in tokenizer.model_input_names:
             __snake_case:						List[str]   =						None

           features.append(
               InputFeatures(
                   input_ids=A							,				attention_mask=A							,				token_type_ids=A							,				label_ids=A		)		)
         return features


if is_torch_available():
      import torch
      from torch import nn
      from torch.utils.data import Dataset







      class     __snake_case       (							__lowerCamelCase			):





             '''simple docstring'''







             lowerCAmelCase__      		=     42
             lowerCAmelCase__      		=     nn.CrossEntropyLoss().ignore_index




             def __init__(  self :    Tuple							,				A :    TokenClassificationTask							,				A :    str							,				A :    PreTrainedTokenizer							,				A :    List[str]							,				A :    str							,				A :    Optional[int] = None							,				A :    str=False							,				A :    Split = Split.train							,				):
               # Load data features from cache or dataset file
               __snake_case:						Union[str, Any]   =						os.path.join(
                   A							,				"""cached_{}_{}_{}""".format(mode.value							,				tokenizer.__class__.__name__							,				str(A		)		)							,				)

               # Make sure only the first process in distributed training processes the dataset,
               # and the others will use the cache.
               __snake_case:						Optional[int]   =						cached_features_file + """.lock"""
               with FileLock(A		):
                 if os.path.exists(A		) and not overwrite_cache:
                   logger.info(f'''Loading features from cached file {cached_features_file}'''		)
                   __snake_case:						Dict   =						torch.load(A		)
                 else:
                   logger.info(f'''Creating features from dataset file at {data_dir}'''		)
                   __snake_case:						Optional[Any]   =						token_classification_task.read_examples_from_file(A							,				A		)
                   # TODO clean up all this to leverage built-in features of tokenizers
                   __snake_case:						Optional[int]   =						token_classification_task.convert_examples_to_features(
                       A							,				A							,				A							,				A							,				cls_token_at_end=bool(model_type in ["""xlnet"""]		)							,				cls_token=tokenizer.cls_token							,				cls_token_segment_id=2 if model_type in ["""xlnet"""] else 0							,				sep_token=tokenizer.sep_token							,				sep_token_extra=A							,				pad_on_left=bool(tokenizer.padding_side == """left"""		)							,				pad_token=tokenizer.pad_token_id							,				pad_token_segment_id=tokenizer.pad_token_type_id							,				pad_token_label_id=self.pad_token_label_id							,				)
                   logger.info(f'''Saving features into cached file {cached_features_file}'''		)
                   torch.save(self.features							,				A		)




             def __len__(  self :    Optional[int]		):
               return len(self.features		)






             def __getitem__(  self :    List[Any]							,				A :    Dict		):
               return self.features[i]


if is_tf_available():
      import tensorflow as tf







      class     __snake_case       :





             '''simple docstring'''







             lowerCAmelCase__      		=     42
             lowerCAmelCase__      		=     -1_00




             def __init__(  self :    List[str]							,				A :    TokenClassificationTask							,				A :    str							,				A :    PreTrainedTokenizer							,				A :    List[str]							,				A :    str							,				A :    Optional[int] = None							,				A :    Optional[int]=False							,				A :    Split = Split.train							,				):
               __snake_case:						Union[str, Any]   =						token_classification_task.read_examples_from_file(A							,				A		)
               # TODO clean up all this to leverage built-in features of tokenizers
               __snake_case:						int   =						token_classification_task.convert_examples_to_features(
                   A							,				A							,				A							,				A							,				cls_token_at_end=bool(model_type in ["""xlnet"""]		)							,				cls_token=tokenizer.cls_token							,				cls_token_segment_id=2 if model_type in ["""xlnet"""] else 0							,				sep_token=tokenizer.sep_token							,				sep_token_extra=A							,				pad_on_left=bool(tokenizer.padding_side == """left"""		)							,				pad_token=tokenizer.pad_token_id							,				pad_token_segment_id=tokenizer.pad_token_type_id							,				pad_token_label_id=self.pad_token_label_id							,				)

               def gen():
                 for ex in self.features:
                   if ex.token_type_ids is None:
                     yield (
                         {"input_ids": ex.input_ids, "attention_mask": ex.attention_mask},
                         ex.label_ids,
                     )
                   else:
                     yield (
                         {
                             "input_ids": ex.input_ids,
                             "attention_mask": ex.attention_mask,
                             "token_type_ids": ex.token_type_ids,
                         },
                         ex.label_ids,
                     )

               if "token_type_ids" not in tokenizer.model_input_names:
                 __snake_case:						List[str]   =						tf.data.Dataset.from_generator(
                     A							,				({"""input_ids""": tf.intaa, """attention_mask""": tf.intaa}, tf.intaa)							,				(
                         {"""input_ids""": tf.TensorShape([None]		), """attention_mask""": tf.TensorShape([None]		)},
                         tf.TensorShape([None]		),
                     )							,				)
               else:
                 __snake_case:						List[Any]   =						tf.data.Dataset.from_generator(
                     A							,				({"""input_ids""": tf.intaa, """attention_mask""": tf.intaa, """token_type_ids""": tf.intaa}, tf.intaa)							,				(
                         {
                             """input_ids""": tf.TensorShape([None]		),
                             """attention_mask""": tf.TensorShape([None]		),
                             """token_type_ids""": tf.TensorShape([None]		),
                         },
                         tf.TensorShape([None]		),
                     )							,				)




             def 							UpperCAmelCase__ (  self :    str		):
               __snake_case:						List[Any]   =						self.dataset.apply(tf.data.experimental.assert_cardinality(len(self.features		)		)		)

               return self.dataset




             def __len__(  self :    Any		):
               return len(self.features		)






             def __getitem__(  self :    int							,				A :    Tuple		):
               return self.features[i]













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


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







class     __snake_case       (							__lowerCamelCase			):





       '''simple docstring'''







       lowerCAmelCase__      		=     """token-classification"""




       def __init__(  self :    Union[str, Any]							,				A :    List[Any]		):
         if type(A		) == dict:
           __snake_case:						str   =						Namespace(**A		)
         __snake_case:						str   =						import_module("""tasks"""		)
         try:
           __snake_case:						Tuple   =						getattr(A							,				hparams.task_type		)
           __snake_case:						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__()}'''		)
         __snake_case:						Optional[int]   =						self.token_classification_task.get_labels(hparams.labels		)
         __snake_case:						str   =						CrossEntropyLoss().ignore_index
         super().__init__(A							,				len(self.labels		)							,				self.mode		)




       def 							UpperCAmelCase__ (  self :    Union[str, Any]							,				**A :    Union[str, Any]		):
         return self.model(**A		)




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

         __snake_case:						Any   =						self(**A		)
         __snake_case:						Any   =						outputs[0]
         # tensorboard_logs = {"loss": loss, "rate": self.lr_scheduler.get_last_lr()[-1]}
         return {"loss": loss}




       def 							UpperCAmelCase__ (  self :    List[str]		):
         __snake_case:						Tuple   =						self.hparams
         for mode in ["train", "dev", "test"]:
           __snake_case:						Union[str, Any]   =						self._feature_file(A		)
           if os.path.exists(A		) and not args.overwrite_cache:
             logger.info("""Loading features from cached file %s"""							,				A		)
             __snake_case:						Tuple   =						torch.load(A		)
           else:
             logger.info("""Creating features from dataset file at %s"""							,				args.data_dir		)
             __snake_case:						List[Any]   =						self.token_classification_task.read_examples_from_file(args.data_dir							,				A		)
             __snake_case:						Optional[int]   =						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 :    List[str]							,				A :    int							,				A :    int							,				A :    bool = False		):
         __snake_case:						List[str]   =						self._feature_file(A		)
         logger.info("""Loading features from cached file %s"""							,				A		)
         __snake_case:						int   =						torch.load(A		)
         __snake_case:						Optional[int]   =						torch.tensor([f.input_ids for f in features]							,				dtype=torch.long		)
         __snake_case:						Optional[Any]   =						torch.tensor([f.attention_mask for f in features]							,				dtype=torch.long		)
         if features[0].token_type_ids is not None:
           __snake_case:						Dict   =						torch.tensor([f.token_type_ids for f in features]							,				dtype=torch.long		)
         else:
           __snake_case:						Tuple   =						torch.tensor([0 for f in features]							,				dtype=torch.long		)
           # HACK(we will not use this anymore soon)
         __snake_case:						Optional[int]   =						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 :    Tuple							,				A :    Optional[Any]							,				A :    int		):
         """Compute validation""" ""
         __snake_case:						List[str]   =						{"""input_ids""": batch[0], """attention_mask""": batch[1], """labels""": batch[3]}
         if self.config.model_type != "distilbert":
           __snake_case:						Union[str, Any]   =						(
               batch[2] if self.config.model_type in ["""bert""", """xlnet"""] else None
           )  # XLM and RoBERTa don"t use token_type_ids
         __snake_case:						List[str]   =						self(**A		)
         __snake_case  ,   __snake_case:						int   =						outputs[:2]
         __snake_case:						List[str]   =						logits.detach().cpu().numpy()
         __snake_case:						Any   =						inputs["""labels"""].detach().cpu().numpy()
         return {"val_loss": tmp_eval_loss.detach().cpu(), "pred": preds, "target": out_label_ids}




       def 							UpperCAmelCase__ (  self :    Dict							,				A :    Dict		):
         __snake_case:						Union[str, Any]   =						torch.stack([x["""val_loss"""] for x in outputs]		).mean()
         __snake_case:						Tuple   =						np.concatenate([x["""pred"""] for x in outputs]							,				axis=0		)
         __snake_case:						List[str]   =						np.argmax(A							,				axis=2		)
         __snake_case:						Optional[Any]   =						np.concatenate([x["""target"""] for x in outputs]							,				axis=0		)

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

         __snake_case:						Dict   =						{
             """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		),
         }

         __snake_case:						Dict   =						dict(results.items()		)
         __snake_case:						Dict   =						results
         return ret, preds_list, out_label_list




       def 							UpperCAmelCase__ (  self :    Tuple							,				A :    int		):
         # when stable
         __snake_case  ,   __snake_case  ,   __snake_case:						List[str]   =						self._eval_end(A		)
         __snake_case:						int   =						ret["""log"""]
         return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}




       def 							UpperCAmelCase__ (  self :    Dict							,				A :    Optional[int]		):
         # updating to test_epoch_end instead of deprecated test_end
         __snake_case  ,   __snake_case  ,   __snake_case:						int   =						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
         __snake_case:						List[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__ (  A :    Any							,				A :    int		):
         # Add NER specific options
         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__":
      __UpperCAmelCase       :   Tuple				       =							argparse.ArgumentParser()
      add_generic_args(parser, os.getcwd())
      __UpperCAmelCase       :   Tuple				       =							NERTransformer.add_model_specific_args(parser, os.getcwd())
      __UpperCAmelCase       :   List[str]				       =							parser.parse_args()
      __UpperCAmelCase       :   Optional[int]				       =							NERTransformer(args)
      __UpperCAmelCase       :   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
            __UpperCAmelCase       :   int				       =							sorted(glob.glob(os.path.join(args.output_dir, "checkpoint-epoch=*.ckpt"), recursive=True))
            __UpperCAmelCase       :   Any				       =							model.load_from_checkpoint(checkpoints[-1])
            trainer.test(model)






