






import gc
import random
import unittest

import numpy as np
import torch

from diffusers import DDIMScheduler, KandinskyVaaPipeline, KandinskyVaaPriorPipeline, UNetaDConditionModel, VQModel
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  (  _A				,							unittest.TestCase		):




						"""simple docstring"""



						A          =    KandinskyVaaPipeline
						A          =    [
						    '''image_embeds''',
						    '''negative_image_embeds''',
						]
						A          =    ['''image_embeds''', '''negative_image_embeds''']
						A          =    [
						    '''generator''',
						    '''height''',
						    '''width''',
						    '''latents''',
						    '''guidance_scale''',
						    '''num_inference_steps''',
						    '''return_dict''',
						    '''guidance_scale''',
						    '''num_images_per_prompt''',
						    '''output_type''',
						    '''return_dict''',
						]
						A          =    False
						@property
						def 			snake_case_      (   self		):



										'''simple docstring'''
										return 32
						@property
						def 			snake_case_      (   self		):



										'''simple docstring'''
										return 32
						@property
						def 			snake_case_      (   self		):



										'''simple docstring'''
										return self.time_input_dim
						@property
						def 			snake_case_      (   self		):



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



										'''simple docstring'''
										return 100
						@property
						def 			snake_case_      (   self		):



										'''simple docstring'''
										torch.manual_seed(0		)

										lowerCAmelCase__   :List[str]      							=				{
										    "in_channels": 4,
										    # Out channels is double in channels because predicts mean and variance
										    "out_channels": 8,
										    "addition_embed_type": "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": "image_proj",
										    "cross_attention_dim": self.cross_attention_dim,
										    "attention_head_dim": 4,
										    "resnet_time_scale_shift": "scale_shift",
										    "class_embed_type": None,
										}

										lowerCAmelCase__   :List[str]      							=				UNetaDConditionModel(**_lowerCAmelCase		)
										return model
						@property
						def 			snake_case_      (   self		):



										'''simple docstring'''
										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		):



										'''simple docstring'''
										torch.manual_seed(0		)
										lowerCAmelCase__   :Optional[Any]      							=				VQModel(**self.dummy_movq_kwargs		)
										return model
						def 			snake_case_      (   self		):



										'''simple docstring'''
										lowerCAmelCase__   :int      							=				self.dummy_unet
										lowerCAmelCase__   :List[str]      							=				self.dummy_movq

										lowerCAmelCase__   :List[Any]      							=				DDIMScheduler(
										    num_train_timesteps=1_000    ,	beta_schedule="linear"    ,	beta_start=0.0_0085    ,	beta_end=0.012    ,	clip_sample=_lowerCAmelCase    ,	set_alpha_to_one=_lowerCAmelCase    ,	steps_offset=1    ,	prediction_type="epsilon"    ,	thresholding=_lowerCAmelCase    ,	)

										lowerCAmelCase__   :Optional[Any]      							=				{
										    "unet": unet,
										    "scheduler": scheduler,
										    "movq": movq,
										}
										return components
						def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase=0		):



										'''simple docstring'''
										lowerCAmelCase__   :List[Any]      							=				floats_tensor((1, self.text_embedder_hidden_size)    ,	rng=random.Random(_lowerCAmelCase		)		).to(_lowerCAmelCase		)
										lowerCAmelCase__   :Any      							=				floats_tensor((1, self.text_embedder_hidden_size)    ,	rng=random.Random(seed + 1		)		).to(
										    _lowerCAmelCase		)
										if str(_lowerCAmelCase		).startswith("mps"		):
														lowerCAmelCase__   :List[Any]      							=				torch.manual_seed(_lowerCAmelCase		)
										else:
														lowerCAmelCase__   :Optional[int]      							=				torch.Generator(device=_lowerCAmelCase		).manual_seed(_lowerCAmelCase		)
										lowerCAmelCase__   :Optional[int]      							=				{
										    "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		):



										'''simple docstring'''
										lowerCAmelCase__   :int      							=				"cpu"

										lowerCAmelCase__   :Dict      							=				self.get_dummy_components()

										lowerCAmelCase__   :Tuple      							=				self.pipeline_class(**_lowerCAmelCase		)
										lowerCAmelCase__   :List[Any]      							=				pipe.to(_lowerCAmelCase		)

										pipe.set_progress_bar_config(disable=_lowerCAmelCase		)

										lowerCAmelCase__   :str      							=				pipe(**self.get_dummy_inputs(_lowerCAmelCase		)		)
										lowerCAmelCase__   :Dict      							=				output.images

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

										lowerCAmelCase__   :Tuple      							=				image[0, -3:, -3:, -1]
										lowerCAmelCase__   :List[Any]      							=				image_from_tuple[0, -3:, -3:, -1]

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

										lowerCAmelCase__   :Tuple      							=				np.array(
										    [0.623_7976, 1.0, 0.3644_1332, 1.0, 0.7063_9634, 0.2987_7186, 0.8565_2125, 0.521_6843, 0.5445_4046]		)

										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		):




						"""simple docstring"""



						def 			snake_case_      (   self		):



										'''simple docstring'''
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()



						def 			snake_case_      (   self		):



										'''simple docstring'''
										lowerCAmelCase__   :int      							=				load_numpy(
										    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
										    "/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy"		)

										lowerCAmelCase__   :str      							=				KandinskyVaaPriorPipeline.from_pretrained(
										    "kandinsky-community/kandinsky-2-2-prior"    ,	torch_dtype=torch.floataa		)
										pipe_prior.to(_lowerCAmelCase		)

										lowerCAmelCase__   :str      							=				KandinskyVaaPipeline.from_pretrained(
										    "kandinsky-community/kandinsky-2-2-decoder"    ,	torch_dtype=torch.floataa		)
										lowerCAmelCase__   :Union[str, Any]      							=				pipeline.to(_lowerCAmelCase		)
										pipeline.set_progress_bar_config(disable=_lowerCAmelCase		)

										lowerCAmelCase__   :Optional[int]      							=				"red cat, 4k photo"

										lowerCAmelCase__   :Tuple      							=				torch.Generator(device="cuda"		).manual_seed(0		)
										lowerCAmelCase__   :Dict      							=				pipe_prior(
										    _lowerCAmelCase    ,	generator=_lowerCAmelCase    ,	num_inference_steps=5    ,	negative_prompt=""    ,	).to_tuple()

										lowerCAmelCase__   :List[Any]      							=				torch.Generator(device="cuda"		).manual_seed(0		)
										lowerCAmelCase__   :List[str]      							=				pipeline(
										    image_embeds=_lowerCAmelCase    ,	negative_image_embeds=_lowerCAmelCase    ,	generator=_lowerCAmelCase    ,	num_inference_steps=100    ,	output_type="np"    ,	)

										lowerCAmelCase__   :int      							=				output.images[0]

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

										assert_mean_pixel_difference(_lowerCAmelCase    ,	_lowerCAmelCase		)











