





import dataclasses
import enum
import functools
import math
import re

# The following script is adapted from the script of TaPas.
# Original: https://github.com/google-research/tapas/master/wikisql_utils.py
from typing import Any, List, Text


_SCREAMING_SNAKE_CASE   						=		"""none"""
_SCREAMING_SNAKE_CASE   						=		"""none"""


def 				SCREAMING_SNAKE_CASE__    (     __a ,			__a ):
	snake_case_		:      List[str]           =						value.split(__a )
	return len(__a ) > 1 and any((len(__a ) == 3 for x in split) )


def 				SCREAMING_SNAKE_CASE__    (     __a ):
	if isinstance(__a ,			__a ):
		return value
	if isinstance(__a ,			__a ):
		return float(__a )
	if not isinstance(__a ,			__a ):
		raise ValueError('Argument value is not a string. Can\'t parse it as float' )
	snake_case_		:      Tuple           =						value

	try:
		# Example: 1,000.7
		if "." in sanitized and "," in sanitized:
			return float(sanitized.replace(',' ,			'' ) )
		# 1,000
		if "," in sanitized and _split_thousands(',' ,			__a ):
			return float(sanitized.replace(',' ,			'' ) )
		# 5,5556
		if "," in sanitized and sanitized.count(',' ) == 1 and not _split_thousands(',' ,			__a ):
			return float(sanitized.replace(',' ,			'.' ) )
		# 0.0.0.1
		if sanitized.count('.' ) > 1:
			return float(sanitized.replace('.' ,			'' ) )
		# 0,0,0,1
		if sanitized.count(',' ) > 1:
			return float(sanitized.replace(',' ,			'' ) )
		return float(__a )
	except ValueError:
		# Avoid adding the sanitized value in the error message.
		raise ValueError('Unable to convert value to float' )


def 				SCREAMING_SNAKE_CASE__    (     __a ):
	if answer is None:
		return None
	try:
		snake_case_		:      str           =						convert_to_float(__a )
		if isinstance(__a ,			__a ) and math.isnan(__a ):
			return None
		return value
	except ValueError:
		return answer.lower()


_SCREAMING_SNAKE_CASE   						=		{
    """text""": lambda x: x,
    """real""": convert_to_float,
}
class SCREAMING_SNAKE_CASE_    (       enum.Enum       ):
						__magic_name__:      Union[str, Any]			   =							0
						__magic_name__:      Optional[int]			   =							1
						__magic_name__:      List[str]			   =							2
						__magic_name__:      Union[str, Any]			   =							3
						__magic_name__:      Tuple			   =							4
						__magic_name__:      Optional[Any]			   =							5
class SCREAMING_SNAKE_CASE_    (       enum.Enum       ):
						__magic_name__:      List[str]			   =							0
						__magic_name__:      Any			   =							1
						__magic_name__:      Optional[Any]			   =							2
@dataclasses.dataclass
class SCREAMING_SNAKE_CASE_    :
						__magic_name__:      Text
						__magic_name__:      _Operator
						__magic_name__:      Any
_SCREAMING_SNAKE_CASE   						=		re.compile(R"""\w+|[^\w\s]+""", re.UNICODE | re.MULTILINE | re.DOTALL)


def 				SCREAMING_SNAKE_CASE__    (     __a ):
	return list(_TOKENIZER.findall(x.lower() ) )


def 				SCREAMING_SNAKE_CASE__    (     __a ,			__a ,			__a ):
	if operator == _Operator.EQUALS:
		return src == tgt
	elif operator == _Operator.GREATER:
		return src > tgt
	elif operator == _Operator.LESSER:
		return src < tgt
	raise ValueError(f"""Unknown operator: {operator}""" )


def 				SCREAMING_SNAKE_CASE__    (     __a ,			__a ,			__a ):
	snake_case_		:      List[str]           =						table['types']
	return _TYPE_CONVERTER[types[column]](__a )


def 				SCREAMING_SNAKE_CASE__    (     __a ):
	return isinstance(__a ,			__a )


