




'''simple docstring'''




import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
    import torch_xla.core.xla_model as xm
    import torch_xla.debug.metrics as met

class   lowerCAmelCase_      (       lowerCamelCase_      ):







  '''simple docstring'''







  def __init__(  self			:		Tuple  ,						*_UpperCAmelCase			:		Union[str, Any]  ,						_UpperCAmelCase			:		str=None  ,						_UpperCAmelCase			:		str=None  ,						**_UpperCAmelCase			:		Union[str, Any]       ):






   """simple docstring"""






   super().__init__(*_lowerCAmelCase  ,						**_lowerCAmelCase       )
   UpperCAmelCase__								=			eval_examples
   UpperCAmelCase__								=			post_process_function


  def 							SCREAMING_SNAKE_CASE__	(  self			:		List[Any]  ,						_UpperCAmelCase			:		Optional[Dataset] = None  ,						_UpperCAmelCase			:		Any=None  ,						_UpperCAmelCase			:		Optional[List[str]] = None  ,						_UpperCAmelCase			:		str = "eval"  ,						**_UpperCAmelCase			:		Optional[int]  ,						):






   """simple docstring"""






   UpperCAmelCase__								=			gen_kwargs.copy()
   UpperCAmelCase__								=			(
       gen_kwargs["max_length"] if gen_kwargs.get("""max_length"""       ) is not None else self.args.generation_max_length
   )
   UpperCAmelCase__								=			(
       gen_kwargs["num_beams"] if gen_kwargs.get("""num_beams"""       ) is not None else self.args.generation_num_beams
   )
   UpperCAmelCase__								=			gen_kwargs

   UpperCAmelCase__								=			self.eval_dataset if eval_dataset is None else eval_dataset
   UpperCAmelCase__								=			self.get_eval_dataloader(_lowerCAmelCase       )
   UpperCAmelCase__								=			self.eval_examples if eval_examples is None else eval_examples

   # Temporarily disable metric computation, we will do it in the loop here.
   UpperCAmelCase__								=			self.compute_metrics
   UpperCAmelCase__								=			None
   UpperCAmelCase__								=			time.time()
   UpperCAmelCase__								=			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
   try:
    UpperCAmelCase__								=			eval_loop(
        _lowerCAmelCase  ,						description="""Evaluation"""  ,						prediction_loss_only=True if compute_metrics is None else None  ,						ignore_keys=_lowerCAmelCase  ,						metric_key_prefix=_lowerCAmelCase  ,						)
   finally:
    UpperCAmelCase__								=			compute_metrics
   UpperCAmelCase__								=			self.args.eval_batch_size * self.args.world_size
   if f'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
    start_time += output.metrics[f'''{metric_key_prefix}_jit_compilation_time''']
   output.metrics.update(
       speed_metrics(
           _lowerCAmelCase  ,						_lowerCAmelCase  ,						num_samples=output.num_samples  ,						num_steps=math.ceil(output.num_samples / total_batch_size       )  ,						)       )

   if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
    # Only the main node write the results by default
    UpperCAmelCase__								=			self.post_process_function(_lowerCAmelCase  ,						_lowerCAmelCase  ,						_lowerCAmelCase       )
    UpperCAmelCase__								=			self.compute_metrics(_lowerCAmelCase       )

    # Prefix all keys with metric_key_prefix + '_'
    for key in list(metrics.keys()       ):
     if not key.startswith(f'''{metric_key_prefix}_'''       ):
      UpperCAmelCase__								=			metrics.pop(_lowerCAmelCase       )

    metrics.update(output.metrics       )
   else:
    UpperCAmelCase__								=			output.metrics

   if self.args.should_log:
    # Only the main node log the results by default
    self.log(_lowerCAmelCase       )

   if self.args.tpu_metrics_debug or self.args.debug:
    # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
    xm.master_print(met.metrics_report()       )

   UpperCAmelCase__								=			self.callback_handler.on_evaluate(self.args  ,						self.state  ,						self.control  ,						_lowerCAmelCase       )
   return metrics



  def 							SCREAMING_SNAKE_CASE__	(  self			:		Optional[Any]  ,						_UpperCAmelCase			:		Union[str, Any]  ,						_UpperCAmelCase			:		Union[str, Any]  ,						_UpperCAmelCase			:		List[Any]=None  ,						_UpperCAmelCase			:		str = "test"  ,						**_UpperCAmelCase			:		Tuple       ):






   """simple docstring"""






   UpperCAmelCase__								=			gen_kwargs.copy()

   UpperCAmelCase__								=			self.get_test_dataloader(_lowerCAmelCase       )

   # Temporarily disable metric computation, we will do it in the loop here.
   UpperCAmelCase__								=			self.compute_metrics
   UpperCAmelCase__								=			None
   UpperCAmelCase__								=			time.time()
   UpperCAmelCase__								=			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
   try:
    UpperCAmelCase__								=			eval_loop(
        _lowerCAmelCase  ,						description="""Prediction"""  ,						prediction_loss_only=True if compute_metrics is None else None  ,						ignore_keys=_lowerCAmelCase  ,						metric_key_prefix=_lowerCAmelCase  ,						)
   finally:
    UpperCAmelCase__								=			compute_metrics

   UpperCAmelCase__								=			self.args.eval_batch_size * self.args.world_size
   if f'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
    start_time += output.metrics[f'''{metric_key_prefix}_jit_compilation_time''']
   output.metrics.update(
       speed_metrics(
           _lowerCAmelCase  ,						_lowerCAmelCase  ,						num_samples=output.num_samples  ,						num_steps=math.ceil(output.num_samples / total_batch_size       )  ,						)       )
   if self.post_process_function is None or self.compute_metrics is None:
    return output

   UpperCAmelCase__								=			self.post_process_function(_lowerCAmelCase  ,						_lowerCAmelCase  ,						_lowerCAmelCase  ,						"""predict"""       )
   UpperCAmelCase__								=			self.compute_metrics(_lowerCAmelCase       )

   # Prefix all keys with metric_key_prefix + '_'
   for key in list(metrics.keys()       ):
    if not key.startswith(f'''{metric_key_prefix}_'''       ):
     UpperCAmelCase__								=			metrics.pop(_lowerCAmelCase       )
   metrics.update(output.metrics       )
   return PredictionOutput(predictions=predictions.predictions  ,						label_ids=predictions.label_ids  ,						metrics=_lowerCAmelCase       )






