





'''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 		lowerCamelCase  (      __UpperCAmelCase      ,   __UpperCAmelCase      ,   __UpperCAmelCase      ,   unittest.TestCase       ):
	snake_case_      =   StableUnCLIPImgaImgPipeline
	snake_case_      =   TEXT_GUIDED_IMAGE_VARIATION_PARAMS
	snake_case_      =   TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	snake_case_      =   frozenset(
	    []       )  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
	snake_case_      =   frozenset([]       )



	def   _lowerCamelCase    (   self	):
			lowerCAmelCase     :    List[Any]						=   32
			lowerCAmelCase     :    Tuple						=   embedder_hidden_size

			# image encoding components

			lowerCAmelCase     :    Union[str, Any]						=   CLIPImageProcessor(crop_size=32		,      size=32	)

			torch.manual_seed(0	)
			lowerCAmelCase     :    List[Any]						=   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	)
			lowerCAmelCase     :    List[Any]						=   StableUnCLIPImageNormalizer(embedding_dim=lowerCAmelCase_	)
			lowerCAmelCase     :    Dict						=   DDPMScheduler(beta_schedule="squaredcos_cap_v2"	)

			torch.manual_seed(0	)
			lowerCAmelCase     :    Optional[Any]						=   CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"	)

			torch.manual_seed(0	)
			lowerCAmelCase     :    List[Any]						=   CLIPTextModel(
			    CLIPTextConfig(
			        bos_token_id=0		,      eos_token_id=2		,      hidden_size=lowerCAmelCase_		,      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=1_000		,      )	)

			torch.manual_seed(0	)
			lowerCAmelCase     :    Optional[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=lowerCAmelCase_		,      layers_per_block=1		,      upcast_attention=lowerCAmelCase_		,      use_linear_projection=lowerCAmelCase_		,      )

			torch.manual_seed(0	)
			lowerCAmelCase     :    str						=   DDIMScheduler(
			    beta_schedule="scaled_linear"		,      beta_start=0.00085		,      beta_end=0.012		,      prediction_type="v_prediction"		,      set_alpha_to_one=lowerCAmelCase_		,      steps_offset=1		,      )

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

			lowerCAmelCase     :    List[str]						=   {
			    # 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		,      a_		,      a_=0		,      a_=True	):







			if str(lowerCAmelCase_	).startswith("mps"	):
					lowerCAmelCase     :    List[Any]						=   torch.manual_seed(lowerCAmelCase_	)
			else:
					lowerCAmelCase     :    Tuple						=   torch.Generator(device=lowerCAmelCase_	).manual_seed(lowerCAmelCase_	)

			lowerCAmelCase     :    str						=   floats_tensor((1, 3, 32, 32)		,      rng=random.Random(lowerCAmelCase_	)	).to(lowerCAmelCase_	)

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

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

			assert image.shape == (1, 32, 32, 3)
			lowerCAmelCase     :    Optional[Any]						=   np.array([0.3872, 0.7224, 0.5601, 0.4741, 0.6872, 0.5814, 0.4636, 0.3867, 0.5078]	)

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



	def   _lowerCamelCase    (   self	):
			lowerCAmelCase     :    int						=   torch_device in ["cpu", "mps"]

			self._test_attention_slicing_forward_pass(test_max_difference=lowerCAmelCase_	)



	def   _lowerCamelCase    (   self	):
			lowerCAmelCase     :    str						=   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   _lowerCamelCase    (   self	):
			self._test_xformers_attention_forwardGenerator_pass(test_max_difference=lowerCAmelCase_	)






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



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



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

			lowerCAmelCase     :    List[Any]						=   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"	)

			lowerCAmelCase     :    Dict						=   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()

			lowerCAmelCase     :    List[str]						=   torch.Generator(device="cpu"	).manual_seed(0	)
			lowerCAmelCase     :    Dict						=   pipe(lowerCAmelCase_		,      "anime turle"		,      generator=lowerCAmelCase_		,      output_type="np"	)

			lowerCAmelCase     :    Optional[Any]						=   output.images[0]

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

			assert_mean_pixel_difference(lowerCAmelCase_		,      lowerCAmelCase_	)



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

			lowerCAmelCase     :    List[Any]						=   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"	)

			lowerCAmelCase     :    Optional[Any]						=   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()

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

			lowerCAmelCase     :    Dict						=   output.images[0]

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

			assert_mean_pixel_difference(lowerCAmelCase_		,      lowerCAmelCase_	)





	def   _lowerCamelCase    (   self	):
			lowerCAmelCase     :    List[str]						=   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()

			lowerCAmelCase     :    Union[str, Any]						=   StableUnCLIPImgaImgPipeline.from_pretrained(
			    "fusing/stable-unclip-2-1-h-img2img"		,      torch_dtype=torch.floataa	)
			lowerCAmelCase     :    Tuple						=   pipe.to(lowerCAmelCase_	)
			pipe.set_progress_bar_config(disable=lowerCAmelCase_	)
			pipe.enable_attention_slicing()
			pipe.enable_sequential_cpu_offload()

			lowerCAmelCase     :    Optional[Any]						=   pipe(
			    lowerCAmelCase_		,      "anime turtle"		,      num_inference_steps=2		,      output_type="np"		,      )

			lowerCAmelCase     :    Optional[int]						=   torch.cuda.max_memory_allocated()
			# make sure that less than 7 GB is allocated
			assert mem_bytes < 7 * 10**9










'''simple docstring'''






from collections import namedtuple

lowerCAmelCase       		=      namedtuple("""from_to""", """from_ to""")

lowerCAmelCase       		=      {
    """cubicmeter""": from_to(1, 1),
    """litre""": from_to(0.001, 10_00),
    """kilolitre""": from_to(1, 1),
    """gallon""": from_to(0.00454, 264.172),
    """cubicyard""": from_to(0.76455, 1.30795),
    """cubicfoot""": from_to(0.028, 35.3147),
    """cup""": from_to(0.000236588, 4226.75),
}






def      __A						(							a_						:  float      ,a_						:  str      ,a_						:  str	):
		if from_type not in METRIC_CONVERSION:
				raise ValueError(
				    f'''Invalid \'from_type\' value: {from_type!r}  Supported values are:\n'''
				    + ", ".join(a_	)	)
		if to_type not in METRIC_CONVERSION:
				raise ValueError(
				    f'''Invalid \'to_type\' value: {to_type!r}.  Supported values are:\n'''
				    + ", ".join(a_	)	)
		return value * METRIC_CONVERSION[from_type].from_ * METRIC_CONVERSION[to_type].to


if __name__ == "__main__":
					import doctest

					doctest.testmod()



