




'''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 			_snake_case		(     lowercase_						,						lowercase_						,						lowercase_						,						unittest.TestCase							):
			lowerCAmelCase_					:				Optional[int]							=   StableUnCLIPPipeline
			lowerCAmelCase_					:				int							=   TEXT_TO_IMAGE_PARAMS
			lowerCAmelCase_					:				str							=   TEXT_TO_IMAGE_BATCH_PARAMS
			lowerCAmelCase_					:				Tuple							=   TEXT_TO_IMAGE_IMAGE_PARAMS
			lowerCAmelCase_					:				int							=   TEXT_TO_IMAGE_IMAGE_PARAMS

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




			def        lowerCAmelCase__    ( self							)    ->    Dict:

										'''simple docstring'''




										snake_case_      =				32
										snake_case_      =				embedder_hidden_size

										# prior components

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

										torch.manual_seed(0							)
										snake_case_      =				CLIPTextModelWithProjection(
										    CLIPTextConfig(
										        bos_token_id=0    ,							eos_token_id=2    ,							hidden_size=a__    ,							projection_dim=a__    ,							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_      =				PriorTransformer(
										    num_attention_heads=2    ,							attention_head_dim=12    ,							embedding_dim=a__    ,							num_layers=1    ,							)

										torch.manual_seed(0							)
										snake_case_      =				DDPMScheduler(
										    variance_type="fixed_small_log"    ,							prediction_type="sample"    ,							num_train_timesteps=1_000    ,							clip_sample=a__    ,							clip_sample_range=5.0    ,							beta_schedule="squaredcos_cap_v2"    ,							)

										# regular denoising components

										torch.manual_seed(0							)
										snake_case_      =				StableUnCLIPImageNormalizer(embedding_dim=a__							)
										snake_case_      =				DDPMScheduler(beta_schedule="squaredcos_cap_v2"							)

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

										torch.manual_seed(0							)
										snake_case_      =				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_      =				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_      =				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=a__    ,							steps_offset=1    ,							)

										torch.manual_seed(0							)
										snake_case_      =				AutoencoderKL()

										snake_case_      =				{
										    # 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        lowerCAmelCase__    ( self    ,							a__    ,							a__=0							)    ->    List[Any]:

										'''simple docstring'''




										if str(a__							).startswith("mps"							):
																	snake_case_      =				torch.manual_seed(a__							)
										else:
																	snake_case_      =				torch.Generator(device=a__							).manual_seed(a__							)
										snake_case_      =				{
										    "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        lowerCAmelCase__    ( self							)    ->    List[str]:

										'''simple docstring'''




										snake_case_      =				torch_device == "cpu"

										self._test_attention_slicing_forward_pass(test_max_difference=a__							)




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

										'''simple docstring'''




										snake_case_      =				torch_device in ["cpu", "mps"]

										self._test_inference_batch_single_identical(test_max_difference=a__							)




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




			def        lowerCAmelCase__    ( self							)    ->    int:

										'''simple docstring'''




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




			def        lowerCAmelCase__    ( self							)    ->    str:

										'''simple docstring'''




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

										snake_case_      =				StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"    ,							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_      =				torch.Generator(device="cpu"							).manual_seed(0							)
										snake_case_      =				pipe("anime turle"    ,							generator=a__    ,							output_type="np"							)

										snake_case_      =				output.images[0]

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

										assert_mean_pixel_difference(a__    ,							a__							)




			def        lowerCAmelCase__    ( self							)    ->    List[Any]:

										'''simple docstring'''




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

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

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

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










'''simple docstring'''





from random import shuffle

import tensorflow as tf
from numpy import array


def 					UpperCamelCase_(     snake_case							:	Optional[int]							,    snake_case							:	Optional[int]					):

							'''simple docstring'''






							snake_case_      =				int(snake_case					)
							assert noofclusters < len(snake_case					)

							# Find out the dimensionality
							snake_case_      =				len(vectors[0]					)

							# Will help select random centroids from among the available vectors
							snake_case_      =				list(range(len(snake_case					)					)					)
							shuffle(snake_case					)

							# GRAPH OF COMPUTATION
							# We initialize a new graph and set it as the default during each run
							# of this algorithm. This ensures that as this function is called
							# multiple times, the default graph doesn't keep getting crowded with
							# unused ops and Variables from previous function calls.

							snake_case_      =				tf.Graph()

							with graph.as_default():
														# SESSION OF COMPUTATION

														snake_case_      =				tf.Session()

														##CONSTRUCTING THE ELEMENTS OF COMPUTATION

														##First lets ensure we have a Variable vector for each centroid,
														##initialized to one of the vectors from the available data points
														snake_case_      =				[
														    tf.Variable(vectors[vector_indices[i]]					) for i in range(snake_case					)
														]
														##These nodes will assign the centroid Variables the appropriate
														##values
														snake_case_      =				tf.placeholder("float64"							,    [dim]					)
														snake_case_      =				[]
														for centroid in centroids:
																					cent_assigns.append(tf.assign(snake_case							,    snake_case					)					)

														##Variables for cluster assignments of individual vectors(initialized
														##to 0 at first)
														snake_case_      =				[tf.Variable(0					) for i in range(len(snake_case					)					)]
														##These nodes will assign an assignment Variable the appropriate
														##value
														snake_case_      =				tf.placeholder("int32"					)
														snake_case_      =				[]
														for assignment in assignments:
																					cluster_assigns.append(tf.assign(snake_case							,    snake_case					)					)

														##Now lets construct the node that will compute the mean
														# The placeholder for the input
														snake_case_      =				tf.placeholder("float"							,    [None, dim]					)
														# The Node/op takes the input and computes a mean along the 0th
														# dimension, i.e. the list of input vectors
														snake_case_      =				tf.reduce_mean(snake_case							,    0					)

														##Node for computing Euclidean distances
														# Placeholders for input
														snake_case_      =				tf.placeholder("float"							,    [dim]					)
														snake_case_      =				tf.placeholder("float"							,    [dim]					)
														snake_case_      =				tf.sqrt(tf.reduce_sum(tf.pow(tf.sub(snake_case							,    snake_case					)							,    2					)					)					)

														##This node will figure out which cluster to assign a vector to,
														##based on Euclidean distances of the vector from the centroids.
														# Placeholder for input
														snake_case_      =				tf.placeholder("float"							,    [noofclusters]					)
														snake_case_      =				tf.argmin(snake_case							,    0					)

														##INITIALIZING STATE VARIABLES

														##This will help initialization of all Variables defined with respect
														##to the graph. The Variable-initializer should be defined after
														##all the Variables have been constructed, so that each of them
														##will be included in the initialization.
														snake_case_      =				tf.initialize_all_variables()

														# Initialize all variables
														sess.run(snake_case					)

														##CLUSTERING ITERATIONS

														# Now perform the Expectation-Maximization steps of K-Means clustering
														# iterations. To keep things simple, we will only do a set number of
														# iterations, instead of using a Stopping Criterion.
														snake_case_      =				1_0_0
														for _ in range(snake_case					):
																					##EXPECTATION STEP
																					##Based on the centroid locations till last iteration, compute
																					##the _expected_ centroid assignments.
																					# Iterate over each vector
																					for vector_n in range(len(snake_case					)					):
																												snake_case_      =				vectors[vector_n]
																												# Compute Euclidean distance between this vector and each
																												# centroid. Remember that this list cannot be named
																												#'centroid_distances', since that is the input to the
																												# cluster assignment node.
																												snake_case_      =				[
																												    sess.run(snake_case							,    feed_dict={va: vect, va: sess.run(snake_case					)}					)
																												    for centroid in centroids
																												]
																												# Now use the cluster assignment node, with the distances
																												# as the input
																												snake_case_      =				sess.run(
																												    snake_case							,    feed_dict={centroid_distances: distances}					)
																												# Now assign the value to the appropriate state variable
																												sess.run(
																												    cluster_assigns[vector_n]							,    feed_dict={assignment_value: assignment}					)

																					##MAXIMIZATION STEP
																					# Based on the expected state computed from the Expectation Step,
																					# compute the locations of the centroids so as to maximize the
																					# overall objective of minimizing within-cluster Sum-of-Squares
																					for cluster_n in range(snake_case					):
																												# Collect all the vectors assigned to this cluster
																												snake_case_      =				[
																												    vectors[i]
																												    for i in range(len(snake_case					)					)
																												    if sess.run(assignments[i]					) == cluster_n
																												]
																												# Compute new centroid location
																												snake_case_      =				sess.run(
																												    snake_case							,    feed_dict={mean_input: array(snake_case					)}					)
																												# Assign value to appropriate variable
																												sess.run(
																												    cent_assigns[cluster_n]							,    feed_dict={centroid_value: new_location}					)

        # Return centroids and assignments
														snake_case_      =				sess.run(snake_case					)
														snake_case_      =				sess.run(snake_case					)
														return centroids, assignments




