


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


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





class      snake_case    (    lowercase			):





  """simple docstring"""




  _lowerCamelCase		    =  "token-classification"






  def __init__(    self			,  UpperCamelCase     ):







   """simple docstring"""
   if type(UpperCamelCase     ) == dict:
    lowerCamelCase_      =  Namespace(**UpperCamelCase     )
   lowerCamelCase_      =  import_module("tasks"     )
   try:
    lowerCamelCase_      =  getattr(UpperCamelCase			,  hparams.task_type     )
    lowerCamelCase_      =  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_      =  self.token_classification_task.get_labels(hparams.labels     )
   lowerCamelCase_      =  CrossEntropyLoss().ignore_index
   super().__init__(UpperCamelCase			,  len(self.labels     )			,  self.mode     )






  def   snake_case		(    self			,  **UpperCamelCase     ):







   """simple docstring"""
   return self.model(**UpperCamelCase     )






  def   snake_case		(    self			,  UpperCamelCase			,  UpperCamelCase     ):







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

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






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  self.hparams
   for mode in ["train", "dev", "test"]:
    lowerCamelCase_      =  self._feature_file(UpperCamelCase     )
    if os.path.exists(UpperCamelCase     ) and not args.overwrite_cache:
     logger.info("Loading features from cached file %s"			,  UpperCamelCase     )
     lowerCamelCase_      =  torch.load(UpperCamelCase     )
    else:
     logger.info("Creating features from dataset file at %s"			,  args.data_dir     )
     lowerCamelCase_      =  self.token_classification_task.read_examples_from_file(args.data_dir			,  UpperCamelCase     )
     lowerCamelCase_      =  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   snake_case		(    self			,  UpperCamelCase			,  UpperCamelCase			,  UpperCamelCase = False     ):







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






  def   snake_case		(    self			,  UpperCamelCase			,  UpperCamelCase     ):







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






  def   snake_case		(    self			,  UpperCamelCase     ):







   """simple docstring"""
   lowerCamelCase_      =  torch.stack([x["val_loss"] for x in outputs]     ).mean()
   lowerCamelCase_      =  np.concatenate([x["pred"] for x in outputs]			,  axis=0     )
   lowerCamelCase_      =  np.argmax(UpperCamelCase			,  axis=2     )
   lowerCamelCase_      =  np.concatenate([x["target"] for x in outputs]			,  axis=0     )

   lowerCamelCase_      =  dict(enumerate(self.labels     )     )
   lowerCamelCase_      =  [[] for _ in range(out_label_ids.shape[0]     )]
   lowerCamelCase_      =  [[] 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_      =  {
       "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_      =  dict(results.items()     )
   lowerCamelCase_      =  results
   return ret, preds_list, out_label_list






  def   snake_case		(    self			,  UpperCamelCase     ):







   """simple docstring"""
   # when stable
   lowerCamelCase_ ,lowerCamelCase_ ,lowerCamelCase_      =  self._eval_end(UpperCamelCase     )
   lowerCamelCase_      =  ret["log"]
   return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}






  def   snake_case		(    self			,  UpperCamelCase     ):







   """simple docstring"""
   # updating to test_epoch_end instead of deprecated test_end
   lowerCamelCase_ ,lowerCamelCase_ ,lowerCamelCase_      =  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_      =  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		(    UpperCamelCase			,  UpperCamelCase     ):







   """simple docstring"""
   # 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=128			,  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__":
     a_  :				Dict						= argparse.ArgumentParser()
     add_generic_args(parser, os.getcwd())
     a_  :				Optional[int]						= NERTransformer.add_model_specific_args(parser, os.getcwd())
     a_  :				Any						= parser.parse_args()
     a_  :				int						= NERTransformer(args)
     a_  :				Union[str, Any]						= 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
          a_  :				Optional[Any]						= sorted(glob.glob(os.path.join(args.output_dir, """checkpoint-epoch=*.ckpt"""), recursive=True))
          a_  :				Optional[int]						= model.load_from_checkpoint(checkpoints[-1])
          trainer.test(model)







'''simple docstring'''

import json
import os
import unittest

from transformers.models.blenderbot_small.tokenization_blenderbot_small import (
    VOCAB_FILES_NAMES,
    BlenderbotSmallTokenizer,
)

from ...test_tokenization_common import TokenizerTesterMixin





class      snake_case    (    lowercase			,						unittest.TestCase			):





  """simple docstring"""




  _lowerCamelCase		    =  BlenderbotSmallTokenizer
  _lowerCamelCase		    =  False






  def   snake_case		(    self     ):







   """simple docstring"""
   super().setUp()

   lowerCamelCase_      =  ["__start__", "adapt", "act", "ap@@", "te", "__end__", "__unk__"]
   lowerCamelCase_      =  dict(zip(UpperCamelCase			,  range(len(UpperCamelCase     )     )     )     )

   lowerCamelCase_      =  ["#version: 0.2", "a p", "t e</w>", "ap t</w>", "a d", "ad apt</w>", "a c", "ac t</w>", ""]
   lowerCamelCase_      =  {"unk_token": "__unk__", "bos_token": "__start__", "eos_token": "__end__"}

   lowerCamelCase_      =  os.path.join(self.tmpdirname			,  VOCAB_FILES_NAMES["vocab_file"]     )
   lowerCamelCase_      =  os.path.join(self.tmpdirname			,  VOCAB_FILES_NAMES["merges_file"]     )
   with open(self.vocab_file			,  "w"			,  encoding="utf-8"     ) as fp:
    fp.write(json.dumps(UpperCamelCase     ) + "\n"     )
   with open(self.merges_file			,  "w"			,  encoding="utf-8"     ) as fp:
    fp.write("\n".join(UpperCamelCase     )     )






  def   snake_case		(    self			,  **UpperCamelCase     ):







   """simple docstring"""
   kwargs.update(self.special_tokens_map     )
   return BlenderbotSmallTokenizer.from_pretrained(self.tmpdirname			,  **UpperCamelCase     )






  def   snake_case		(    self			,  UpperCamelCase     ):







   """simple docstring"""
   lowerCamelCase_      =  "adapt act apte"
   lowerCamelCase_      =  "adapt act apte"
   return input_text, output_text






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  BlenderbotSmallTokenizer(self.vocab_file			,  self.merges_file			,  **self.special_tokens_map     )
   lowerCamelCase_      =  "adapt act apte"
   lowerCamelCase_      =  ["adapt", "act", "ap@@", "te"]
   lowerCamelCase_      =  tokenizer.tokenize(UpperCamelCase     )
   self.assertListEqual(UpperCamelCase			,  UpperCamelCase     )

   lowerCamelCase_      =  [tokenizer.bos_token] + tokens + [tokenizer.eos_token]

   lowerCamelCase_      =  [0, 1, 2, 3, 4, 5]
   self.assertListEqual(tokenizer.convert_tokens_to_ids(UpperCamelCase     )			,  UpperCamelCase     )






  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  BlenderbotSmallTokenizer.from_pretrained("facebook/blenderbot-90M"     )
   assert tok("sam"     ).input_ids == [1384]
   lowerCamelCase_      =  "I am a small frog."
   lowerCamelCase_      =  tok([src_text]			,  padding=UpperCamelCase			,  truncation=UpperCamelCase     )["input_ids"]
   lowerCamelCase_      =  tok.batch_decode(UpperCamelCase			,  skip_special_tokens=UpperCamelCase			,  clean_up_tokenization_spaces=UpperCamelCase     )[0]
   assert src_text != decoded  # I wish it did!
   assert decoded == "i am a small frog ."







  def   snake_case		(    self     ):







   """simple docstring"""
   lowerCamelCase_      =  BlenderbotSmallTokenizer.from_pretrained("facebook/blenderbot-90M"     )
   lowerCamelCase_      =  "I am a small frog ."
   lowerCamelCase_      =  "."
   lowerCamelCase_      =  tok(UpperCamelCase     )["input_ids"]
   lowerCamelCase_      =  tok(UpperCamelCase     )["input_ids"]

   assert encoded[-1] == encoded_dot[0]



