


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"""


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

       def 					__snake_case      (							self :						Any       )			->					List[Any]:
              __snake_case	:    List[Any]											=     32
              __snake_case	:    Optional[Any]											=     embedder_hidden_size

              # image encoding components

              __snake_case	:    List[str]											=     CLIPImageProcessor(crop_size=32					,			size=32       )

              torch.manual_seed(0       )
              __snake_case	:    str											=     CLIPVisionModelWithProjection(
                  CLIPVisionConfig(
                      hidden_size=lowerCamelCase					,			projection_dim=lowerCamelCase					,			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	:    int											=     StableUnCLIPImageNormalizer(embedding_dim=lowerCamelCase       )
              __snake_case	:    Union[str, Any]											=     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	:    Tuple											=     CLIPTextModel(
                  CLIPTextConfig(
                      bos_token_id=0					,			eos_token_id=2					,			hidden_size=lowerCamelCase					,			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=1000					,			)       )

              torch.manual_seed(0       )
              __snake_case	:    int											=     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=lowerCamelCase					,			layers_per_block=1					,			upcast_attention=lowerCamelCase					,			use_linear_projection=lowerCamelCase					,			)

              torch.manual_seed(0       )
              __snake_case	:    int											=     DDIMScheduler(
                  beta_schedule="scaled_linear"					,			beta_start=0.0_00_85					,			beta_end=0.0_12					,			prediction_type="v_prediction"					,			set_alpha_to_one=lowerCamelCase					,			steps_offset=1					,			)

              torch.manual_seed(0       )
              __snake_case	:    str											=     AutoencoderKL()

              __snake_case	:    int											=     {
                  # 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 :						Any					,			lowerCamelCase :						List[str]					,			lowerCamelCase :						List[Any]=0					,			lowerCamelCase :						int=True       )			->					Optional[Any]:

              if str(lowerCamelCase       ).startswith("mps"       ):
                     __snake_case	:    List[str]											=     torch.manual_seed(lowerCamelCase       )
              else:
                     __snake_case	:    Any											=     torch.Generator(device=lowerCamelCase       ).manual_seed(lowerCamelCase       )

              __snake_case	:    Any											=     floats_tensor((1, 3, 32, 32)					,			rng=random.Random(lowerCamelCase       )       ).to(lowerCamelCase       )

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

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

              assert image.shape == (1, 32, 32, 3)
              __snake_case	:    str											=     np.array([0.38_72, 0.72_24, 0.56_01, 0.47_41, 0.68_72, 0.58_14, 0.46_36, 0.38_67, 0.50_78]       )

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

       def 					__snake_case      (							self :						Dict       )			->					Tuple:
              __snake_case	:    Union[str, Any]											=     torch_device in ['''cpu''', '''mps''']

              self._test_attention_slicing_forward_pass(test_max_difference=lowerCamelCase       )

       def 					__snake_case      (							self :						List[Any]       )			->					Optional[int]:
              __snake_case	:    int											=     torch_device in ['''cpu''', '''mps''']

              self._test_inference_batch_single_identical(test_max_difference=lowerCamelCase       )






       @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[Any]       )			->					Union[str, Any]:
              self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowerCamelCase       )

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




       """simple docstring"""


       def 					__snake_case      (							self :						Optional[int]       )			->					List[Any]:
              super().tearDown()
              gc.collect()
              torch.cuda.empty_cache()

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

              __snake_case	:    Dict											=     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	:    Union[str, Any]											=     StableUnCLIPImgaImgPipeline.from_pretrained(
                  "fusing/stable-unclip-2-1-l-img2img"					,			torch_dtype=torch.floataa       )
              pipe.to(lowerCamelCase       )
              pipe.set_progress_bar_config(disable=lowerCamelCase       )
              # 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	:    Tuple											=     torch.Generator(device="cpu"       ).manual_seed(0       )
              __snake_case	:    Dict											=     pipe(lowerCamelCase					,			"anime turle"					,			generator=lowerCamelCase					,			output_type="np"       )

              __snake_case	:    int											=     output.images[0]

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

              assert_mean_pixel_difference(lowerCamelCase					,			lowerCamelCase       )

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

              __snake_case	:    Optional[int]											=     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	:    Tuple											=     StableUnCLIPImgaImgPipeline.from_pretrained(
                  "fusing/stable-unclip-2-1-h-img2img"					,			torch_dtype=torch.floataa       )
              pipe.to(lowerCamelCase       )
              pipe.set_progress_bar_config(disable=lowerCamelCase       )
              # 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	:    Any											=     torch.Generator(device="cpu"       ).manual_seed(0       )
              __snake_case	:    Optional[Any]											=     pipe(lowerCamelCase					,			"anime turle"					,			generator=lowerCamelCase					,			output_type="np"       )

              __snake_case	:    str											=     output.images[0]

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

              assert_mean_pixel_difference(lowerCamelCase					,			lowerCamelCase       )






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

              __snake_case	:    List[str]											=     pipe(
                  lowerCamelCase					,			"anime turtle"					,			num_inference_steps=2					,			output_type="np"					,			)

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





from maths.prime_check import is_prime







def        lowerCAmelCase_  (      __lowerCamelCase			):
							if not isinstance(__lowerCamelCase							,			__lowerCamelCase			):
														__snake_case	:    List[str]											=     F'Input value of [number={number}] must be an integer'
														raise TypeError(__lowerCamelCase			)
							if is_prime(__lowerCamelCase			) and is_prime(number + 2			):
														return number + 2
							else:
														return -1


if __name__ == "__main__":
			import doctest

			doctest.testmod()

