

'''simple docstring'''



import argparse
import collections
import json
import os
import re
import string
import sys

import numpy as np


__snake_case			:     Optional[Any]										=      re.compile(r'\b(a|an|the)\b', re.UNICODE)

__snake_case			:     List[str]										=      None
def 						__lowerCamelCase	(       )	->							Dict:






						"""simple docstring"""





						A__					:       Tuple     =argparse.ArgumentParser("""Official evaluation script for SQuAD version 2.0."""	)
						parser.add_argument("""data_file""",		metavar="""data.json""",		help="""Input data JSON file."""	)
						parser.add_argument("""pred_file""",		metavar="""pred.json""",		help="""Model predictions."""	)
						parser.add_argument(
						    """--out-file""",		"""-o""",		metavar="""eval.json""",		help="""Write accuracy metrics to file (default is stdout)."""	)
						parser.add_argument(
						    """--na-prob-file""",		"""-n""",		metavar="""na_prob.json""",		help="""Model estimates of probability of no answer."""	)
						parser.add_argument(
						    """--na-prob-thresh""",		"""-t""",		type=__snake_case,		default=1.0,		help="""Predict \"\" if no-answer probability exceeds this (default = 1.0).""",		)
						parser.add_argument(
						    """--out-image-dir""",		"""-p""",		metavar="""out_images""",		default=__snake_case,		help="""Save precision-recall curves to directory."""	)
						parser.add_argument("""--verbose""",		"""-v""",		action="""store_true"""	)
						if len(sys.argv	) == 1:
												parser.print_help()
												sys.exit(1	)
						return parser.parse_args()
def 						__lowerCamelCase	(       __snake_case					:		int	)	->							Union[str, Any]:






						"""simple docstring"""





						A__					:       int     ={}
						for article in dataset:
												for p in article["paragraphs"]:
																		for qa in p["qas"]:
																								A__					:       List[Any]     =bool(qa["""answers"""]["""text"""]	)
						return qid_to_has_ans
def 						__lowerCamelCase	(       __snake_case					:		Union[str, Any]	)	->							Tuple:






						"""simple docstring"""





						def remove_articles(__snake_case					:		Union[str, Any]	):
												return ARTICLES_REGEX.sub(""" """,		__snake_case	)

						def white_space_fix(__snake_case					:		Optional[Any]	):
												return " ".join(text.split()	)

						def remove_punc(__snake_case					:		Any	):
												A__					:       Optional[Any]     =set(string.punctuation	)
												return "".join(ch for ch in text if ch not in exclude	)

						def lower(__snake_case					:		int	):
												return text.lower()

						return white_space_fix(remove_articles(remove_punc(lower(__snake_case	)	)	)	)
def 						__lowerCamelCase	(       __snake_case					:		str	)	->							Union[str, Any]:






						"""simple docstring"""





						if not s:
												return []
						return normalize_answer(__snake_case	).split()
def 						__lowerCamelCase	(       __snake_case					:		Optional[Any],		__snake_case					:		Union[str, Any]	)	->							str:






						"""simple docstring"""





						return int(normalize_answer(__snake_case	) == normalize_answer(__snake_case	)	)
def 						__lowerCamelCase	(       __snake_case					:		Any,		__snake_case					:		Dict	)	->							Union[str, Any]:






						"""simple docstring"""





						A__					:       Optional[Any]     =get_tokens(__snake_case	)
						A__					:       Union[str, Any]     =get_tokens(__snake_case	)
						A__					:       Optional[Any]     =collections.Counter(__snake_case	) & collections.Counter(__snake_case	)
						A__					:       Dict     =sum(common.values()	)
						if len(__snake_case	) == 0 or len(__snake_case	) == 0:
												# If either is no-answer, then F1 is 1 if they agree, 0 otherwise
												return int(gold_toks == pred_toks	)
						if num_same == 0:
												return 0
						A__					:       Tuple     =1.0 * num_same / len(__snake_case	)
						A__					:       Dict     =1.0 * num_same / len(__snake_case	)
						A__					:       Tuple     =(2 * precision * recall) / (precision + recall)
						return fa
