



import os
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


_lowercase     							=						logging.get_logger(__name__)
_lowercase     							=						{'''vocab_file''': '''spiece.model'''}

_lowercase     							=						{
    '''vocab_file''': {
        '''albert-base-v1''': '''https://huggingface.co/albert-base-v1/resolve/main/spiece.model''',
        '''albert-large-v1''': '''https://huggingface.co/albert-large-v1/resolve/main/spiece.model''',
        '''albert-xlarge-v1''': '''https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model''',
        '''albert-xxlarge-v1''': '''https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model''',
        '''albert-base-v2''': '''https://huggingface.co/albert-base-v2/resolve/main/spiece.model''',
        '''albert-large-v2''': '''https://huggingface.co/albert-large-v2/resolve/main/spiece.model''',
        '''albert-xlarge-v2''': '''https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model''',
        '''albert-xxlarge-v2''': '''https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model''',
    }
}

_lowercase     							=						{
    '''albert-base-v1''': 5_12,
    '''albert-large-v1''': 5_12,
    '''albert-xlarge-v1''': 5_12,
    '''albert-xxlarge-v1''': 5_12,
    '''albert-base-v2''': 5_12,
    '''albert-large-v2''': 5_12,
    '''albert-xlarge-v2''': 5_12,
    '''albert-xxlarge-v2''': 5_12,
}

