






# DISCLAIMER: This code is strongly influenced by https://github.com/pesser/pytorch_diffusion
# and https://github.com/hojonathanho/diffusion
import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.schedulers.scheduling_utils import SchedulerMixin
from diffusers.utils import BaseOutput, deprecate


@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->DDIM
class    _lowerCAmelCase			(		lowercase_      ):



				"""simple docstring"""



				_lowercase				:   torch.FloatTensor
				_lowercase				:   Optional[torch.FloatTensor]      =  None


def _UpperCAmelCase   (	a						:	Optional[int]  ,					a						:	Optional[int]=0.999  ,					a						:	str="cosine"  ,					):
							if alpha_transform_type == "cosine":

														def alpha_bar_fn(a						:	List[Any] ):
																					return math.cos((t + 0.008) / 1.008 * math.pi / 2 ) ** 2

							elif alpha_transform_type == "exp":

														def alpha_bar_fn(a						:	Union[str, Any] ):
																					return math.exp(t * -12.0 )

							else:
														raise ValueError(F'''Unsupported alpha_tranform_type: {alpha_transform_type}''' )

							snake_case__	  =       []
							for i in range(a ):
														snake_case__	  =       i / num_diffusion_timesteps
														snake_case__	  =       (i + 1) / num_diffusion_timesteps
														betas.append(min(1 - alpha_bar_fn(a ) / alpha_bar_fn(a )  ,					a ) )
							return torch.tensor(a  ,					dtype=torch.floataa )



