


'''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   :    List[str]											= logging.getLogger(__name__)






class   __snake_case			(  _SCREAMING_SNAKE_CASE):







    """simple docstring"""



    lowercase  					=   'token-classification'

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

    def __lowercase  (  self       :    Optional[int]       ,		**lowerCamelCase       :    Union[str, Any]						)  ->    Any:
      return self.model(**lowerCamelCase						)

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

      lowerCAmelCase_     : Union[str, Any]     =  self(**lowerCamelCase						)
      lowerCAmelCase_     : Dict     =  outputs[0]
      # tensorboard_logs = {"loss": loss, "rate": self.lr_scheduler.get_last_lr()[-1]}
      return {"loss": loss}

    def __lowercase  (  self       :    List[str]						)  ->    Optional[int]:
      lowerCAmelCase_     : List[str]     =  self.hparams
      for mode in ["train", "dev", "test"]:
        lowerCAmelCase_     : Optional[int]     =  self._feature_file(lowerCamelCase						)
        if os.path.exists(lowerCamelCase						) and not args.overwrite_cache:
          logger.info("""Loading features from cached file %s"""       ,		lowerCamelCase						)
          lowerCAmelCase_     : str     =  torch.load(lowerCamelCase						)
        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       ,		lowerCamelCase						)
          lowerCAmelCase_     : int     =  self.token_classification_task.convert_examples_to_features(
              lowerCamelCase       ,		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=lowerCamelCase       ,		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"""       ,		lowerCamelCase						)
          torch.save(lowerCamelCase       ,		lowerCamelCase						)

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

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

    def __lowercase  (  self       :    Optional[Any]       ,		lowerCamelCase       :    Tuple						)  ->    Tuple:
      lowerCAmelCase_     : Dict     =  torch.stack([x["""val_loss"""] for x in outputs]						).mean()
      lowerCAmelCase_     : List[Any]     =  np.concatenate([x["""pred"""] for x in outputs]       ,		axis=0						)
      lowerCAmelCase_     : Optional[int]     =  np.argmax(lowerCamelCase       ,		axis=2						)
      lowerCAmelCase_     : str     =  np.concatenate([x["""target"""] for x in outputs]       ,		axis=0						)

      lowerCAmelCase_     : Optional[Any]     =  dict(enumerate(self.labels						)						)
      lowerCAmelCase_     : List[Any]     =  [[] for _ in range(out_label_ids.shape[0]						)]
      lowerCAmelCase_     : str     =  [[] 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_     : List[str]     =  {
          """val_loss""": val_loss_mean,
          """accuracy_score""": accuracy_score(lowerCamelCase       ,		lowerCamelCase						),
          """precision""": precision_score(lowerCamelCase       ,		lowerCamelCase						),
          """recall""": recall_score(lowerCamelCase       ,		lowerCamelCase						),
          """f1""": fa_score(lowerCamelCase       ,		lowerCamelCase						),
      }

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

    def __lowercase  (  self       :    Optional[int]       ,		lowerCamelCase       :    List[str]						)  ->    Optional[Any]:
      # when stable
      lowerCAmelCase_,				lowerCAmelCase_,				lowerCAmelCase_     : str     =  self._eval_end(lowerCamelCase						)
      lowerCAmelCase_     : Union[str, Any]     =  ret["""log"""]
      return {"val_loss": logs["val_loss"], "log": logs, "progress_bar": logs}

    def __lowercase  (  self       :    Union[str, Any]       ,		lowerCamelCase       :    int						)  ->    List[str]:
      # updating to test_epoch_end instead of deprecated test_end
      lowerCAmelCase_,				lowerCAmelCase_,				lowerCAmelCase_     : str     =  self._eval_end(lowerCamelCase						)

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







'''simple docstring'''






from math import factorial

__A   :    dict[str, int]											= {str(digit): factorial(digit) for digit in range(10)}







def 							UpperCamelCase_     (       A__   :  int   ):
  '''simple docstring'''





  if not isinstance(A__				,       A__   ):
    raise TypeError("""Parameter number must be int"""   )

  if number < 0:
    raise ValueError("""Parameter number must be greater than or equal to 0"""   )

  # Converts number in string to iterate on its digits and adds its factorial.
  return sum(DIGIT_FACTORIAL[digit] for digit in str(A__   )   )







def 							UpperCamelCase_     (       A__   :  int = 60				,       A__   :  int = 1_00_00_00   ):
  '''simple docstring'''





  if not isinstance(A__				,       A__   ) or not isinstance(A__				,       A__   ):
    raise TypeError("""Parameters chain_length and number_limit must be int"""   )

  if chain_length <= 0 or number_limit <= 0:
    raise ValueError(
        """Parameters chain_length and number_limit must be greater than 0"""   )

  # the counter for the chains with the exact desired length
  lowerCAmelCase_     : Optional[int]     =  0
  # the cached sizes of the previous chains
  lowerCAmelCase_     : dict[int, int]     =  {}

  for start_chain_element in range(1				,       A__   ):
    # The temporary set will contain the elements of the chain
    lowerCAmelCase_     : List[str]     =  set()
    lowerCAmelCase_     : Optional[int]     =  0

    # Stop computing the chain when you find a cached size, a repeating item or the
    # length is greater then the desired one.
    lowerCAmelCase_     : int     =  start_chain_element
    while (
        chain_element not in chain_sets_lengths
        and chain_element not in chain_set
        and chain_set_length <= chain_length
    ):
      chain_set.add(A__   )
      chain_set_length += 1
      lowerCAmelCase_     : Union[str, Any]     =  digit_factorial_sum(A__   )

    if chain_element in chain_sets_lengths:
      chain_set_length += chain_sets_lengths[chain_element]

    lowerCAmelCase_     : int     =  chain_set_length

    # If chain contains the exact amount of elements increase the counter
    if chain_set_length == chain_length:
      chains_counter += 1

  return chains_counter


if __name__ == "__main__":
  import doctest

  doctest.testmod()
  print(F'''{solution()}''')



