
"""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_		(				A__				,				A__				,				A__				,				unittest.TestCase ):



							'''simple docstring'''





							_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    A__						( self				)       ->  str:
													__lowerCAmelCase     =  32
													__lowerCAmelCase     =  embedder_hidden_size

													# image encoding components

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

													torch.manual_seed(0				)
													__lowerCAmelCase     =  CLIPVisionModelWithProjection(
													    CLIPVisionConfig(
													        hidden_size=snake_case_    ,				projection_dim=snake_case_    ,				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     =  StableUnCLIPImageNormalizer(embedding_dim=snake_case_				)
													__lowerCAmelCase     =  DDPMScheduler(beta_schedule="""squaredcos_cap_v2"""				)

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

													torch.manual_seed(0				)
													__lowerCAmelCase     =  CLIPTextModel(
													    CLIPTextConfig(
													        bos_token_id=0    ,				eos_token_id=2    ,				hidden_size=snake_case_    ,				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     =  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=snake_case_    ,				layers_per_block=1    ,				upcast_attention=snake_case_    ,				use_linear_projection=snake_case_    ,				)

													torch.manual_seed(0				)
													__lowerCAmelCase     =  DDIMScheduler(
													    beta_schedule="""scaled_linear"""    ,				beta_start=0.00_085    ,				beta_end=0.012    ,				prediction_type="""v_prediction"""    ,				set_alpha_to_one=snake_case_    ,				steps_offset=1    ,				)

													torch.manual_seed(0				)
													__lowerCAmelCase     =  AutoencoderKL()

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

													__lowerCAmelCase     =  floats_tensor((1, 3, 32, 32)    ,				rng=random.Random(snake_case_				)				).to(snake_case_				)

													if pil_image:
																			__lowerCAmelCase     =  input_image * 0.5 + 0.5
																			__lowerCAmelCase     =  input_image.clamp(0    ,				1				)
																			__lowerCAmelCase     =  input_image.cpu().permute(0    ,				2    ,				3    ,				1				).float().numpy()
																			__lowerCAmelCase     =  DiffusionPipeline.numpy_to_pil(snake_case_				)[0]

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




							@skip_mps
							def    A__						( self				)       ->  Optional[int]:
													__lowerCAmelCase     =  """cpu"""  # ensure determinism for the device-dependent torch.Generator
													__lowerCAmelCase     =  self.get_dummy_components()
													__lowerCAmelCase     =  StableUnCLIPImgaImgPipeline(**snake_case_				)
													__lowerCAmelCase     =  sd_pipe.to(snake_case_				)
													sd_pipe.set_progress_bar_config(disable=snake_case_				)

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

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

													self._test_attention_slicing_forward_pass(test_max_difference=snake_case_				)




							def    A__						( self				)       ->  List[Any]:
													__lowerCAmelCase     =  torch_device in ["""cpu""", """mps"""]

													self._test_inference_batch_single_identical(test_max_difference=snake_case_				)




							@unittest.skipIf(
							    torch_device != """cuda""" or not is_xformers_available()    ,				reason="""XFormers attention is only available with CUDA and `xformers` installed"""    ,				)
							def    A__						( self				)       ->  Dict:
													self._test_xformers_attention_forwardGenerator_pass(test_max_difference=snake_case_				)
@slow
@require_torch_gpu
class       lowerCAmelCase_		(				unittest.TestCase ):



							'''simple docstring'''





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




							def    A__						( self				)       ->  str:
													__lowerCAmelCase     =  load_image(
													    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""				)

													__lowerCAmelCase     =  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     =  StableUnCLIPImgaImgPipeline.from_pretrained(
													    """fusing/stable-unclip-2-1-l-img2img"""    ,				torch_dtype=torch.floataa				)
													pipe.to(snake_case_				)
													pipe.set_progress_bar_config(disable=snake_case_				)
													# 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     =  torch.Generator(device="""cpu"""				).manual_seed(0				)
													__lowerCAmelCase     =  pipe(snake_case_    ,				"""anime turle"""    ,				generator=snake_case_    ,				output_type="""np"""				)

													__lowerCAmelCase     =  output.images[0]

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

													assert_mean_pixel_difference(snake_case_    ,				snake_case_				)




							def    A__						( self				)       ->  Any:
													__lowerCAmelCase     =  load_image(
													    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/turtle.png"""				)

													__lowerCAmelCase     =  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     =  StableUnCLIPImgaImgPipeline.from_pretrained(
													    """fusing/stable-unclip-2-1-h-img2img"""    ,				torch_dtype=torch.floataa				)
													pipe.to(snake_case_				)
													pipe.set_progress_bar_config(disable=snake_case_				)
													# 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     =  torch.Generator(device="""cpu"""				).manual_seed(0				)
													__lowerCAmelCase     =  pipe(snake_case_    ,				"""anime turle"""    ,				generator=snake_case_    ,				output_type="""np"""				)

													__lowerCAmelCase     =  output.images[0]

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

													assert_mean_pixel_difference(snake_case_    ,				snake_case_				)




							def    A__						( self				)       ->  int:
													__lowerCAmelCase     =  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     =  StableUnCLIPImgaImgPipeline.from_pretrained(
													    """fusing/stable-unclip-2-1-h-img2img"""    ,				torch_dtype=torch.floataa				)
													__lowerCAmelCase     =  pipe.to(snake_case_				)
													pipe.set_progress_bar_config(disable=snake_case_				)
													pipe.enable_attention_slicing()
													pipe.enable_sequential_cpu_offload()

													__lowerCAmelCase     =  pipe(
													    snake_case_    ,				"""anime turtle"""    ,				num_inference_steps=2    ,				output_type="""np"""    ,				)

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


