
"""simple docstring"""






import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, logging, randn_tensor
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin


_lowercase							 =							logging.get_logger(__name__)  # pylint: disable=invalid-name

@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->EulerAncestralDiscrete
class 				__a    (   __a					):



							'''simple docstring'''







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





def     lowerCAmelCase__ (		__magic_name__ ,      __magic_name__=0.999 ,      __magic_name__="cosine" ,      )	->Tuple:
			if alpha_transform_type == "cosine":

						def alpha_bar_fn(__magic_name__			):
									return math.cos((t + 0.008) / 1.008 * math.pi / 2			) ** 2

			elif alpha_transform_type == "exp":

						def alpha_bar_fn(__magic_name__			):
									return math.exp(t * -12.0			)

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

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







class 				__a    (   __a				,      __a					):



							'''simple docstring'''







							_lowerCamelCase				: Any					      =      [e.name for e in KarrasDiffusionSchedulers]
							_lowerCamelCase				: Tuple					      =      1
							@register_to_config
							def __init__(    self     ,						_lowerCamelCase = 1_000     ,						_lowerCamelCase = 0.0001     ,						_lowerCamelCase = 0.02     ,						_lowerCamelCase = "linear"     ,						_lowerCamelCase = None     ,						_lowerCamelCase = "epsilon"     ,						_lowerCamelCase = "linspace"     ,						_lowerCamelCase = 0     ,						) ->       Tuple:




										'''simple docstring'''
										if trained_betas is not None:
													__lowercase						      =					torch.tensor(_lowerCamelCase     ,						dtype=torch.floataa					)
										elif beta_schedule == "linear":
													__lowercase						      =					torch.linspace(_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						dtype=torch.floataa					)
										elif beta_schedule == "scaled_linear":
													# this schedule is very specific to the latent diffusion model.
													__lowercase						      =					(
													    torch.linspace(beta_start**0.5     ,						beta_end**0.5     ,						_lowerCamelCase     ,						dtype=torch.floataa					) ** 2
													)
										elif beta_schedule == "squaredcos_cap_v2":
													# Glide cosine schedule
													__lowercase						      =					betas_for_alpha_bar(_lowerCamelCase					)
										else:
													raise NotImplementedError(f'''{beta_schedule} does is not implemented for {self.__class__}'''					)

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

										__lowercase						      =					np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5					)
										__lowercase						      =					np.concatenate([sigmas[::-1], [0.0]]					).astype(np.floataa					)
										__lowercase						      =					torch.from_numpy(_lowerCamelCase					)

										# setable values
										__lowercase						      =					None
										__lowercase						      =					np.linspace(0     ,						num_train_timesteps - 1     ,						_lowerCamelCase     ,						dtype=_lowerCamelCase					)[::-1].copy()
										__lowercase						      =					torch.from_numpy(_lowerCamelCase					)
										__lowercase						      =					False
							@property
							def      SCREAMING_SNAKE_CASE	(    self					) ->       int:




										'''simple docstring'''
										# 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      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase					) ->       torch.FloatTensor:




										'''simple docstring'''
										if isinstance(_lowerCamelCase     ,						torch.Tensor					):
													__lowercase						      =					timestep.to(self.timesteps.device					)
										__lowercase						      =					(self.timesteps == timestep).nonzero().item()
										__lowercase						      =					self.sigmas[step_index]
										__lowercase						      =					sample / ((sigma**2 + 1) ** 0.5)
										__lowercase						      =					True
										return sample
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase = None					) ->       Optional[int]:




										'''simple docstring'''
										__lowercase						      =					num_inference_steps

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

										__lowercase						      =					np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5					)
										__lowercase						      =					np.interp(_lowerCamelCase     ,						np.arange(0     ,						len(_lowerCamelCase					)					)     ,						_lowerCamelCase					)
										__lowercase						      =					np.concatenate([sigmas, [0.0]]					).astype(np.floataa					)
										__lowercase						      =					torch.from_numpy(_lowerCamelCase					).to(device=_lowerCamelCase					)
										if str(_lowerCamelCase					).startswith("mps"					):
													# mps does not support float64
													__lowercase						      =					torch.from_numpy(_lowerCamelCase					).to(_lowerCamelCase     ,						dtype=torch.floataa					)
										else:
													__lowercase						      =					torch.from_numpy(_lowerCamelCase					).to(device=_lowerCamelCase					)
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase = None     ,						_lowerCamelCase = True     ,						) ->       Union[EulerAncestralDiscreteSchedulerOutput, Tuple]:




										'''simple docstring'''

										if (
										    isinstance(_lowerCamelCase     ,						_lowerCamelCase					)
										    or isinstance(_lowerCamelCase     ,						torch.IntTensor					)
										    or isinstance(_lowerCamelCase     ,						torch.LongTensor					)
										):
													raise ValueError(
													    (
													        "Passing integer indices (e.g. from `enumerate(timesteps)`) as timesteps to"
													        " `EulerDiscreteScheduler.step()` is not supported. Make sure to pass"
													        " one of the `scheduler.timesteps` as a timestep."
													    )     ,						)

										if not self.is_scale_input_called:
													logger.warning(
													    "The `scale_model_input` function should be called before `step` to ensure correct denoising. "
													    "See `StableDiffusionPipeline` for a usage example."					)

										if isinstance(_lowerCamelCase     ,						torch.Tensor					):
													__lowercase						      =					timestep.to(self.timesteps.device					)

										__lowercase						      =					(self.timesteps == timestep).nonzero().item()
										__lowercase						      =					self.sigmas[step_index]

										# 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
										if self.config.prediction_type == "epsilon":
													__lowercase						      =					sample - sigma * model_output
										elif self.config.prediction_type == "v_prediction":
													# * c_out + input * c_skip
													__lowercase						      =					model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
										elif self.config.prediction_type == "sample":
													raise NotImplementedError("prediction_type not implemented yet: sample"					)
										else:
													raise ValueError(
													    f'''prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`'''					)

										__lowercase						      =					self.sigmas[step_index]
										__lowercase						      =					self.sigmas[step_index + 1]
										__lowercase						      =					(sigma_to**2 * (sigma_from**2 - sigma_to**2) / sigma_from**2) ** 0.5
										__lowercase						      =					(sigma_to**2 - sigma_up**2) ** 0.5

										# 2. Convert to an ODE derivative
										__lowercase						      =					(sample - pred_original_sample) / sigma

										__lowercase						      =					sigma_down - sigma

										__lowercase						      =					sample + derivative * dt

										__lowercase						      =					model_output.device
										__lowercase						      =					randn_tensor(model_output.shape     ,						dtype=model_output.dtype     ,						device=_lowerCamelCase     ,						generator=_lowerCamelCase					)

										__lowercase						      =					prev_sample + noise * sigma_up

										if not return_dict:
													return (prev_sample,)

										return EulerAncestralDiscreteSchedulerOutput(
										    prev_sample=_lowerCamelCase     ,						pred_original_sample=_lowerCamelCase					)
							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						) ->       torch.FloatTensor:




										'''simple docstring'''
										# Make sure sigmas and timesteps have the same device and dtype as original_samples
										__lowercase						      =					self.sigmas.to(device=original_samples.device     ,						dtype=original_samples.dtype					)
										if original_samples.device.type == "mps" and torch.is_floating_point(_lowerCamelCase					):
													# mps does not support float64
													__lowercase						      =					self.timesteps.to(original_samples.device     ,						dtype=torch.floataa					)
													__lowercase						      =					timesteps.to(original_samples.device     ,						dtype=torch.floataa					)
										else:
													__lowercase						      =					self.timesteps.to(original_samples.device					)
													__lowercase						      =					timesteps.to(original_samples.device					)

										__lowercase						      =					[(schedule_timesteps == t).nonzero().item() for t in timesteps]

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

										__lowercase						      =					original_samples + noise * sigma
										return noisy_samples





							def __len__(    self					) ->       Dict:




										'''simple docstring'''
										return self.config.num_train_timesteps







