






"""simple docstring"""







from random import shuffle

import tensorflow as tf
from numpy import array


def      UpperCamelCase       ( _A  ,				_A )     ->       Optional[Any]:
				lowercase    :      List[str]       = int(_lowerCAmelCase )
				assert noofclusters < len(_lowerCAmelCase )

				# Find out the dimensionality
				lowercase    :      Dict       = len(vectors[0] )

				# Will help select random centroids from among the available vectors
				lowercase    :      Any       = list(range(len(_lowerCAmelCase ) ) )
				shuffle(_lowerCAmelCase )

				# 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.

				lowercase    :      Dict       = tf.Graph()

				with graph.as_default():
								# SESSION OF COMPUTATION

								lowercase    :      Optional[Any]       = 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
								lowercase    :      Union[str, Any]       = [
								    tf.Variable(vectors[vector_indices[i]] ) for i in range(_lowerCAmelCase )
								]
								##These nodes will assign the centroid Variables the appropriate
								##values
								lowercase    :      List[str]       = tf.placeholder("""float64"""  ,				[dim] )
								lowercase    :      int       = []
								for centroid in centroids:
												cent_assigns.append(tf.assign(_lowerCAmelCase  ,				_lowerCAmelCase ) )

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

								##Now lets construct the node that will compute the mean
								# The placeholder for the input
								lowercase    :      Any       = 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
								lowercase    :      Tuple       = tf.reduce_mean(_lowerCAmelCase  ,				0 )

								##Node for computing Euclidean distances
								# Placeholders for input
								lowercase    :      Optional[Any]       = tf.placeholder("""float"""  ,				[dim] )
								lowercase    :      List[str]       = tf.placeholder("""float"""  ,				[dim] )
								lowercase    :      Tuple       = tf.sqrt(tf.reduce_sum(tf.pow(tf.sub(_lowerCAmelCase  ,				_lowerCAmelCase )  ,				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
								lowercase    :      Optional[Any]       = tf.placeholder("""float"""  ,				[noofclusters] )
								lowercase    :      int       = tf.argmin(_lowerCAmelCase  ,				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.
								lowercase    :      Optional[Any]       = tf.initialize_all_variables()

								# Initialize all variables
								sess.run(_lowerCAmelCase )

								##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.
								lowercase    :      List[str]       = 100
								for _ in range(_lowerCAmelCase ):
												##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(_lowerCAmelCase ) ):
																lowercase    :      Optional[int]       = 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.
																lowercase    :      List[Any]       = [
																    sess.run(_lowerCAmelCase  ,				feed_dict={va: vect, va: sess.run(_lowerCAmelCase )} )
																    for centroid in centroids
																]
																# Now use the cluster assignment node, with the distances
																# as the input
																lowercase    :      Optional[Any]       = sess.run(
																    _lowerCAmelCase  ,				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(_lowerCAmelCase ):
																# Collect all the vectors assigned to this cluster
																lowercase    :      List[Any]       = [
																    vectors[i]
																    for i in range(len(_lowerCAmelCase ) )
																    if sess.run(assignments[i] ) == cluster_n
																]
																# Compute new centroid location
																lowercase    :      Any       = sess.run(
																    _lowerCAmelCase  ,				feed_dict={mean_input: array(_lowerCAmelCase )} )
																# Assign value to appropriate variable
																sess.run(
																    cent_assigns[cluster_n]  ,				feed_dict={centroid_value: new_location} )

        # Return centroids and assignments
								lowercase    :      List[Any]       = sess.run(_lowerCAmelCase )
								lowercase    :      int       = sess.run(_lowerCAmelCase )
								return centroids, assignments


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



							'''simple docstring'''





							_snake_case						      =		StableUnCLIPPipeline
							_snake_case						      =		TEXT_TO_IMAGE_PARAMS
							_snake_case						      =		TEXT_TO_IMAGE_BATCH_PARAMS
							_snake_case						      =		TEXT_TO_IMAGE_IMAGE_PARAMS
							_snake_case						      =		TEXT_TO_IMAGE_IMAGE_PARAMS

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




							def    A__						( self				)       ->  Optional[Any]:
													__lowerCAmelCase     =  32
													__lowerCAmelCase     =  embedder_hidden_size

													# prior components

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

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

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

													# 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     =  {
													    # 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    A__						( self    ,				snake_case_    ,				snake_case_=0				)       ->  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     =  {
													    """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    A__						( self				)       ->  Tuple:
													__lowerCAmelCase     =  torch_device == """cpu"""

													self._test_attention_slicing_forward_pass(test_max_difference=snake_case_				)




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

													self._test_inference_batch_single_identical(test_max_difference=snake_case_				)
@slow
@require_torch_gpu
class       lowerCAmelCase_		(				unittest.TestCase ):



							'''simple docstring'''





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




							def    A__						( self				)       ->  Tuple:
													__lowerCAmelCase     =  load_numpy(
													    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"""				)

													__lowerCAmelCase     =  StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""    ,				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("""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				)       ->  Tuple:
													torch.cuda.empty_cache()
													torch.cuda.reset_max_memory_allocated()
													torch.cuda.reset_peak_memory_stats()

													__lowerCAmelCase     =  StableUnCLIPPipeline.from_pretrained("""fusing/stable-unclip-2-1-l"""    ,				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(
													    """anime turtle"""    ,				prior_num_inference_steps=2    ,				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
