





'''simple docstring'''






from sklearn.metrics import mean_squared_error

import datasets


lowerCAmelCase									=   """\
@article{scikit-learn,
 title={Scikit-learn: Machine Learning in {P}ython},
 author={Pedregosa, F. and Varoquaux, G. and Gramfort, A. and Michel, V.
         and Thirion, B. and Grisel, O. and Blondel, M. and Prettenhofer, P.
         and Weiss, R. and Dubourg, V. and Vanderplas, J. and Passos, A. and
         Cournapeau, D. and Brucher, M. and Perrot, M. and Duchesnay, E.},
 journal={Journal of Machine Learning Research},
 volume={12},
 pages={2825--2830},
 year={2011}
}
"""

lowerCAmelCase									=   """\
Mean Squared Error(MSE) is the average of the square of difference between the predicted
and actual values.
"""


lowerCAmelCase									=   """
Args:
    predictions: array-like of shape (n_samples,) or (n_samples, n_outputs)
        Estimated target values.
    references: array-like of shape (n_samples,) or (n_samples, n_outputs)
        Ground truth (correct) target values.
    sample_weight: array-like of shape (n_samples,), default=None
        Sample weights.
    multioutput: {\"raw_values\", \"uniform_average\"} or array-like of shape (n_outputs,), default=\"uniform_average\"
        Defines aggregating of multiple output values. Array-like value defines weights used to average errors.

        \"raw_values\" : Returns a full set of errors in case of multioutput input.

        \"uniform_average\" : Errors of all outputs are averaged with uniform weight.

    squared : bool, default=True
        If True returns MSE value, if False returns RMSE (Root Mean Squared Error) value.

Returns:
    mse : mean squared error.
Examples:

    >>> mse_metric = datasets.load_metric(\"mse\")
    >>> predictions = [2.5, 0.0, 2, 8]
    >>> references = [3, -0.5, 2, 7]
    >>> results = mse_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'mse': 0.375}
    >>> rmse_result = mse_metric.compute(predictions=predictions, references=references, squared=False)
    >>> print(rmse_result)
    {'mse': 0.6123724356957945}

    If you're using multi-dimensional lists, then set the config as follows :

    >>> mse_metric = datasets.load_metric(\"mse\", \"multilist\")
    >>> predictions = [[0.5, 1], [-1, 1], [7, -6]]
    >>> references = [[0, 2], [-1, 2], [8, -5]]
    >>> results = mse_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'mse': 0.7083333333333334}
    >>> results = mse_metric.compute(predictions=predictions, references=references, multioutput='raw_values')
    >>> print(results) # doctest: +NORMALIZE_WHITESPACE
    {'mse': array([0.41666667, 1. ])}
"""
@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION   , _KWARGS_DESCRIPTION							)
class 		lowerCamelCase	(   datasets.Metric							):



	def 	_lowerCamelCase     ( self						):
			return datasets.MetricInfo(
			    description=_DESCRIPTION		,   citation=_CITATION		,   inputs_description=_KWARGS_DESCRIPTION		,   features=datasets.Features(self._get_feature_types()						)		,   reference_urls=[
			        "https://scikit-learn.org/stable/modules/generated/sklearn.metrics.mean_squared_error.html"
			    ]		,   )



	def 	_lowerCamelCase     ( self						):
			if self.config_name == "multilist":
					return {
					    "predictions": datasets.Sequence(datasets.Value("float"						)						),
					    "references": datasets.Sequence(datasets.Value("float"						)						),
					}
			else:
					return {
					    "predictions": datasets.Value("float"						),
					    "references": datasets.Value("float"						),
					}





	def 	_lowerCamelCase     ( self		,   a_		,   a_		,   a_=None		,   a_="uniform_average"		,   a_=True						):
			lowerCAmelCase  :     str       						= mean_squared_error(
			    a_		,   a_		,   sample_weight=a_		,   multioutput=a_		,   squared=a_						)

			return {"mse": mse}










'''simple docstring'''






import math
from collections import defaultdict
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin, SchedulerOutput






