


'''simple docstring'''






import faiss  # noqa: F401 # Here to have a nice missing dependency error message early on
import numpy  # noqa: F401 # Here to have a nice missing dependency error message early on
import requests  # noqa: F401 # Here to have a nice missing dependency error message early on
import sklearn  # noqa: F401 # Here to have a nice missing dependency error message early on
import tqdm  # noqa: F401 # Here to have a nice missing dependency error message early on
from mauve import compute_mauve  # From: mauve-text

import datasets


_A			  =  """\
@inproceedings{pillutla-etal:mauve:neurips2021,
  title={MAUVE: Measuring the Gap Between Neural Text and Human Text using Divergence Frontiers},
  author={Pillutla, Krishna and Swayamdipta, Swabha and Zellers, Rowan and Thickstun, John and Welleck, Sean and Choi, Yejin and Harchaoui, Zaid},
  booktitle = {NeurIPS},
  year      = {2021}
}

"""

_A			  =  """\
MAUVE is a library built on PyTorch and HuggingFace Transformers to measure the gap between neural text and human text with the eponymous MAUVE measure.

MAUVE summarizes both Type I and Type II errors measured softly using Kullback–Leibler (KL) divergences.

For details, see the MAUVE paper: https://arxiv.org/abs/2102.01454 (Neurips, 2021).

This metrics is a wrapper around the official implementation of MAUVE:
https://github.com/krishnap25/mauve
"""

_A			  =  """
Calculates MAUVE scores between two lists of generated text and reference text.
Args:
    predictions: list of generated text to score. Each predictions
        should be a string with tokens separated by spaces.
    references: list of reference for each prediction. Each
        reference should be a string with tokens separated by spaces.
Optional Args:
    num_buckets: the size of the histogram to quantize P and Q. Options: \'auto\' (default) or an integer
    pca_max_data: the number data points to use for PCA dimensionality reduction prior to clustering. If -1, use all the data. Default -1
    kmeans_explained_var: amount of variance of the data to keep in dimensionality reduction by PCA. Default 0.9
    kmeans_num_redo: number of times to redo k-means clustering (the best objective is kept). Default 5
    kmeans_max_iter: maximum number of k-means iterations. Default 500
    featurize_model_name: name of the model from which features are obtained. Default \'gpt2-large\' Use one of [\'gpt2\', \'gpt2-medium\', \'gpt2-large\', \'gpt2-xl\'].
    device_id: Device for featurization. Supply a GPU id (e.g. 0 or 3) to use GPU. If no GPU with this id is found, use CPU
    max_text_length: maximum number of tokens to consider. Default 1024
    divergence_curve_discretization_size: Number of points to consider on the divergence curve. Default 25
    mauve_scaling_factor: \"c\" from the paper. Default 5.
    verbose: If True (default), print running time updates
    seed: random seed to initialize k-means cluster assignments.
Returns:
    mauve: MAUVE score, a number between 0 and 1. Larger values indicate that P and Q are closer,
    frontier_integral: Frontier Integral, a number between 0 and 1. Smaller values indicate that P and Q are closer,
    divergence_curve: a numpy.ndarray of shape (m, 2); plot it with matplotlib to view the divergence curve,
    p_hist: a discrete distribution, which is a quantized version of the text distribution p_text,
    q_hist: same as above, but with q_text.
Examples:

    >>> # faiss segfaults in doctest for some reason, so the .compute call is not tested with doctest
    >>> import datasets
    >>> mauve = datasets.load_metric(\'mauve\')
    >>> predictions = [\"hello there\", \"general kenobi\"]
    >>> references = [\"hello there\", \"general kenobi\"]
    >>> out = mauve.compute(predictions=predictions, references=references) # doctest: +SKIP
    >>> print(out.mauve) # doctest: +SKIP
    1.0
"""






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




			def       _snake_case (   self							)      ->    int:


								'''simple docstring'''






								return datasets.MetricInfo(
								    description=_DESCRIPTION       ,   citation=_CITATION       ,   homepage='''https://github.com/krishnap25/mauve'''       ,   inputs_description=_KWARGS_DESCRIPTION       ,   features=datasets.Features(
								        {
								            '''predictions''': datasets.Value('''string'''       ,   id='''sequence'''							),
								            '''references''': datasets.Value('''string'''       ,   id='''sequence'''							),
								        }							)       ,   codebase_urls=['''https://github.com/krishnap25/mauve''']       ,   reference_urls=[
								        '''https://arxiv.org/abs/2102.01454''',
								        '''https://github.com/krishnap25/mauve''',
								    ]       ,   )




			def       _snake_case (   self       ,   lowercase       ,   lowercase       ,   lowercase=None       ,   lowercase=None       ,   lowercase=None       ,   lowercase=None       ,   lowercase="auto"       ,   lowercase=-1       ,   lowercase=0.9       ,   lowercase=5       ,   lowercase=5_0_0       ,   lowercase="gpt2-large"       ,   lowercase=-1       ,   lowercase=1_0_2_4       ,   lowercase=2_5       ,   lowercase=5       ,   lowercase=True       ,   lowercase=2_5       ,   )      ->    Union[str, Any]:


								'''simple docstring'''






								__SCREAMING_SNAKE_CASE       :				Any       =			compute_mauve(
								    p_text=_UpperCAmelCase       ,   q_text=_UpperCAmelCase       ,   p_features=_UpperCAmelCase       ,   q_features=_UpperCAmelCase       ,   p_tokens=_UpperCAmelCase       ,   q_tokens=_UpperCAmelCase       ,   num_buckets=_UpperCAmelCase       ,   pca_max_data=_UpperCAmelCase       ,   kmeans_explained_var=_UpperCAmelCase       ,   kmeans_num_redo=_UpperCAmelCase       ,   kmeans_max_iter=_UpperCAmelCase       ,   featurize_model_name=_UpperCAmelCase       ,   device_id=_UpperCAmelCase       ,   max_text_length=_UpperCAmelCase       ,   divergence_curve_discretization_size=_UpperCAmelCase       ,   mauve_scaling_factor=_UpperCAmelCase       ,   verbose=_UpperCAmelCase       ,   seed=_UpperCAmelCase       ,   )
								return out







