


import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()





class    lowerCAmelCase__				(			__lowerCamelCase,    __lowerCamelCase,    unittest.TestCase       ):






						"""simple docstring"""
						__UpperCAmelCase   :		Any  					=	StableDiffusionParadigmsPipeline
						__UpperCAmelCase   :		List[Any]  					=	TEXT_TO_IMAGE_PARAMS
						__UpperCAmelCase   :		int  					=	TEXT_TO_IMAGE_BATCH_PARAMS
						__UpperCAmelCase   :		Optional[Any]  					=	TEXT_TO_IMAGE_IMAGE_PARAMS
						__UpperCAmelCase   :		Optional[Any]  					=	TEXT_TO_IMAGE_IMAGE_PARAMS




						def 					_UpperCamelCase			(   self     ):
							torch.manual_seed(0     )
							lowerCamelCase_			:							str      =       UNetaDConditionModel(
							    block_out_channels=(32, 64)				,							layers_per_block=2				,							sample_size=32				,							in_channels=4				,							out_channels=4				,							down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")				,							up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")				,							cross_attention_dim=32				,							attention_head_dim=(2, 4)				,							use_linear_projection=a_				,							)
							lowerCamelCase_			:							str      =       DDIMParallelScheduler(
							    beta_start=0.0_00_85				,							beta_end=0.0_12				,							beta_schedule="scaled_linear"				,							clip_sample=a_				,							set_alpha_to_one=a_				,							)
							torch.manual_seed(0     )
							lowerCamelCase_			:							str      =       AutoencoderKL(
							    block_out_channels=[32, 64]				,							in_channels=3				,							out_channels=3				,							down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]				,							up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]				,							latent_channels=4				,							sample_size=128				,							)
							torch.manual_seed(0     )
							lowerCamelCase_			:							Optional[int]      =       CLIPTextConfig(
							    bos_token_id=0				,							eos_token_id=2				,							hidden_size=32				,							intermediate_size=37				,							layer_norm_eps=1E-05				,							num_attention_heads=4				,							num_hidden_layers=5				,							pad_token_id=1				,							vocab_size=1000				,							hidden_act="gelu"				,							projection_dim=512				,							)
							lowerCamelCase_			:							int      =       CLIPTextModel(a_     )
							lowerCamelCase_			:							Any      =       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

							lowerCamelCase_			:							Union[str, Any]      =       {
							    "unet": unet,
							    "scheduler": scheduler,
							    "vae": vae,
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "safety_checker": None,
							    "feature_extractor": None,
							}
							return components




						def 					_UpperCamelCase			(   self				,							a_				,							a_=0     ):
							if str(a_     ).startswith("mps"     ):
								lowerCamelCase_			:							Union[str, Any]      =       torch.manual_seed(a_     )
							else:
								lowerCamelCase_			:							Dict      =       torch.Generator(device=a_     ).manual_seed(a_     )
							lowerCamelCase_			:							Optional[Any]      =       {
							    "prompt": "a photograph of an astronaut riding a horse",
							    "generator": generator,
							    "num_inference_steps": 10,
							    "guidance_scale": 6.0,
							    "output_type": "numpy",
							    "parallel": 3,
							    "debug": True,
							}
							return inputs




						def 					_UpperCamelCase			(   self     ):
							lowerCamelCase_			:							Optional[Any]      =       "cpu"  # ensure determinism for the device-dependent torch.Generator
							lowerCamelCase_			:							List[Any]      =       self.get_dummy_components()
							lowerCamelCase_			:							Tuple      =       StableDiffusionParadigmsPipeline(**a_     )
							lowerCamelCase_			:							Optional[Any]      =       sd_pipe.to(a_     )
							sd_pipe.set_progress_bar_config(disable=a_     )

							lowerCamelCase_			:							List[Any]      =       self.get_dummy_inputs(a_     )
							lowerCamelCase_			:							Any      =       sd_pipe(**a_     ).images
							lowerCamelCase_			:							Optional[int]      =       image[0, -3:, -3:, -1]
							assert image.shape == (1, 64, 64, 3)

							lowerCamelCase_			:							Dict      =       np.array([0.47_73, 0.54_17, 0.47_23, 0.49_25, 0.56_31, 0.47_52, 0.52_40, 0.49_35, 0.50_23]     )

							assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2




						def 					_UpperCamelCase			(   self     ):
							lowerCamelCase_			:							int      =       "cpu"  # ensure determinism for the device-dependent torch.Generator
							lowerCamelCase_			:							List[str]      =       self.get_dummy_components()
							torch.manual_seed(0     )
							lowerCamelCase_			:							Any      =       DDPMParallelScheduler()
							torch.manual_seed(0     )
							lowerCamelCase_			:							List[Any]      =       StableDiffusionParadigmsPipeline(**a_     )
							lowerCamelCase_			:							Optional[int]      =       sd_pipe.to(a_     )
							sd_pipe.set_progress_bar_config(disable=a_     )

							lowerCamelCase_			:							List[Any]      =       self.get_dummy_inputs(a_     )
							lowerCamelCase_			:							Union[str, Any]      =       sd_pipe(**a_     ).images
							lowerCamelCase_			:							Optional[Any]      =       image[0, -3:, -3:, -1]
							assert image.shape == (1, 64, 64, 3)

							lowerCamelCase_			:							Optional[int]      =       np.array([0.35_73, 0.44_20, 0.49_60, 0.47_99, 0.37_96, 0.38_79, 0.48_19, 0.43_65, 0.44_68]     )

							assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2




						def 					_UpperCamelCase			(   self     ):
							super().test_inference_batch_consistent(batch_sizes=[1, 2]     )




						def 					_UpperCamelCase			(   self     ):
							super().test_inference_batch_single_identical(batch_size=2				,							expected_max_diff=3E-3     )




						def 					_UpperCamelCase			(   self     ):
							lowerCamelCase_			:							List[Any]      =       "cpu"  # ensure determinism for the device-dependent torch.Generator
							lowerCamelCase_			:							List[Any]      =       self.get_dummy_components()
							lowerCamelCase_			:							List[str]      =       StableDiffusionParadigmsPipeline(**a_     )
							lowerCamelCase_			:							Optional[Any]      =       sd_pipe.to(a_     )
							sd_pipe.set_progress_bar_config(disable=a_     )

							lowerCamelCase_			:							List[str]      =       self.get_dummy_inputs(a_     )
							lowerCamelCase_			:							Optional[Any]      =       "french fries"
							lowerCamelCase_			:							str      =       sd_pipe(**a_				,							negative_prompt=a_     )
							lowerCamelCase_			:							Tuple      =       output.images
							lowerCamelCase_			:							Optional[int]      =       image[0, -3:, -3:, -1]

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

							lowerCamelCase_			:							Optional[Any]      =       np.array([0.47_71, 0.54_20, 0.46_83, 0.49_18, 0.56_36, 0.47_25, 0.52_30, 0.49_23, 0.50_15]     )

							assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1E-2





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






						"""simple docstring"""




						def 					_UpperCamelCase			(   self     ):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()




						def 					_UpperCamelCase			(   self				,							a_=0     ):
							lowerCamelCase_			:							Any      =       torch.Generator(device=a_     ).manual_seed(a_     )
							lowerCamelCase_			:							Union[str, Any]      =       {
							    "prompt": "a photograph of an astronaut riding a horse",
							    "generator": generator,
							    "num_inference_steps": 10,
							    "guidance_scale": 7.5,
							    "output_type": "numpy",
							    "parallel": 3,
							    "debug": True,
							}
							return inputs




						def 					_UpperCamelCase			(   self     ):
							lowerCamelCase_			:							int      =       "stabilityai/stable-diffusion-2-base"
							lowerCamelCase_			:							str      =       DDIMParallelScheduler.from_pretrained(a_				,							subfolder="scheduler"     )
							lowerCamelCase_			:							str      =       StableDiffusionParadigmsPipeline.from_pretrained(a_				,							scheduler=a_				,							safety_checker=a_     )
							pipe.to(a_     )
							pipe.set_progress_bar_config(disable=a_     )
							pipe.enable_attention_slicing()

							lowerCamelCase_			:							Any      =       self.get_inputs()
							lowerCamelCase_			:							Union[str, Any]      =       pipe(**a_     ).images
							lowerCamelCase_			:							List[Any]      =       image[0, -3:, -3:, -1].flatten()

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

							lowerCamelCase_			:							Tuple      =       np.array([0.96_22, 0.96_02, 0.97_48, 0.95_91, 0.96_30, 0.96_91, 0.96_61, 0.96_31, 0.97_41]     )

							assert np.abs(expected_slice - image_slice     ).max() < 1E-2





from __future__ import annotations





class    lowerCAmelCase__				:






						"""simple docstring"""




						def __init__(   self				,							a_     ):
							lowerCamelCase_			:							List[str]      =       value
							lowerCamelCase_			:							Node | None      =       None
							lowerCamelCase_			:							Node | None      =       None





class    lowerCAmelCase__				:






						"""simple docstring"""
						__UpperCAmelCase   :		int




						def __init__(   self     ):
							lowerCamelCase_			:							str      =       0




						def 					_UpperCamelCase			(   self				,							a_				,							a_     ):
							if node is None:
								return

							if path_sum == self.target:
								self.paths += 1

							if node.left:
								self.depth_first_search(node.left				,							path_sum + node.left.value     )
							if node.right:
								self.depth_first_search(node.right				,							path_sum + node.right.value     )




						def 					_UpperCamelCase			(   self				,							a_				,							a_ = None     ):
							if node is None:
								return 0
							if target is not None:
								lowerCamelCase_			:							List[str]      =       target

							self.depth_first_search(a_				,							node.value     )
							self.path_sum(node.left     )
							self.path_sum(node.right     )

							return self.paths


if __name__ == "__main__":
							import doctest

							doctest.testmod()