def   __A    (				a_    :	int   ,a_    :	str=0.9_9_9   ,a_    :	List[str]="cosine"   ,):
		if alpha_transform_type == "cosine":

				def alpha_bar_fn(a_    :	List[Any]						):
						return math.cos((t + 0.0_0_8) / 1.0_0_8 * math.pi / 2						) ** 2

		elif alpha_transform_type == "exp":

				def alpha_bar_fn(a_    :	List[str]						):
						return math.exp(t * -1_2.0						)

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

		lowerCAmelCase  :     Optional[int]       						= []
		for i in range(a_						):
				lowerCAmelCase  :     Dict       						= i / num_diffusion_timesteps
				lowerCAmelCase  :     List[Any]       						= (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	(   _A   , _A							):
	snake_case_   = [e.name for e in KarrasDiffusionSchedulers]
	snake_case_   = 2



	@register_to_config
	def __init__( self		,   a_ = 1_000		,   a_ = 0.00085		,   a_ = 0.012		,   a_ = "linear"		,   a_ = None		,   a_ = "epsilon"		,   a_ = False		,   a_ = False		,   a_ = 1.0		,   a_ = "linspace"		,   a_ = 0		,   ):
			if trained_betas is not None:
					lowerCAmelCase  :     List[Any]       						= torch.tensor(a_		,   dtype=torch.floataa						)
			elif beta_schedule == "linear":
					lowerCAmelCase  :     List[str]       						= torch.linspace(a_		,   a_		,   a_		,   dtype=torch.floataa						)
			elif beta_schedule == "scaled_linear":
					# this schedule is very specific to the latent diffusion model.
					lowerCAmelCase  :     List[str]       						= (
					    torch.linspace(beta_start**0.5		,   beta_end**0.5		,   a_		,   dtype=torch.floataa						) ** 2
					)
			elif beta_schedule == "squaredcos_cap_v2":
					# Glide cosine schedule
					lowerCAmelCase  :     List[Any]       						= betas_for_alpha_bar(a_		,   alpha_transform_type="cosine"						)
			elif beta_schedule == "exp":
					lowerCAmelCase  :     Any       						= betas_for_alpha_bar(a_		,   alpha_transform_type="exp"						)
			else:
					raise NotImplementedError(F'''{beta_schedule} does is not implemented for {self.__class__}'''						)

			lowerCAmelCase  :     Optional[int]       						= 1.0 - self.betas
			lowerCAmelCase  :     Optional[Any]       						= torch.cumprod(self.alphas		,   dim=0						)

			#  set all values
			self.set_timesteps(a_		,   a_		,   a_						)
			lowerCAmelCase  :     Union[str, Any]       						= use_karras_sigmas



	def 	_lowerCamelCase     ( self		,   a_		,   a_=None						):
			if schedule_timesteps is None:
					lowerCAmelCase  :     List[str]       						= self.timesteps

			lowerCAmelCase  :     int       						= (schedule_timesteps == timestep).nonzero()

			# The sigma index that is taken for the **very** first `step`
			# is always the second index (or the last index if there is only 1)
			# This way we can ensure we don't accidentally skip a sigma in
			# case we start in the middle of the denoising schedule (e.g. for image-to-image)
			if len(self._index_counter						) == 0:
					lowerCAmelCase  :     Union[str, Any]       						= 1 if len(a_						) > 1 else 0
			else:
					lowerCAmelCase  :     Tuple       						= timestep.cpu().item() if torch.is_tensor(a_						) else timestep
					lowerCAmelCase  :     Any       						= self._index_counter[timestep_int]

			return indices[pos].item()



	@property
	def 	_lowerCamelCase     ( self						):
			# standard deviation of the initial noise distribution
			if self.config.timestep_spacing in ["linspace", "trailing"]:
					return self.sigmas.max()

			return (self.sigmas.max() ** 2 + 1) ** 0.5



	def 	_lowerCamelCase     ( self		,   a_		,   a_		,   ):
			lowerCAmelCase  :     Union[str, Any]       						= self.index_for_timestep(a_						)

			lowerCAmelCase  :     Any       						= self.sigmas[step_index]
			lowerCAmelCase  :     int       						= sample / ((sigma**2 + 1) ** 0.5)
			return sample



	def 	_lowerCamelCase     ( self		,   a_		,   a_ = None		,   a_ = None		,   ):
			lowerCAmelCase  :     List[Any]       						= num_inference_steps

			lowerCAmelCase  :     Optional[int]       						= num_train_timesteps or self.config.num_train_timesteps

			# "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
			if self.config.timestep_spacing == "linspace":
					lowerCAmelCase  :     Optional[Any]       						= np.linspace(0		,   num_train_timesteps - 1		,   a_		,   dtype=a_						)[::-1].copy()
			elif self.config.timestep_spacing == "leading":
					lowerCAmelCase  :     str       						= 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
					lowerCAmelCase  :     Dict       						= (np.arange(0		,   a_						) * step_ratio).round()[::-1].copy().astype(a_						)
					timesteps += self.config.steps_offset
			elif self.config.timestep_spacing == "trailing":
					lowerCAmelCase  :     Tuple       						= 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
					lowerCAmelCase  :     int       						= (np.arange(a_		,   0		,   -step_ratio						)).round().copy().astype(a_						)
					timesteps -= 1
			else:
					raise ValueError(
					    F'''{self.config.timestep_spacing} is not supported. Please make sure to choose one of \'linspace\', \'leading\' or \'trailing\'.'''						)

			lowerCAmelCase  :     Any       						= np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5						)
			lowerCAmelCase  :     Any       						= np.log(a_						)
			lowerCAmelCase  :     int       						= np.interp(a_		,   np.arange(0		,   len(a_						)						)		,   a_						)

			if self.config.use_karras_sigmas:
					lowerCAmelCase  :     int       						= self._convert_to_karras(in_sigmas=a_		,   num_inference_steps=self.num_inference_steps						)
					lowerCAmelCase  :     Any       						= np.array([self._sigma_to_t(a_		,   a_						) for sigma in sigmas]						)

			lowerCAmelCase  :     int       						= np.concatenate([sigmas, [0.0]]						).astype(np.floataa						)
			lowerCAmelCase  :     int       						= torch.from_numpy(a_						).to(device=a_						)
			lowerCAmelCase  :     Any       						= torch.cat([sigmas[:1], sigmas[1:-1].repeat_interleave(2						), sigmas[-1:]]						)

			lowerCAmelCase  :     Union[str, Any]       						= torch.from_numpy(a_						)
			lowerCAmelCase  :     Union[str, Any]       						= torch.cat([timesteps[:1], timesteps[1:].repeat_interleave(2						)]						)

			if str(a_						).startswith("mps"						):
					# mps does not support float64
					lowerCAmelCase  :     str       						= timesteps.to(a_		,   dtype=torch.floataa						)
			else:
					lowerCAmelCase  :     int       						= timesteps.to(device=a_						)

			# empty dt and derivative
			lowerCAmelCase  :     Union[str, Any]       						= None
			lowerCAmelCase  :     Optional[Any]       						= None

			# for exp beta schedules, such as the one for `pipeline_shap_e.py`
			# we need an index counter
			lowerCAmelCase  :     int       						= defaultdict(a_						)



	def 	_lowerCamelCase     ( self		,   a_		,   a_						):
			# get log sigma
			lowerCAmelCase  :     str       						= np.log(a_						)

			# get distribution
			lowerCAmelCase  :     int       						= log_sigma - log_sigmas[:, np.newaxis]

			# get sigmas range
			lowerCAmelCase  :     int       						= np.cumsum((dists >= 0)		,   axis=0						).argmax(axis=0						).clip(max=log_sigmas.shape[0] - 2						)
			lowerCAmelCase  :     Any       						= low_idx + 1

			lowerCAmelCase  :     str       						= log_sigmas[low_idx]
			lowerCAmelCase  :     Optional[Any]       						= log_sigmas[high_idx]

			# interpolate sigmas
			lowerCAmelCase  :     Optional[Any]       						= (low - log_sigma) / (low - high)
			lowerCAmelCase  :     List[str]       						= np.clip(a_		,   0		,   1						)

			# transform interpolation to time range
			lowerCAmelCase  :     int       						= (1 - w) * low_idx + w * high_idx
			lowerCAmelCase  :     Dict       						= t.reshape(sigma.shape						)
			return t



	def 	_lowerCamelCase     ( self		,   a_		,   a_						):
			lowerCAmelCase  :     float       						= in_sigmas[-1].item()
			lowerCAmelCase  :     float       						= in_sigmas[0].item()

			lowerCAmelCase  :     str       						= 7.0  # 7.0 is the value used in the paper
			lowerCAmelCase  :     Optional[int]       						= np.linspace(0		,   1		,   a_						)
			lowerCAmelCase  :     Optional[int]       						= sigma_min ** (1 / rho)
			lowerCAmelCase  :     str       						= sigma_max ** (1 / rho)
			lowerCAmelCase  :     List[str]       						= (max_inv_rho + ramp * (min_inv_rho - max_inv_rho)) ** rho
			return sigmas



	@property
	def 	_lowerCamelCase     ( self						):
			return self.dt is None



	def 	_lowerCamelCase     ( self		,   a_		,   a_		,   a_		,   a_ = True		,   ):
			lowerCAmelCase  :     Optional[int]       						= self.index_for_timestep(a_						)

			# advance index counter by 1
			lowerCAmelCase  :     List[str]       						= timestep.cpu().item() if torch.is_tensor(a_						) else timestep
			self._index_counter[timestep_int] += 1

			if self.state_in_first_order:
					lowerCAmelCase  :     Optional[Any]       						= self.sigmas[step_index]
					lowerCAmelCase  :     Dict       						= self.sigmas[step_index + 1]
			else:
					# 2nd order / Heun's method
					lowerCAmelCase  :     List[str]       						= self.sigmas[step_index - 1]
					lowerCAmelCase  :     int       						= self.sigmas[step_index]

			# currently only gamma=0 is supported. This usually works best anyways.
			# We can support gamma in the future but then need to scale the timestep before
			# passing it to the model which requires a change in API
			lowerCAmelCase  :     int       						= 0
			lowerCAmelCase  :     Optional[int]       						= sigma * (gamma + 1)  # Note: sigma_hat == sigma for now

			# 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
			if self.config.prediction_type == "epsilon":
					lowerCAmelCase  :     int       						= sigma_hat if self.state_in_first_order else sigma_next
					lowerCAmelCase  :     Tuple       						= sample - sigma_input * model_output
			elif self.config.prediction_type == "v_prediction":
					lowerCAmelCase  :     List[str]       						= sigma_hat if self.state_in_first_order else sigma_next
					lowerCAmelCase  :     int       						= model_output * (-sigma_input / (sigma_input**2 + 1) ** 0.5) + (
					    sample / (sigma_input**2 + 1)
					)
			elif self.config.prediction_type == "sample":
					lowerCAmelCase  :     Optional[int]       						= model_output
			else:
					raise ValueError(
					    F'''prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`'''						)

			if self.config.clip_sample:
					lowerCAmelCase  :     Optional[int]       						= pred_original_sample.clamp(
					    -self.config.clip_sample_range		,   self.config.clip_sample_range						)

			if self.state_in_first_order:
					# 2. Convert to an ODE derivative for 1st order
					lowerCAmelCase  :     Union[str, Any]       						= (sample - pred_original_sample) / sigma_hat
					# 3. delta timestep
					lowerCAmelCase  :     Optional[int]       						= sigma_next - sigma_hat

					# store for 2nd order step
					lowerCAmelCase  :     List[str]       						= derivative
					lowerCAmelCase  :     Optional[int]       						= dt
					lowerCAmelCase  :     Dict       						= sample
			else:
					# 2. 2nd order / Heun's method
					lowerCAmelCase  :     Dict       						= (sample - pred_original_sample) / sigma_next
					lowerCAmelCase  :     Optional[int]       						= (self.prev_derivative + derivative) / 2

					# 3. take prev timestep & sample
					lowerCAmelCase  :     Optional[int]       						= self.dt
					lowerCAmelCase  :     Tuple       						= self.sample

					# free dt and derivative
					# Note, this puts the scheduler in "first order mode"
					lowerCAmelCase  :     Tuple       						= None
					lowerCAmelCase  :     Optional[Any]       						= None
					lowerCAmelCase  :     Union[str, Any]       						= None

			lowerCAmelCase  :     Any       						= sample + derivative * dt

			if not return_dict:
					return (prev_sample,)

			return SchedulerOutput(prev_sample=a_						)



	def 	_lowerCamelCase     ( self		,   a_		,   a_		,   a_		,   ):
			# Make sure sigmas and timesteps have the same device and dtype as original_samples
			lowerCAmelCase  :     Tuple       						= self.sigmas.to(device=original_samples.device		,   dtype=original_samples.dtype						)
			if original_samples.device.type == "mps" and torch.is_floating_point(a_						):
					# mps does not support float64
					lowerCAmelCase  :     Optional[int]       						= self.timesteps.to(original_samples.device		,   dtype=torch.floataa						)
					lowerCAmelCase  :     Dict       						= timesteps.to(original_samples.device		,   dtype=torch.floataa						)
			else:
					lowerCAmelCase  :     List[Any]       						= self.timesteps.to(original_samples.device						)
					lowerCAmelCase  :     str       						= timesteps.to(original_samples.device						)

			lowerCAmelCase  :     Tuple       						= [self.index_for_timestep(a_		,   a_						) for t in timesteps]

			lowerCAmelCase  :     Dict       						= sigmas[step_indices].flatten()
			while len(sigma.shape						) < len(original_samples.shape						):
					lowerCAmelCase  :     Any       						= sigma.unsqueeze(-1						)

			lowerCAmelCase  :     str       						= original_samples + noise * sigma
			return noisy_samples





	def __len__( self						):
			return self.config.num_train_timesteps



