"""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     UpperCamelCase__(			_UpperCamelCase		,						_UpperCamelCase		,						_UpperCamelCase		,						unittest.TestCase ):
						lowerCAmelCase__  :						str													=		StableUnCLIPImgaImgPipeline
						lowerCAmelCase__  :						List[Any]													=		TEXT_GUIDED_IMAGE_VARIATION_PARAMS
						lowerCAmelCase__  :						Any													=		TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
						lowerCAmelCase__  :						Optional[Any]													=		frozenset(
						    [] )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
						lowerCAmelCase__  :						List[Any]													=		frozenset([] )





						def 				snake_case__	(       self     )					->				Union[str, Any]:
													A__	     =							32
													A__	     =							embedder_hidden_size

													# image encoding components

													A__	     =							CLIPImageProcessor(crop_size=32		,size=32     )

													torch.manual_seed(0     )
													A__	     =							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__	     =							StableUnCLIPImageNormalizer(embedding_dim=_UpperCAmelCase     )
													A__	     =							DDPMScheduler(beta_schedule='squaredcos_cap_v2'     )

													torch.manual_seed(0     )
													A__	     =							CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

													torch.manual_seed(0     )
													A__	     =							CLIPTextModel(
													    CLIPTextConfig(
													        bos_token_id=0		,eos_token_id=2		,hidden_size=_UpperCAmelCase		,projection_dim=32		,intermediate_size=37		,layer_norm_eps=1e-05		,num_attention_heads=4		,num_hidden_layers=5		,pad_token_id=1		,vocab_size=10_00		,)     )

													torch.manual_seed(0     )
													A__	     =							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__	     =							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__	     =							AutoencoderKL()

													A__	     =							{
													    # 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 				snake_case__	(       self		,__UpperCAmelCase		,__UpperCAmelCase=0		,__UpperCAmelCase=True     )					->				List[Any]:
													if str(_UpperCAmelCase     ).startswith('mps'     ):
																				A__	     =							torch.manual_seed(_UpperCAmelCase     )
													else:
																				A__	     =							torch.Generator(device=_UpperCAmelCase     ).manual_seed(_UpperCAmelCase     )

													A__	     =							floats_tensor((1, 3, 32, 32)		,rng=random.Random(_UpperCAmelCase     )     ).to(_UpperCAmelCase     )

													if pil_image:
																				A__	     =							input_image * 0.5 + 0.5
																				A__	     =							input_image.clamp(0		,1     )
																				A__	     =							input_image.cpu().permute(0		,2		,3		,1     ).float().numpy()
																				A__	     =							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 				snake_case__	(       self     )					->				Optional[Any]:
													A__	     =							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
													A__	     =							self.get_dummy_components()
													A__	     =							StableUnCLIPImgaImgPipeline(**_UpperCAmelCase     )
													A__	     =							sd_pipe.to(_UpperCAmelCase     )
													sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

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

													assert image.shape == (1, 32, 32, 3)
													A__	     =							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 				snake_case__	(       self     )					->				int:
													A__	     =							torch_device in ['''cpu''', '''mps''']

													self._test_attention_slicing_forward_pass(test_max_difference=_UpperCAmelCase     )





						def 				snake_case__	(       self     )					->				Optional[Any]:
													A__	     =							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 				snake_case__	(       self     )					->				Dict:
													self._test_xformers_attention_forwardGenerator_pass(test_max_difference=_UpperCAmelCase     )






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





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





						def 				snake_case__	(       self     )					->				List[str]:
													A__	     =							load_image(
													    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'     )

													A__	     =							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__	     =							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__	     =							torch.Generator(device='cpu'     ).manual_seed(0     )
													A__	     =							pipe(_UpperCAmelCase		,'anime turle'		,generator=_UpperCAmelCase		,output_type='np'     )

													A__	     =							output.images[0]

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

													assert_mean_pixel_difference(_UpperCAmelCase		,_UpperCAmelCase     )





						def 				snake_case__	(       self     )					->				Optional[int]:
													A__	     =							load_image(
													    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png'     )

													A__	     =							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__	     =							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__	     =							torch.Generator(device='cpu'     ).manual_seed(0     )
													A__	     =							pipe(_UpperCAmelCase		,'anime turle'		,generator=_UpperCAmelCase		,output_type='np'     )

													A__	     =							output.images[0]

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

													assert_mean_pixel_difference(_UpperCAmelCase		,_UpperCAmelCase     )





						def 				snake_case__	(       self     )					->				Union[str, Any]:
													A__	     =							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__	     =							StableUnCLIPImgaImgPipeline.from_pretrained(
													    'fusing/stable-unclip-2-1-h-img2img'		,torch_dtype=torch.floataa     )
													A__	     =							pipe.to(_UpperCAmelCase     )
													pipe.set_progress_bar_config(disable=_UpperCAmelCase     )
													pipe.enable_attention_slicing()
													pipe.enable_sequential_cpu_offload()

													A__	     =							pipe(
													    _UpperCAmelCase		,'anime turtle'		,num_inference_steps=2		,output_type='np'		,)

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






"""simple docstring"""




import os


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_doctest_list.py
__lowerCamelCase											=  "."


if __name__ == "__main__":
				__lowerCamelCase											=  os.path.join(REPO_PATH, "utils/documentation_tests.txt")
				__lowerCamelCase											=  []
				__lowerCamelCase											=  []
				with open(doctest_file_path) as fp:
								for line in fp:
												__lowerCamelCase											=  line.strip()
												__lowerCamelCase											=  os.path.join(REPO_PATH, line)
												if not (os.path.isfile(path) or os.path.isdir(path)):
																non_existent_paths.append(line)
												all_paths.append(path)
				if len(non_existent_paths) > 0:
								__lowerCamelCase											=  "\n".join(non_existent_paths)
								raise ValueError(F'''`utils/documentation_tests.txt` contains non-existent paths:\n{non_existent_paths}''')
				if all_paths != sorted(all_paths):
								raise ValueError("Files in `utils/documentation_tests.txt` are not in alphabetical order.")





