



'''simple docstring'''





import gc
import random
import tempfile
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMInverseScheduler,
    DDIMScheduler,
    DPMSolverMultistepInverseScheduler,
    DPMSolverMultistepScheduler,
    StableDiffusionDiffEditPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_image, slow
from diffusers.utils.testing_utils import enable_full_determinism, floats_tensor, require_torch_gpu, torch_device

from ..pipeline_params import TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS, TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




class 			_snake_case		(     lowercase_						,						lowercase_						,						unittest.TestCase							):
			lowerCAmelCase_					:				Optional[int]							=   StableDiffusionDiffEditPipeline
			lowerCAmelCase_					:				Optional[Any]							=   TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {"height", "width", "image"} | {"image_latents"}
			lowerCAmelCase_					:				Union[str, Any]							=   TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS - {"image"} | {"image_latents"}
			lowerCAmelCase_					:				str							=   frozenset(
			    []							)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
			lowerCAmelCase_					:				Optional[Any]							=   frozenset([]							)




			def        lowerCAmelCase__    ( self							)    ->    Tuple:

										'''simple docstring'''




										torch.manual_seed(0							)
										snake_case_      =				UNetaDConditionModel(
										    block_out_channels=(32, 64)    ,							layers_per_block=2    ,							sample_size=32    ,							in_channels=4    ,							out_channels=4    ,							down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")    ,							up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")    ,							cross_attention_dim=32    ,							attention_head_dim=(2, 4)    ,							use_linear_projection=a__    ,							)
										snake_case_      =				DDIMScheduler(
										    beta_start=0.0_0_0_8_5    ,							beta_end=0.0_1_2    ,							beta_schedule="scaled_linear"    ,							clip_sample=a__    ,							set_alpha_to_one=a__    ,							)
										snake_case_      =				DDIMInverseScheduler(
										    beta_start=0.0_0_0_8_5    ,							beta_end=0.0_1_2    ,							beta_schedule="scaled_linear"    ,							clip_sample=a__    ,							set_alpha_to_zero=a__    ,							)
										torch.manual_seed(0							)
										snake_case_      =				AutoencoderKL(
										    block_out_channels=[32, 64]    ,							in_channels=3    ,							out_channels=3    ,							down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]    ,							up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]    ,							latent_channels=4    ,							sample_size=128    ,							)
										torch.manual_seed(0							)
										snake_case_      =				CLIPTextConfig(
										    bos_token_id=0    ,							eos_token_id=2    ,							hidden_size=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    ,							hidden_act="gelu"    ,							projection_dim=512    ,							)
										snake_case_      =				CLIPTextModel(a__							)
										snake_case_      =				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"							)

										snake_case_      =				{
										    "unet": unet,
										    "scheduler": scheduler,
										    "inverse_scheduler": inverse_scheduler,
										    "vae": vae,
										    "text_encoder": text_encoder,
										    "tokenizer": tokenizer,
										    "safety_checker": None,
										    "feature_extractor": None,
										}

										return components




			def        lowerCAmelCase__    ( self    ,							a__    ,							a__=0							)    ->    Any:

										'''simple docstring'''




										snake_case_      =				floats_tensor((1, 16, 16)    ,							rng=random.Random(a__							)							).to(a__							)
										snake_case_      =				floats_tensor((1, 2, 4, 16, 16)    ,							rng=random.Random(a__							)							).to(a__							)
										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 dog and a newt",
										    "mask_image": mask,
										    "image_latents": latents,
										    "generator": generator,
										    "num_inference_steps": 2,
										    "inpaint_strength": 1.0,
										    "guidance_scale": 6.0,
										    "output_type": "numpy",
										}

										return inputs




			def        lowerCAmelCase__    ( self    ,							a__    ,							a__=0							)    ->    Optional[int]:

										'''simple docstring'''




										snake_case_      =				floats_tensor((1, 3, 32, 32)    ,							rng=random.Random(a__							)							).to(a__							)
										snake_case_      =				image.cpu().permute(0    ,							2    ,							3    ,							1							)[0]
										snake_case_      =				Image.fromarray(np.uinta(a__							)							).convert("RGB"							)
										if str(a__							).startswith("mps"							):
																	snake_case_      =				torch.manual_seed(a__							)
										else:
																	snake_case_      =				torch.Generator(device=a__							).manual_seed(a__							)
										snake_case_      =				{
										    "image": image,
										    "source_prompt": "a cat and a frog",
										    "target_prompt": "a dog and a newt",
										    "generator": generator,
										    "num_inference_steps": 2,
										    "num_maps_per_mask": 2,
										    "mask_encode_strength": 1.0,
										    "guidance_scale": 6.0,
										    "output_type": "numpy",
										}

										return inputs




			def        lowerCAmelCase__    ( self    ,							a__    ,							a__=0							)    ->    Union[str, Any]:

										'''simple docstring'''




										snake_case_      =				floats_tensor((1, 3, 32, 32)    ,							rng=random.Random(a__							)							).to(a__							)
										snake_case_      =				image.cpu().permute(0    ,							2    ,							3    ,							1							)[0]
										snake_case_      =				Image.fromarray(np.uinta(a__							)							).convert("RGB"							)
										if str(a__							).startswith("mps"							):
																	snake_case_      =				torch.manual_seed(a__							)
										else:
																	snake_case_      =				torch.Generator(device=a__							).manual_seed(a__							)
										snake_case_      =				{
										    "image": image,
										    "prompt": "a cat and a frog",
										    "generator": generator,
										    "num_inference_steps": 2,
										    "inpaint_strength": 1.0,
										    "guidance_scale": 6.0,
										    "decode_latents": True,
										    "output_type": "numpy",
										}
										return inputs




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

										'''simple docstring'''




										if not hasattr(self.pipeline_class    ,							"_optional_components"							):
																	return

										snake_case_      =				self.get_dummy_components()
										snake_case_      =				self.pipeline_class(**a__							)
										pipe.to(a__							)
										pipe.set_progress_bar_config(disable=a__							)

										# set all optional components to None and update pipeline config accordingly
										for optional_component in pipe._optional_components:
																	setattr(a__    ,							a__    ,							a__							)
										pipe.register_modules(**{optional_component: None for optional_component in pipe._optional_components}							)

										snake_case_      =				self.get_dummy_inputs(a__							)
										snake_case_      =				pipe(**a__							)[0]

										with tempfile.TemporaryDirectory() as tmpdir:
																	pipe.save_pretrained(a__							)
																	snake_case_      =				self.pipeline_class.from_pretrained(a__							)
																	pipe_loaded.to(a__							)
																	pipe_loaded.set_progress_bar_config(disable=a__							)

										for optional_component in pipe._optional_components:
																	self.assertTrue(
																	    getattr(a__    ,							a__							) is None    ,							F'`{optional_component}` did not stay set to None after loading.'    ,							)

										snake_case_      =				self.get_dummy_inputs(a__							)
										snake_case_      =				pipe_loaded(**a__							)[0]

										snake_case_      =				np.abs(output - output_loaded							).max()
										self.assertLess(a__    ,							1e-4							)




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

										'''simple docstring'''




										snake_case_      =				"cpu"

										snake_case_      =				self.get_dummy_components()
										snake_case_      =				self.pipeline_class(**a__							)
										pipe.to(a__							)
										pipe.set_progress_bar_config(disable=a__							)

										snake_case_      =				self.get_dummy_mask_inputs(a__							)
										snake_case_      =				pipe.generate_mask(**a__							)
										snake_case_      =				mask[0, -3:, -3:]

										self.assertEqual(mask.shape    ,							(1, 16, 16)							)
										snake_case_      =				np.array([0] * 9							)
										snake_case_      =				np.abs(mask_slice.flatten() - expected_slice							).max()
										self.assertLessEqual(a__    ,							1e-3							)
										self.assertEqual(mask[0, -3, -4]    ,							0							)




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

										'''simple docstring'''




										snake_case_      =				"cpu"

										snake_case_      =				self.get_dummy_components()
										snake_case_      =				self.pipeline_class(**a__							)
										pipe.to(a__							)
										pipe.set_progress_bar_config(disable=a__							)

										snake_case_      =				self.get_dummy_inversion_inputs(a__							)
										snake_case_      =				pipe.invert(**a__							).images
										snake_case_      =				image[0, -1, -3:, -3:]

										self.assertEqual(image.shape    ,							(2, 32, 32, 3)							)
										snake_case_      =				np.array(
										    [0.5_1_5_0, 0.5_1_3_4, 0.5_0_4_3, 0.5_3_7_6, 0.4_6_9_4, 0.5_1_0_5_0, 0.5_0_1_5, 0.4_4_0_7, 0.4_7_9_9]    ,							)
										snake_case_      =				np.abs(image_slice.flatten() - expected_slice							).max()
										self.assertLessEqual(a__    ,							1e-3							)




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

										'''simple docstring'''




										super().test_inference_batch_single_identical(expected_max_diff=5e-3							)




			def        lowerCAmelCase__    ( self							)    ->    Dict:

										'''simple docstring'''




										snake_case_      =				"cpu"

										snake_case_      =				self.get_dummy_components()

										snake_case_      =				{"beta_start": 0.0_0_0_8_5, "beta_end": 0.0_1_2, "beta_schedule": "scaled_linear"}
										snake_case_      =				DPMSolverMultistepScheduler(**a__							)
										snake_case_      =				DPMSolverMultistepInverseScheduler(**a__							)

										snake_case_      =				self.pipeline_class(**a__							)
										pipe.to(a__							)
										pipe.set_progress_bar_config(disable=a__							)

										snake_case_      =				self.get_dummy_inversion_inputs(a__							)
										snake_case_      =				pipe.invert(**a__							).images
										snake_case_      =				image[0, -1, -3:, -3:]

										self.assertEqual(image.shape    ,							(2, 32, 32, 3)							)
										snake_case_      =				np.array(
										    [0.5_1_5_0, 0.5_1_3_4, 0.5_0_4_3, 0.5_3_7_6, 0.4_6_9_4, 0.5_1_0_5_0, 0.5_0_1_5, 0.4_4_0_7, 0.4_7_9_9]    ,							)
										snake_case_      =				np.abs(image_slice.flatten() - expected_slice							).max()
										self.assertLessEqual(a__    ,							1e-3							)




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




			def        lowerCAmelCase__    ( self							)    ->    Union[str, Any]:

										'''simple docstring'''




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




			@classmethod
			def        lowerCAmelCase__    ( cls							)    ->    List[Any]:

										'''simple docstring'''




										snake_case_      =				load_image(
										    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/diffedit/fruit.png"							)

										snake_case_      =				raw_image.convert("RGB"							).resize((768, 768)							)

										snake_case_      =				raw_image




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

										'''simple docstring'''




										snake_case_      =				torch.manual_seed(0							)

										snake_case_      =				StableDiffusionDiffEditPipeline.from_pretrained(
										    "stabilityai/stable-diffusion-2-1"    ,							safety_checker=a__    ,							torch_dtype=torch.floataa							)
										snake_case_      =				DDIMScheduler.from_config(pipe.scheduler.config							)
										snake_case_      =				DDIMInverseScheduler.from_config(pipe.scheduler.config							)
										pipe.enable_model_cpu_offload()
										pipe.set_progress_bar_config(disable=a__							)

										snake_case_      =				"a bowl of fruit"
										snake_case_      =				"a bowl of pears"

										snake_case_      =				pipe.generate_mask(
										    image=self.raw_image    ,							source_prompt=a__    ,							target_prompt=a__    ,							generator=a__    ,							)

										snake_case_      =				pipe.invert(
										    prompt=a__    ,							image=self.raw_image    ,							inpaint_strength=0.7    ,							generator=a__							).latents

										snake_case_      =				pipe(
										    prompt=a__    ,							mask_image=a__    ,							image_latents=a__    ,							generator=a__    ,							negative_prompt=a__    ,							inpaint_strength=0.7    ,							output_type="numpy"    ,							).images[0]

										snake_case_      =				(
										    np.array(
										        load_image(
										            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
										            "/diffedit/pears.png"							).resize((768, 768)							)							)
										    / 255
										)
										assert np.abs((expected_image - image).max()							) < 5e-1




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

										'''simple docstring'''




										snake_case_      =				torch.manual_seed(0							)

										snake_case_      =				StableDiffusionDiffEditPipeline.from_pretrained(
										    "stabilityai/stable-diffusion-2-1"    ,							safety_checker=a__    ,							torch_dtype=torch.floataa							)
										snake_case_      =				DPMSolverMultistepScheduler.from_config(pipe.scheduler.config							)
										snake_case_      =				DPMSolverMultistepInverseScheduler.from_config(pipe.scheduler.config							)
										pipe.enable_model_cpu_offload()
										pipe.set_progress_bar_config(disable=a__							)

										snake_case_      =				"a bowl of fruit"
										snake_case_      =				"a bowl of pears"

										snake_case_      =				pipe.generate_mask(
										    image=self.raw_image    ,							source_prompt=a__    ,							target_prompt=a__    ,							generator=a__    ,							)

										snake_case_      =				pipe.invert(
										    prompt=a__    ,							image=self.raw_image    ,							inpaint_strength=0.7    ,							generator=a__    ,							num_inference_steps=25    ,							).latents

										snake_case_      =				pipe(
										    prompt=a__    ,							mask_image=a__    ,							image_latents=a__    ,							generator=a__    ,							negative_prompt=a__    ,							inpaint_strength=0.7    ,							num_inference_steps=25    ,							output_type="numpy"    ,							).images[0]

										snake_case_      =				(
										    np.array(
										        load_image(
										            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
										            "/diffedit/pears.png"							).resize((768, 768)							)							)
										    / 255
										)
										assert np.abs((expected_image - image).max()							) < 5e-1










'''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




