



"""simple docstring"""
import argparse
import collections
import json
import os
import re
import string
import sys

import numpy as np


SCREAMING_SNAKE_CASE_				= re.compile(r"""\b(a|an|the)\b""", re.UNICODE)

SCREAMING_SNAKE_CASE_				= None



def        lowerCAmelCase_						(							)			->      Dict:
		a_     :      int 						=     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=lowerCAmelCase_, 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=lowerCAmelCase_, 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_						(							SCREAMING_SNAKE_CASE__	)			->      List[str]:
		a_     :      List[str] 						=     {}
		for article in dataset:
				for p in article["paragraphs"]:
						for qa in p["qas"]:
								a_     :      Union[str, Any] 						=     bool(qa["answers"]["text"]	)
		return qid_to_has_ans



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__	)			->      Tuple:



		def remove_articles(SCREAMING_SNAKE_CASE__	):
				return ARTICLES_REGEX.sub(" ", lowerCAmelCase_	)

		def white_space_fix(SCREAMING_SNAKE_CASE__	):
				return " ".join(text.split()	)

		def remove_punc(SCREAMING_SNAKE_CASE__	):
				a_     :      List[str] 						=     set(string.punctuation	)
				return "".join(ch for ch in text if ch not in exclude	)

		def lower(SCREAMING_SNAKE_CASE__	):
				return text.lower()

		return white_space_fix(remove_articles(remove_punc(lower(lowerCAmelCase_	)	)	)	)



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__	)			->      Any:



		if not s:
				return []
		return normalize_answer(lowerCAmelCase_	).split()



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)			->      str:
		return int(normalize_answer(lowerCAmelCase_	) == normalize_answer(lowerCAmelCase_	)	)



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)			->      Dict:
		a_     :      Tuple 						=     get_tokens(lowerCAmelCase_	)
		a_     :      int 						=     get_tokens(lowerCAmelCase_	)
		a_     :      str 						=     collections.Counter(lowerCAmelCase_	) & collections.Counter(lowerCAmelCase_	)
		a_     :      List[Any] 						=     sum(common.values()	)
		if len(lowerCAmelCase_	) == 0 or len(lowerCAmelCase_	) == 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_     :      Union[str, Any] 						=     1.0 * num_same / len(lowerCAmelCase_	)
		a_     :      Any 						=     1.0 * num_same / len(lowerCAmelCase_	)
		a_     :      List[Any] 						=     (2 * precision * recall) / (precision + recall)
		return fa



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)			->      str:
		a_     :      int 						=     {}
		a_     :      Optional[int] 						=     {}
		for article in dataset:
				for p in article["paragraphs"]:
						for qa in p["qas"]:
								a_     :      str 						=     qa['''id''']
								a_     :      List[Any] 						=     [t for t in qa['''answers''']['''text'''] if normalize_answer(lowerCAmelCase_	)]
								if not gold_answers:
										# For unanswerable questions, only correct answer is empty string
										a_     :      Any 						=     ['''''']
								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(lowerCAmelCase_, lowerCAmelCase_	) for a in gold_answers	)
								a_     :      Dict 						=     max(compute_fa(lowerCAmelCase_, lowerCAmelCase_	) for a in gold_answers	)
		return exact_scores, fa_scores



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)			->      str:
		a_     :      int 						=     {}
		for qid, s in scores.items():
				a_     :      Any 						=     na_probs[qid] > na_prob_thresh
				if pred_na:
						a_     :      Optional[Any] 						=     float(not qid_to_has_ans[qid]	)
				else:
						a_     :      int 						=     s
		return new_scores



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__=None	)			->      Tuple:



		if not qid_list:
				a_     :      List[Any] 						=     len(lowerCAmelCase_	)
				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_     :      Tuple 						=     len(lowerCAmelCase_	)
				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_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)			->      str:



		for k in new_eval:
				a_     :      Dict 						=     new_eval[k]



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)			->      Union[str, Any]:
		plt.step(lowerCAmelCase_, lowerCAmelCase_, color="b", alpha=0.2, where="post"	)
		plt.fill_between(lowerCAmelCase_, lowerCAmelCase_, 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(lowerCAmelCase_	)
		plt.savefig(lowerCAmelCase_	)
		plt.clf()



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=None	)			->      Union[str, Any]:
		a_     :      Union[str, Any] 						=     sorted(lowerCAmelCase_, key=lambda SCREAMING_SNAKE_CASE__	: na_probs[k]	)
		a_     :      Any 						=     0.0
		a_     :      Optional[Any] 						=     1.0
		a_     :      str 						=     0.0
		a_     :      List[str] 						=     [1.0]
		a_     :      str 						=     [0.0]
		a_     :      Optional[Any] 						=     0.0
		for i, qid in enumerate(lowerCAmelCase_	):
				if qid_to_has_ans[qid]:
						true_pos += scores[qid]
				a_     :      str 						=     true_pos / float(i + 1	)
				a_     :      List[str] 						=     true_pos / float(lowerCAmelCase_	)
				if i == len(lowerCAmelCase_	) - 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(lowerCAmelCase_	)
						recalls.append(lowerCAmelCase_	)
		if out_image:
				plot_pr_curve(lowerCAmelCase_, lowerCAmelCase_, lowerCAmelCase_, lowerCAmelCase_	)
		return {"ap": 1_00.0 * avg_prec}



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)			->      List[Any]:



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



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)			->      Optional[Any]:



		if not qid_list:
				return
		a_     :      int 						=     [na_probs[k] for k in qid_list]
		a_     :      List[Any] 						=     np.ones_like(lowerCAmelCase_	) / float(len(lowerCAmelCase_	)	)
		plt.hist(lowerCAmelCase_, weights=lowerCAmelCase_, 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(lowerCAmelCase_, F"""na_prob_hist_{name}.png"""	)	)
		plt.clf()



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



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)			->      Union[str, Any]:
		a_     :      Dict 						=     find_best_thresh(lowerCAmelCase_, lowerCAmelCase_, lowerCAmelCase_, lowerCAmelCase_	)
		a_     :      str 						=     find_best_thresh(lowerCAmelCase_, lowerCAmelCase_, lowerCAmelCase_, lowerCAmelCase_	)
		a_     :      str 						=     best_exact
		a_     :      Any 						=     exact_thresh
		a_     :      Union[str, Any] 						=     best_fa
		a_     :      List[str] 						=     fa_thresh



