
"""simple docstring"""




# Lint as: python3
import os
import re
import urllib.parse
from pathlib import Path
from typing import Callable, List, Optional, Union
from zipfile import ZipFile

from ..utils.file_utils import cached_path, hf_github_url
from ..utils.logging import get_logger
from ..utils.version import Version


A__   :					str							=   get_logger(__name__)


class 			__magic_name__       :
       UpperCamelCase_   =		'''dummy_data'''
       UpperCamelCase_   =		'''datasets'''
       UpperCamelCase_   =		False
       def __init__(     self ,				A_ ,				A_ ,				A_ ,				A_ = None ,				A_ = False ,				A_ = True ,				A_ = None ,				)     ->	Optional[Any]:





         """simple docstring"""




         _lowercase:			List[str]       =     0
         _lowercase:			int       =     dataset_name
         _lowercase:			Optional[Any]       =     cache_dir
         _lowercase:			Union[str, Any]       =     use_local_dummy_data
         _lowercase:			int       =     config
         # download_callbacks take a single url as input
         _lowercase:			List[Callable]       =     download_callbacks or []
         # if False, it doesn't load existing files and it returns the paths of the dummy files relative
         # to the dummy_data zip file root
         _lowercase:			Optional[Any]       =     load_existing_dummy_data

         # TODO(PVP, QL) might need to make this more general
         _lowercase:			List[str]       =     str(A_       )
         # to be downloaded
         _lowercase:			Tuple       =     None
         _lowercase:			int       =     None
       @property
       def 							lowercase_							(     self       )     ->	Any:





         """simple docstring"""




         if self._dummy_file is None:
           _lowercase:			Optional[Any]       =     self.download_dummy_data()
         return self._dummy_file
       @property
       def 							lowercase_							(     self       )     ->	List[Any]:





         """simple docstring"""




         if self.config is not None:
           # structure is dummy / config_name / version_name
           return os.path.join('''dummy''' ,				self.config.name ,				self.version_name       )
         # structure is dummy / version_name
         return os.path.join('''dummy''' ,				self.version_name       )
       @property
       def 							lowercase_							(     self       )     ->	Optional[int]:





         """simple docstring"""




         return os.path.join(self.dummy_data_folder ,				'''dummy_data.zip'''       )
       def 							lowercase_							(     self       )     ->	Any:





         """simple docstring"""




         _lowercase:			Optional[Any]       =     (
             self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
         )

         _lowercase:			Tuple       =     cached_path(
             A_ ,				cache_dir=self.cache_dir ,				extract_compressed_file=A_ ,				force_extract=A_       )

         return os.path.join(A_ ,				self.dummy_file_name       )
       @property
       def 							lowercase_							(     self       )     ->	Union[str, Any]:





         """simple docstring"""




         return os.path.join(self.datasets_scripts_dir ,				self.dataset_name ,				self.dummy_zip_file       )
       @property
       def 							lowercase_							(     self       )     ->	Optional[int]:





         """simple docstring"""




         if self._bucket_url is None:
           _lowercase:			Any       =     hf_github_url(self.dataset_name ,				self.dummy_zip_file.replace(os.sep ,				'''/'''       )       )
         return self._bucket_url
       @property
       def 							lowercase_							(     self       )     ->	Dict:





         """simple docstring"""




         if os.path.isdir(self.dummy_file       ):
           return self.dummy_file
         # else cut off path to file -> example `xsum`.
         return "/".join(self.dummy_file.replace(os.sep ,				'''/'''       ).split('''/'''       )[:-1]       )
       def 							lowercase_							(     self ,				A_ ,				*A_       )     ->	str:





         """simple docstring"""




         if self.load_existing_dummy_data:
           # dummy data is downloaded and tested
           _lowercase:			str       =     self.dummy_file
         else:
           # dummy data cannot be downloaded and only the path to dummy file is returned
           _lowercase:			Dict       =     self.dummy_file_name

         # special case when data_url is a dict
         if isinstance(A_ ,				A_       ):
           return self.create_dummy_data_dict(A_ ,				A_       )
         elif isinstance(A_ ,				(list, tuple)       ):
           return self.create_dummy_data_list(A_ ,				A_       )
         else:
           return self.create_dummy_data_single(A_ ,				A_       )
       def 							lowercase_							(     self ,				A_ ,				*A_       )     ->	Dict:





         """simple docstring"""




         return self.download_and_extract(A_       )
       def 							lowercase_							(     self ,				A_ ,				A_       )     ->	int:





         """simple docstring"""




         return self.download_and_extract(A_       )
       def 							lowercase_							(     self ,				A_ ,				*A_ ,				**A_       )     ->	List[str]:





         """simple docstring"""




         return path
       def 							lowercase_							(     self       )     ->	str:





         """simple docstring"""




         return {}
       def 							lowercase_							(     self ,				A_ ,				A_       )     ->	str:





         """simple docstring"""




         _lowercase:			List[str]       =     {}
         for key, single_urls in data_url.items():
           for download_callback in self.download_callbacks:
             if isinstance(A_ ,				A_       ):
               for single_url in single_urls:
                 download_callback(A_       )
             else:
               _lowercase:			str       =     single_urls
               download_callback(A_       )
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
           if isinstance(A_ ,				A_       ):
             _lowercase:			str       =     [os.path.join(A_ ,				urllib.parse.quote_plus(Path(A_       ).name       )       ) for x in single_urls]
           else:
             _lowercase:			Optional[int]       =     single_urls
             _lowercase:			Tuple       =     os.path.join(A_ ,				urllib.parse.quote_plus(Path(A_       ).name       )       )
           _lowercase:			Dict       =     value

         # make sure that values are unique
         if all(isinstance(A_ ,				A_       ) for i in dummy_data_dict.values()       ) and len(set(dummy_data_dict.values()       )       ) < len(
             dummy_data_dict.values()       ):
           # append key to value to make its name unique
           _lowercase:			str       =     {key: value + key for key, value in dummy_data_dict.items()}

         return dummy_data_dict
       def 							lowercase_							(     self ,				A_ ,				A_       )     ->	Dict:





         """simple docstring"""




         _lowercase:			Dict       =     []
         # trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
         _lowercase:			Optional[Any]       =     all(bool(re.findall('''[0-9]{3,}-of-[0-9]{3,}''' ,				A_       )       ) for url in data_url       )
         _lowercase:			Tuple       =     all(
             url.startswith('''https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/pubmed'''       ) for url in data_url       )
         if data_url and (is_tf_records or is_pubmed_records):
           _lowercase:			Tuple       =     [data_url[0]] * len(A_       )
         for single_url in data_url:
           for download_callback in self.download_callbacks:
             download_callback(A_       )
           # we force the name of each key to be the last file / folder name of the url path
           # if the url has arguments, we need to encode them with urllib.parse.quote_plus
           _lowercase:			int       =     os.path.join(A_ ,				urllib.parse.quote_plus(single_url.split('''/'''       )[-1]       )       )
           dummy_data_list.append(A_       )
         return dummy_data_list
       def 							lowercase_							(     self ,				A_ ,				A_       )     ->	List[Any]:





         """simple docstring"""




         for download_callback in self.download_callbacks:
           download_callback(A_       )
         # we force the name of each key to be the last file / folder name of the url path
         # if the url has arguments, we need to encode them with urllib.parse.quote_plus
         _lowercase:			Tuple       =     os.path.join(A_ ,				urllib.parse.quote_plus(data_url.split('''/'''       )[-1]       )       )
         if os.path.exists(A_       ) or not self.load_existing_dummy_data:
           return value
         else:
           # Backward compatibility, maybe deprecate at one point.
           # For many datasets with single url calls to dl_manager.download_and_extract,
           # the dummy_data.zip file is actually the zipped downloaded file
           # while now we expected the dummy_data.zip file to be a directory containing
           # the downloaded file.
           return path_to_dummy_data
       def 							lowercase_							(     self       )     ->	Any:





         """simple docstring"""




         pass
       def 							lowercase_							(     self       )     ->	int:





         """simple docstring"""




         pass
       def 							lowercase_							(     self ,				A_       )     ->	Union[str, Any]:





         """simple docstring"""




         def _iter_archive_members(A_       ):
           # this preserves the order of the members inside the ZIP archive
           _lowercase:			str       =     Path(self.dummy_file       ).parent
           _lowercase:			Optional[int]       =     path.relative_to(A_       )
           with ZipFile(self.local_path_to_dummy_data       ) as zip_file:
             _lowercase:			Any       =     zip_file.namelist()
           for member in members:
             if member.startswith(relative_path.as_posix()       ):
               yield dummy_parent_path.joinpath(A_       )

         _lowercase:			Any       =     Path(A_       )
         _lowercase:			List[str]       =     _iter_archive_members(A_       ) if self.use_local_dummy_data else path.rglob('''*'''       )
         for file_path in file_paths:
           if file_path.is_file() and not file_path.name.startswith(('''.''', '''__''')       ):
             yield file_path.relative_to(A_       ).as_posix(), file_path.open('''rb'''       )



       def 							lowercase_							(     self ,				A_       )     ->	Union[str, Any]:





         """simple docstring"""




         if not isinstance(A_ ,				A_       ):
           _lowercase:			Any       =     [paths]
         for path in paths:
           if os.path.isfile(A_       ):
             if os.path.basename(A_       ).startswith(('''.''', '''__''')       ):
               return
             yield path
           else:
             for dirpath, dirnames, filenames in os.walk(A_       ):
               if os.path.basename(A_       ).startswith(('''.''', '''__''')       ):
                 continue
               dirnames.sort()
               for filename in sorted(A_       ):
                 if filename.startswith(('''.''', '''__''')       ):
                   continue
                 yield os.path.join(A_ ,				A_       )


