

from copy import deepcopy
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn
from torch.utils.data import Dataset

from .deepspeed import is_deepspeed_zeroa_enabled
from .generation.configuration_utils import GenerationConfig
from .trainer import Trainer
from .utils import logging


if TYPE_CHECKING:
      from .data.data_collator import DataCollator
      from .modeling_utils import PreTrainedModel
      from .tokenization_utils_base import PreTrainedTokenizerBase
      from .trainer_callback import TrainerCallback
      from .trainer_utils import EvalPrediction, PredictionOutput
      from .training_args import TrainingArguments


__a						:Any            = logging.get_logger(__name__)

class      _a				( _a							):




      """simple docstring"""

      def __init__(      self       :							Union[str, Any]	, UpperCAmelCase       :							List[str] = None	, UpperCAmelCase       :							Union[str, Any] = None	, UpperCAmelCase       :							Tuple = None	, UpperCAmelCase       :							Any = None	, UpperCAmelCase       :							Tuple = None	, UpperCAmelCase       :							int = None	, UpperCAmelCase       :							Any = None	, UpperCAmelCase       :							Dict = None	, UpperCAmelCase       :							Dict = None	, UpperCAmelCase       :							List[Any] = (None, None)	, UpperCAmelCase       :							Union[str, Any] = None	, ):
            super().__init__(
                model=lowerCamelCase__	, args=lowerCamelCase__	, data_collator=lowerCamelCase__	, train_dataset=lowerCamelCase__	, eval_dataset=lowerCamelCase__	, tokenizer=lowerCamelCase__	, model_init=lowerCamelCase__	, compute_metrics=lowerCamelCase__	, callbacks=lowerCamelCase__	, optimizers=lowerCamelCase__	, preprocess_logits_for_metrics=lowerCamelCase__	, )

            # Override self.model.generation_config if a GenerationConfig is specified in args.
            # Priority: args.generation_config > model.generation_config > default GenerationConfig.
            if self.args.generation_config is not None:
                  A_	     =							self.load_generation_config(self.args.generation_config  )
                  A_	     =							gen_config





      @staticmethod
      def 		__A  (      UpperCAmelCase       :							Any  ):

            # GenerationConfig provided, nothing to do
            if isinstance(lowerCamelCase__	, lowerCamelCase__  ):
                  return deepcopy(lowerCamelCase__  )

            # str or Path
            A_	     =							Path(lowerCamelCase__  ) if isinstance(lowerCamelCase__	, lowerCamelCase__  ) else gen_config_arg
            A_	     =							None

            # Figuring if it is path pointing to a file, pointing to a directory or else a model id or URL
            # This step is required in order to determine config_file_name
            if pretrained_model_name.is_file():
                  A_	     =							pretrained_model_name.name
                  A_	     =							pretrained_model_name.parent
            # dir path
            elif pretrained_model_name.is_dir():
                  pass
            # model id or URL
            else:
                  A_	     =							gen_config_arg

            A_	     =							GenerationConfig.from_pretrained(lowerCamelCase__	, lowerCamelCase__  )
            return gen_config





      def 		__A  (      self       :							str	, UpperCAmelCase       :							Dict = None	, UpperCAmelCase       :							Any = None	, UpperCAmelCase       :							List[Any] = "eval"	, **UpperCAmelCase       :							List[Any]	, ):
            A_	     =							gen_kwargs.copy()
            if gen_kwargs.get("max_length"  ) is None and gen_kwargs.get("max_new_tokens"  ) is None:
                  A_	     =							self.args.generation_max_length
            A_	     =							(
                gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"  ) is not None else self.args.generation_num_beams
            )
            A_	     =							gen_kwargs

            return super().evaluate(lowerCamelCase__	, ignore_keys=lowerCamelCase__	, metric_key_prefix=lowerCamelCase__  )





      def 		__A  (      self       :							List[Any]	, UpperCAmelCase       :							Union[str, Any]	, UpperCAmelCase       :							Union[str, Any] = None	, UpperCAmelCase       :							Tuple = "test"	, **UpperCAmelCase       :							str	, ):
            A_	     =							gen_kwargs.copy()
            if gen_kwargs.get("max_length"  ) is None and gen_kwargs.get("max_new_tokens"  ) is None:
                  A_	     =							self.args.generation_max_length
            A_	     =							(
                gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"  ) is not None else self.args.generation_num_beams
            )
            A_	     =							gen_kwargs

            return super().predict(lowerCamelCase__	, ignore_keys=lowerCamelCase__	, metric_key_prefix=lowerCamelCase__  )





      def 		__A  (      self       :							Optional[int]	, UpperCAmelCase       :							Any	, UpperCAmelCase       :							str	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							Optional[Any] = None	, **UpperCAmelCase       :							List[Any]	, ):

            if not self.args.predict_with_generate or prediction_loss_only:
                  return super().prediction_step(
                      lowerCamelCase__	, lowerCamelCase__	, prediction_loss_only=lowerCamelCase__	, ignore_keys=lowerCamelCase__  )

            A_	     =							"labels" in inputs
            A_	     =							self._prepare_inputs(lowerCamelCase__  )

            # XXX: adapt synced_gpus for fairscale as well
            # Priority (handled in generate):
            # gen_kwargs > model.generation_config > default GenerationConfig()

            if len(lowerCamelCase__  ) == 0 and hasattr(self	, "_gen_kwargs"  ):
                  A_	     =							self._gen_kwargs.copy()

            if gen_kwargs.get("max_length"  ) is None and gen_kwargs.get("max_new_tokens"  ) is None:
                  A_	     =							self.model.config.max_length
            A_	     =							(
                gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"  ) is not None else self.model.config.num_beams
            )
            A_	     =							True if is_deepspeed_zeroa_enabled() else False
            A_	     =							(
                gen_kwargs["synced_gpus"] if gen_kwargs.get("synced_gpus"  ) is not None else default_synced_gpus
            )

            # If the `decoder_input_ids` was created from `labels`, evict the former, so that the model can freely generate
            # (otherwise, it would continue generating from the padded `decoder_input_ids`)
            if (
                "labels" in inputs
                and "decoder_input_ids" in inputs
                and inputs["labels"].shape == inputs["decoder_input_ids"].shape
            ):
                  A_	     =							{k: v for k, v in inputs.items() if k != "decoder_input_ids"}
            A_	     =							self.model.generate(**lowerCamelCase__	, **lowerCamelCase__  )

            # Temporary hack to ensure the generation config is not initialized for each iteration of the evaluation loop
            # TODO: remove this hack when the legacy code that initializes generation_config from a model config is
            # removed in https://github.com/huggingface/transformers/blob/98d88b23f54e5a23e741833f1e973fdf600cc2c5/src/transformers/generation/utils.py#L1183
            if self.model.generation_config._from_model_config:
                  A_	     =							False

            # Retrieves GenerationConfig from model.generation_config
            A_	     =							self.model.generation_config
            # in case the batch is shorter than max length, the output should be padded
            if generated_tokens.shape[-1] < gen_config.max_length:
                  A_	     =							self._pad_tensors_to_max_len(lowerCamelCase__	, gen_config.max_length  )
            elif gen_config.max_new_tokens is not None and generated_tokens.shape[-1] < gen_config.max_new_tokens + 1:
                  A_	     =							self._pad_tensors_to_max_len(lowerCamelCase__	, gen_config.max_new_tokens + 1  )

            with torch.no_grad():
                  if has_labels:
                        with self.compute_loss_context_manager():
                              A_	     =							model(**lowerCamelCase__  )
                        if self.label_smoother is not None:
                              A_	     =							self.label_smoother(lowerCamelCase__	, inputs["labels"]  ).mean().detach()
                        else:
                              A_	     =							(outputs["loss"] if isinstance(lowerCamelCase__	, lowerCamelCase__  ) else outputs[0]).mean().detach()
                  else:
                        A_	     =							None

            if self.args.prediction_loss_only:
                  return loss, None, None

            if has_labels:
                  A_	     =							inputs["labels"]
                  if labels.shape[-1] < gen_config.max_length:
                        A_	     =							self._pad_tensors_to_max_len(lowerCamelCase__	, gen_config.max_length  )
                  elif gen_config.max_new_tokens is not None and labels.shape[-1] < gen_config.max_new_tokens + 1:
                        A_	     =							self._pad_tensors_to_max_len(lowerCamelCase__	, gen_config.max_new_tokens + 1  )
            else:
                  A_	     =							None

            return loss, generated_tokens, labels







      def 		__A  (      self       :							str	, UpperCAmelCase       :							int	, UpperCAmelCase       :							Optional[int]  ):
            if self.tokenizer is not None and hasattr(self.tokenizer	, "pad_token_id"  ):
                  # If PAD token is not defined at least EOS token has to be defined
                  A_	     =							(
                      self.tokenizer.pad_token_id if self.tokenizer.pad_token_id is not None else self.tokenizer.eos_token_id
                  )
            else:
                  if self.model.config.pad_token_id is not None:
                        A_	     =							self.model.config.pad_token_id
                  else:
                        raise ValueError("Pad_token_id must be set in the configuration of the model, in order to pad tensors"  )

            A_	     =							pad_token_id * torch.ones(
                (tensor.shape[0], max_length)	, dtype=tensor.dtype	, device=tensor.device  )
            A_	     =							tensor
            return padded_tensor