def 						__lowerCamelCase	(       __snake_case					:		int,		__snake_case					:		Optional[Any]	)	->							Any:






						"""simple docstring"""





						A__					:       str     ={}
						A__					:       str     ={}
						for article in dataset:
												for p in article["paragraphs"]:
																		for qa in p["qas"]:
																								A__					:       int     =qa["""id"""]
																								A__					:       str     =[t for t in qa["""answers"""]["""text"""] if normalize_answer(__snake_case	)]
																								if not gold_answers:
																														# For unanswerable questions, only correct answer is empty string
																														A__					:       int     =[""""""]
																								if qid not in preds:
																														print(f"Missing prediction for {qid}"	)
																														continue
																								A__					:       Any     =preds[qid]
																								# Take max over all gold answers
																								A__					:       Union[str, Any]     =max(compute_exact(__snake_case,		__snake_case	) for a in gold_answers	)
																								A__					:       int     =max(compute_fa(__snake_case,		__snake_case	) for a in gold_answers	)
						return exact_scores, fa_scores
def 						__lowerCamelCase	(       __snake_case					:		List[Any],		__snake_case					:		Tuple,		__snake_case					:		int,		__snake_case					:		List[str]	)	->							List[str]:






						"""simple docstring"""





						A__					:       Union[str, Any]     ={}
						for qid, s in scores.items():
												A__					:       Dict     =na_probs[qid] > na_prob_thresh
												if pred_na:
																		A__					:       Union[str, Any]     =float(not qid_to_has_ans[qid]	)
												else:
																		A__					:       Tuple     =s
						return new_scores
def 						__lowerCamelCase	(       __snake_case					:		List[Any],		__snake_case					:		Any,		__snake_case					:		List[Any]=None	)	->							Union[str, Any]:






						"""simple docstring"""





						if not qid_list:
												A__					:       str     =len(__snake_case	)
												return collections.OrderedDict(
												    [
												        ("""exact""", 1_00.0 * sum(exact_scores.values()	) / total),
												        ("""f1""", 1_00.0 * sum(fa_scores.values()	) / total),
												        ("""total""", total),
												    ]	)
						else:
												A__					:       List[str]     =len(__snake_case	)
												return collections.OrderedDict(
												    [
												        ("""exact""", 1_00.0 * sum(exact_scores[k] for k in qid_list	) / total),
												        ("""f1""", 1_00.0 * sum(fa_scores[k] for k in qid_list	) / total),
												        ("""total""", total),
												    ]	)
def 						__lowerCamelCase	(       __snake_case					:		List[Any],		__snake_case					:		Dict,		__snake_case					:		Tuple	)	->							List[Any]:






						"""simple docstring"""





						for k in new_eval:
												A__					:       str     =new_eval[k]
def 						__lowerCamelCase	(       __snake_case					:		int,		__snake_case					:		List[str],		__snake_case					:		Any,		__snake_case					:		Optional[int]	)	->							Optional[int]:






						"""simple docstring"""





						plt.step(__snake_case,		__snake_case,		color="""b""",		alpha=0.2,		where="""post"""	)
						plt.fill_between(__snake_case,		__snake_case,		step="""post""",		alpha=0.2,		color="""b"""	)
						plt.xlabel("""Recall"""	)
						plt.ylabel("""Precision"""	)
						plt.xlim([0.0, 1.05]	)
						plt.ylim([0.0, 1.05]	)
						plt.title(__snake_case	)
						plt.savefig(__snake_case	)
						plt.clf()