def        lowerCAmelCase_						(							)			->      Optional[Any]:



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


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

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




"""simple docstring"""
import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()




class 			snake_case_						(						a_     ,unittest.TestCase				):
							__lowerCAmelCase    					=					KandinskyImgaImgPipeline
							__lowerCAmelCase    					=					["prompt", "image_embeds", "negative_image_embeds", "image"]
							__lowerCAmelCase    					=					[
							    "prompt",
							    "negative_prompt",
							    "image_embeds",
							    "negative_image_embeds",
							    "image",
							]
							__lowerCAmelCase    					=					[
							    "generator",
							    "height",
							    "width",
							    "strength",
							    "guidance_scale",
							    "negative_prompt",
							    "num_inference_steps",
							    "return_dict",
							    "guidance_scale",
							    "num_images_per_prompt",
							    "output_type",
							    "return_dict",
							]
							__lowerCAmelCase    					=					False






							@property
							def snake_case_  (   self      ):
									return 3_2






							@property
							def snake_case_  (   self      ):
									return 3_2






							@property
							def snake_case_  (   self      ):
									return self.time_input_dim






							@property
							def snake_case_  (   self      ):
									return self.time_input_dim * 4






							@property
							def snake_case_  (   self      ):
									return 1_0_0






							@property
							def snake_case_  (   self      ):
									a_     :      List[str] 						=     XLMRobertaTokenizerFast.from_pretrained("YiYiXu/tiny-random-mclip-base"      )
									return tokenizer






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )
									a_     :      List[Any] 						=     MCLIPConfig(
									    numDims=self.cross_attention_dim			,     transformerDimensions=self.text_embedder_hidden_size			,     hidden_size=self.text_embedder_hidden_size			,     intermediate_size=3_7			,     num_attention_heads=4			,     num_hidden_layers=5			,     vocab_size=1_0_0_5			,     )

									a_     :      str 						=     MultilingualCLIP(a_      )
									a_     :      Any 						=     text_encoder.eval()

									return text_encoder






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )

									a_     :      Union[str, Any] 						=     {
									    "in_channels": 4,
									    # Out channels is double in channels because predicts mean and variance
									    "out_channels": 8,
									    "addition_embed_type": "text_image",
									    "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
									    "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
									    "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
									    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
									    "layers_per_block": 1,
									    "encoder_hid_dim": self.text_embedder_hidden_size,
									    "encoder_hid_dim_type": "text_image_proj",
									    "cross_attention_dim": self.cross_attention_dim,
									    "attention_head_dim": 4,
									    "resnet_time_scale_shift": "scale_shift",
									    "class_embed_type": None,
									}

									a_     :      Dict 						=     UNetaDConditionModel(**a_      )
									return model






							@property
							def snake_case_  (   self      ):
									return {
									    "block_out_channels": [3_2, 6_4],
									    "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
									    "in_channels": 3,
									    "latent_channels": 4,
									    "layers_per_block": 1,
									    "norm_num_groups": 8,
									    "norm_type": "spatial",
									    "num_vq_embeddings": 1_2,
									    "out_channels": 3,
									    "up_block_types": [
									        "AttnUpDecoderBlock2D",
									        "UpDecoderBlock2D",
									    ],
									    "vq_embed_dim": 4,
									}






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )
									a_     :      int 						=     VQModel(**self.dummy_movq_kwargs      )
									return model






							def snake_case_  (   self      ):
									a_     :      Dict 						=     self.dummy_text_encoder
									a_     :      Dict 						=     self.dummy_tokenizer
									a_     :      Optional[int] 						=     self.dummy_unet
									a_     :      Dict 						=     self.dummy_movq

									a_     :      List[str] 						=     {
									    "num_train_timesteps": 1_0_0_0,
									    "beta_schedule": "linear",
									    "beta_start": 0.00_085,
									    "beta_end": 0.012,
									    "clip_sample": False,
									    "set_alpha_to_one": False,
									    "steps_offset": 0,
									    "prediction_type": "epsilon",
									    "thresholding": False,
									}

									a_     :      List[Any] 						=     DDIMScheduler(**a_      )

									a_     :      Union[str, Any] 						=     {
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									    "unet": unet,
									    "scheduler": scheduler,
									    "movq": movq,
									}

									return components






							def snake_case_  (   self			,     a_			,     a_=0      ):
									a_     :      Optional[int] 						=     floats_tensor((1, self.cross_attention_dim)			,     rng=random.Random(a_      )      ).to(a_      )
									a_     :      Tuple 						=     floats_tensor((1, self.cross_attention_dim)			,     rng=random.Random(seed + 1      )      ).to(a_      )
									# create init_image
									a_     :      Optional[int] 						=     floats_tensor((1, 3, 6_4, 6_4)			,     rng=random.Random(a_      )      ).to(a_      )
									a_     :      Any 						=     image.cpu().permute(0			,     2			,     3			,     1      )[0]
									a_     :      int 						=     Image.fromarray(np.uinta(a_      )      ).convert("RGB"      ).resize((2_5_6, 2_5_6)      )

									if str(a_      ).startswith("mps"      ):
											a_     :      Any 						=     torch.manual_seed(a_      )
									else:
											a_     :      Any 						=     torch.Generator(device=a_      ).manual_seed(a_      )
									a_     :      List[Any] 						=     {
									    "prompt": "horse",
									    "image": init_image,
									    "image_embeds": image_embeds,
									    "negative_image_embeds": negative_image_embeds,
									    "generator": generator,
									    "height": 6_4,
									    "width": 6_4,
									    "num_inference_steps": 1_0,
									    "guidance_scale": 7.0,
									    "strength": 0.2,
									    "output_type": "np",
									}
									return inputs






							def snake_case_  (   self      ):
									a_     :      Optional[Any] 						=     "cpu"

									a_     :      List[Any] 						=     self.get_dummy_components()

									a_     :      Union[str, Any] 						=     self.pipeline_class(**a_      )
									a_     :      Tuple 						=     pipe.to(a_      )

									pipe.set_progress_bar_config(disable=a_      )

									a_     :      Union[str, Any] 						=     pipe(**self.get_dummy_inputs(a_      )      )
									a_     :      Any 						=     output.images

									a_     :      str 						=     pipe(
									    **self.get_dummy_inputs(a_      )			,     return_dict=a_			,     )[0]

									a_     :      List[str] 						=     image[0, -3:, -3:, -1]
									a_     :      Optional[Any] 						=     image_from_tuple[0, -3:, -3:, -1]

									assert image.shape == (1, 6_4, 6_4, 3)

									a_     :      Optional[int] 						=     np.array(
									    [0.61_474_943, 0.6_073_539, 0.43_308_544, 0.5_928_269, 0.47_493_595, 0.46_755_973, 0.4_613_838, 0.45_368_797, 0.50_119_233]      )
									assert (
									    np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-2
									), F""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""
									assert (
									    np.abs(image_from_tuple_slice.flatten() - expected_slice      ).max() < 1e-2
									), F""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""




