





"""simple docstring"""






import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModel,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableUnCLIPImgaImgPipeline, UNetaDConditionModel
from diffusers.pipelines.pipeline_utils import DiffusionPipeline
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    floats_tensor,
    load_image,
    load_numpy,
    require_torch_gpu,
    skip_mps,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()




class 				__lowercase			(			_UpperCamelCase   ,      _UpperCamelCase   ,      _UpperCamelCase   ,      unittest.TestCase							):







					'''simple docstring'''




					__lowerCAmelCase  =  StableUnCLIPImgaImgPipeline
					__lowerCAmelCase  =  TEXT_GUIDED_IMAGE_VARIATION_PARAMS
					__lowerCAmelCase  =  TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
					__lowerCAmelCase  =  frozenset(
					    []							)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
					__lowerCAmelCase  =  frozenset([]							)







					def  _lowerCamelCase		(						self     ):
									__a :						Optional[Any]			       =  32
									__a :						int			       =  embedder_hidden_size

									# image encoding components

									__a :						Tuple			       =  CLIPImageProcessor(crop_size=32	,			size=32     )

									torch.manual_seed(0     )
									__a :						Union[str, Any]			       =  CLIPVisionModelWithProjection(
									    CLIPVisionConfig(
									        hidden_size=_UpperCAmelCase	,			projection_dim=_UpperCAmelCase	,			num_hidden_layers=5	,			num_attention_heads=4	,			image_size=32	,			intermediate_size=37	,			patch_size=1	,			)     )

									# regular denoising components

									torch.manual_seed(0     )
									__a :						List[str]			       =  StableUnCLIPImageNormalizer(embedding_dim=_UpperCAmelCase     )
									__a :						Tuple			       =  DDPMScheduler(beta_schedule='''squaredcos_cap_v2'''     )

									torch.manual_seed(0     )
									__a :						List[Any]			       =  CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''     )

									torch.manual_seed(0     )
									__a :						Dict			       =  CLIPTextModel(
									    CLIPTextConfig(
									        bos_token_id=0	,			eos_token_id=2	,			hidden_size=_UpperCAmelCase	,			projection_dim=32	,			intermediate_size=37	,			layer_norm_eps=1e-0_5	,			num_attention_heads=4	,			num_hidden_layers=5	,			pad_token_id=1	,			vocab_size=1000	,			)     )

									torch.manual_seed(0     )
									__a :						Any			       =  UNetaDConditionModel(
									    sample_size=32	,			in_channels=4	,			out_channels=4	,			down_block_types=('''CrossAttnDownBlock2D''', '''DownBlock2D''')	,			up_block_types=('''UpBlock2D''', '''CrossAttnUpBlock2D''')	,			block_out_channels=(32, 64)	,			attention_head_dim=(2, 4)	,			class_embed_type='''projection'''	,			projection_class_embeddings_input_dim=embedder_projection_dim * 2	,			cross_attention_dim=_UpperCAmelCase	,			layers_per_block=1	,			upcast_attention=_UpperCAmelCase	,			use_linear_projection=_UpperCAmelCase	,			)

									torch.manual_seed(0     )
									__a :						Union[str, Any]			       =  DDIMScheduler(
									    beta_schedule='''scaled_linear'''	,			beta_start=0.0_0_0_8_5	,			beta_end=0.0_1_2	,			prediction_type='''v_prediction'''	,			set_alpha_to_one=_UpperCAmelCase	,			steps_offset=1	,			)

									torch.manual_seed(0     )
									__a :						List[Any]			       =  AutoencoderKL()

									__a :						Any			       =  {
									    # image encoding components
									    '''feature_extractor''': feature_extractor,
									    '''image_encoder''': image_encoder.eval(),
									    # image noising components
									    '''image_normalizer''': image_normalizer.eval(),
									    '''image_noising_scheduler''': image_noising_scheduler,
									    # regular denoising components
									    '''tokenizer''': tokenizer,
									    '''text_encoder''': text_encoder.eval(),
									    '''unet''': unet.eval(),
									    '''scheduler''': scheduler,
									    '''vae''': vae.eval(),
									}

									return components







					def  _lowerCamelCase		(						self	,			_UpperCAmelCase	,			_UpperCAmelCase=0	,			_UpperCAmelCase=True     ):
									if str(_UpperCAmelCase     ).startswith('''mps'''     ):
													__a :						int			       =  torch.manual_seed(_UpperCAmelCase     )
									else:
													__a :						Dict			       =  torch.Generator(device=_UpperCAmelCase     ).manual_seed(_UpperCAmelCase     )

									__a :						int			       =  floats_tensor((1, 3, 32, 32)	,			rng=random.Random(_UpperCAmelCase     )     ).to(_UpperCAmelCase     )

									if pil_image:
													__a :						int			       =  input_image * 0.5 + 0.5
													__a :						List[str]			       =  input_image.clamp(0	,			1     )
													__a :						Tuple			       =  input_image.cpu().permute(0	,			2	,			3	,			1     ).float().numpy()
													__a :						int			       =  DiffusionPipeline.numpy_to_pil(_UpperCAmelCase     )[0]

									return {
									    "prompt": "An anime racoon running a marathon",
									    "image": input_image,
									    "generator": generator,
									    "num_inference_steps": 2,
									    "output_type": "np",
									}







					@skip_mps
					def  _lowerCamelCase		(						self     ):
									__a :						int			       =  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
									__a :						Any			       =  self.get_dummy_components()
									__a :						Union[str, Any]			       =  StableUnCLIPImgaImgPipeline(**_UpperCAmelCase     )
									__a :						str			       =  sd_pipe.to(_UpperCAmelCase     )
									sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

									__a :						List[Any]			       =  self.get_dummy_inputs(_UpperCAmelCase     )
									inputs.update({'''image_embeds''': None}     )
									__a :						Optional[Any]			       =  sd_pipe(**_UpperCAmelCase     ).images
									__a :						str			       =  image[0, -3:, -3:, -1]

									assert image.shape == (1, 32, 32, 3)
									__a :						Dict			       =  np.array([0.3_8_7_2, 0.7_2_2_4, 0.5_6_0_1, 0.4_7_4_1, 0.6_8_7_2, 0.5_8_1_4, 0.4_6_3_6, 0.3_8_6_7, 0.5_0_7_8]     )

									assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-3







					def  _lowerCamelCase		(						self     ):
									__a :						Any			       =  torch_device in ['''cpu''', '''mps''']

									self._test_attention_slicing_forward_pass(test_max_difference=_UpperCAmelCase     )







					def  _lowerCamelCase		(						self     ):
									__a :						Tuple			       =  torch_device in ['''cpu''', '''mps''']

									self._test_inference_batch_single_identical(test_max_difference=_UpperCAmelCase     )







					@unittest.skipIf(
					    torch_device != '''cuda''' or not is_xformers_available()	,			reason='''XFormers attention is only available with CUDA and `xformers` installed'''	,			)
					def  _lowerCamelCase		(						self     ):
									self._test_xformers_attention_forwardGenerator_pass(test_max_difference=_UpperCAmelCase     )






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







					'''simple docstring'''




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







					def  _lowerCamelCase		(						self     ):
									__a :						Tuple			       =  load_image(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''     )

									__a :						int			       =  load_numpy(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_img2img_anime_turtle_fp16.npy'''     )

									__a :						Dict			       =  StableUnCLIPImgaImgPipeline.from_pretrained(
									    '''fusing/stable-unclip-2-1-l-img2img'''	,			torch_dtype=torch.floataa     )
									pipe.to(_UpperCAmelCase     )
									pipe.set_progress_bar_config(disable=_UpperCAmelCase     )
									# stable unclip will oom when integration tests are run on a V100,
									# so turn on memory savings
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									__a :						Optional[int]			       =  torch.Generator(device='''cpu'''     ).manual_seed(0     )
									__a :						Union[str, Any]			       =  pipe(_UpperCAmelCase	,			'''anime turle'''	,			generator=_UpperCAmelCase	,			output_type='''np'''     )

									__a :						Tuple			       =  output.images[0]

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

									assert_mean_pixel_difference(_UpperCAmelCase	,			_UpperCAmelCase     )







					def  _lowerCamelCase		(						self     ):
									__a :						Any			       =  load_image(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''     )

									__a :						List[str]			       =  load_numpy(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_h_img2img_anime_turtle_fp16.npy'''     )

									__a :						Dict			       =  StableUnCLIPImgaImgPipeline.from_pretrained(
									    '''fusing/stable-unclip-2-1-h-img2img'''	,			torch_dtype=torch.floataa     )
									pipe.to(_UpperCAmelCase     )
									pipe.set_progress_bar_config(disable=_UpperCAmelCase     )
									# stable unclip will oom when integration tests are run on a V100,
									# so turn on memory savings
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									__a :						List[str]			       =  torch.Generator(device='''cpu'''     ).manual_seed(0     )
									__a :						Optional[Any]			       =  pipe(_UpperCAmelCase	,			'''anime turle'''	,			generator=_UpperCAmelCase	,			output_type='''np'''     )

									__a :						int			       =  output.images[0]

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

									assert_mean_pixel_difference(_UpperCAmelCase	,			_UpperCAmelCase     )







					def  _lowerCamelCase		(						self     ):
									__a :						Tuple			       =  load_image(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'''     )

									torch.cuda.empty_cache()
									torch.cuda.reset_max_memory_allocated()
									torch.cuda.reset_peak_memory_stats()

									__a :						Union[str, Any]			       =  StableUnCLIPImgaImgPipeline.from_pretrained(
									    '''fusing/stable-unclip-2-1-h-img2img'''	,			torch_dtype=torch.floataa     )
									__a :						Optional[Any]			       =  pipe.to(_UpperCAmelCase     )
									pipe.set_progress_bar_config(disable=_UpperCAmelCase     )
									pipe.enable_attention_slicing()
									pipe.enable_sequential_cpu_offload()

									__a :						List[Any]			       =  pipe(
									    _UpperCAmelCase	,			'''anime turtle'''	,			num_inference_steps=2	,			output_type='''np'''	,			)

									__a :						Any			       =  torch.cuda.max_memory_allocated()
									# make sure that less than 7 GB is allocated
									assert mem_bytes < 7 * 10**9






"""simple docstring"""






def     __A      (			a_						:int   ,       a_						:float   ,       a_						:float)		->		float:
				return round(float(moles / volume) * nfactor)

def     __A      (			a_						:float   ,       a_						:float   ,       a_						:float)		->		float:
				return round(float((moles * 0.0_8_2_1 * temperature) / (volume)))

def     __A      (			a_						:float   ,       a_						:float   ,       a_						:float)		->		float:
				return round(float((moles * 0.0_8_2_1 * temperature) / (pressure)))

def     __A      (			a_						:float   ,       a_						:float   ,       a_						:float)		->		float:
				return round(float((pressure * volume) / (0.0_8_2_1 * moles)))


if __name__ == "__main__":
		import doctest

		doctest.testmod()