def 						__lowerCamelCase	(       __snake_case					:		List[Any],		__snake_case					:		str,		__snake_case					:		Tuple,		__snake_case					:		Dict,		__snake_case					:		Any=None,		__snake_case					:		str=None	)	->							Tuple:






						"""simple docstring"""





						A__					:       Optional[int]     =sorted(__snake_case,		key=lambda __snake_case	: na_probs[k]	)
						A__					:       Any     =0.0
						A__					:       Any     =1.0
						A__					:       str     =0.0
						A__					:       Tuple     =[1.0]
						A__					:       Dict     =[0.0]
						A__					:       Optional[int]     =0.0
						for i, qid in enumerate(__snake_case	):
												if qid_to_has_ans[qid]:
																		true_pos += scores[qid]
												A__					:       List[Any]     =true_pos / float(i + 1	)
												A__					:       Dict     =true_pos / float(__snake_case	)
												if i == len(__snake_case	) - 1 or na_probs[qid] != na_probs[qid_list[i + 1]]:
																		# i.e., if we can put a threshold after this point
																		avg_prec += cur_p * (cur_r - recalls[-1])
																		precisions.append(__snake_case	)
																		recalls.append(__snake_case	)
						if out_image:
												plot_pr_curve(__snake_case,		__snake_case,		__snake_case,		__snake_case	)
						return {"ap": 1_00.0 * avg_prec}
def 						__lowerCamelCase	(       __snake_case					:		Optional[int],		__snake_case					:		List[Any],		__snake_case					:		Union[str, Any],		__snake_case					:		Tuple,		__snake_case					:		List[str],		__snake_case					:		Union[str, Any]	)	->							List[str]:






						"""simple docstring"""





						if out_image_dir and not os.path.exists(__snake_case	):
												os.makedirs(__snake_case	)
						A__					:       str     =sum(1 for v in qid_to_has_ans.values() if v	)
						if num_true_pos == 0:
												return
						A__					:       Dict     =make_precision_recall_eval(
						    __snake_case,		__snake_case,		__snake_case,		__snake_case,		out_image=os.path.join(__snake_case,		"""pr_exact.png"""	),		title="""Precision-Recall curve for Exact Match score""",		)
						A__					:       Union[str, Any]     =make_precision_recall_eval(
						    __snake_case,		__snake_case,		__snake_case,		__snake_case,		out_image=os.path.join(__snake_case,		"""pr_f1.png"""	),		title="""Precision-Recall curve for F1 score""",		)
						A__					:       Union[str, Any]     ={k: float(__snake_case	) for k, v in qid_to_has_ans.items()}
						A__					:       Any     =make_precision_recall_eval(
						    __snake_case,		__snake_case,		__snake_case,		__snake_case,		out_image=os.path.join(__snake_case,		"""pr_oracle.png"""	),		title="""Oracle Precision-Recall curve (binary task of HasAns vs. NoAns)""",		)
						merge_eval(__snake_case,		__snake_case,		"""pr_exact"""	)
						merge_eval(__snake_case,		__snake_case,		"""pr_f1"""	)
						merge_eval(__snake_case,		__snake_case,		"""pr_oracle"""	)
def 						__lowerCamelCase	(       __snake_case					:		Any,		__snake_case					:		str,		__snake_case					:		Dict,		__snake_case					:		Dict	)	->							Tuple:






						"""simple docstring"""





						if not qid_list:
												return
						A__					:       Any     =[na_probs[k] for k in qid_list]
						A__					:       List[Any]     =np.ones_like(__snake_case	) / float(len(__snake_case	)	)
						plt.hist(__snake_case,		weights=__snake_case,		bins=20,		range=(0.0, 1.0)	)
						plt.xlabel("""Model probability of no-answer"""	)
						plt.ylabel("""Proportion of dataset"""	)
						plt.title(f"Histogram of no-answer probability: {name}"	)
						plt.savefig(os.path.join(__snake_case,		f"na_prob_hist_{name}.png"	)	)
						plt.clf()