@slow
@require_torch_gpu
class 			snake_case_						(						unittest.TestCase				):






							def snake_case_  (   self      ):
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()






							def snake_case_  (   self      ):
									a_     :      Optional[int] 						=     load_numpy(
									    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
									    "/kandinsky/kandinsky_img2img_frog.npy"      )

									a_     :      Union[str, Any] 						=     load_image(
									    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"      )
									a_     :      Optional[Any] 						=     "A red cartoon frog, 4k"

									a_     :      int 						=     KandinskyPriorPipeline.from_pretrained(
									    "kandinsky-community/kandinsky-2-1-prior"			,     torch_dtype=torch.floataa      )
									pipe_prior.to(a_      )

									a_     :      int 						=     KandinskyImgaImgPipeline.from_pretrained(
									    "kandinsky-community/kandinsky-2-1"			,     torch_dtype=torch.floataa      )
									a_     :      List[Any] 						=     pipeline.to(a_      )

									pipeline.set_progress_bar_config(disable=a_      )

									a_     :      int 						=     torch.Generator(device="cpu"      ).manual_seed(0      )
									a_						,				a_     :      Optional[int] 						=     pipe_prior(
									    a_			,     generator=a_			,     num_inference_steps=5			,     negative_prompt=""			,     ).to_tuple()

									a_     :      List[Any] 						=     pipeline(
									    a_			,     image=a_			,     image_embeds=a_			,     negative_image_embeds=a_			,     generator=a_			,     num_inference_steps=1_0_0			,     height=7_6_8			,     width=7_6_8			,     strength=0.2			,     output_type="np"			,     )

									a_     :      int 						=     output.images[0]

									assert image.shape == (7_6_8, 7_6_8, 3)

									assert_mean_pixel_difference(a_			,     a_      )