




'''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 					_a (a__,		a__,		a__,		unittest.TestCase	):





     '''simple docstring'''



     lowerCAmelCase_  :     Optional[Any]          =					StableUnCLIPImgaImgPipeline
     lowerCAmelCase_  :     str          =					TEXT_GUIDED_IMAGE_VARIATION_PARAMS
     lowerCAmelCase_  :     int          =					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
     lowerCAmelCase_  :     Tuple          =					frozenset(
         []	)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
     lowerCAmelCase_  :     str          =					frozenset([]	)







     def 			snake_case_      (				self							)      ->				List[Any]:
          snake_case			:      List[str]        =			32
          snake_case			:      List[Any]        =			embedder_hidden_size

          # image encoding components

          snake_case			:      int        =			CLIPImageProcessor(crop_size=32							,size=32							)

          torch.manual_seed(0							)
          snake_case			:      Optional[int]        =			CLIPVisionModelWithProjection(
              CLIPVisionConfig(
                  hidden_size=__a							,projection_dim=__a							,num_hidden_layers=5							,num_attention_heads=4							,image_size=32							,intermediate_size=37							,patch_size=1							,)							)

          # regular denoising components

          torch.manual_seed(0							)
          snake_case			:      Any        =			StableUnCLIPImageNormalizer(embedding_dim=__a							)
          snake_case			:      List[str]        =			DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""							)

          torch.manual_seed(0							)
          snake_case			:      str        =			CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""							)

          torch.manual_seed(0							)
          snake_case			:      int        =			CLIPTextModel(
              CLIPTextConfig(
                  bos_token_id=0							,eos_token_id=2							,hidden_size=__a							,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=1_000							,)							)

          torch.manual_seed(0							)
          snake_case			:      List[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=__a							,layers_per_block=1							,upcast_attention=__a							,use_linear_projection=__a							,)

          torch.manual_seed(0							)
          snake_case			:      Dict        =			DDIMScheduler(
              beta_schedule="""scaled_linear"""							,beta_start=0.00_085							,beta_end=0.012							,prediction_type="""v_prediction"""							,set_alpha_to_one=__a							,steps_offset=1							,)

          torch.manual_seed(0							)
          snake_case			:      Optional[Any]        =			AutoencoderKL()

          snake_case			:      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 			snake_case_      (				self							,__a							,__a=0							,__a=True							)      ->				Optional[Any]:
          if str(__a							).startswith("""mps"""							):
               snake_case			:      int        =			torch.manual_seed(__a							)
          else:
               snake_case			:      Dict        =			torch.Generator(device=__a							).manual_seed(__a							)

          snake_case			:      Optional[Any]        =			floats_tensor((1, 3, 32, 32)							,rng=random.Random(__a							)							).to(__a							)

          if pil_image:
               snake_case			:      List[Any]        =			input_image * 0.5 + 0.5
               snake_case			:      Tuple        =			input_image.clamp(0							,1							)
               snake_case			:      str        =			input_image.cpu().permute(0							,2							,3							,1							).float().numpy()
               snake_case			:      int        =			DiffusionPipeline.numpy_to_pil(__a							)[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							)      ->				List[Any]:
          snake_case			:      Optional[Any]        =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
          snake_case			:      Tuple        =			self.get_dummy_components()
          snake_case			:      Tuple        =			StableUnCLIPImgaImgPipeline(**__a							)
          snake_case			:      Dict        =			sd_pipe.to(__a							)
          sd_pipe.set_progress_bar_config(disable=__a							)

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

          assert image.shape == (1, 32, 32, 3)
          snake_case			:      Optional[Any]        =			np.array([0.3_872, 0.7_224, 0.5_601, 0.4_741, 0.6_872, 0.5_814, 0.4_636, 0.3_867, 0.5_078]							)

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







     def 			snake_case_      (				self							)      ->				Optional[Any]:
          snake_case			:      List[str]        =			torch_device in ["""cpu""", """mps"""]

          self._test_attention_slicing_forward_pass(test_max_difference=__a							)







     def 			snake_case_      (				self							)      ->				Dict:
          snake_case			:      Optional[Any]        =			torch_device in ["""cpu""", """mps"""]

          self._test_inference_batch_single_identical(test_max_difference=__a							)







     @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							)      ->				Optional[int]:
          self._test_xformers_attention_forwardGenerator_pass(test_max_difference=__a							)







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





     '''simple docstring'''



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







     def 			snake_case_      (				self							)      ->				Union[str, Any]:
          snake_case			:      Tuple        =			load_image(
              """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""							)

          snake_case			:      str        =			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"""							)

          snake_case			:      Dict        =			StableUnCLIPImgaImgPipeline.from_pretrained(
              """fusing/stable-unclip-2-1-l-img2img"""							,torch_dtype=torch.floataa							)
          pipe.to(__a							)
          pipe.set_progress_bar_config(disable=__a							)
          # 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()

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

          snake_case			:      Any        =			output.images[0]

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

          assert_mean_pixel_difference(__a							,__a							)







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

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

          snake_case			:      Optional[Any]        =			StableUnCLIPImgaImgPipeline.from_pretrained(
              """fusing/stable-unclip-2-1-h-img2img"""							,torch_dtype=torch.floataa							)
          pipe.to(__a							)
          pipe.set_progress_bar_config(disable=__a							)
          # 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()

          snake_case			:      List[Any]        =			torch.Generator(device="""cpu"""							).manual_seed(0							)
          snake_case			:      Tuple        =			pipe(__a							,"""anime turle"""							,generator=__a							,output_type="""np"""							)

          snake_case			:      str        =			output.images[0]

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

          assert_mean_pixel_difference(__a							,__a							)







     def 			snake_case_      (				self							)      ->				List[str]:
          snake_case			:      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()

          snake_case			:      Dict        =			StableUnCLIPImgaImgPipeline.from_pretrained(
              """fusing/stable-unclip-2-1-h-img2img"""							,torch_dtype=torch.floataa							)
          snake_case			:      List[Any]        =			pipe.to(__a							)
          pipe.set_progress_bar_config(disable=__a							)
          pipe.enable_attention_slicing()
          pipe.enable_sequential_cpu_offload()

          snake_case			:      Dict        =			pipe(
              __a							,"""anime turtle"""							,num_inference_steps=2							,output_type="""np"""							,)

          snake_case			:      List[Any]        =			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
lowercase      :   List[str]      =       """."""


if __name__ == "__main__":
      lowercase      :   List[str]      =       os.path.join(REPO_PATH, """utils/documentation_tests.txt""")
      lowercase      :   Union[str, Any]      =       []
      lowercase      :   List[str]      =       []
      with open(doctest_file_path) as fp:
            for line in fp:
                  lowercase      :   List[str]      =       line.strip()
                  lowercase      :   Optional[Any]      =       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:
            lowercase      :   int      =       """\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.""")