def 						__lowerCamelCase	(       __snake_case					:		Optional[Any],		__snake_case					:		List[str],		__snake_case					:		Optional[Any],		__snake_case					:		str	)	->							str:






						"""simple docstring"""





						A__					:       List[str]     =sum(1 for k in qid_to_has_ans if not qid_to_has_ans[k]	)
						A__					:       Optional[Any]     =num_no_ans
						A__					:       Dict     =cur_score
						A__					:       Optional[Any]     =0.0
						A__					:       Optional[Any]     =sorted(__snake_case,		key=lambda __snake_case	: na_probs[k]	)
						for i, qid in enumerate(__snake_case	):
												if qid not in scores:
																		continue
												if qid_to_has_ans[qid]:
																		A__					:       Any     =scores[qid]
												else:
																		if preds[qid]:
																								A__					:       int     =-1
																		else:
																								A__					:       Any     =0
												cur_score += diff
												if cur_score > best_score:
																		A__					:       str     =cur_score
																		A__					:       Dict     =na_probs[qid]
						return 1_00.0 * best_score / len(__snake_case	), best_thresh
def 						__lowerCamelCase	(       __snake_case					:		Any,		__snake_case					:		Dict,		__snake_case					:		str,		__snake_case					:		Dict,		__snake_case					:		List[Any],		__snake_case					:		List[Any]	)	->							int:






						"""simple docstring"""





						A__   ,						A__					:       Optional[int]     =find_best_thresh(__snake_case,		__snake_case,		__snake_case,		__snake_case	)
						A__   ,						A__					:       List[Any]     =find_best_thresh(__snake_case,		__snake_case,		__snake_case,		__snake_case	)
						A__					:       int     =best_exact
						A__					:       int     =exact_thresh
						A__					:       List[str]     =best_fa
						A__					:       str     =fa_thresh
def 						__lowerCamelCase	(       )	->							Union[str, Any]:






						"""simple docstring"""





						with open(OPTS.data_file	) as f:
												A__					:       int     =json.load(__snake_case	)
												A__					:       Optional[Any]     =dataset_json["""data"""]
						with open(OPTS.pred_file	) as f:
												A__					:       Dict     =json.load(__snake_case	)
						if OPTS.na_prob_file:
												with open(OPTS.na_prob_file	) as f:
																		A__					:       Optional[int]     =json.load(__snake_case	)
						else:
												A__					:       List[Any]     ={k: 0.0 for k in preds}
						A__					:       int     =make_qid_to_has_ans(__snake_case	)  # maps qid to True/False
						A__					:       Optional[int]     =[k for k, v in qid_to_has_ans.items() if v]
						A__					:       Dict     =[k for k, v in qid_to_has_ans.items() if not v]
						A__   ,						A__					:       str     =get_raw_scores(__snake_case,		__snake_case	)
						A__					:       Any     =apply_no_ans_threshold(__snake_case,		__snake_case,		__snake_case,		OPTS.na_prob_thresh	)
						A__					:       List[Any]     =apply_no_ans_threshold(__snake_case,		__snake_case,		__snake_case,		OPTS.na_prob_thresh	)
						A__					:       Optional[int]     =make_eval_dict(__snake_case,		__snake_case	)
						if has_ans_qids:
												A__					:       Any     =make_eval_dict(__snake_case,		__snake_case,		qid_list=__snake_case	)
												merge_eval(__snake_case,		__snake_case,		"""HasAns"""	)
						if no_ans_qids:
												A__					:       Any     =make_eval_dict(__snake_case,		__snake_case,		qid_list=__snake_case	)
												merge_eval(__snake_case,		__snake_case,		"""NoAns"""	)
						if OPTS.na_prob_file:
												find_all_best_thresh(__snake_case,		__snake_case,		__snake_case,		__snake_case,		__snake_case,		__snake_case	)
						if OPTS.na_prob_file and OPTS.out_image_dir:
												run_precision_recall_analysis(__snake_case,		__snake_case,		__snake_case,		__snake_case,		__snake_case,		OPTS.out_image_dir	)
												histogram_na_prob(__snake_case,		__snake_case,		OPTS.out_image_dir,		"""hasAns"""	)
												histogram_na_prob(__snake_case,		__snake_case,		OPTS.out_image_dir,		"""noAns"""	)
						if OPTS.out_file:
												with open(OPTS.out_file,		"""w"""	) as f:
																		json.dump(__snake_case,		__snake_case	)
						else:
												print(json.dumps(__snake_case,		indent=2	)	)


