





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    A__  (			__snake_case			,  __snake_case			,  __snake_case			,  unittest.TestCase     ):
      _UpperCAmelCase      :List[str]		       =  StableUnCLIPPipeline
      _UpperCAmelCase      :Optional[int]		       =  TEXT_TO_IMAGE_PARAMS
      _UpperCAmelCase      :Optional[Any]		       =  TEXT_TO_IMAGE_BATCH_PARAMS
      _UpperCAmelCase      :str		       =  TEXT_TO_IMAGE_IMAGE_PARAMS
      _UpperCAmelCase      :int		       =  TEXT_TO_IMAGE_IMAGE_PARAMS

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





      def 			__UpperCamelCase(					self  ):
       '''simple docstring'''




       UpperCamelCase      :   Any			      =  32
       UpperCamelCase      :   int			      =  embedder_hidden_size

       # prior components

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

       torch.manual_seed(0  )
       UpperCamelCase      :   Union[str, Any]			      =  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=1000			,				)  )

       torch.manual_seed(0  )
       UpperCamelCase      :   Tuple			      =  PriorTransformer(
           num_attention_heads=2			,				attention_head_dim=12			,				embedding_dim=A_			,				num_layers=1			,				)

       torch.manual_seed(0  )
       UpperCamelCase      :   Dict			      =  DDPMScheduler(
           variance_type="fixed_small_log"			,				prediction_type="sample"			,				num_train_timesteps=1000			,				clip_sample=A_			,				clip_sample_range=5.0			,				beta_schedule="squaredcos_cap_v2"			,				)

       # regular denoising components

       torch.manual_seed(0  )
       UpperCamelCase      :   List[str]			      =  StableUnCLIPImageNormalizer(embedding_dim=A_  )
       UpperCamelCase      :   Optional[int]			      =  DDPMScheduler(beta_schedule="squaredcos_cap_v2"  )

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

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

       torch.manual_seed(0  )
       UpperCamelCase      :   int			      =  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  )
       UpperCamelCase      :   Union[str, Any]			      =  DDIMScheduler(
           beta_schedule="scaled_linear"			,				beta_start=0.0_00_85			,				beta_end=0.0_12			,				prediction_type="v_prediction"			,				set_alpha_to_one=A_			,				steps_offset=1			,				)

       torch.manual_seed(0  )
       UpperCamelCase      :   Optional[Any]			      =  AutoencoderKL()

       UpperCamelCase      :   Union[str, Any]			      =  {
           # 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			,				A_			,				A_=0  ):
       '''simple docstring'''




       if str(A_  ).startswith("mps"  ):
        UpperCamelCase      :   Union[str, Any]			      =  torch.manual_seed(A_  )
       else:
        UpperCamelCase      :   Dict			      =  torch.Generator(device=A_  ).manual_seed(A_  )
       UpperCamelCase      :   List[str]			      =  {
           "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  ):
       '''simple docstring'''




       UpperCamelCase      :   Tuple			      =  torch_device == "cpu"

       self._test_attention_slicing_forward_pass(test_max_difference=A_  )





      def 			__UpperCamelCase(					self  ):
       '''simple docstring'''




       UpperCamelCase      :   Optional[int]			      =  torch_device in ["cpu", "mps"]

       self._test_inference_batch_single_identical(test_max_difference=A_  )



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





      def 			__UpperCamelCase(					self  ):
       '''simple docstring'''




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





      def 			__UpperCamelCase(					self  ):
       '''simple docstring'''




       UpperCamelCase      :   Optional[int]			      =  load_numpy(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy"  )

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

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

       UpperCamelCase      :   int			      =  output.images[0]

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

       assert_mean_pixel_difference(A_			,				A_  )





      def 			__UpperCamelCase(					self  ):
       '''simple docstring'''




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

       UpperCamelCase      :   Union[str, Any]			      =  StableUnCLIPPipeline.from_pretrained("fusing/stable-unclip-2-1-l"			,				torch_dtype=torch.floataa  )
       UpperCamelCase      :   Dict			      =  pipe.to(A_  )
       pipe.set_progress_bar_config(disable=A_  )
       pipe.enable_attention_slicing()
       pipe.enable_sequential_cpu_offload()

       UpperCamelCase      :   Optional[int]			      =  pipe(
           "anime turtle"			,				prior_num_inference_steps=2			,				num_inference_steps=2			,				output_type="np"			,				)

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







from random import shuffle

import tensorflow as tf
from numpy import array



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

 # Find out the dimensionality
 UpperCamelCase      :   str			      =  len(vectors[0]	)

 # Will help select random centroids from among the available vectors
 UpperCamelCase      :   List[str]			      =  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.

 UpperCamelCase      :   Any			      =  tf.Graph()

 with graph.as_default():
  # SESSION OF COMPUTATION

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

  ##Variables for cluster assignments of individual vectors(initialized
  ##to 0 at first)
  UpperCamelCase      :   List[str]			      =  [tf.Variable(0	) for i in range(len(_lowerCAmelCase	)	)]
  ##These nodes will assign an assignment Variable the appropriate
  ##value
  UpperCamelCase      :   Any			      =  tf.placeholder("int32"	)
  UpperCamelCase      :   Optional[int]			      =  []
  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
  UpperCamelCase      :   Tuple			      =  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
  UpperCamelCase      :   List[str]			      =  tf.reduce_mean(_lowerCAmelCase				,  0	)

  ##Node for computing Euclidean distances
  # Placeholders for input
  UpperCamelCase      :   Optional[int]			      =  tf.placeholder("float"				,  [dim]	)
  UpperCamelCase      :   Any			      =  tf.placeholder("float"				,  [dim]	)
  UpperCamelCase      :   List[str]			      =  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
  UpperCamelCase      :   Tuple			      =  tf.placeholder("float"				,  [noofclusters]	)
  UpperCamelCase      :   str			      =  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.
  UpperCamelCase      :   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.
  UpperCamelCase      :   Any			      =  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	)	):
    UpperCamelCase      :   Dict			      =  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.
    UpperCamelCase      :   Optional[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
    UpperCamelCase      :   Tuple			      =  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
    UpperCamelCase      :   str			      =  [
        vectors[i]
        for i in range(len(_lowerCAmelCase	)	)
        if sess.run(assignments[i]	) == cluster_n
    ]
    # Compute new centroid location
    UpperCamelCase      :   Optional[int]			      =  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
  UpperCamelCase      :   Union[str, Any]			      =  sess.run(_lowerCAmelCase	)
  UpperCamelCase      :   Tuple			      =  sess.run(_lowerCAmelCase	)
  return centroids, assignments
