



import numpy as np

import datasets


A_		:     Tuple        =     '''
Compute the Mahalanobis Distance

Mahalonobis distance is the distance between a point and a distribution.
And not between two distinct points. It is effectively a multivariate equivalent of the Euclidean distance.
It was introduced by Prof. P. C. Mahalanobis in 1936
and has been used in various statistical applications ever since
[source: https://www.machinelearningplus.com/statistics/mahalanobis-distance/]
'''

A_		:     List[Any]        =     '''\
@article{de2000mahalanobis,
  title={The mahalanobis distance},
  author={De Maesschalck, Roy and Jouan-Rimbaud, Delphine and Massart, D{\'e}sir{\'e} L},
  journal={Chemometrics and intelligent laboratory systems},
  volume={50},
  number={1},
  pages={1--18},
  year={2000},
  publisher={Elsevier}
}
'''

A_		:     Optional[Any]        =     '''
Args:
    X: List of datapoints to be compared with the `reference_distribution`.
    reference_distribution: List of datapoints from the reference distribution we want to compare to.
Returns:
    mahalanobis: The Mahalonobis distance for each datapoint in `X`.
Examples:

    >>> mahalanobis_metric = datasets.load_metric("mahalanobis")
    >>> results = mahalanobis_metric.compute(reference_distribution=[[0, 1], [1, 0]], X=[[0, 1]])
    >>> print(results)
    {\'mahalanobis\': array([0.5])}
'''



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

   def 		A							(      self			:       List[Any]							)       ->				List[Any]:


    """simple docstring"""






    return datasets.MetricInfo(
        description=_DESCRIPTION							,   citation=_CITATION							,   inputs_description=_KWARGS_DESCRIPTION							,   features=datasets.Features(
            {
                "X": datasets.Sequence(datasets.Value("float"							,   id="sequence"							)							,   id="X"							),
            }							)							,   )




   def 		A							(      self			:       Any							,   __lowerCAmelCase			:       Dict							,   __lowerCAmelCase			:       Dict							)       ->				Union[str, Any]:


    """simple docstring"""






    a						=		np.array(__lowerCAmelCase							)
    a						=		np.array(__lowerCAmelCase							)

    # Assert that arrays are 2D
    if len(X.shape							) != 2:
     raise ValueError("Expected `X` to be a 2D vector"							)
    if len(reference_distribution.shape							) != 2:
     raise ValueError("Expected `reference_distribution` to be a 2D vector"							)
    if reference_distribution.shape[0] < 2:
     raise ValueError(
         "Expected `reference_distribution` to be a 2D vector with more than one element in the first dimension"							)

    # Get mahalanobis distance for each prediction
    a						=		X - np.mean(__lowerCAmelCase							)
    a						=		np.cov(reference_distribution.T							)
    try:
     a						=		np.linalg.inv(__lowerCAmelCase							)
    except np.linalg.LinAlgError:
     a						=		np.linalg.pinv(__lowerCAmelCase							)
    a						=		np.dot(__lowerCAmelCase							,   __lowerCAmelCase							)
    a						=		np.dot(__lowerCAmelCase							,   X_minus_mu.T							).diagonal()

    return {"mahalanobis": mahal_dist}






from typing import List, Optional, Tuple, Union

import torch

from ...models import UNetaDModel
from ...schedulers import ScoreSdeVeScheduler
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput



class 		_lowercase		(       UpperCAmelCase__	):
   _UpperCAmelCase     	=			42
   _UpperCAmelCase     	=			42

   def __init__(      self			:       Optional[Any]							,   __lowerCAmelCase			:       UNetaDModel							,   __lowerCAmelCase			:       ScoreSdeVeScheduler							)       ->				str:


    """simple docstring"""






    super().__init__()
    self.register_modules(unet=__lowerCAmelCase							,   scheduler=__lowerCAmelCase							)




   @torch.no_grad()
   def __call__(      self			:       int							,   __lowerCAmelCase			:       int = 1							,   __lowerCAmelCase			:       int = 2000							,   __lowerCAmelCase			:       Optional[Union[torch.Generator, List[torch.Generator]]] = None							,   __lowerCAmelCase			:       Optional[str] = "pil"							,   __lowerCAmelCase			:       bool = True							,   **__lowerCAmelCase			:       Any							,   )       ->				Union[ImagePipelineOutput, Tuple]:


    """simple docstring"""






    a						=		self.unet.config.sample_size
    a						=		(batch_size, 3, img_size, img_size)

    a						=		self.unet

    a						=		randn_tensor(__lowerCAmelCase							,   generator=__lowerCAmelCase							) * self.scheduler.init_noise_sigma
    a						=		sample.to(self.device							)

    self.scheduler.set_timesteps(__lowerCAmelCase							)
    self.scheduler.set_sigmas(__lowerCAmelCase							)

    for i, t in enumerate(self.progress_bar(self.scheduler.timesteps							)							):
     a						=		self.scheduler.sigmas[i] * torch.ones(shape[0]							,   device=self.device							)

     # correction step
     for _ in range(self.scheduler.config.correct_steps							):
      a						=		self.unet(__lowerCAmelCase							,   __lowerCAmelCase							).sample
      a						=		self.scheduler.step_correct(__lowerCAmelCase							,   __lowerCAmelCase							,   generator=__lowerCAmelCase							).prev_sample

     # prediction step
     a						=		model(__lowerCAmelCase							,   __lowerCAmelCase							).sample
     a						=		self.scheduler.step_pred(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   generator=__lowerCAmelCase							)

     a		,  a						=		output.prev_sample, output.prev_sample_mean

    a						=		sample_mean.clamp(0							,   1							)
    a						=		sample.cpu().permute(0							,   2							,   3							,   1							).numpy()
    if output_type == "pil":
     a						=		self.numpy_to_pil(__lowerCAmelCase							)

    if not return_dict:
     return (sample,)

    return ImagePipelineOutput(images=__lowerCAmelCase							)