def 				SCREAMING_SNAKE_CASE__    (     __a ,			__a ,			__a ):
	for cond in conditions:
		snake_case_		:      List[Any]           =						row[cond.column]

		snake_case_		:      Dict           =						_parse_value(__a ,			cond.column ,			cond.cmp_value )

		if _is_string(__a ) and _is_string(__a ):
			snake_case_		:      Optional[int]           =						_normalize_for_match(__a )
			snake_case_		:      Tuple           =						_normalize_for_match(__a )

		if not isinstance(__a ,			type(__a ) ):
			raise ValueError('Type difference {} != {}'.format(type(__a ) ,			type(__a ) ) )

		if not _compare(cond.operator ,			__a ,			__a ):
			return False
	return True


def 				SCREAMING_SNAKE_CASE__    (     __a ,			__a ,			__a ):
	if not answer_coordinates:
		if aggregation_op == _Aggregation.COUNT:
			return 0.0
		else:
			return EMPTY_ANSWER_AGG

    # Count can support non numeric answers.
	if aggregation_op == _Aggregation.COUNT:
		return float(len(__a ) )

	# If we have just one answer, if float returns it or try a conversion.
	snake_case_		:      Optional[int]           =						[table['rows'][i][j] for (i, j) in answer_coordinates]
	if len(__a ) == 1:
		try:
			return convert_to_float(values[0] )
		except ValueError as e:
			if aggregation_op != _Aggregation.NONE:
				raise e

	if aggregation_op == _Aggregation.NONE:
		return None

	# Other aggregation only support numeric values. Bail out if we have strings.
	if not all((isinstance(__a ,			(int, float) ) for v in values) ):
		return None

	if aggregation_op == _Aggregation.SUM:
		return float(sum(__a ) )
	elif aggregation_op == _Aggregation.AVERAGE:
		return sum(__a ) / len(__a )
	else:
		raise ValueError(f"""Unknown aggregation: {aggregation_op}""" )


def 				SCREAMING_SNAKE_CASE__    (     __a ,			__a ):
	snake_case_		:      Dict           =						sql_query['agg']
	if aggregation_op_index >= 3:
		snake_case_		:      List[Any]           =						_Aggregation(__a )
	else:
		snake_case_		:      int           =						_Aggregation.NONE

	snake_case_		:      Dict           =						sql_query['sel']
	snake_case_		:      Optional[int]           =						[
	    _Condition(__a ,			_Operator(__a ) ,			__a )
	    for column, operator, cmp_value in zip(
	        sql_query['conds']['column_index'] ,			sql_query['conds']['operator_index'] ,			sql_query['conds']['condition'] )
	]

	snake_case_		:      List[str]           =						[]
	for row in range(len(table['rows'] ) ):
		if _respect_conditions(__a ,			table['rows'][row] ,			__a ):
			indices.append((row, target_column) )

	if not indices:
		return [], aggregation_op

	if len(__a ) == 1:
		return indices, aggregation_op

	# Parsing of MIN/MAX.
	if aggregation_op_index in (1, 2):
		snake_case_		:      Union[str, Any]           =						{2: min, 1: max}
		snake_case_		:      Tuple           =						[(table['rows'][i][j], index) for index, (i, j) in enumerate(__a )]
		snake_case_		:      Optional[int]           =						functools.reduce(operators[sql_query['agg']] ,			__a )

		snake_case_		:      str           =						[indices[reduced[1]]]
		return ret, _Aggregation.NONE

	return indices, aggregation_op


def 				SCREAMING_SNAKE_CASE__    (     __a ,			__a ,			__a ):
	if float_answer is not None:
		return [str(__a )]
	return [str(table['real_rows'][r][c] ) for r, c in answer_coordinates]


def 				SCREAMING_SNAKE_CASE__    (     __a ,			__a ):
	snake_case_    ,snake_case_		:      Dict           =						_get_answer_coordinates(__a ,			__a )
	snake_case_		:      Any           =						_get_float_answer(__a ,			__a ,			__a )
	snake_case_		:      Any           =						_get_answer_text(__a ,			__a ,			__a )
	# keep the original data the same with TaPas
	if len(__a ) == 0:
		snake_case_		:      Optional[Any]           =						[EMPTY_ANSWER]
	return answer_text