import os
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


_UpperCAmelCase		:  Any	=   logging.get_logger(__name__)

_UpperCAmelCase		:  List[Any]	=   {"vocab_file": "sentencepiece.bpe.model"}

_UpperCAmelCase		:  List[str]	=   {
    "vocab_file": {
        "camembert-base": "https://huggingface.co/camembert-base/resolve/main/sentencepiece.bpe.model",
    }
}

_UpperCAmelCase		:  Any	=   {
    "camembert-base": 512,
}

_UpperCAmelCase		:  List[Any]	=   "▁"




class       __lowerCAmelCase				(							lowerCAmelCase):
       _a           =  VOCAB_FILES_NAMES
       _a           =  PRETRAINED_VOCAB_FILES_MAP
       _a           =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       _a           =  ['''input_ids''', '''attention_mask''']






       def __init__(      self:    List[str]				,    _lowerCAmelCase:    Optional[int]				,    _lowerCAmelCase:    Dict="<s>"				,    _lowerCAmelCase:    Union[str, Any]="</s>"				,    _lowerCAmelCase:    Optional[int]="</s>"				,    _lowerCAmelCase:    List[Any]="<s>"				,    _lowerCAmelCase:    Tuple="<unk>"				,    _lowerCAmelCase:    Union[str, Any]="<pad>"				,    _lowerCAmelCase:    str="<mask>"				,    _lowerCAmelCase:    int=["<s>NOTUSED", "</s>NOTUSED"]				,    _lowerCAmelCase:    Optional[Dict[str, Any]] = None				,    **_lowerCAmelCase:    Any				,    ):
         # Mask token behave like a normal word, i.e. include the space before it
         lowercase			:Dict							=  AddedToken(_lowerCAmelCase				,    lstrip=_lowerCAmelCase				,    rstrip=_lowerCAmelCase ) if isinstance(_lowerCAmelCase				,    _lowerCAmelCase ) else mask_token

         lowercase			:Optional[int]							=  {} if sp_model_kwargs is None else sp_model_kwargs

         super().__init__(
             bos_token=_lowerCAmelCase				,    eos_token=_lowerCAmelCase				,    unk_token=_lowerCAmelCase				,    sep_token=_lowerCAmelCase				,    cls_token=_lowerCAmelCase				,    pad_token=_lowerCAmelCase				,    mask_token=_lowerCAmelCase				,    additional_special_tokens=_lowerCAmelCase				,    sp_model_kwargs=self.sp_model_kwargs				,    **_lowerCAmelCase				,    )
         lowercase			:Dict							=  spm.SentencePieceProcessor(**self.sp_model_kwargs )
         self.sp_model.Load(str(_lowerCAmelCase ) )
         lowercase			:Tuple							=  vocab_file
         # HACK: These tokens were added by fairseq but don't seem to be actually used when duplicated in the actual
         # sentencepiece vocabulary (this is the case for <s> and </s>
         lowercase			:int							=  {"<s>NOTUSED": 0, "<pad>": 1, "</s>NOTUSED": 2, "<unk>": 3}
         lowercase			:Tuple							=  len(self.fairseq_tokens_to_ids )
         lowercase			:int							=  len(self.sp_model ) + len(self.fairseq_tokens_to_ids )
         lowercase			:Any							=  {v: k for k, v in self.fairseq_tokens_to_ids.items()}






       def     SCREAMING_SNAKE_CASE			(      self:    Tuple				,    _lowerCAmelCase:    List[int]				,    _lowerCAmelCase:    Optional[List[int]] = None ):

         if token_ids_a is None:
           return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
         lowercase			:List[Any]							=  [self.cls_token_id]
         lowercase			:List[str]							=  [self.sep_token_id]
         return cls + token_ids_a + sep + sep + token_ids_a + sep






       def     SCREAMING_SNAKE_CASE			(      self:    Dict				,    _lowerCAmelCase:    List[int]				,    _lowerCAmelCase:    Optional[List[int]] = None				,    _lowerCAmelCase:    bool = False ):
         if already_has_special_tokens:
           return super().get_special_tokens_mask(
               token_ids_a=_lowerCAmelCase				,    token_ids_a=_lowerCAmelCase				,    already_has_special_tokens=_lowerCAmelCase )

         if token_ids_a is None:
           return [1] + ([0] * len(_lowerCAmelCase )) + [1]
         return [1] + ([0] * len(_lowerCAmelCase )) + [1, 1] + ([0] * len(_lowerCAmelCase )) + [1]






       def     SCREAMING_SNAKE_CASE			(      self:    Union[str, Any]				,    _lowerCAmelCase:    List[int]				,    _lowerCAmelCase:    Optional[List[int]] = None ):
         lowercase			:Any							=  [self.sep_token_id]
         lowercase			:str							=  [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]






       @property
       def     SCREAMING_SNAKE_CASE			(      self:    Union[str, Any] ):
         return len(self.fairseq_tokens_to_ids ) + len(self.sp_model )






       def     SCREAMING_SNAKE_CASE			(      self:    Tuple ):
         lowercase			:Any							=  {self.convert_ids_to_tokens(_lowerCAmelCase ): i for i in range(self.vocab_size )}
         vocab.update(self.added_tokens_encoder )
         return vocab






       def     SCREAMING_SNAKE_CASE			(      self:    int				,    _lowerCAmelCase:    str ):
         return self.sp_model.encode(_lowerCAmelCase				,    out_type=_lowerCAmelCase )






       def     SCREAMING_SNAKE_CASE			(      self:    Dict				,    _lowerCAmelCase:    List[str] ):
         if token in self.fairseq_tokens_to_ids:
           return self.fairseq_tokens_to_ids[token]
         elif self.sp_model.PieceToId(_lowerCAmelCase ) == 0:
           # Convert sentence piece unk token to fairseq unk token index
           return self.unk_token_id
         return self.fairseq_offset + self.sp_model.PieceToId(_lowerCAmelCase )






       def     SCREAMING_SNAKE_CASE			(      self:    str				,    _lowerCAmelCase:    Tuple ):
         if index in self.fairseq_ids_to_tokens:
           return self.fairseq_ids_to_tokens[index]
         return self.sp_model.IdToPiece(index - self.fairseq_offset )






       def     SCREAMING_SNAKE_CASE			(      self:    List[Any]				,    _lowerCAmelCase:    List[str] ):
         lowercase			:Tuple							=  []
         lowercase			:Any							=  ""
         lowercase			:str							=  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(_lowerCAmelCase ) + token
             lowercase			:Optional[int]							=  True
             lowercase			:Any							=  []
           else:
             current_sub_tokens.append(_lowerCAmelCase )
             lowercase			:str							=  False
         out_string += self.sp_model.decode(_lowerCAmelCase )
         return out_string.strip()






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






       def __setstate__(      self:    Optional[Any]				,    _lowerCAmelCase:    Dict ):
         lowercase			:int							=  d

         # for backward compatibility
         if not hasattr(self				,    "sp_model_kwargs" ):
           lowercase			:Optional[int]							=  {}

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






       def     SCREAMING_SNAKE_CASE			(      self:    Optional[int]				,    _lowerCAmelCase:    str				,    _lowerCAmelCase:    Optional[str] = None ):
         if not os.path.isdir(_lowerCAmelCase ):
           logger.error(F"Vocabulary path ({save_directory}) should be a directory" )
           return
         lowercase			:Any							=  os.path.join(
             _lowerCAmelCase				,    (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"] )

         if os.path.abspath(self.vocab_file ) != os.path.abspath(_lowerCAmelCase ) and os.path.isfile(self.vocab_file ):
           copyfile(self.vocab_file				,    _lowerCAmelCase )
         elif not os.path.isfile(self.vocab_file ):
           with open(_lowerCAmelCase				,    "wb" ) as fi:
             lowercase			:Dict							=  self.sp_model.serialized_model_proto()
             fi.write(_lowerCAmelCase )

         return (out_vocab_file,)