'''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


__lowerCAmelCase		     =  get_logger(__name__)

class       __magic_name__		:
							lowerCAmelCase			:       Optional[Any]  							=   'dummy_data'
							lowerCAmelCase			:       List[str]  							=   'datasets'
							lowerCAmelCase			:       str  							=   False







							def __init__(				self   :			Tuple      ,_UpperCAmelCase   :			str      ,_UpperCAmelCase   :			str      ,_UpperCAmelCase   :			Union[Version, str]      ,_UpperCAmelCase   :			Optional[str] = None      ,_UpperCAmelCase   :			bool = False      ,_UpperCAmelCase   :			bool = True      ,_UpperCAmelCase   :			Optional[List[Callable]] = None      ,):
									_a	: int								=				0
									_a	: str								=				dataset_name
									_a	: Optional[int]								=				cache_dir
									_a	: int								=				use_local_dummy_data
									_a	: Optional[int]								=				config
									# download_callbacks take a single url as input
									_a	: 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
									_a	: str								=				load_existing_dummy_data

									# TODO(PVP, QL) might need to make this more general
									_a	: Optional[int]								=				str(_UpperCAmelCase						)
									# to be downloaded
									_a	: Union[str, Any]								=				None
									_a	: Dict								=				None







							@property
							def 				__lowercase     (				self   :			Any						):
									if self._dummy_file is None:
											_a	: Dict								=				self.download_dummy_data()
									return self._dummy_file







							@property
							def 				__lowercase     (				self   :			Optional[int]						):
									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   :			Any						):
									return os.path.join(self.dummy_data_folder      ,'dummy_data.zip'						)







							def 				__lowercase     (				self   :			List[str]						):
									_a	: Dict								=				(
									    self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
									)

									_a	: Dict								=				cached_path(
									    _UpperCAmelCase      ,cache_dir=self.cache_dir      ,extract_compressed_file=_UpperCAmelCase      ,force_extract=_UpperCAmelCase						)

									return os.path.join(_UpperCAmelCase      ,self.dummy_file_name						)







							@property
							def 				__lowercase     (				self   :			List[str]						):
									return os.path.join(self.datasets_scripts_dir      ,self.dataset_name      ,self.dummy_zip_file						)







							@property
							def 				__lowercase     (				self   :			Optional[Any]						):
									if self._bucket_url is None:
											_a	: Any								=				hf_github_url(self.dataset_name      ,self.dummy_zip_file.replace(os.sep      ,'/'						)						)
									return self._bucket_url







							@property
							def 				__lowercase     (				self   :			int						):
									# return full path if its a dir
									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   :			Dict      ,_UpperCAmelCase   :			str      ,*_UpperCAmelCase   :			int						):
									if self.load_existing_dummy_data:
											# dummy data is downloaded and tested
											_a	: List[Any]								=				self.dummy_file
									else:
											# dummy data cannot be downloaded and only the path to dummy file is returned
											_a	: str								=				self.dummy_file_name

									# special case when data_url is a dict
									if isinstance(_UpperCAmelCase      ,_UpperCAmelCase						):
											return self.create_dummy_data_dict(_UpperCAmelCase      ,_UpperCAmelCase						)
									elif isinstance(_UpperCAmelCase      ,(list, tuple)						):
											return self.create_dummy_data_list(_UpperCAmelCase      ,_UpperCAmelCase						)
									else:
											return self.create_dummy_data_single(_UpperCAmelCase      ,_UpperCAmelCase						)







							def 				__lowercase     (				self   :			List[str]      ,_UpperCAmelCase   :			Optional[int]      ,*_UpperCAmelCase   :			Optional[int]						):
									return self.download_and_extract(_UpperCAmelCase						)







							def 				__lowercase     (				self   :			Tuple      ,_UpperCAmelCase   :			List[Any]      ,_UpperCAmelCase   :			Optional[Any]						):
									return self.download_and_extract(_UpperCAmelCase						)







							def 				__lowercase     (				self   :			List[str]      ,_UpperCAmelCase   :			List[str]      ,*_UpperCAmelCase   :			int      ,**_UpperCAmelCase   :			Dict						):
									return path







							def 				__lowercase     (				self   :			Optional[Any]						):
									return {}







							def 				__lowercase     (				self   :			Optional[Any]      ,_UpperCAmelCase   :			List[Any]      ,_UpperCAmelCase   :			str						):
									_a	: int								=				{}
									for key, single_urls in data_url.items():
											for download_callback in self.download_callbacks:
													if isinstance(_UpperCAmelCase      ,_UpperCAmelCase						):
															for single_url in single_urls:
																	download_callback(_UpperCAmelCase						)
													else:
															_a	: Union[str, Any]								=				single_urls
															download_callback(_UpperCAmelCase						)
            # 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(_UpperCAmelCase      ,_UpperCAmelCase						):
													_a	: Optional[int]								=				[os.path.join(_UpperCAmelCase      ,urllib.parse.quote_plus(Path(_UpperCAmelCase						).name						)						) for x in single_urls]
											else:
													_a	: Tuple								=				single_urls
													_a	: Optional[Any]								=				os.path.join(_UpperCAmelCase      ,urllib.parse.quote_plus(Path(_UpperCAmelCase						).name						)						)
											_a	: Union[str, Any]								=				value

									# make sure that values are unique
									if all(isinstance(_UpperCAmelCase      ,_UpperCAmelCase						) 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
											_a	: Dict								=				{key: value + key for key, value in dummy_data_dict.items()}

									return dummy_data_dict







							def 				__lowercase     (				self   :			List[Any]      ,_UpperCAmelCase   :			List[str]      ,_UpperCAmelCase   :			List[str]						):
									_a	: str								=				[]
									# trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
									_a	: Tuple								=				all(bool(re.findall('[0-9]{3,}-of-[0-9]{3,}'      ,_UpperCAmelCase						)						) for url in data_url						)
									_a	: str								=				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):
											_a	: str								=				[data_url[0]] * len(_UpperCAmelCase						)
									for single_url in data_url:
											for download_callback in self.download_callbacks:
													download_callback(_UpperCAmelCase						)
											# 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
											_a	: List[Any]								=				os.path.join(_UpperCAmelCase      ,urllib.parse.quote_plus(single_url.split('/'						)[-1]						)						)
											dummy_data_list.append(_UpperCAmelCase						)
									return dummy_data_list







							def 				__lowercase     (				self   :			List[Any]      ,_UpperCAmelCase   :			Union[str, Any]      ,_UpperCAmelCase   :			List[str]						):
									for download_callback in self.download_callbacks:
											download_callback(_UpperCAmelCase						)
									# 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
									_a	: List[Any]								=				os.path.join(_UpperCAmelCase      ,urllib.parse.quote_plus(data_url.split('/'						)[-1]						)						)
									if os.path.exists(_UpperCAmelCase						) 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   :			Tuple						):
									pass







							def 				__lowercase     (				self   :			Optional[int]						):
									pass







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

									_a	: Dict								=				Path(_UpperCAmelCase						)
									_a	: Dict								=				_iter_archive_members(_UpperCAmelCase						) 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(_UpperCAmelCase						).as_posix(), file_path.open('rb'						)







							def 				__lowercase     (				self   :			Any      ,_UpperCAmelCase   :			Union[str, Any]						):
									if not isinstance(_UpperCAmelCase      ,_UpperCAmelCase						):
											_a	: str								=				[paths]
									for path in paths:
											if os.path.isfile(_UpperCAmelCase						):
													if os.path.basename(_UpperCAmelCase						).startswith(('.', '__')						):
															return
													yield path
											else:
													for dirpath, dirnames, filenames in os.walk(_UpperCAmelCase						):
															if os.path.basename(_UpperCAmelCase						).startswith(('.', '__')						):
																	continue
															dirnames.sort()
															for filename in sorted(_UpperCAmelCase						):
																	if filename.startswith(('.', '__')						):
																			continue
																	yield os.path.join(_UpperCAmelCase      ,_UpperCAmelCase						)