import json
import os
from typing import List, Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


__a						:Optional[Any]            = logging.get_logger(__name__)

__a						:Union[str, Any]            = {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

__a						:Union[str, Any]            = {
    'vocab_file': {
        'microsoft/biogpt': 'https://huggingface.co/microsoft/biogpt/resolve/main/vocab.json',
    },
    'merges_file': {'microsoft/biogpt': 'https://huggingface.co/microsoft/biogpt/resolve/main/merges.txt'},
}

__a						:str            = {
    'microsoft/biogpt': 1024,
}

def 			__snake_case	(   __UpperCamelCase  :   Any						):



      """simple docstring"""

      A_	     =							set()
      A_	     =							word[0]
      for char in word[1:]:
            pairs.add((prev_char, char)						)
            A_	     =							char
      return pairs



class      _a				( snake_case_							):




      """simple docstring"""

      _lowerCamelCase					: Tuple					=					VOCAB_FILES_NAMES
      _lowerCamelCase					: Any					=					PRETRAINED_VOCAB_FILES_MAP
      _lowerCamelCase					: str					=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _lowerCamelCase					: List[Any]					=					['input_ids', 'attention_mask']





      def __init__(      self       :							Union[str, Any]	, UpperCAmelCase       :							Union[str, Any]	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							List[Any]="<unk>"	, UpperCAmelCase       :							Dict="<s>"	, UpperCAmelCase       :							Tuple="</s>"	, UpperCAmelCase       :							Optional[int]="</s>"	, UpperCAmelCase       :							int="<pad>"	, **UpperCAmelCase       :							List[str]	, ):
            super().__init__(
                bos_token=UpperCAmelCase	, eos_token=UpperCAmelCase	, sep_token=UpperCAmelCase	, unk_token=UpperCAmelCase	, pad_token=UpperCAmelCase	, **UpperCAmelCase	, )

            try:
                  import sacremoses
            except ImportError:
                  raise ImportError(
                      "You need to install sacremoses to use BioGptTokenizer. "
                      "See https://pypi.org/project/sacremoses/ for installation."  )

            A_	     =							"en"
            A_	     =							sacremoses
            # cache of sm.MosesTokenizer instance
            A_	     =							{}
            A_	     =							{}
            with open(UpperCAmelCase	, encoding="utf-8"  ) as vocab_handle:
                  A_	     =							json.load(UpperCAmelCase  )
            A_	     =							{v: k for k, v in self.encoder.items()}
            with open(UpperCAmelCase	, encoding="utf-8"  ) as merges_handle:
                  A_	     =							merges_handle.read().split("\n"  )[:-1]
            A_	     =							[tuple(merge.split()[:2]  ) for merge in merges]
            A_	     =							dict(zip(UpperCAmelCase	, range(len(UpperCAmelCase  )  )  )  )
            A_	     =							{}





      @property
      def 		__A  (      self       :							int  ):
            return len(self.encoder  )





      def 		__A  (      self       :							Union[str, Any]  ):
            return dict(self.encoder	, **self.added_tokens_encoder  )





      def 		__A  (      self       :							Any	, UpperCAmelCase       :							Union[str, Any]	, UpperCAmelCase       :							Optional[Any]  ):
            if lang not in self.cache_moses_tokenizer:
                  A_	     =							self.sm.MosesTokenizer(lang=UpperCAmelCase  )
                  A_	     =							moses_tokenizer
            return self.cache_moses_tokenizer[lang].tokenize(
                UpperCAmelCase	, aggressive_dash_splits=UpperCAmelCase	, return_str=UpperCAmelCase	, escape=UpperCAmelCase  )





      def 		__A  (      self       :							List[Any]	, UpperCAmelCase       :							Tuple	, UpperCAmelCase       :							List[Any]  ):
            if lang not in self.cache_moses_detokenizer:
                  A_	     =							self.sm.MosesDetokenizer(lang=UpperCAmelCase  )
                  A_	     =							moses_detokenizer
            return self.cache_moses_detokenizer[lang].detokenize(UpperCAmelCase  )





      def 		__A  (      self       :							Optional[Any]	, UpperCAmelCase       :							int  ):
            A_	     =							tuple(token[:-1]  ) + (token[-1] + "</w>",)
            if token in self.cache:
                  return self.cache[token]
            A_	     =							get_pairs(UpperCAmelCase  )

            if not pairs:
                  return token + "</w>"

            while True:
                  A_	     =							min(UpperCAmelCase	, key=lambda UpperCAmelCase  : self.bpe_ranks.get(UpperCAmelCase	, float("inf"  )  )  )
                  if bigram not in self.bpe_ranks:
                        break
                  A_     ,    A_	     =							bigram
                  A_	     =							[]
                  A_	     =							0
                  while i < len(UpperCAmelCase  ):
                        try:
                              A_	     =							word.index(UpperCAmelCase	, UpperCAmelCase  )
                        except ValueError:
                              new_word.extend(word[i:]  )
                              break
                        else:
                              new_word.extend(word[i:j]  )
                              A_	     =							j

                        if word[i] == first and i < len(UpperCAmelCase  ) - 1 and word[i + 1] == second:
                              new_word.append(first + second  )
                              i += 2
                        else:
                              new_word.append(word[i]  )
                              i += 1
                  A_	     =							tuple(UpperCAmelCase  )
                  A_	     =							new_word
                  if len(UpperCAmelCase  ) == 1:
                        break
                  else:
                        A_	     =							get_pairs(UpperCAmelCase  )
            A_	     =							" ".join(UpperCAmelCase  )
            if word == "\n  </w>":
                  A_	     =							"\n</w>"
            A_	     =							word
            return word





      def 		__A  (      self       :							Optional[int]	, UpperCAmelCase       :							str	, UpperCAmelCase       :							Optional[Any]=False  ):
            if bypass_tokenizer:
                  A_	     =							text.split()
            else:
                  A_	     =							self.moses_tokenize(UpperCAmelCase	, self.lang  )

            A_	     =							[]
            for token in text:
                  if token:
                        split_tokens.extend(list(self.bpe(UpperCAmelCase  ).split(" "  )  )  )

            return split_tokens





      def 		__A  (      self       :							Dict	, UpperCAmelCase       :							Union[str, Any]  ):
            return self.encoder.get(UpperCAmelCase	, self.encoder.get(self.unk_token  )  )





      def 		__A  (      self       :							Optional[Any]	, UpperCAmelCase       :							List[str]  ):
            return self.decoder.get(UpperCAmelCase	, self.unk_token  )





      def 		__A  (      self       :							Dict	, UpperCAmelCase       :							Any  ):
            A_	     =							[t.replace(" "	, ""  ).replace("</w>"	, " "  ) for t in tokens]
            A_	     =							"".join(UpperCAmelCase  ).split()
            # detokenize
            A_	     =							self.moses_detokenize(UpperCAmelCase	, self.lang  )
            return text





      def 		__A  (      self       :							List[str]	, UpperCAmelCase       :							List[int]	, UpperCAmelCase       :							Optional[List[int]] = None  ):
            if token_ids_a is None:
                  return [self.sep_token_id] + token_ids_a
            A_	     =							[self.sep_token_id]
            return sep + token_ids_a + sep + token_ids_a





      def 		__A  (      self       :							Any	, UpperCAmelCase       :							List[int]	, UpperCAmelCase       :							Optional[List[int]] = None	, UpperCAmelCase       :							bool = 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  )
            # no bos used in fairseq
            if token_ids_a is not None:
                  return [1] + ([0] * len(UpperCAmelCase  )) + [1] + ([0] * len(UpperCAmelCase  ))
            return [1] + ([0] * len(UpperCAmelCase  ))





      def 		__A  (      self       :							Optional[Any]	, UpperCAmelCase       :							List[int]	, UpperCAmelCase       :							Optional[List[int]] = None  ):
            A_	     =							[self.sep_token_id]

            # no bos used in fairseq
            if token_ids_a is None:
                  return len(token_ids_a + sep  ) * [0]
            return len(token_ids_a + sep  ) * [0] + len(token_ids_a + sep  ) * [1]





      def 		__A  (      self       :							Union[str, Any]	, UpperCAmelCase       :							str	, UpperCAmelCase       :							Optional[str] = None  ):
            if not os.path.isdir(UpperCAmelCase  ):
                  logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''  )
                  return
            A_	     =							os.path.join(
                UpperCAmelCase	, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]  )
            A_	     =							os.path.join(
                UpperCAmelCase	, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]  )

            with open(UpperCAmelCase	, "w"	, encoding="utf-8"  ) as f:
                  f.write(json.dumps(self.encoder	, indent=2	, sort_keys=UpperCAmelCase	, ensure_ascii=UpperCAmelCase  ) + "\n"  )

            A_	     =							0
            with open(UpperCAmelCase	, "w"	, encoding="utf-8"  ) as writer:
                  for bpe_tokens, token_index in sorted(self.bpe_ranks.items()	, key=lambda UpperCAmelCase  : kv[1]  ):
                        if index != token_index:
                              logger.warning(
                                  f'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
                                  " Please check that the tokenizer is not corrupted!"  )
                              A_	     =							token_index
                        writer.write(" ".join(UpperCAmelCase  ) + "\n"  )
                        index += 1

            return vocab_file, merge_file





      def __getstate__(      self       :							Union[str, Any]  ):
            A_	     =							self.__dict__.copy()
            A_	     =							None
            return state







      def __setstate__(      self       :							List[Any]	, UpperCAmelCase       :							Dict  ):
            A_	     =							d

            try:
                  import sacremoses
            except ImportError:
                  raise ImportError(
                      "You need to install sacremoses to use XLMTokenizer. "
                      "See https://pypi.org/project/sacremoses/ for installation."  )

            A_	     =							sacremoses