from collections.abc import Callable
from math import pi, sqrt
from random import uniform
from statistics import mean


def 				snake_case__							(							UpperCAmelCase							:  int       ):

    # A local function to see if a dot lands in the circle.
    def is_in_circle(UpperCAmelCase							:  float    ,			UpperCAmelCase							:  float       ) -> bool:
        lowerCAmelCase__   :Optional[int]      							=				sqrt((x**2) + (y**2)       )
        # Our circle has a radius of 1, so a distance
        # greater than 1 would land outside the circle.
        return distance_from_centre <= 1

    # The proportion of guesses that landed in the circle
    lowerCAmelCase__   :Union[str, Any]      							=				mean(
        int(is_in_circle(uniform(-1.0    ,			1.0       )    ,			uniform(-1.0    ,			1.0       )       )       )
        for _ in range(UpperCAmelCase       )       )
    # The ratio of the area for circle to square is pi/4.
    lowerCAmelCase__   :List[Any]      							=				proportion * 4
    print(F'''The estimated value of pi is {pi_estimate}'''       )
    print(F'''The numpy value of pi is {pi}'''       )
    print(F'''The total error is {abs(pi - pi_estimate       )}'''       )


def 				snake_case__							(							UpperCAmelCase							:  int    ,			UpperCAmelCase							:  Callable[[float], float]    ,			UpperCAmelCase							:  float = 0.0    ,			UpperCAmelCase							:  float = 1.0    ,			):
    return mean(
        function_to_integrate(uniform(UpperCAmelCase    ,			UpperCAmelCase       )       ) for _ in range(UpperCAmelCase       )       ) * (max_value - min_value)


def 				snake_case__							(							UpperCAmelCase							:  int    ,			UpperCAmelCase							:  float = 0.0    ,			UpperCAmelCase							:  float = 1.0       ):

    def identity_function(UpperCAmelCase							:  float       ) -> float:
        return x

    lowerCAmelCase__   :List[str]      							=				area_under_curve_estimator(
        UpperCAmelCase    ,			UpperCAmelCase    ,			UpperCAmelCase    ,			UpperCAmelCase       )
    lowerCAmelCase__   :Tuple      							=				(max_value * max_value - min_value * min_value) / 2

    print("******************"       )
    print(F'''Estimating area under y=x where x varies from {min_value} to {max_value}'''       )
    print(F'''Estimated value is {estimated_value}'''       )
    print(F'''Expected value is {expected_value}'''       )
    print(F'''Total error is {abs(estimated_value - expected_value       )}'''       )
    print("******************"       )


def 				snake_case__							(							UpperCAmelCase							:  int       ):

    def function_to_integrate(UpperCAmelCase							:  float       ) -> float:
        return sqrt(4.0 - x * x       )

    lowerCAmelCase__   :int      							=				area_under_curve_estimator(
        UpperCAmelCase    ,			UpperCAmelCase    ,			0.0    ,			2.0       )

    print("******************"       )
    print("Estimating pi using area_under_curve_estimator"       )
    print(F'''Estimated value is {estimated_value}'''       )
    print(F'''Expected value is {pi}'''       )
    print(F'''Total error is {abs(estimated_value - pi       )}'''       )
    print("******************"       )


if __name__ == "__main__":
 import doctest

 doctest.testmod()