_lowercase     							=						'''▁'''
class 				__A					( lowercase_      ):
			UpperCamelCase					:List[Any]     =			VOCAB_FILES_NAMES
			UpperCamelCase					:Dict     =			PRETRAINED_VOCAB_FILES_MAP
			UpperCamelCase					:Any     =			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES





			def __init__(self							,    __magic_name__							,    __magic_name__=True							,    __magic_name__=True							,    __magic_name__=False							,    __magic_name__="[CLS]"							,    __magic_name__="[SEP]"							,    __magic_name__="<unk>"							,    __magic_name__="[SEP]"							,    __magic_name__="<pad>"							,    __magic_name__="[CLS]"							,    __magic_name__="[MASK]"							,    __magic_name__ = None							,    **__magic_name__							,    ):
						lowerCamelCase__ :      str								=		(
						    AddedToken(UpperCamelCase__							,    lstrip=UpperCamelCase__							,    rstrip=UpperCamelCase__							,    normalized=UpperCamelCase__				)
						    if isinstance(UpperCamelCase__							,    UpperCamelCase__				)
						    else mask_token
						)

						lowerCamelCase__ :      int								=		{} if sp_model_kwargs is None else sp_model_kwargs

						super().__init__(
						    do_lower_case=UpperCamelCase__							,    remove_space=UpperCamelCase__							,    keep_accents=UpperCamelCase__							,    bos_token=UpperCamelCase__							,    eos_token=UpperCamelCase__							,    unk_token=UpperCamelCase__							,    sep_token=UpperCamelCase__							,    pad_token=UpperCamelCase__							,    cls_token=UpperCamelCase__							,    mask_token=UpperCamelCase__							,    sp_model_kwargs=self.sp_model_kwargs							,    **UpperCamelCase__							,    )

						lowerCamelCase__ :      Dict								=		do_lower_case
						lowerCamelCase__ :      Optional[int]								=		remove_space
						lowerCamelCase__ :      List[Any]								=		keep_accents
						lowerCamelCase__ :      int								=		vocab_file

						lowerCamelCase__ :      List[str]								=		spm.SentencePieceProcessor(**self.sp_model_kwargs				)
						self.sp_model.Load(UpperCamelCase__				)





			@property
			def        _snake_case    (self				):
						return len(self.sp_model				)





			def        _snake_case    (self				):
						lowerCamelCase__ :      str								=		{self.convert_ids_to_tokens(UpperCamelCase__				): i for i in range(self.vocab_size				)}
						vocab.update(self.added_tokens_encoder				)
						return vocab





			def __getstate__(self				):
						lowerCamelCase__ :      List[str]								=		self.__dict__.copy()
						lowerCamelCase__ :      Optional[int]								=		None
						return state





			def __setstate__(self							,    __magic_name__				):
						lowerCamelCase__ :      str								=		d

						# for backward compatibility
						if not hasattr(self							,    """sp_model_kwargs"""				):
									lowerCamelCase__ :      Dict								=		{}

						lowerCamelCase__ :      Optional[Any]								=		spm.SentencePieceProcessor(**self.sp_model_kwargs				)
						self.sp_model.Load(self.vocab_file				)





			def        _snake_case    (self							,    __magic_name__				):







						if self.remove_space:
									lowerCamelCase__ :      Any								=		""" """.join(inputs.strip().split()				)
						else:
									lowerCamelCase__ :      int								=		inputs
						lowerCamelCase__ :      Union[str, Any]								=		outputs.replace("""``"""							,    """\""""				).replace("""''"""							,    """\""""				)

						if not self.keep_accents:
									lowerCamelCase__ :      Any								=		unicodedata.normalize("""NFKD"""							,    UpperCamelCase__				)
									lowerCamelCase__ :      Optional[int]								=		"""""".join([c for c in outputs if not unicodedata.combining(UpperCamelCase__				)]				)
						if self.do_lower_case:
									lowerCamelCase__ :      str								=		outputs.lower()

						return outputs





			def        _snake_case    (self							,    __magic_name__				):
						lowerCamelCase__ :      Dict								=		self.preprocess_text(UpperCamelCase__				)
						lowerCamelCase__ :      List[Any]								=		self.sp_model.encode(UpperCamelCase__							,    out_type=UpperCamelCase__				)
						lowerCamelCase__ :      int								=		[]
						for piece in pieces:
									if len(UpperCamelCase__				) > 1 and piece[-1] == str(""","""				) and piece[-2].isdigit():
												lowerCamelCase__ :      List[Any]								=		self.sp_model.EncodeAsPieces(piece[:-1].replace(UpperCamelCase__							,    """"""				)				)
												if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
															if len(cur_pieces[0]				) == 1:
																		lowerCamelCase__ :      List[str]								=		cur_pieces[1:]
															else:
																		lowerCamelCase__ :      Optional[Any]								=		cur_pieces[0][1:]
												cur_pieces.append(piece[-1]				)
												new_pieces.extend(UpperCamelCase__				)
									else:
												new_pieces.append(UpperCamelCase__				)

						return new_pieces





			def        _snake_case    (self							,    __magic_name__				):
						return self.sp_model.PieceToId(UpperCamelCase__				)





			def        _snake_case    (self							,    __magic_name__				):
						return self.sp_model.IdToPiece(UpperCamelCase__				)





			def        _snake_case    (self							,    __magic_name__				):
						lowerCamelCase__ :      List[Any]								=		[]
						lowerCamelCase__ :      List[Any]								=		""""""
						lowerCamelCase__ :      int								=		False
						for token in tokens:
									# make sure that special tokens are not decoded using sentencepiece model
									if token in self.all_special_tokens:
												if not prev_is_special:
															out_string += " "
												out_string += self.sp_model.decode(UpperCamelCase__				) + token
												lowerCamelCase__ :      Optional[int]								=		True
												lowerCamelCase__ :      int								=		[]
									else:
												current_sub_tokens.append(UpperCamelCase__				)
												lowerCamelCase__ :      Any								=		False
						out_string += self.sp_model.decode(UpperCamelCase__				)
						return out_string.strip()





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__ = None				):
						lowerCamelCase__ :      int								=		[self.sep_token_id]
						lowerCamelCase__ :      Optional[int]								=		[self.cls_token_id]
						if token_ids_a is None:
									return cls + token_ids_a + sep
						return cls + token_ids_a + sep + token_ids_a + sep





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__ = None							,    __magic_name__ = False				):







						if already_has_special_tokens:
									return super().get_special_tokens_mask(
									    token_ids_a=UpperCamelCase__							,    token_ids_a=UpperCamelCase__							,    already_has_special_tokens=UpperCamelCase__				)

						if token_ids_a is not None:
									return [1] + ([0] * len(UpperCamelCase__				)) + [1] + ([0] * len(UpperCamelCase__				)) + [1]
						return [1] + ([0] * len(UpperCamelCase__				)) + [1]





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__ = None				):
						lowerCamelCase__ :      Union[str, Any]								=		[self.sep_token_id]
						lowerCamelCase__ :      Union[str, Any]								=		[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				) * [0] + len(token_ids_a + sep				) * [1]





			def        _snake_case    (self							,    __magic_name__							,    __magic_name__ = None				):







						if not os.path.isdir(UpperCamelCase__				):
									logger.error(f"Vocabulary path ({save_directory}) should be a directory"				)
									return
						lowerCamelCase__ :      int								=		os.path.join(
						    UpperCamelCase__							,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]				)

						if os.path.abspath(self.vocab_file				) != os.path.abspath(UpperCamelCase__				) and os.path.isfile(self.vocab_file				):
									copyfile(self.vocab_file							,    UpperCamelCase__				)
						elif not os.path.isfile(self.vocab_file				):
									with open(UpperCamelCase__							,    """wb"""				) as fi:
												lowerCamelCase__ :      Optional[Any]								=		self.sp_model.serialized_model_proto()
												fi.write(UpperCamelCase__				)

						return (out_vocab_file,)








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


