






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


A:   Union[str, Any]						=					"\\n@inproceedings{Rajpurkar2016SQuAD10,\n  title={SQuAD: 100, 000+ Questions for Machine Comprehension of Text},\n  author={Pranav Rajpurkar and Jian Zhang and Konstantin Lopyrev and Percy Liang},\n  booktitle={EMNLP},\n  year={2016}\n}\n"

A:   Tuple						=					"\nThis metric wrap the official scoring script for version 2 of the Stanford Question\nAnswering Dataset (SQuAD).\n\nStanford Question Answering Dataset (SQuAD) is a reading comprehension dataset, consisting of questions posed by\ncrowdworkers on a set of Wikipedia articles, where the answer to every question is a segment of text, or span,\nfrom the corresponding reading passage, or the question might be unanswerable.\n\nSQuAD2.0 combines the 100,000 questions in SQuAD1.1 with over 50,000 unanswerable questions\nwritten adversarially by crowdworkers to look similar to answerable ones.\nTo do well on SQuAD2.0, systems must not only answer questions when possible, but also\ndetermine when no answer is supported by the paragraph and abstain from answering.\n"

A:   Dict						=					"\nComputes SQuAD v2 scores (F1 and EM).\nArgs:\n    predictions: List of triple for question-answers to score with the following elements:\n        - the question-answer 'id' field as given in the references (see below)\n        - the text of the answer\n        - the probability that the question has no answer\n    references: List of question-answers dictionaries with the following key-values:\n            - 'id': id of the question-answer pair (see above),\n            - 'answers': a list of Dict {'text': text of the answer as a string}\n    no_answer_threshold: float\n        Probability threshold to decide that a question has no answer.\nReturns:\n    'exact': Exact match (the normalized answer exactly match the gold answer)\n    'f1': The F-score of predicted tokens versus the gold answer\n    'total': Number of score considered\n    'HasAns_exact': Exact match (the normalized answer exactly match the gold answer)\n    'HasAns_f1': The F-score of predicted tokens versus the gold answer\n    'HasAns_total': Number of score considered\n    'NoAns_exact': Exact match (the normalized answer exactly match the gold answer)\n    'NoAns_f1': The F-score of predicted tokens versus the gold answer\n    'NoAns_total': Number of score considered\n    'best_exact': Best exact match (with varying threshold)\n    'best_exact_thresh': No-answer probability threshold associated to the best exact match\n    'best_f1': Best F1 (with varying threshold)\n    'best_f1_thresh': No-answer probability threshold associated to the best F1\nExamples:\n\n    >>> predictions = [{'prediction_text': '1976', 'id': '56e10a3be3433e1400422b22', 'no_answer_probability': 0.}]\n    >>> references = [{'answers': {'answer_start': [97], 'text': ['1976']}, 'id': '56e10a3be3433e1400422b22'}]\n    >>> squad_v2_metric = datasets.load_metric(\"squad_v2\")\n    >>> results = squad_v2_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {'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}\n"






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







	"""simple docstring"""



	def 							lowerCamelCase__		(				self							)    ->	Union[str, Any]:
							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 							lowerCamelCase__		(				self	,   _lowercase	,   _lowercase	,   _lowercase=1.0							)    ->	List[Any]:
							lowercase_   : Dict   	=      {p['id']: p['no_answer_probability'] for p in predictions}
							lowercase_   : Optional[Any]   	=      [{'paragraphs': [{'qas': references}]}]
							lowercase_   : Any   	=      {p['id']: p['prediction_text'] for p in predictions}

							lowercase_   : Dict   	=      make_qid_to_has_ans(_lowercase							)  # maps qid to True/False
							lowercase_   : List[str]   	=      [k for k, v in qid_to_has_ans.items() if v]
							lowercase_   : int   	=      [k for k, v in qid_to_has_ans.items() if not v]

							lowercase_	,   lowercase_   : Optional[Any]   	=      get_raw_scores(_lowercase	,   _lowercase							)
							lowercase_   : List[Any]   	=      apply_no_ans_threshold(_lowercase	,   _lowercase	,   _lowercase	,   _lowercase							)
							lowercase_   : Optional[int]   	=      apply_no_ans_threshold(_lowercase	,   _lowercase	,   _lowercase	,   _lowercase							)
							lowercase_   : Tuple   	=      make_eval_dict(_lowercase	,   _lowercase							)

							if has_ans_qids:
													lowercase_   : Optional[int]   	=      make_eval_dict(_lowercase	,   _lowercase	,   qid_list=_lowercase							)
													merge_eval(_lowercase	,   _lowercase	,   'HasAns'							)
							if no_ans_qids:
													lowercase_   : int   	=      make_eval_dict(_lowercase	,   _lowercase	,   qid_list=_lowercase							)
													merge_eval(_lowercase	,   _lowercase	,   'NoAns'							)
							find_all_best_thresh(_lowercase	,   _lowercase	,   _lowercase	,   _lowercase	,   _lowercase	,   _lowercase							)
							return dict(_lowercase							)








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


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






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







	"""simple docstring"""



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