"""simple docstring"""






import datasets

from .evaluate import (
    apply_no_ans_threshold,
    find_all_best_thresh,
    get_raw_scores,
    make_eval_dict,
    make_qid_to_has_ans,
    merge_eval,
)


_lowercase							 =							'''\
@inproceedings{Rajpurkar2016SQuAD10,
  title={SQuAD: 100, 000+ Questions for Machine Comprehension of Text},
  author={Pranav Rajpurkar and Jian Zhang and Konstantin Lopyrev and Percy Liang},
  booktitle={EMNLP},
  year={2016}
}
'''

_lowercase							 =							'''
This metric wrap the official scoring script for version 2 of the Stanford Question
Answering Dataset (SQuAD).

Stanford Question Answering Dataset (SQuAD) is a reading comprehension dataset, consisting of questions posed by
crowdworkers on a set of Wikipedia articles, where the answer to every question is a segment of text, or span,
from the corresponding reading passage, or the question might be unanswerable.

SQuAD2.0 combines the 100,000 questions in SQuAD1.1 with over 50,000 unanswerable questions
written adversarially by crowdworkers to look similar to answerable ones.
To do well on SQuAD2.0, systems must not only answer questions when possible, but also
determine when no answer is supported by the paragraph and abstain from answering.
'''

_lowercase							 =							'''
Computes SQuAD v2 scores (F1 and EM).
Args:
    predictions: List of triple for question-answers to score with the following elements:
        - the question-answer \'id\' field as given in the references (see below)
        - the text of the answer
        - the probability that the question has no answer
    references: List of question-answers dictionaries with the following key-values:
            - \'id\': id of the question-answer pair (see above),
            - \'answers\': a list of Dict {\'text\': text of the answer as a string}
    no_answer_threshold: float
        Probability threshold to decide that a question has no answer.
Returns:
    \'exact\': Exact match (the normalized answer exactly match the gold answer)
    \'f1\': The F-score of predicted tokens versus the gold answer
    \'total\': Number of score considered
    \'HasAns_exact\': Exact match (the normalized answer exactly match the gold answer)
    \'HasAns_f1\': The F-score of predicted tokens versus the gold answer
    \'HasAns_total\': Number of score considered
    \'NoAns_exact\': Exact match (the normalized answer exactly match the gold answer)
    \'NoAns_f1\': The F-score of predicted tokens versus the gold answer
    \'NoAns_total\': Number of score considered
    \'best_exact\': Best exact match (with varying threshold)
    \'best_exact_thresh\': No-answer probability threshold associated to the best exact match
    \'best_f1\': Best F1 (with varying threshold)
    \'best_f1_thresh\': No-answer probability threshold associated to the best F1
Examples:

    >>> predictions = [{\'prediction_text\': \'1976\', \'id\': \'56e10a3be3433e1400422b22\', \'no_answer_probability\': 0.}]
    >>> references = [{\'answers\': {\'answer_start\': [97], \'text\': [\'1976\']}, \'id\': \'56e10a3be3433e1400422b22\'}]
    >>> squad_v2_metric = datasets.load_metric("squad_v2")
    >>> results = squad_v2_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {\'exact\': 100.0, \'f1\': 100.0, \'total\': 1, \'HasAns_exact\': 100.0, \'HasAns_f1\': 100.0, \'HasAns_total\': 1, \'best_exact\': 100.0, \'best_exact_thresh\': 0.0, \'best_f1\': 100.0, \'best_f1_thresh\': 0.0}
'''

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



							'''simple docstring'''







							def      SCREAMING_SNAKE_CASE	(    self					) ->       Tuple:




										'''simple docstring'''
										return datasets.MetricInfo(
										    description=_DESCRIPTION     ,						citation=_CITATION     ,						inputs_description=_KWARGS_DESCRIPTION     ,						features=datasets.Features(
										        {
										            "predictions": {
										                "id": datasets.Value("string"					),
										                "prediction_text": datasets.Value("string"					),
										                "no_answer_probability": datasets.Value("float32"					),
										            },
										            "references": {
										                "id": datasets.Value("string"					),
										                "answers": datasets.features.Sequence(
										                    {"text": datasets.Value("string"					), "answer_start": datasets.Value("int32"					)}					),
										            },
										        }					)     ,						codebase_urls=["https://rajpurkar.github.io/SQuAD-explorer/"]     ,						reference_urls=["https://rajpurkar.github.io/SQuAD-explorer/"]     ,						)





							def      SCREAMING_SNAKE_CASE	(    self     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase=1.0					) ->       Tuple:




										'''simple docstring'''
										__lowercase						      =					{p["id"]: p["no_answer_probability"] for p in predictions}
										__lowercase						      =					[{"paragraphs": [{"qas": references}]}]
										__lowercase						      =					{p["id"]: p["prediction_text"] for p in predictions}

										__lowercase						      =					make_qid_to_has_ans(_lowerCamelCase					)  # maps qid to True/False
										__lowercase						      =					[k for k, v in qid_to_has_ans.items() if v]
										__lowercase						      =					[k for k, v in qid_to_has_ans.items() if not v]

										__lowercase      ,				__lowercase						      =					get_raw_scores(_lowerCamelCase     ,						_lowerCamelCase					)
										__lowercase						      =					apply_no_ans_threshold(_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					)
										__lowercase						      =					apply_no_ans_threshold(_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					)
										__lowercase						      =					make_eval_dict(_lowerCamelCase     ,						_lowerCamelCase					)

										if has_ans_qids:
													__lowercase						      =					make_eval_dict(_lowerCamelCase     ,						_lowerCamelCase     ,						qid_list=_lowerCamelCase					)
													merge_eval(_lowerCamelCase     ,						_lowerCamelCase     ,						"HasAns"					)
										if no_ans_qids:
													__lowercase						      =					make_eval_dict(_lowerCamelCase     ,						_lowerCamelCase     ,						qid_list=_lowerCamelCase					)
													merge_eval(_lowerCamelCase     ,						_lowerCamelCase     ,						"NoAns"					)
										find_all_best_thresh(_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase     ,						_lowerCamelCase					)
										return dict(_lowerCamelCase					)