"""simple docstring"""
import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
			from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
			import torch_xla.core.xla_model as xm

def  lowercase     (_lowerCAmelCase						):
						if is_torch_version("""<""" ,    """2.0.0"""						) or not hasattr(_lowerCAmelCase ,    """_dynamo"""						):
												return False
						return isinstance(_lowerCAmelCase ,    torch._dynamo.eval_frame.OptimizedModule						)

def  lowercase     (_lowerCAmelCase ,    _lowerCAmelCase = True						):
						__lowerCAmelCase     =  (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

						__lowerCAmelCase     =  is_compiled_module(_lowerCAmelCase						)
						if is_compiled:
												__lowerCAmelCase     =  model
												__lowerCAmelCase     =  model._orig_mod

						if is_deepspeed_available():
												options += (DeepSpeedEngine,)

						while isinstance(_lowerCAmelCase ,    _lowerCAmelCase						):
												__lowerCAmelCase     =  model.module

						if not keep_fpaa_wrapper:
												__lowerCAmelCase     =  getattr(_lowerCAmelCase ,    """forward"""						)
												__lowerCAmelCase     =  model.__dict__.pop("""_original_forward""" ,    _lowerCAmelCase						)
												if original_forward is not None:
																		while hasattr(_lowerCAmelCase ,    """__wrapped__"""						):
																								__lowerCAmelCase     =  forward.__wrapped__
																								if forward == original_forward:
																														break
																		__lowerCAmelCase     =  forward
												if getattr(_lowerCAmelCase ,    """_converted_to_transformer_engine""" ,    _lowerCAmelCase						):
																		convert_model(_lowerCAmelCase ,    to_transformer_engine=_lowerCAmelCase						)

						if is_compiled:
												__lowerCAmelCase     =  model
												__lowerCAmelCase     =  compiled_model

						return model

def  lowercase     ():
						PartialState().wait_for_everyone()

def  lowercase     (_lowerCAmelCase ,    _lowerCAmelCase						):
						if PartialState().distributed_type == DistributedType.TPU:
												xm.save(_lowerCAmelCase ,    _lowerCAmelCase						)
						elif PartialState().local_process_index == 0:
												torch.save(_lowerCAmelCase ,    _lowerCAmelCase						)

@contextmanager
def  lowercase     (**_lowerCAmelCase						):
						for key, value in kwargs.items():
												__lowerCAmelCase     =  str(_lowerCAmelCase						)

						yield

						for key in kwargs:
												if key.upper() in os.environ:
																		del os.environ[key.upper()]

def  lowercase     (_lowerCAmelCase						):
						if not hasattr(_lowerCAmelCase ,    """__qualname__"""						) and not hasattr(_lowerCAmelCase ,    """__name__"""						):
												__lowerCAmelCase     =  getattr(_lowerCAmelCase ,    """__class__""" ,    _lowerCAmelCase						)
						if hasattr(_lowerCAmelCase ,    """__qualname__"""						):
												return obj.__qualname__
						if hasattr(_lowerCAmelCase ,    """__name__"""						):
												return obj.__name__
						return str(_lowerCAmelCase						)

def  lowercase     (_lowerCAmelCase ,    _lowerCAmelCase						):
						for key, value in source.items():
												if isinstance(_lowerCAmelCase ,    _lowerCAmelCase						):
																		__lowerCAmelCase     =  destination.setdefault(_lowerCAmelCase ,    {}						)
																		merge_dicts(_lowerCAmelCase ,    _lowerCAmelCase						)
												else:
																		__lowerCAmelCase     =  value

						return destination



def  lowercase     (_lowerCAmelCase = None						):
						if port is None:
												__lowerCAmelCase     =  2_9500
						with socket.socket(socket.AF_INET ,    socket.SOCK_STREAM						) as s:
												return s.connect_ex(("""localhost""", port)						) == 0