def 						_UpperCAmelCase     (	a		:						Any    ,      a		:						Dict=0.9_99    ,      a		:						Optional[int]="cosine"    ,      )      ->      Tuple:





						"""simple docstring"""
						if alpha_transform_type == "cosine":

												def alpha_bar_fn(a		:						Optional[int]				):
																		return math.cos((t + 0.0_08) / 1.0_08 * math.pi / 2				) ** 2

						elif alpha_transform_type == "exp":

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

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

						lowercase_   : Optional[int]   	=      []
						for i in range(a				):
												lowercase_   : List[Any]   	=      i / num_diffusion_timesteps
												lowercase_   : Union[str, 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 				__magic_name__  (						UpperCAmelCase_,    UpperCAmelCase_					):







	"""simple docstring"""



	SCREAMING_SNAKE_CASE_       :   Union[str, Any]   =	[e.name for e in KarrasDiffusionSchedulers]
	SCREAMING_SNAKE_CASE_       :   Tuple   =	1
	@register_to_config
	def __init__(				self	,   _lowercase = 1000	,   _lowercase = 0.00_01	,   _lowercase = 0.02	,   _lowercase = "linear"	,   _lowercase = None	,   _lowercase = "epsilon"	,   _lowercase = "linspace"	,   _lowercase = 0	,   )    ->	List[str]:
							if trained_betas is not None:
													lowercase_   : Any   	=      torch.tensor(_lowercase	,   dtype=torch.floataa							)
							elif beta_schedule == "linear":
													lowercase_   : Optional[int]   	=      torch.linspace(_lowercase	,   _lowercase	,   _lowercase	,   dtype=torch.floataa							)
							elif beta_schedule == "scaled_linear":
													# this schedule is very specific to the latent diffusion model.
													lowercase_   : Dict   	=      (
													    torch.linspace(beta_start**0.5	,   beta_end**0.5	,   _lowercase	,   dtype=torch.floataa							) ** 2
													)
							elif beta_schedule == "squaredcos_cap_v2":
													# Glide cosine schedule
													lowercase_   : Any   	=      betas_for_alpha_bar(_lowercase							)
							else:
													raise NotImplementedError(f"{beta_schedule} does is not implemented for {self.__class__}"							)

							lowercase_   : Dict   	=      1.0 - self.betas
							lowercase_   : int   	=      torch.cumprod(self.alphas	,   dim=0							)

							lowercase_   : Optional[int]   	=      np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5							)
							lowercase_   : str   	=      np.concatenate([sigmas[::-1], [0.0]]							).astype(np.floataa							)
							lowercase_   : Tuple   	=      torch.from_numpy(_lowercase							)

							# setable values
							lowercase_   : Union[str, Any]   	=      None
							lowercase_   : Optional[Any]   	=      np.linspace(0	,   num_train_timesteps - 1	,   _lowercase	,   dtype=_lowercase							)[::-1].copy()
							lowercase_   : Dict   	=      torch.from_numpy(_lowercase							)
							lowercase_   : int   	=      False
	@property
	def 							lowerCamelCase__		(				self							)    ->	Union[str, Any]:
							# 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	,   _lowercase	,   _lowercase							)    ->	torch.FloatTensor:
							if isinstance(_lowercase	,   torch.Tensor							):
													lowercase_   : List[str]   	=      timestep.to(self.timesteps.device							)
							lowercase_   : List[str]   	=      (self.timesteps == timestep).nonzero().item()
							lowercase_   : Any   	=      self.sigmas[step_index]
							lowercase_   : Tuple   	=      sample / ((sigma**2 + 1) ** 0.5)
							lowercase_   : List[str]   	=      True
							return sample
	def 							lowerCamelCase__		(				self	,   _lowercase	,   _lowercase = None							)    ->	Tuple:
							lowercase_   : Any   	=      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_   : Tuple   	=      np.linspace(0	,   self.config.num_train_timesteps - 1	,   _lowercase	,   dtype=_lowercase							)[
													    ::-1
													].copy()
							elif self.config.timestep_spacing == "leading":
													lowercase_   : int   	=      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_   : Optional[int]   	=      (np.arange(0	,   _lowercase							) * step_ratio).round()[::-1].copy().astype(_lowercase							)
													timesteps += self.config.steps_offset
							elif self.config.timestep_spacing == "trailing":
													lowercase_   : Optional[Any]   	=      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_   : Dict   	=      (np.arange(self.config.num_train_timesteps	,   0	,   -step_ratio							)).round().copy().astype(_lowercase							)
													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_   : str   	=      np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5							)
							lowercase_   : Dict   	=      np.interp(_lowercase	,   np.arange(0	,   len(_lowercase							)							)	,   _lowercase							)
							lowercase_   : Any   	=      np.concatenate([sigmas, [0.0]]							).astype(np.floataa							)
							lowercase_   : Dict   	=      torch.from_numpy(_lowercase							).to(device=_lowercase							)
							if str(_lowercase							).startswith('mps'							):
													# mps does not support float64
													lowercase_   : Union[str, Any]   	=      torch.from_numpy(_lowercase							).to(_lowercase	,   dtype=torch.floataa							)
							else:
													lowercase_   : Optional[int]   	=      torch.from_numpy(_lowercase							).to(device=_lowercase							)
	def 							lowerCamelCase__		(				self	,   _lowercase	,   _lowercase	,   _lowercase	,   _lowercase = None	,   _lowercase = True	,   )    ->	Union[EulerAncestralDiscreteSchedulerOutput, Tuple]:

							if (
							    isinstance(_lowercase	,   _lowercase							)
							    or isinstance(_lowercase	,   torch.IntTensor							)
							    or isinstance(_lowercase	,   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(_lowercase	,   torch.Tensor							):
													lowercase_   : Optional[int]   	=      timestep.to(self.timesteps.device							)

							lowercase_   : List[Any]   	=      (self.timesteps == timestep).nonzero().item()
							lowercase_   : List[str]   	=      self.sigmas[step_index]

							# 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
							if self.config.prediction_type == "epsilon":
													lowercase_   : List[str]   	=      sample - sigma * model_output
							elif self.config.prediction_type == "v_prediction":
													# * c_out + input * c_skip
													lowercase_   : int   	=      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_   : Tuple   	=      self.sigmas[step_index]
							lowercase_   : str   	=      self.sigmas[step_index + 1]
							lowercase_   : Dict   	=      (sigma_to**2 * (sigma_from**2 - sigma_to**2) / sigma_from**2) ** 0.5
							lowercase_   : Optional[Any]   	=      (sigma_to**2 - sigma_up**2) ** 0.5

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

							lowercase_   : Tuple   	=      sigma_down - sigma

							lowercase_   : str   	=      sample + derivative * dt

							lowercase_   : Optional[int]   	=      model_output.device
							lowercase_   : Union[str, Any]   	=      randn_tensor(model_output.shape	,   dtype=model_output.dtype	,   device=_lowercase	,   generator=_lowercase							)

							lowercase_   : int   	=      prev_sample + noise * sigma_up

							if not return_dict:
													return (prev_sample,)

							return EulerAncestralDiscreteSchedulerOutput(
							    prev_sample=_lowercase	,   pred_original_sample=_lowercase							)
	def 							lowerCamelCase__		(				self	,   _lowercase	,   _lowercase	,   _lowercase	,   )    ->	torch.FloatTensor:
							# Make sure sigmas and timesteps have the same device and dtype as original_samples
							lowercase_   : List[str]   	=      self.sigmas.to(device=original_samples.device	,   dtype=original_samples.dtype							)
							if original_samples.device.type == "mps" and torch.is_floating_point(_lowercase							):
													# mps does not support float64
													lowercase_   : Dict   	=      self.timesteps.to(original_samples.device	,   dtype=torch.floataa							)
													lowercase_   : List[str]   	=      timesteps.to(original_samples.device	,   dtype=torch.floataa							)
							else:
													lowercase_   : List[Any]   	=      self.timesteps.to(original_samples.device							)
													lowercase_   : List[Any]   	=      timesteps.to(original_samples.device							)

							lowercase_   : Optional[Any]   	=      [(schedule_timesteps == t).nonzero().item() for t in timesteps]

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

							lowercase_   : str   	=      original_samples + noise * sigma
							return noisy_samples







	def __len__(				self							)    ->	Dict:
							return self.config.num_train_timesteps