_lowercase     							=						logging.getLogger(__name__)
class 				__A					( A_      ):
			UpperCamelCase					:Optional[int]     =			'''token-classification'''





			def __init__(self							,    __magic_name__				):
						if type(__magic_name__				) == dict:
									lowerCamelCase__ :      Any								=		Namespace(**__magic_name__				)
						lowerCamelCase__ :      str								=		import_module("""tasks"""				)
						try:
									lowerCamelCase__ :      Optional[Any]								=		getattr(__magic_name__							,    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__ :      Tuple								=		CrossEntropyLoss().ignore_index
						super().__init__(__magic_name__							,    len(self.labels				)							,    self.mode				)





			def        _snake_case    (self							,    **__magic_name__				):
						return self.model(**__magic_name__				)





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

						lowerCamelCase__ :      List[str]								=		self(**__magic_name__				)
						lowerCamelCase__ :      Dict								=		outputs[0]
						# tensorboard_logs = {"loss": loss, "rate": self.lr_scheduler.get_last_lr()[-1]}
						return {"loss": loss}





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





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





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





			def        _snake_case    (self							,    __magic_name__				):
						lowerCamelCase__ :      List[str]								=		torch.stack([x["""val_loss"""] for x in outputs]				).mean()
						lowerCamelCase__ :      Optional[int]								=		np.concatenate([x["""pred"""] for x in outputs]							,    axis=0				)
						lowerCamelCase__ :      List[str]								=		np.argmax(__magic_name__							,    axis=2				)
						lowerCamelCase__ :      Optional[Any]								=		np.concatenate([x["""target"""] for x in outputs]							,    axis=0				)

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

						lowerCamelCase__ :      Tuple								=		{
						    """val_loss""": val_loss_mean,
						    """accuracy_score""": accuracy_score(__magic_name__							,    __magic_name__				),
						    """precision""": precision_score(__magic_name__							,    __magic_name__				),
						    """recall""": recall_score(__magic_name__							,    __magic_name__				),
						    """f1""": fa_score(__magic_name__							,    __magic_name__				),
						}

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





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





			def        _snake_case    (self							,    __magic_name__				):
						# updating to test_epoch_end instead of deprecated test_end
						lowerCamelCase__							,lowerCamelCase__							,lowerCamelCase__ :      Union[str, Any]								=		self._eval_end(__magic_name__				)

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



