


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

import sentencepiece as spm

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


UpperCAmelCase__   	=							logging.get_logger(__name__)

UpperCAmelCase__   	=							'''▁'''

UpperCAmelCase__   	=							{'''vocab_file''': '''sentencepiece.bpe.model'''}

UpperCAmelCase__   	=							{
    '''vocab_file''': {
        '''facebook/xglm-564M''': '''https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model''',
    }
}

UpperCAmelCase__   	=							{
    '''facebook/xglm-564M''': 2048,
}







class       lowerCamelCase__   (       lowerCAmelCase):
     SCREAMING_SNAKE_CASE__      			=			VOCAB_FILES_NAMES
     SCREAMING_SNAKE_CASE__      			=			PRETRAINED_VOCAB_FILES_MAP
     SCREAMING_SNAKE_CASE__      			=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     SCREAMING_SNAKE_CASE__      			=			['''input_ids''', '''attention_mask''']





     def __init__(self		,     UpperCAmelCase		,     UpperCAmelCase="<s>"		,     UpperCAmelCase="</s>"		,     UpperCAmelCase="</s>"		,     UpperCAmelCase="<s>"		,     UpperCAmelCase="<unk>"		,     UpperCAmelCase="<pad>"		,     UpperCAmelCase = None		,     **UpperCAmelCase		,     )     ->   None:
      _lowercase							      ={} if sp_model_kwargs is None else sp_model_kwargs

      # Compatibility with the original tokenizer
      _lowercase							      =7
      _lowercase							      =[f"<madeupword{i}>" for i in range(self.num_madeup_words       )]

      _lowercase							      =kwargs.get('''additional_special_tokens'''		,     []       )
      kwargs["additional_special_tokens"] += [
          word for word in madeup_words if word not in kwargs["additional_special_tokens"]
      ]

      super().__init__(
          bos_token=UpperCAmelCase		,     eos_token=UpperCAmelCase		,     unk_token=UpperCAmelCase		,     sep_token=UpperCAmelCase		,     cls_token=UpperCAmelCase		,     pad_token=UpperCAmelCase		,     sp_model_kwargs=self.sp_model_kwargs		,     **UpperCAmelCase		,     )

      _lowercase							      =spm.SentencePieceProcessor(**self.sp_model_kwargs       )
      self.sp_model.Load(str(UpperCAmelCase       )       )
      _lowercase							      =vocab_file

      # Original fairseq vocab and spm vocab must be "aligned":
      # Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
      # -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
      # fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
      # spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

      # The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
      _lowercase							      =1

      # Mimic fairseq token-to-id alignment for the first 4 token
      _lowercase							      ={'''<s>''': 0, '''<pad>''': 1, '''</s>''': 2, '''<unk>''': 3}

      _lowercase							      =len(self.sp_model       )
      _lowercase							      ={f"<madeupword{i}>": sp_size + i + self.fairseq_offset for i in range(self.num_madeup_words       )}
      self.fairseq_tokens_to_ids.update(UpperCAmelCase       )

      _lowercase							      ={v: k for k, v in self.fairseq_tokens_to_ids.items()}





     def __getstate__(self       )     ->   List[str]:
      _lowercase							      =self.__dict__.copy()
      _lowercase							      =None
      _lowercase							      =self.sp_model.serialized_model_proto()
      return state





     def __setstate__(self		,     UpperCAmelCase       )     ->   Any:
      _lowercase							      =d

      # for backward compatibility
      if not hasattr(self		,     '''sp_model_kwargs'''       ):
       _lowercase							      ={}

      _lowercase							      =spm.SentencePieceProcessor(**self.sp_model_kwargs       )
      self.sp_model.LoadFromSerializedProto(self.sp_model_proto       )





     def      __A					(self		,     UpperCAmelCase		,     UpperCAmelCase = None       )     ->   List[int]:

      if token_ids_a is None:
       return [self.sep_token_id] + token_ids_a
      _lowercase							      =[self.sep_token_id]
      return sep + token_ids_a + sep + sep + token_ids_a





     def      __A					(self		,     UpperCAmelCase		,     UpperCAmelCase = None		,     UpperCAmelCase = False       )     ->   List[int]:

      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 None:
       return [1] + ([0] * len(UpperCAmelCase       ))
      return [1] + ([0] * len(UpperCAmelCase       )) + [1, 1] + ([0] * len(UpperCAmelCase       ))





     def      __A					(self		,     UpperCAmelCase		,     UpperCAmelCase = None       )     ->   List[int]:
      _lowercase							      =[self.sep_token_id]

      if token_ids_a is None:
       return len(sep + token_ids_a       ) * [0]
      return len(sep + token_ids_a + sep + sep + token_ids_a       ) * [0]





     @property
     def      __A					(self       )     ->   int:
      return len(self.sp_model       ) + self.fairseq_offset + self.num_madeup_words





     def      __A					(self       )     ->   Union[str, Any]:
      _lowercase							      ={self.convert_ids_to_tokens(UpperCAmelCase       ): i for i in range(self.vocab_size       )}
      vocab.update(self.added_tokens_encoder       )
      return vocab





     def      __A					(self		,     UpperCAmelCase       )     ->   List[str]:
      return self.sp_model.encode(UpperCAmelCase		,     out_type=UpperCAmelCase       )





     def      __A					(self		,     UpperCAmelCase       )     ->   List[str]:
      if token in self.fairseq_tokens_to_ids:
       return self.fairseq_tokens_to_ids[token]
      _lowercase							      =self.sp_model.PieceToId(UpperCAmelCase       )

      # Need to return unknown token if the SP model returned 0
      return spm_id + self.fairseq_offset if spm_id else self.unk_token_id





     def      __A					(self		,     UpperCAmelCase       )     ->   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      __A					(self		,     UpperCAmelCase       )     ->   Dict:
      _lowercase							      =''''''.join(UpperCAmelCase       ).replace(UpperCAmelCase		,     ''' '''       ).strip()
      return out_string





     def      __A					(self		,     UpperCAmelCase		,     UpperCAmelCase = None       )     ->   Tuple[str]:
      if not os.path.isdir(UpperCAmelCase       ):
       logger.error(f"Vocabulary path ({save_directory}) should be a directory"       )
       return
      _lowercase							      =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:
        _lowercase							      =self.sp_model.serialized_model_proto()
        fi.write(UpperCAmelCase       )

      return (out_vocab_file,)





