






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


__lowerCamelCase							=  logging.getLogger(__name__)



class       __A		(   SCREAMING_SNAKE_CASE_			):
					UpperCAmelCase__								=				"token-classification"

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

					def   lowerCamelCase__   (			self  :      Any     ,		**__snake_case  :      Union[str, Any]					)			->							Any:
											return self.model(**__snake_case					)

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

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

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

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

					def   lowerCamelCase__   (			self  :      Union[str, Any]     ,		__snake_case  :      List[Any]     ,		__snake_case  :      List[str]					)			->							Union[str, Any]:
											"""Compute validation""" ""
											__magic_name__:	Optional[Any]							=       {"""input_ids""": batch[0], """attention_mask""": batch[1], """labels""": batch[3]}
											if self.config.model_type != "distilbert":
																	__magic_name__:	Dict							=       (
																	    batch[2] if self.config.model_type in ["""bert""", """xlnet"""] else None
																	)  # XLM and RoBERTa don"t use token_type_ids
											__magic_name__:	int							=       self(**__snake_case					)
											__magic_name__,			__magic_name__:	Any							=       outputs[:2]
											__magic_name__:	Any							=       logits.detach().cpu().numpy()
											__magic_name__:	Dict							=       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]     ,		__snake_case  :      str					)			->							List[Any]:
											__magic_name__:	Union[str, Any]							=       torch.stack([x["""val_loss"""] for x in outputs]					).mean()
											__magic_name__:	Optional[int]							=       np.concatenate([x["""pred"""] for x in outputs]     ,		axis=0					)
											__magic_name__:	Union[str, Any]							=       np.argmax(__snake_case     ,		axis=2					)
											__magic_name__:	Dict							=       np.concatenate([x["""target"""] for x in outputs]     ,		axis=0					)

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

											__magic_name__:	List[Any]							=       {
											    """val_loss""": val_loss_mean,
											    """accuracy_score""": accuracy_score(__snake_case     ,		__snake_case					),
											    """precision""": precision_score(__snake_case     ,		__snake_case					),
											    """recall""": recall_score(__snake_case     ,		__snake_case					),
											    """f1""": fa_score(__snake_case     ,		__snake_case					),
											}

											__magic_name__:	Dict							=       dict(results.items()					)
											__magic_name__:	List[str]							=       results
											return ret, preds_list, out_label_list

					def   lowerCamelCase__   (			self  :      Tuple     ,		__snake_case  :      List[str]					)			->							Any:
											# when stable
											__magic_name__,			__magic_name__,			__magic_name__:	List[Any]							=       self._eval_end(__snake_case					)
											__magic_name__:	Optional[int]							=       ret["""log"""]
											return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}

					def   lowerCamelCase__   (			self  :      List[str]     ,		__snake_case  :      Optional[Any]					)			->							int:
											# updating to test_epoch_end instead of deprecated test_end
											__magic_name__,			__magic_name__,			__magic_name__:	Any							=       self._eval_end(__snake_case					)

											# Converting to the dict required by pl
											# https://github.com/PyTorchLightning/pytorch-lightning/blob/master/\
											# pytorch_lightning/trainer/logging.py#L139
											__magic_name__:	Dict							=       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__   (			__snake_case  :      Optional[int]     ,		__snake_case  :      List[Any]					)			->							str:
											# Add NER specific options
											BaseTransformer.add_model_specific_args(__snake_case     ,		__snake_case					)
											parser.add_argument(
											    """--task_type"""     ,		default="""NER"""     ,		type=__snake_case     ,		help="""Task type to fine tune in training (e.g. NER, POS, etc)"""					)
											parser.add_argument(
											    """--max_seq_length"""     ,		default=1_2_8     ,		type=__snake_case     ,		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=__snake_case     ,		help="""Path to a file containing all labels. If not specified, CoNLL-2003 labels are used."""     ,		)
											parser.add_argument(
											    """--gpus"""     ,		default=0     ,		type=__snake_case     ,		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							=  argparse.ArgumentParser()
							add_generic_args(parser, os.getcwd())
							__lowerCamelCase							=  NERTransformer.add_model_specific_args(parser, os.getcwd())
							__lowerCamelCase							=  parser.parse_args()
							__lowerCamelCase							=  NERTransformer(args)
							__lowerCamelCase							=  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							=  sorted(glob.glob(os.path.join(args.output_dir, 'checkpoint-epoch=*.ckpt'), recursive=True))
														__lowerCamelCase							=  model.load_from_checkpoint(checkpoints[-1])
														trainer.test(model)










"""simple docstring"""





import json
from typing import List, Optional, Tuple

from tokenizers import pre_tokenizers, processors

from ...tokenization_utils_base import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roberta import RobertaTokenizer


__A											=						logging.get_logger(__name__)

__A											=						{"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt""", """tokenizer_file""": """tokenizer.json"""}

__A											=						{
    """vocab_file""": {
        """roberta-base""": """https://huggingface.co/roberta-base/resolve/main/vocab.json""",
        """roberta-large""": """https://huggingface.co/roberta-large/resolve/main/vocab.json""",
        """roberta-large-mnli""": """https://huggingface.co/roberta-large-mnli/resolve/main/vocab.json""",
        """distilroberta-base""": """https://huggingface.co/distilroberta-base/resolve/main/vocab.json""",
        """roberta-base-openai-detector""": """https://huggingface.co/roberta-base-openai-detector/resolve/main/vocab.json""",
        """roberta-large-openai-detector""": (
            """https://huggingface.co/roberta-large-openai-detector/resolve/main/vocab.json"""
        ),
    },
    """merges_file""": {
        """roberta-base""": """https://huggingface.co/roberta-base/resolve/main/merges.txt""",
        """roberta-large""": """https://huggingface.co/roberta-large/resolve/main/merges.txt""",
        """roberta-large-mnli""": """https://huggingface.co/roberta-large-mnli/resolve/main/merges.txt""",
        """distilroberta-base""": """https://huggingface.co/distilroberta-base/resolve/main/merges.txt""",
        """roberta-base-openai-detector""": """https://huggingface.co/roberta-base-openai-detector/resolve/main/merges.txt""",
        """roberta-large-openai-detector""": (
            """https://huggingface.co/roberta-large-openai-detector/resolve/main/merges.txt"""
        ),
    },
    """tokenizer_file""": {
        """roberta-base""": """https://huggingface.co/roberta-base/resolve/main/tokenizer.json""",
        """roberta-large""": """https://huggingface.co/roberta-large/resolve/main/tokenizer.json""",
        """roberta-large-mnli""": """https://huggingface.co/roberta-large-mnli/resolve/main/tokenizer.json""",
        """distilroberta-base""": """https://huggingface.co/distilroberta-base/resolve/main/tokenizer.json""",
        """roberta-base-openai-detector""": (
            """https://huggingface.co/roberta-base-openai-detector/resolve/main/tokenizer.json"""
        ),
        """roberta-large-openai-detector""": (
            """https://huggingface.co/roberta-large-openai-detector/resolve/main/tokenizer.json"""
        ),
    },
}

__A											=						{
    """roberta-base""": 512,
    """roberta-large""": 512,
    """roberta-large-mnli""": 512,
    """distilroberta-base""": 512,
    """roberta-base-openai-detector""": 512,
    """roberta-large-openai-detector""": 512,
}






class    _lowerCAmelCase  (	a			):
       """simple docstring"""




       __magic_name__	:str						    =   VOCAB_FILES_NAMES
       __magic_name__	:List[Any]						    =   PRETRAINED_VOCAB_FILES_MAP
       __magic_name__	:Optional[Any]						    =   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       __magic_name__	:str						    =   ["""input_ids""", """attention_mask"""]
       __magic_name__	:Any						    =   RobertaTokenizer






       def __init__(       self       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase="replace"       ,					__UpperCAmelCase="<s>"       ,					__UpperCAmelCase="</s>"       ,					__UpperCAmelCase="</s>"       ,					__UpperCAmelCase="<s>"       ,					__UpperCAmelCase="<unk>"       ,					__UpperCAmelCase="<pad>"       ,					__UpperCAmelCase="<mask>"       ,					__UpperCAmelCase=False       ,					__UpperCAmelCase=True       ,					**__UpperCAmelCase       ,					):



         '''simple docstring'''

         super().__init__(
             __UpperCAmelCase       ,					__UpperCAmelCase       ,					tokenizer_file=__UpperCAmelCase       ,					errors=__UpperCAmelCase       ,					bos_token=__UpperCAmelCase       ,					eos_token=__UpperCAmelCase       ,					sep_token=__UpperCAmelCase       ,					cls_token=__UpperCAmelCase       ,					unk_token=__UpperCAmelCase       ,					pad_token=__UpperCAmelCase       ,					mask_token=__UpperCAmelCase       ,					add_prefix_space=__UpperCAmelCase       ,					trim_offsets=__UpperCAmelCase       ,					**__UpperCAmelCase       ,					)

         lowerCAmelCase__     :Dict											=				json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__() )
         if pre_tok_state.get('add_prefix_space'       ,					__UpperCAmelCase ) != add_prefix_space:
           lowerCAmelCase__     :Optional[int]											=				getattr(__UpperCAmelCase       ,					pre_tok_state.pop('type' ) )
           lowerCAmelCase__     :List[Any]											=				add_prefix_space
           lowerCAmelCase__     :str											=				pre_tok_class(**__UpperCAmelCase )

         lowerCAmelCase__     :List[str]											=				add_prefix_space

         lowerCAmelCase__     :str											=				'post_processor'
         lowerCAmelCase__     :Optional[Any]											=				getattr(self.backend_tokenizer       ,					__UpperCAmelCase       ,					__UpperCAmelCase )
         if tokenizer_component_instance:
           lowerCAmelCase__     :Optional[Any]											=				json.loads(tokenizer_component_instance.__getstate__() )

           # The lists 'sep' and 'cls' must be cased in tuples for the object `post_processor_class`
           if "sep" in state:
             lowerCAmelCase__     :Any											=				tuple(state['sep'] )
           if "cls" in state:
             lowerCAmelCase__     :int											=				tuple(state['cls'] )

           lowerCAmelCase__     :List[Any]											=				False

           if state.get('add_prefix_space'       ,					__UpperCAmelCase ) != add_prefix_space:
             lowerCAmelCase__     :Union[str, Any]											=				add_prefix_space
             lowerCAmelCase__     :Any											=				True

           if state.get('trim_offsets'       ,					__UpperCAmelCase ) != trim_offsets:
             lowerCAmelCase__     :Union[str, Any]											=				trim_offsets
             lowerCAmelCase__     :Optional[int]											=				True

           if changes_to_apply:
             lowerCAmelCase__     :str											=				getattr(__UpperCAmelCase       ,					state.pop('type' ) )
             lowerCAmelCase__     :Any											=				component_class(**__UpperCAmelCase )
             setattr(self.backend_tokenizer       ,					__UpperCAmelCase       ,					__UpperCAmelCase )






       @property
       def 	snake_case		(       self ):



         '''simple docstring'''

         if self._mask_token is None:
           if self.verbose:
             logger.error('Using mask_token, but it is not set yet.' )
           return None
         return str(self._mask_token )






       @mask_token.setter
       def 	snake_case		(       self       ,					__UpperCAmelCase ):



         '''simple docstring'''

         lowerCAmelCase__     :List[Any]											=				AddedToken(__UpperCAmelCase       ,					lstrip=__UpperCAmelCase       ,					rstrip=__UpperCAmelCase ) if isinstance(__UpperCAmelCase       ,					__UpperCAmelCase ) else value
         lowerCAmelCase__     :List[str]											=				value






       def 	snake_case		(       self       ,					*__UpperCAmelCase       ,					**__UpperCAmelCase ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[Any]											=				kwargs.get('is_split_into_words'       ,					__UpperCAmelCase )
         assert self.add_prefix_space or not is_split_into_words, (
             F"You need to instantiate {self.__class__.__name__} with add_prefix_space=True "
             "to use it with pretokenized inputs."
         )

         return super()._batch_encode_plus(*__UpperCAmelCase       ,					**__UpperCAmelCase )






       def 	snake_case		(       self       ,					*__UpperCAmelCase       ,					**__UpperCAmelCase ):



         '''simple docstring'''

         lowerCAmelCase__     :Any											=				kwargs.get('is_split_into_words'       ,					__UpperCAmelCase )

         assert self.add_prefix_space or not is_split_into_words, (
             F"You need to instantiate {self.__class__.__name__} with add_prefix_space=True "
             "to use it with pretokenized inputs."
         )

         return super()._encode_plus(*__UpperCAmelCase       ,					**__UpperCAmelCase )






       def 	snake_case		(       self       ,					__UpperCAmelCase       ,					__UpperCAmelCase = None ):



         '''simple docstring'''

         lowerCAmelCase__     :Union[str, Any]											=				self._tokenizer.model.save(__UpperCAmelCase       ,					name=__UpperCAmelCase )
         return tuple(__UpperCAmelCase )






       def 	snake_case		(       self       ,					__UpperCAmelCase       ,					__UpperCAmelCase=None ):



         '''simple docstring'''

         lowerCAmelCase__     :str											=				[self.bos_token_id] + token_ids_a + [self.eos_token_id]
         if token_ids_a is None:
           return output

         return output + [self.eos_token_id] + token_ids_a + [self.eos_token_id]






       def 	snake_case		(       self       ,					__UpperCAmelCase       ,					__UpperCAmelCase = None ):



         '''simple docstring'''

         lowerCAmelCase__     :List[Any]											=				[self.sep_token_id]
         lowerCAmelCase__     :int											=				[self.cls_token_id]

         if token_ids_a is None:
           return len(cls + token_ids_a + sep ) * [0]
         return len(cls + token_ids_a + sep + sep + token_ids_a + sep ) * [0]