import gc
import random
import unittest

import numpy as np
import torch
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, 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 	_UpperCAmelCase						( _lowercase			,							unittest.TestCase       ):
					__A        = KandinskyPipeline
					__A        = [
					    "prompt",
					    "image_embeds",
					    "negative_image_embeds",
					]
					__A        = ["prompt", "negative_prompt", "image_embeds", "negative_image_embeds"]
					__A        = [
					    "generator",
					    "height",
					    "width",
					    "latents",
					    "guidance_scale",
					    "negative_prompt",
					    "num_inference_steps",
					    "return_dict",
					    "guidance_scale",
					    "num_images_per_prompt",
					    "output_type",
					    "return_dict",
					]
					__A        = False




					@property
					def 						snake_case_		(    self							)    ->	Any:
							return 32




					@property
					def 						snake_case_		(    self							)    ->	Optional[Any]:
							return 32




					@property
					def 						snake_case_		(    self							)    ->	List[Any]:
							return self.time_input_dim




					@property
					def 						snake_case_		(    self							)    ->	Union[str, Any]:
							return self.time_input_dim * 4




					@property
					def 						snake_case_		(    self							)    ->	Dict:
							return 100




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




					@property
					def 						snake_case_		(    self							)    ->	Union[str, Any]:
							torch.manual_seed(0							)
							UpperCamelCase__								=						MCLIPConfig(
							    numDims=self.cross_attention_dim     ,   transformerDimensions=self.text_embedder_hidden_size     ,   hidden_size=self.text_embedder_hidden_size     ,   intermediate_size=37     ,   num_attention_heads=4     ,   num_hidden_layers=5     ,   vocab_size=1_005     ,   )

							UpperCamelCase__								=						MultilingualCLIP(__lowerCAmelCase							)
							UpperCamelCase__								=						text_encoder.eval()

							return text_encoder




					@property
					def 						snake_case_		(    self							)    ->	Optional[Any]:
							torch.manual_seed(0							)

							UpperCamelCase__								=						{
							    """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,
							}

							UpperCamelCase__								=						UNetaDConditionModel(**__lowerCAmelCase							)
							return model




					@property
					def 						snake_case_		(    self							)    ->	Union[str, Any]:
							return {
							    "block_out_channels": [32, 64],
							    "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": 12,
							    "out_channels": 3,
							    "up_block_types": [
							        "AttnUpDecoderBlock2D",
							        "UpDecoderBlock2D",
							    ],
							    "vq_embed_dim": 4,
							}




					@property
					def 						snake_case_		(    self							)    ->	Optional[int]:
							torch.manual_seed(0							)
							UpperCamelCase__								=						VQModel(**self.dummy_movq_kwargs							)
							return model




					def 						snake_case_		(    self							)    ->	Dict:
							UpperCamelCase__								=						self.dummy_text_encoder
							UpperCamelCase__								=						self.dummy_tokenizer
							UpperCamelCase__								=						self.dummy_unet
							UpperCamelCase__								=						self.dummy_movq

							UpperCamelCase__								=						DDIMScheduler(
							    num_train_timesteps=1_000     ,   beta_schedule="""linear"""     ,   beta_start=0.0_0_0_8_5     ,   beta_end=0.0_1_2     ,   clip_sample=__lowerCAmelCase     ,   set_alpha_to_one=__lowerCAmelCase     ,   steps_offset=1     ,   prediction_type="""epsilon"""     ,   thresholding=__lowerCAmelCase     ,   )

							UpperCamelCase__								=						{
							    """text_encoder""": text_encoder,
							    """tokenizer""": tokenizer,
							    """unet""": unet,
							    """scheduler""": scheduler,
							    """movq""": movq,
							}
							return components




					def 						snake_case_		(    self     ,   __lowerCAmelCase     ,   __lowerCAmelCase=0							)    ->	str:
							UpperCamelCase__								=						floats_tensor((1, self.cross_attention_dim)     ,   rng=random.Random(__lowerCAmelCase							)							).to(__lowerCAmelCase							)
							UpperCamelCase__								=						floats_tensor((1, self.cross_attention_dim)     ,   rng=random.Random(seed + 1							)							).to(__lowerCAmelCase							)
							if str(__lowerCAmelCase							).startswith("""mps"""							):
									UpperCamelCase__								=						torch.manual_seed(__lowerCAmelCase							)
							else:
									UpperCamelCase__								=						torch.Generator(device=__lowerCAmelCase							).manual_seed(__lowerCAmelCase							)
							UpperCamelCase__								=						{
							    """prompt""": """horse""",
							    """image_embeds""": image_embeds,
							    """negative_image_embeds""": negative_image_embeds,
							    """generator""": generator,
							    """height""": 64,
							    """width""": 64,
							    """guidance_scale""": 4.0,
							    """num_inference_steps""": 2,
							    """output_type""": """np""",
							}
							return inputs




					def 						snake_case_		(    self							)    ->	Tuple:
							UpperCamelCase__								=						"""cpu"""

							UpperCamelCase__								=						self.get_dummy_components()

							UpperCamelCase__								=						self.pipeline_class(**__lowerCAmelCase							)
							UpperCamelCase__								=						pipe.to(__lowerCAmelCase							)

							pipe.set_progress_bar_config(disable=__lowerCAmelCase							)

							UpperCamelCase__								=						pipe(**self.get_dummy_inputs(__lowerCAmelCase							)							)
							UpperCamelCase__								=						output.images

							UpperCamelCase__								=						pipe(
							    **self.get_dummy_inputs(__lowerCAmelCase							)     ,   return_dict=__lowerCAmelCase     ,   )[0]

							UpperCamelCase__								=						image[0, -3:, -3:, -1]
							UpperCamelCase__								=						image_from_tuple[0, -3:, -3:, -1]

							assert image.shape == (1, 64, 64, 3)

							UpperCamelCase__								=						np.array(
							    [0.3_2_8_6_6_3, 1.0, 0.2_3_2_1_6_8_7_3, 1.0, 0.9_2_7_1_7_5_6_4, 0.4_6_3_9_0_4_6, 0.9_6_8_9_4_7_7_7, 0.3_1_7_1_3_3_7_8, 0.6_2_9_3_9_5_3]							)

							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 	_UpperCAmelCase						( unittest.TestCase       ):




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




					def 						snake_case_		(    self							)    ->	Any:
							UpperCamelCase__								=						load_numpy(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
							    """/kandinsky/kandinsky_text2img_cat_fp16.npy"""							)

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

							UpperCamelCase__								=						KandinskyPipeline.from_pretrained("""kandinsky-community/kandinsky-2-1"""     ,   torch_dtype=torch.floataa							)
							UpperCamelCase__								=						pipeline.to(__lowerCAmelCase							)
							pipeline.set_progress_bar_config(disable=__lowerCAmelCase							)

							UpperCamelCase__								=						"""red cat, 4k photo"""

							UpperCamelCase__								=						torch.Generator(device="""cuda"""							).manual_seed(0							)
							UpperCamelCase__	,      UpperCamelCase__								=						pipe_prior(
							    __lowerCAmelCase     ,   generator=__lowerCAmelCase     ,   num_inference_steps=5     ,   negative_prompt=""""""     ,   ).to_tuple()

							UpperCamelCase__								=						torch.Generator(device="""cuda"""							).manual_seed(0							)
							UpperCamelCase__								=						pipeline(
							    __lowerCAmelCase     ,   image_embeds=__lowerCAmelCase     ,   negative_image_embeds=__lowerCAmelCase     ,   generator=__lowerCAmelCase     ,   num_inference_steps=100     ,   output_type="""np"""     ,   )

							UpperCamelCase__								=						output.images[0]

							assert image.shape == (512, 512, 3)

							assert_mean_pixel_difference(__lowerCAmelCase     ,   __lowerCAmelCase							)