if __name__ == "__main__":
		__snake_case			:     Tuple										=      parse_args()
		if OPTS.out_image_dir:
				import matplotlib

				matplotlib.use('Agg')
				import matplotlib.pyplot as plt
		main()




'''simple docstring'''



import torch

from diffusers import CMStochasticIterativeScheduler

from .test_schedulers import SchedulerCommonTest



class lowerCamelCase			(  lowercase_						):



			'''simple docstring'''






			__snake_case        =	(CMStochasticIterativeScheduler,)
			__snake_case        =	10


			def lowercase__						(				self	:						List[str]		,      **lowerCAmelCase_	:						Dict			)      ->		Dict:





									'''simple docstring'''
									A__					:       int     ={
									    """num_train_timesteps""": 2_01,
									    """sigma_min""": 0.002,
									    """sigma_max""": 80.0,
									}

									config.update(**lowerCAmelCase_			)
									return config


			def lowercase__						(				self	:						Optional[Any]			)      ->		Optional[int]:





									'''simple docstring'''
									A__					:       Dict     =10

									A__					:       str     =self.get_scheduler_config()
									A__					:       Any     =self.scheduler_classes[0](**lowerCAmelCase_			)

									scheduler.set_timesteps(lowerCAmelCase_			)

									A__					:       List[Any]     =scheduler.timesteps[0]
									A__					:       Union[str, Any]     =scheduler.timesteps[1]

									A__					:       Optional[int]     =self.dummy_sample
									A__					:       Union[str, Any]     =0.1 * sample

									A__					:       Dict     =scheduler.step(lowerCAmelCase_		,      lowerCAmelCase_		,      lowerCAmelCase_			).prev_sample
									A__					:       int     =scheduler.step(lowerCAmelCase_		,      lowerCAmelCase_		,      lowerCAmelCase_			).prev_sample

									self.assertEqual(output_a.shape		,      sample.shape			)
									self.assertEqual(output_a.shape		,      output_a.shape			)


			def lowercase__						(				self	:						Optional[int]			)      ->		Optional[int]:





									'''simple docstring'''
									for timesteps in [10, 50, 1_00, 10_00]:
															self.check_over_configs(num_train_timesteps=lowerCAmelCase_			)


			def lowercase__						(				self	:						Tuple			)      ->		Optional[int]:





									'''simple docstring'''
									for clip_denoised in [True, False]:
															self.check_over_configs(clip_denoised=lowerCAmelCase_			)


			def lowercase__						(				self	:						List[Any]			)      ->		Dict:





									'''simple docstring'''
									A__					:       Union[str, Any]     =self.scheduler_classes[0]
									A__					:       Dict     =self.get_scheduler_config()
									A__					:       Any     =scheduler_class(**lowerCAmelCase_			)

									A__					:       int     =1
									scheduler.set_timesteps(lowerCAmelCase_			)
									A__					:       Any     =scheduler.timesteps

									A__					:       Optional[int]     =torch.manual_seed(0			)

									A__					:       List[Any]     =self.dummy_model()
									A__					:       Optional[int]     =self.dummy_sample_deter * scheduler.init_noise_sigma

									for i, t in enumerate(lowerCAmelCase_			):
															# 1. scale model input
															A__					:       Optional[Any]     =scheduler.scale_model_input(lowerCAmelCase_		,      lowerCAmelCase_			)

															# 2. predict noise residual
															A__					:       Union[str, Any]     =model(lowerCAmelCase_		,      lowerCAmelCase_			)

															# 3. predict previous sample x_t-1
															A__					:       Tuple     =scheduler.step(lowerCAmelCase_		,      lowerCAmelCase_		,      lowerCAmelCase_		,      generator=lowerCAmelCase_			).prev_sample

															A__					:       Dict     =pred_prev_sample

									A__					:       Union[str, Any]     =torch.sum(torch.abs(lowerCAmelCase_			)			)
									A__					:       Optional[Any]     =torch.mean(torch.abs(lowerCAmelCase_			)			)

									assert abs(result_sum.item() - 192.7614			) < 1e-2
									assert abs(result_mean.item() - 0.2510			) < 1e-3


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





									'''simple docstring'''
									A__					:       Dict     =self.scheduler_classes[0]
									A__					:       Dict     =self.get_scheduler_config()
									A__					:       Tuple     =scheduler_class(**lowerCAmelCase_			)

									A__					:       Tuple     =[1_06, 0]
									scheduler.set_timesteps(timesteps=lowerCAmelCase_			)
									A__					:       List[Any]     =scheduler.timesteps

									A__					:       Optional[Any]     =torch.manual_seed(0			)

									A__					:       int     =self.dummy_model()
									A__					:       int     =self.dummy_sample_deter * scheduler.init_noise_sigma

									for t in timesteps:
															# 1. scale model input
															A__					:       Any     =scheduler.scale_model_input(lowerCAmelCase_		,      lowerCAmelCase_			)

															# 2. predict noise residual
															A__					:       List[str]     =model(lowerCAmelCase_		,      lowerCAmelCase_			)

															# 3. predict previous sample x_t-1
															A__					:       Dict     =scheduler.step(lowerCAmelCase_		,      lowerCAmelCase_		,      lowerCAmelCase_		,      generator=lowerCAmelCase_			).prev_sample

															A__					:       Union[str, Any]     =pred_prev_sample

									A__					:       Union[str, Any]     =torch.sum(torch.abs(lowerCAmelCase_			)			)
									A__					:       List[Any]     =torch.mean(torch.abs(lowerCAmelCase_			)			)

									assert abs(result_sum.item() - 347.6357			) < 1e-2
									assert abs(result_mean.item() - 0.4527			) < 1e-3


			def lowercase__						(				self	:						Optional[Any]			)      ->		Dict:





									'''simple docstring'''
									A__					:       Optional[Any]     =self.scheduler_classes[0]
									A__					:       Union[str, Any]     =self.get_scheduler_config()
									A__					:       List[Any]     =scheduler_class(**lowerCAmelCase_			)

									A__					:       Tuple     =[39, 30, 12, 15, 0]

									with self.assertRaises(lowerCAmelCase_		,      msg="""`timesteps` must be in descending order."""			):
															scheduler.set_timesteps(timesteps=lowerCAmelCase_			)


			def lowercase__						(				self	:						Union[str, Any]			)      ->		List[str]:





									'''simple docstring'''
									A__					:       Union[str, Any]     =self.scheduler_classes[0]
									A__					:       List[str]     =self.get_scheduler_config()
									A__					:       Tuple     =scheduler_class(**lowerCAmelCase_			)

									A__					:       Dict     =[39, 30, 12, 1, 0]
									A__					:       int     =len(lowerCAmelCase_			)

									with self.assertRaises(lowerCAmelCase_		,      msg="""Can only pass one of `num_inference_steps` or `timesteps`."""			):
															scheduler.set_timesteps(num_inference_steps=lowerCAmelCase_		,      timesteps=lowerCAmelCase_			)



			def lowercase__						(				self	:						Any			)      ->		Optional[int]:





									'''simple docstring'''
									A__					:       Optional[Any]     =self.scheduler_classes[0]
									A__					:       Any     =self.get_scheduler_config()
									A__					:       Optional[int]     =scheduler_class(**lowerCAmelCase_			)

									A__					:       List[str]     =[scheduler.config.num_train_timesteps]

									with self.assertRaises(
									    lowerCAmelCase_		,      msg="""`timesteps` must start before `self.config.train_timesteps`: {scheduler.config.num_train_timesteps}}"""		,      ):
															scheduler.set_timesteps(timesteps=lowerCAmelCase_			)