import comet  # From: unbabel-comet
import torch

import datasets


UpperCAmelCase__   	=							datasets.logging.get_logger(__name__)

UpperCAmelCase__   	=							'''\
@inproceedings{rei-EtAl:2020:WMT,
   author    = {Rei, Ricardo  and  Stewart, Craig  and  Farinha, Ana C  and  Lavie, Alon},
   title     = {Unbabel\'s Participation in the WMT20 Metrics Shared Task},
   booktitle      = {Proceedings of the Fifth Conference on Machine Translation},
   month          = {November},
   year           = {2020},
   address        = {Online},
   publisher      = {Association for Computational Linguistics},
   pages     = {909--918},
}
@inproceedings{rei-etal-2020-comet,
   title = "{COMET}: A Neural Framework for {MT} Evaluation",
   author = "Rei, Ricardo  and
      Stewart, Craig  and
      Farinha, Ana C  and
      Lavie, Alon",
   booktitle = "Proceedings of the 2020 Conference on Empirical Methods in Natural Language Processing (EMNLP)",
   month = nov,
   year = "2020",
   address = "Online",
   publisher = "Association for Computational Linguistics",
   url = "https://www.aclweb.org/anthology/2020.emnlp-main.213",
   pages = "2685--2702",
}
'''

UpperCAmelCase__   	=							'''\
Crosslingual Optimized Metric for Evaluation of Translation (COMET) is an open-source framework used to train Machine Translation metrics that achieve high levels of correlation with different types of human judgments (HTER, DA\'s or MQM).
With the release of the framework the authors also released fully trained models that were used to compete in the WMT20 Metrics Shared Task achieving SOTA in that years competition.

See the [README.md] file at https://unbabel.github.io/COMET/html/models.html for more information.
'''

UpperCAmelCase__   	=							'''
COMET score.

Args:

`sources` (list of str): Source sentences
`predictions` (list of str): candidate translations
`references` (list of str): reference translations
`cuda` (bool): If set to True, runs COMET using GPU
`show_progress` (bool): Shows progress
`model`: COMET model to be used. Will default to `wmt-large-da-estimator-1719` if None.

Returns:
    `samples`: List of dictionaries with `src`, `mt`, `ref` and `score`.
    `scores`: List of scores.

Examples:

    >>> comet_metric = datasets.load_metric(\'comet\')
    >>> # comet_metric = load_metric(\'comet\', \'wmt20-comet-da\')  # you can also choose which model to use
    >>> source = ["Dem Feuer konnte Einhalt geboten werden", "Schulen und Kindergärten wurden eröffnet."]
    >>> hypothesis = ["The fire could be stopped", "Schools and kindergartens were open"]
    >>> reference = ["They were able to control the fire.", "Schools and kindergartens opened"]
    >>> results = comet_metric.compute(predictions=hypothesis, references=reference, sources=source)
    >>> print([round(v, 2) for v in results["scores"]])
    [0.19, 0.92]
'''







@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION   ,							_KWARGS_DESCRIPTION)
class       lowerCamelCase__   (       datasets.Metric):





     def      __A					(self       )     ->   Optional[int]:
      return datasets.MetricInfo(
          description=_DESCRIPTION		,     citation=_CITATION		,     homepage='''https://unbabel.github.io/COMET/html/index.html'''		,     inputs_description=_KWARGS_DESCRIPTION		,     features=datasets.Features(
              {
                  '''sources''': datasets.Value('''string'''		,     id='''sequence'''       ),
                  '''predictions''': datasets.Value('''string'''		,     id='''sequence'''       ),
                  '''references''': datasets.Value('''string'''		,     id='''sequence'''       ),
              }       )		,     codebase_urls=['''https://github.com/Unbabel/COMET''']		,     reference_urls=[
              '''https://github.com/Unbabel/COMET''',
              '''https://www.aclweb.org/anthology/2020.emnlp-main.213/''',
              '''http://www.statmt.org/wmt20/pdf/2020.wmt-1.101.pdf6''',
          ]		,     )





     def      __A					(self		,     UpperCAmelCase       )     ->   Dict:
      if self.config_name == "default":
       _lowercase							      =comet.load_from_checkpoint(comet.download_model('''wmt20-comet-da'''       )       )
      else:
       _lowercase							      =comet.load_from_checkpoint(comet.download_model(self.config_name       )       )





     def      __A					(self		,     UpperCAmelCase		,     UpperCAmelCase		,     UpperCAmelCase		,     UpperCAmelCase=None		,     UpperCAmelCase=False       )     ->   int:
      if gpus is None:
       _lowercase							      =1 if torch.cuda.is_available() else 0
      _lowercase							      ={'''src''': sources, '''mt''': predictions, '''ref''': references}
      _lowercase							      =[dict(zip(UpperCAmelCase		,     UpperCAmelCase       )       ) for t in zip(*data.values()       )]
      _lowercase      ,    _lowercase							      =self.scorer.predict(UpperCAmelCase		,     gpus=UpperCAmelCase		,     progress_bar=UpperCAmelCase       )
      return {"mean_score": mean_score, "scores": scores}