class    _lowerCAmelCase			(		lowercase_     ,      lowercase_      ):



				"""simple docstring"""



				_lowercase				:   Optional[Any]      =  1



				@register_to_config
				def __init__(      self    :    Optional[int]			,		UpperCamelCase__    :    int = 1_0_0_0			,		UpperCamelCase__    :    float = 0.00_01			,		UpperCamelCase__    :    float = 0.02			,		UpperCamelCase__    :    str = "linear"			,		UpperCamelCase__    :    Optional[Union[np.ndarray, List[float]]] = None			,		UpperCamelCase__    :    bool = True			,		UpperCamelCase__    :    bool = True			,		UpperCamelCase__    :    int = 0			,		UpperCamelCase__    :    str = "epsilon"			,		UpperCamelCase__    :    float = 1.0			,		**UpperCamelCase__    :    Any			,		):






											'''simple docstring'''







											if kwargs.get("""set_alpha_to_one"""			,		UpperCamelCase__) is not None:
																		snake_case__	  =       (
																		    """The `set_alpha_to_one` argument is deprecated. Please use `set_alpha_to_zero` instead."""
																		)
																		deprecate("""set_alpha_to_one"""			,		"""1.0.0"""			,		UpperCamelCase__			,		standard_warn=UpperCamelCase__)
																		snake_case__	  =       kwargs["""set_alpha_to_one"""]
											if trained_betas is not None:
																		snake_case__	  =       torch.tensor(UpperCamelCase__			,		dtype=torch.floataa)
											elif beta_schedule == "linear":
																		snake_case__	  =       torch.linspace(UpperCamelCase__			,		UpperCamelCase__			,		UpperCamelCase__			,		dtype=torch.floataa)
											elif beta_schedule == "scaled_linear":
																		# this schedule is very specific to the latent diffusion model.
																		snake_case__	  =       (
																		    torch.linspace(beta_start**0.5			,		beta_end**0.5			,		UpperCamelCase__			,		dtype=torch.floataa) ** 2
																		)
											elif beta_schedule == "squaredcos_cap_v2":
																		# Glide cosine schedule
																		snake_case__	  =       betas_for_alpha_bar(UpperCamelCase__)
											else:
																		raise NotImplementedError(F'''{beta_schedule} does is not implemented for {self.__class__}''')

											snake_case__	  =       1.0 - self.betas
											snake_case__	  =       torch.cumprod(self.alphas			,		dim=0)

											# At every step in inverted ddim, we are looking into the next alphas_cumprod
											# For the final step, there is no next alphas_cumprod, and the index is out of bounds
											# `set_alpha_to_zero` decides whether we set this parameter simply to zero
											# in this case, self.step() just output the predicted noise
											# or whether we use the final alpha of the "non-previous" one.
											snake_case__	  =       torch.tensor(0.0) if set_alpha_to_zero else self.alphas_cumprod[-1]

											# standard deviation of the initial noise distribution
											snake_case__	  =       1.0

											# setable values
											snake_case__	  =       None
											snake_case__	  =       torch.from_numpy(np.arange(0			,		UpperCamelCase__).copy().astype(np.intaa))



				def 					__magic_name__							(      self    :    Union[str, Any]			,		UpperCamelCase__    :    torch.FloatTensor			,		UpperCamelCase__    :    Optional[int] = None):






											'''simple docstring'''







											return sample



				def 					__magic_name__							(      self    :    Optional[Any]			,		UpperCamelCase__    :    int			,		UpperCamelCase__    :    Union[str, torch.device] = None):






											'''simple docstring'''







											if num_inference_steps > self.config.num_train_timesteps:
																		raise ValueError(
																		    F'''`num_inference_steps`: {num_inference_steps} cannot be larger than `self.config.train_timesteps`:'''
																		    F''' {self.config.num_train_timesteps} as the unet model trained with this scheduler can only handle'''
																		    F''' maximal {self.config.num_train_timesteps} timesteps.''')

											snake_case__	  =       num_inference_steps
											snake_case__	  =       self.config.num_train_timesteps // self.num_inference_steps
											# creates integer timesteps by multiplying by ratio
											# casting to int to avoid issues when num_inference_step is power of 3
											snake_case__	  =       (np.arange(0			,		UpperCamelCase__) * step_ratio).round().copy().astype(np.intaa)
											snake_case__	  =       torch.from_numpy(UpperCamelCase__).to(UpperCamelCase__)
											self.timesteps += self.config.steps_offset



				def 					__magic_name__							(      self    :    List[Any]			,		UpperCamelCase__    :    torch.FloatTensor			,		UpperCamelCase__    :    int			,		UpperCamelCase__    :    torch.FloatTensor			,		UpperCamelCase__    :    float = 0.0			,		UpperCamelCase__    :    bool = False			,		UpperCamelCase__    :    Optional[torch.FloatTensor] = None			,		UpperCamelCase__    :    bool = True			,		):






											'''simple docstring'''







											snake_case__	  =       timestep + self.config.num_train_timesteps // self.num_inference_steps

											# 2. compute alphas, betas
											# change original implementation to exactly match noise levels for analogous forward process
											snake_case__	  =       self.alphas_cumprod[timestep]
											snake_case__	  =       (
											    self.alphas_cumprod[prev_timestep]
											    if prev_timestep < self.config.num_train_timesteps
											    else self.final_alpha_cumprod
											)

											snake_case__	  =       1 - alpha_prod_t

											# 3. compute predicted original sample from predicted noise also called
											# "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
											if self.config.prediction_type == "epsilon":
																		snake_case__	  =       (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
																		snake_case__	  =       model_output
											elif self.config.prediction_type == "sample":
																		snake_case__	  =       model_output
																		snake_case__	  =       (sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
											elif self.config.prediction_type == "v_prediction":
																		snake_case__	  =       (alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
																		snake_case__	  =       (alpha_prod_t**0.5) * model_output + (beta_prod_t**0.5) * sample
											else:
																		raise ValueError(
																		    F'''prediction_type given as {self.config.prediction_type} must be one of `epsilon`, `sample`, or'''
																		    """ `v_prediction`""")

											# 4. Clip or threshold "predicted x_0"
											if self.config.clip_sample:
																		snake_case__	  =       pred_original_sample.clamp(
																		    -self.config.clip_sample_range			,		self.config.clip_sample_range)

											# 5. compute "direction pointing to x_t" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
											snake_case__	  =       (1 - alpha_prod_t_prev) ** 0.5 * pred_epsilon

											# 6. compute x_t without "random noise" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
											snake_case__	  =       alpha_prod_t_prev ** 0.5 * pred_original_sample + pred_sample_direction

											if not return_dict:
																		return (prev_sample, pred_original_sample)
											return DDIMSchedulerOutput(prev_sample=UpperCamelCase__			,		pred_original_sample=UpperCamelCase__)



				def __len__(      self    :    Tuple):






											'''simple docstring'''







											return self.config.num_train_timesteps








import comet  # From: unbabel-comet
import torch

import datasets


a__       =     datasets.logging.get_logger(__name__)

a__       =     """\
@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\",
}
"""

a__       =     """\
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.
"""

a__       =     """
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      ):



				"""simple docstring"""



				def 					__magic_name__							(      self    :    Optional[Any]):






											'''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 					__magic_name__							(      self    :    str			,		UpperCamelCase__    :    Dict):






											'''simple docstring'''







											if self.config_name == "default":
																		snake_case__	  =       comet.load_from_checkpoint(comet.download_model("""wmt20-comet-da"""))
											else:
																		snake_case__	  =       comet.load_from_checkpoint(comet.download_model(self.config_name))



				def 					__magic_name__							(      self    :    List[str]			,		UpperCamelCase__    :    str			,		UpperCamelCase__    :    Tuple			,		UpperCamelCase__    :    List[Any]			,		UpperCamelCase__    :    int=None			,		UpperCamelCase__    :    List[Any]=False):






											'''simple docstring'''







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