





'''simple docstring'''



import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()






class 							__lowercase       (      lowerCAmelCase__ ,      lowerCAmelCase__ ,      lowerCAmelCase__ ,      unittest.TestCase						):







	'''simple docstring'''

	a    :				Optional[Any]											=       StableUnCLIPPipeline
	a    :				Any											=       TEXT_TO_IMAGE_PARAMS
	a    :				Optional[int]											=       TEXT_TO_IMAGE_BATCH_PARAMS
	a    :				str											=       TEXT_TO_IMAGE_IMAGE_PARAMS
	a    :				int											=       TEXT_TO_IMAGE_IMAGE_PARAMS

	# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
	a    :				str											=       False


	def 						_UpperCAmelCase     (self )							->							Optional[Any]:







				'''simple docstring'''





				__lowercase										=							32
				__lowercase										=							embedder_hidden_size

				# prior components

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

				torch.manual_seed(0 )
				__lowercase										=							CLIPTextModelWithProjection(
				    CLIPTextConfig(
				        bos_token_id=0     ,eos_token_id=2     ,hidden_size=_lowerCamelCase     ,projection_dim=_lowerCamelCase     ,intermediate_size=37     ,layer_norm_eps=1E-0_5     ,num_attention_heads=4     ,num_hidden_layers=5     ,pad_token_id=1     ,vocab_size=1000     ,) )

				torch.manual_seed(0 )
				__lowercase										=							PriorTransformer(
				    num_attention_heads=2     ,attention_head_dim=12     ,embedding_dim=_lowerCamelCase     ,num_layers=1     ,)

				torch.manual_seed(0 )
				__lowercase										=							DDPMScheduler(
				    variance_type='''fixed_small_log'''     ,prediction_type='''sample'''     ,num_train_timesteps=1000     ,clip_sample=_lowerCamelCase     ,clip_sample_range=5.0     ,beta_schedule='''squaredcos_cap_v2'''     ,)

				# regular denoising components

				torch.manual_seed(0 )
				__lowercase										=							StableUnCLIPImageNormalizer(embedding_dim=_lowerCamelCase )
				__lowercase										=							DDPMScheduler(beta_schedule='''squaredcos_cap_v2''' )

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

				torch.manual_seed(0 )
				__lowercase										=							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=1000     ,) )

				torch.manual_seed(0 )
				__lowercase										=							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 )
				__lowercase										=							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=_lowerCamelCase     ,steps_offset=1     ,)

				torch.manual_seed(0 )
				__lowercase										=							AutoencoderKL()

				__lowercase										=							{
				    # prior components
				    '''prior_tokenizer''': prior_tokenizer,
				    '''prior_text_encoder''': prior_text_encoder,
				    '''prior''': prior,
				    '''prior_scheduler''': prior_scheduler,
				    # image noising components
				    '''image_normalizer''': image_normalizer,
				    '''image_noising_scheduler''': image_noising_scheduler,
				    # regular denoising components
				    '''tokenizer''': tokenizer,
				    '''text_encoder''': text_encoder,
				    '''unet''': unet,
				    '''scheduler''': scheduler,
				    '''vae''': vae,
				}

				return components


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase     ,_lowerCamelCase=0 )							->							str:







				'''simple docstring'''





				if str(_lowerCamelCase ).startswith('''mps''' ):
							__lowercase										=							torch.manual_seed(_lowerCamelCase )
				else:
							__lowercase										=							torch.Generator(device=_lowerCamelCase ).manual_seed(_lowerCamelCase )
				__lowercase										=							{
				    '''prompt''': '''A painting of a squirrel eating a burger''',
				    '''generator''': generator,
				    '''num_inference_steps''': 2,
				    '''prior_num_inference_steps''': 2,
				    '''output_type''': '''numpy''',
				}
				return inputs


	def 						_UpperCAmelCase     (self )							->							int:







				'''simple docstring'''





				__lowercase										=							torch_device == '''cpu'''

				self._test_attention_slicing_forward_pass(test_max_difference=_lowerCamelCase )


	def 						_UpperCAmelCase     (self )							->							Tuple:







				'''simple docstring'''





				__lowercase										=							torch_device in ['''cpu''', '''mps''']

				self._test_inference_batch_single_identical(test_max_difference=_lowerCamelCase )






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







	'''simple docstring'''

	def 						_UpperCAmelCase     (self )							->							Dict:







				'''simple docstring'''





				super().tearDown()
				gc.collect()
				torch.cuda.empty_cache()


	def 						_UpperCAmelCase     (self )							->							str:







				'''simple docstring'''





				__lowercase										=							load_numpy(
				    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy''' )

				__lowercase										=							StableUnCLIPPipeline.from_pretrained('''fusing/stable-unclip-2-1-l'''     ,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()

				__lowercase										=							torch.Generator(device='''cpu''' ).manual_seed(0 )
				__lowercase										=							pipe('''anime turle'''     ,generator=_lowerCamelCase     ,output_type='''np''' )

				__lowercase										=							output.images[0]

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

				assert_mean_pixel_difference(_lowerCamelCase     ,_lowerCamelCase )


	def 						_UpperCAmelCase     (self )							->							str:







				'''simple docstring'''





				torch.cuda.empty_cache()
				torch.cuda.reset_max_memory_allocated()
				torch.cuda.reset_peak_memory_stats()

				__lowercase										=							StableUnCLIPPipeline.from_pretrained('''fusing/stable-unclip-2-1-l'''     ,torch_dtype=torch.floataa )
				__lowercase										=							pipe.to(_lowerCamelCase )
				pipe.set_progress_bar_config(disable=_lowerCamelCase )
				pipe.enable_attention_slicing()
				pipe.enable_sequential_cpu_offload()

				__lowercase										=							pipe(
				    '''anime turtle'''     ,prior_num_inference_steps=2     ,num_inference_steps=2     ,output_type='''np'''     ,)

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












'''simple docstring'''



from __future__ import annotations

from typing import Any






class 							__lowercase       :







	'''simple docstring'''

	def __init__(self     ,_lowerCamelCase )							->							None:







				'''simple docstring'''





				__lowercase										=							num_of_nodes
				__lowercase										=							[]
				__lowercase										=							{}


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase     ,_lowerCamelCase     ,_lowerCamelCase )							->							None:







				'''simple docstring'''





				self.m_edges.append([u_node, v_node, weight] )


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							int:







				'''simple docstring'''





				if self.m_component[u_node] == u_node:
							return u_node
				return self.find_component(self.m_component[u_node] )


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase )							->							None:







				'''simple docstring'''





				if self.m_component[u_node] != u_node:
							for k in self.m_component:
										__lowercase										=							self.find_component(_lowerCamelCase )


	def 						_UpperCAmelCase     (self     ,_lowerCamelCase     ,_lowerCamelCase     ,_lowerCamelCase )							->							None:







				'''simple docstring'''





				if component_size[u_node] <= component_size[v_node]:
							__lowercase										=							v_node
							component_size[v_node] += component_size[u_node]
							self.set_component(_lowerCamelCase )

				elif component_size[u_node] >= component_size[v_node]:
							__lowercase										=							self.find_component(_lowerCamelCase )
							component_size[u_node] += component_size[v_node]
							self.set_component(_lowerCamelCase )


	def 						_UpperCAmelCase     (self )							->							None:







				'''simple docstring'''





				__lowercase										=							[]
				__lowercase										=							0

				__lowercase										=							[-1] * self.m_num_of_nodes

				# A list of components (initialized to all of the nodes)
				for node in range(self.m_num_of_nodes ):
							self.m_component.update({node: node} )
							component_size.append(1 )

				__lowercase										=							self.m_num_of_nodes

				while num_of_components > 1:
							for edge in self.m_edges:
										__lowercase						,		__lowercase						,		__lowercase										=							edge

										__lowercase										=							self.m_component[u]
										__lowercase										=							self.m_component[v]

										if u_component != v_component:

													for component in (u_component, v_component):
																if (
																    minimum_weight_edge[component] == -1
																    or minimum_weight_edge[component][2] > w
																):
																			__lowercase										=							[u, v, w]

							for edge in minimum_weight_edge:
										if isinstance(_lowerCamelCase     ,_lowerCamelCase ):
													__lowercase						,		__lowercase						,		__lowercase										=							edge

													__lowercase										=							self.m_component[u]
													__lowercase										=							self.m_component[v]

													if u_component != v_component:
																mst_weight += w
																self.union(_lowerCamelCase     ,_lowerCamelCase     ,_lowerCamelCase )
																print(f"Added edge [{u} - {v}]\nAdded weight: {w}\n" )
																num_of_components -= 1

							__lowercase										=							[-1] * self.m_num_of_nodes
				print(f"The total weight of the minimal spanning tree is: {mst_weight}" )





def    _lowerCAmelCase						(				):
			pass


if __name__ == "__main__":
	import doctest

	doctest.testmod()