"""simple docstring"""




import comet  # From: unbabel-comet
import torch

import datasets


A__   :					int							=   datasets.logging.get_logger(__name__)

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

A__   :					Union[str, Any]							=   '\\nCrosslingual 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).\nWith 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.\n\nSee the [README.md] file at https://unbabel.github.io/COMET/html/models.html for more information.\n'

A__   :					int							=   '\nCOMET score.\n\nArgs:\n\n`sources` (list of str): Source sentences\n`predictions` (list of str): candidate translations\n`references` (list of str): reference translations\n`cuda` (bool): If set to True, runs COMET using GPU\n`show_progress` (bool): Shows progress\n`model`: COMET model to be used. Will default to `wmt-large-da-estimator-1719` if None.\n\nReturns:\n    `samples`: List of dictionaries with `src`, `mt`, `ref` and `score`.\n    `scores`: List of scores.\n\nExamples:\n\n    >>> comet_metric = datasets.load_metric(\'comet\')\n    >>> # comet_metric = load_metric(\'comet\', \'wmt20-comet-da\')  # you can also choose which model to use\n    >>> source = ["Dem Feuer konnte Einhalt geboten werden", "Schulen und Kindergärten wurden eröffnet."]\n    >>> hypothesis = ["The fire could be stopped", "Schools and kindergartens were open"]\n    >>> reference = ["They were able to control the fire.", "Schools and kindergartens opened"]\n    >>> results = comet_metric.compute(predictions=hypothesis, references=reference, sources=source)\n    >>> print([round(v, 2) for v in results["scores"]])\n    [0.19, 0.92]\n'


@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION   ,					_KWARGS_DESCRIPTION )
class 			__magic_name__       (       datasets.Metric ):
       def 							lowercase_							(     self       )     ->	Tuple:





         """simple docstring"""




         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 							lowercase_							(     self ,				A_       )     ->	Dict:





         """simple docstring"""




         if self.config_name == "default":
           _lowercase:			str       =     comet.load_from_checkpoint(comet.download_model('''wmt20-comet-da'''       )       )
         else:
           _lowercase:			Optional[Any]       =     comet.load_from_checkpoint(comet.download_model(self.config_name       )       )



       def 							lowercase_							(     self ,				A_ ,				A_ ,				A_ ,				A_=None ,				A_=False       )     ->	List[str]:





         """simple docstring"""




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