"""simple docstring"""







import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionModelEditingPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

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


enable_full_determinism()


@skip_mps
class       _lowercase		( __lowerCamelCase   ,   __lowerCamelCase   ,   __lowerCamelCase   ,   unittest.TestCase	):
				lowerCamelCase__							=							StableDiffusionModelEditingPipeline
				lowerCamelCase__							=							TEXT_TO_IMAGE_PARAMS
				lowerCamelCase__							=							TEXT_TO_IMAGE_BATCH_PARAMS
				lowerCamelCase__							=							TEXT_TO_IMAGE_IMAGE_PARAMS
				lowerCamelCase__							=							TEXT_TO_IMAGE_IMAGE_PARAMS

				def 							A			(  self							:     int  ):


										'''simple docstring'''
										torch.manual_seed(0  )
										UpperCAmelCase__       						=		UNetaDConditionModel(
										    block_out_channels=(3_2, 6_4)							,     layers_per_block=2							,     sample_size=3_2							,     in_channels=4							,     out_channels=4							,     down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")							,     up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")							,     cross_attention_dim=3_2							,     )
										UpperCAmelCase__       						=		DDIMScheduler()
										torch.manual_seed(0  )
										UpperCAmelCase__       						=		AutoencoderKL(
										    block_out_channels=[3_2, 6_4]							,     in_channels=3							,     out_channels=3							,     down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]							,     up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]							,     latent_channels=4							,     )
										torch.manual_seed(0  )
										UpperCAmelCase__       						=		CLIPTextConfig(
										    bos_token_id=0							,     eos_token_id=2							,     hidden_size=3_2							,     intermediate_size=3_7							,     layer_norm_eps=1e-05							,     num_attention_heads=4							,     num_hidden_layers=5							,     pad_token_id=1							,     vocab_size=1_0_0_0							,     )
										UpperCAmelCase__       						=		CLIPTextModel(_lowercase  )
										UpperCAmelCase__       						=		CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""  )

										UpperCAmelCase__       						=		{
										    """unet""": unet,
										    """scheduler""": scheduler,
										    """vae""": vae,
										    """text_encoder""": text_encoder,
										    """tokenizer""": tokenizer,
										    """safety_checker""": None,
										    """feature_extractor""": None,
										}
										return components

				def 							A			(  self							:     str							,     _lowercase							:     str							,     _lowercase							:     str=0  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		torch.manual_seed(_lowercase  )
										UpperCAmelCase__       						=		{
										    """prompt""": """A field of roses""",
										    """generator""": generator,
										    # Setting height and width to None to prevent OOMs on CPU.
										    """height""": None,
										    """width""": None,
										    """num_inference_steps""": 2,
										    """guidance_scale""": 6.0,
										    """output_type""": """numpy""",
										}
										return inputs

				def 							A			(  self							:     List[str]  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
										UpperCAmelCase__       						=		self.get_dummy_components()
										UpperCAmelCase__       						=		StableDiffusionModelEditingPipeline(**_lowercase  )
										UpperCAmelCase__       						=		sd_pipe.to(_lowercase  )
										sd_pipe.set_progress_bar_config(disable=_lowercase  )

										UpperCAmelCase__       						=		self.get_dummy_inputs(_lowercase  )
										UpperCAmelCase__       						=		sd_pipe(**_lowercase  ).images
										UpperCAmelCase__       						=		image[0, -3:, -3:, -1]
										assert image.shape == (1, 6_4, 6_4, 3)

										UpperCAmelCase__       						=		np.array([0.4755, 0.5132, 0.4976, 0.3904, 0.3554, 0.4765, 0.5139, 0.5158, 0.4889]  )

										assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2

				def 							A			(  self							:     Optional[int]  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
										UpperCAmelCase__       						=		self.get_dummy_components()
										UpperCAmelCase__       						=		StableDiffusionModelEditingPipeline(**_lowercase  )
										UpperCAmelCase__       						=		sd_pipe.to(_lowercase  )
										sd_pipe.set_progress_bar_config(disable=_lowercase  )

										UpperCAmelCase__       						=		self.get_dummy_inputs(_lowercase  )
										UpperCAmelCase__       						=		"""french fries"""
										UpperCAmelCase__       						=		sd_pipe(**_lowercase							,     negative_prompt=_lowercase  )
										UpperCAmelCase__       						=		output.images
										UpperCAmelCase__       						=		image[0, -3:, -3:, -1]

										assert image.shape == (1, 6_4, 6_4, 3)

										UpperCAmelCase__       						=		np.array([0.4992, 0.5101, 0.5004, 0.3949, 0.3604, 0.4735, 0.5216, 0.5204, 0.4913]  )

										assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2

				def 							A			(  self							:     Optional[int]  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
										UpperCAmelCase__       						=		self.get_dummy_components()
										UpperCAmelCase__       						=		EulerAncestralDiscreteScheduler(
										    beta_start=0.0_0085							,     beta_end=0.012							,     beta_schedule="""scaled_linear"""  )
										UpperCAmelCase__       						=		StableDiffusionModelEditingPipeline(**_lowercase  )
										UpperCAmelCase__       						=		sd_pipe.to(_lowercase  )
										sd_pipe.set_progress_bar_config(disable=_lowercase  )

										UpperCAmelCase__       						=		self.get_dummy_inputs(_lowercase  )
										UpperCAmelCase__       						=		sd_pipe(**_lowercase  ).images
										UpperCAmelCase__       						=		image[0, -3:, -3:, -1]

										assert image.shape == (1, 6_4, 6_4, 3)

										UpperCAmelCase__       						=		np.array([0.4747, 0.5372, 0.4779, 0.4982, 0.5543, 0.4816, 0.5238, 0.4904, 0.5027]  )

										assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1e-2

				def 							A			(  self							:     Optional[int]  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
										UpperCAmelCase__       						=		self.get_dummy_components()
										UpperCAmelCase__       						=		PNDMScheduler()
										UpperCAmelCase__       						=		StableDiffusionModelEditingPipeline(**_lowercase  )
										UpperCAmelCase__       						=		sd_pipe.to(_lowercase  )
										sd_pipe.set_progress_bar_config(disable=_lowercase  )

										UpperCAmelCase__       						=		self.get_dummy_inputs(_lowercase  )
										# the pipeline does not expect pndm so test if it raises error.
										with self.assertRaises(_lowercase  ):
																UpperCAmelCase__       						=		sd_pipe(**_lowercase  ).images

				def 							A			(  self							:     Tuple  ):


										'''simple docstring'''
										super().test_inference_batch_single_identical(expected_max_diff=5e-3  )

				def 							A			(  self							:     Any  ):


										'''simple docstring'''
										super().test_attention_slicing_forward_pass(expected_max_diff=5e-3  )


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

				def 							A			(  self							:     Any  ):


										'''simple docstring'''
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()

				def 							A			(  self							:     Optional[Any]							,     _lowercase							:     Any=0  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		torch.manual_seed(_lowercase  )
										UpperCAmelCase__       						=		{
										    """prompt""": """A field of roses""",
										    """generator""": generator,
										    """num_inference_steps""": 3,
										    """guidance_scale""": 7.5,
										    """output_type""": """numpy""",
										}
										return inputs

				def 							A			(  self							:     List[Any]  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		"""CompVis/stable-diffusion-v1-4"""
										UpperCAmelCase__       						=		StableDiffusionModelEditingPipeline.from_pretrained(_lowercase							,     safety_checker=_lowercase  )
										pipe.to(_lowercase  )
										pipe.set_progress_bar_config(disable=_lowercase  )
										pipe.enable_attention_slicing()

										UpperCAmelCase__       						=		self.get_inputs()
										UpperCAmelCase__       						=		pipe(**_lowercase  ).images
										UpperCAmelCase__       						=		image[0, -3:, -3:, -1].flatten()

										assert image.shape == (1, 5_1_2, 5_1_2, 3)

										UpperCAmelCase__       						=		np.array(
										    [0.674_9496, 0.638_6453, 0.5144_3267, 0.6609_4905, 0.6192_1215, 0.549_1332, 0.574_4417, 0.5807_5106, 0.517_4658]  )

										assert np.abs(expected_slice - image_slice  ).max() < 1e-2

										# make sure image changes after editing
										pipe.edit_model("""A pack of roses"""							,     """A pack of blue roses"""  )

										UpperCAmelCase__       						=		pipe(**_lowercase  ).images
										UpperCAmelCase__       						=		image[0, -3:, -3:, -1].flatten()

										assert image.shape == (1, 5_1_2, 5_1_2, 3)

										assert np.abs(expected_slice - image_slice  ).max() > 1e-1

				def 							A			(  self							:     Optional[Any]  ):


										'''simple docstring'''
										torch.cuda.empty_cache()
										torch.cuda.reset_max_memory_allocated()
										torch.cuda.reset_peak_memory_stats()

										UpperCAmelCase__       						=		"""CompVis/stable-diffusion-v1-4"""
										UpperCAmelCase__       						=		DDIMScheduler.from_pretrained(_lowercase							,     subfolder="""scheduler"""  )
										UpperCAmelCase__       						=		StableDiffusionModelEditingPipeline.from_pretrained(
										    _lowercase							,     scheduler=_lowercase							,     safety_checker=_lowercase  )
										UpperCAmelCase__       						=		pipe.to(_lowercase  )
										pipe.set_progress_bar_config(disable=_lowercase  )
										pipe.enable_attention_slicing(1  )
										pipe.enable_sequential_cpu_offload()

										UpperCAmelCase__       						=		self.get_inputs()
										UpperCAmelCase__       						=		pipe(**_lowercase  )

										UpperCAmelCase__       						=		torch.cuda.max_memory_allocated()
										# make sure that less than 4.4 GB is allocated
										assert mem_bytes < 4.4 * 1_0**9


"""simple docstring"""







import numpy as np


def 	a__   (		UpperCAmelCase__							:      List[str]		,      UpperCAmelCase__							:      int		,      UpperCAmelCase__							:      str      )->						Optional[int]:
						UpperCAmelCase__       						=		n_draws * class_counts / class_counts.sum()
						# floored means we don't overshoot n_samples, but probably undershoot
						UpperCAmelCase__       						=		np.floor(UpperCAmelCase__      )
						# we add samples according to how much "left over" probability
						# they had, until we arrive at n_samples
						UpperCAmelCase__       						=		int(n_draws - floored.sum()      )
						if need_to_add > 0:
												UpperCAmelCase__       						=		continuous - floored
												UpperCAmelCase__       						=		np.sort(np.unique(UpperCAmelCase__      )      )[::-1]
												# add according to remainder, but break ties
												# randomly to avoid biases
												for value in values:
																		((UpperCAmelCase__)						,   )       						=		np.where(remainder == value      )
																		# if we need_to_add less than what's in inds
																		# we draw randomly from them.
																		# if we need to add more, we add them all and
																		# go to the next value
																		UpperCAmelCase__       						=		min(len(UpperCAmelCase__      )		,      UpperCAmelCase__      )
																		UpperCAmelCase__       						=		rng.choice(UpperCAmelCase__		,      size=UpperCAmelCase__		,      replace=UpperCAmelCase__      )
																		floored[inds] += 1
																		need_to_add -= add_now
																		if need_to_add == 0:
																								break
						return floored.astype(np.intaa      )







def 	a__   (		UpperCAmelCase__							:      Optional[int]		,      UpperCAmelCase__							:      Optional[int]		,      UpperCAmelCase__							:      Dict		,      UpperCAmelCase__							:      Any		,      UpperCAmelCase__							:      Union[str, Any]=1_0      )->						Tuple:
						UpperCAmelCase__						,   UpperCAmelCase__       						=		np.unique(UpperCAmelCase__		,      return_inverse=UpperCAmelCase__      )
						UpperCAmelCase__       						=		classes.shape[0]
						UpperCAmelCase__       						=		np.bincount(UpperCAmelCase__      )
						if np.min(UpperCAmelCase__      ) < 2:
												raise ValueError("""Minimum class count error"""      )
						if n_train < n_classes:
												raise ValueError(
												    """The train_size = %d should be greater or """ """equal to the number of classes = %d""" % (n_train, n_classes)      )
						if n_test < n_classes:
												raise ValueError(
												    """The test_size = %d should be greater or """ """equal to the number of classes = %d""" % (n_test, n_classes)      )
						UpperCAmelCase__       						=		np.split(np.argsort(UpperCAmelCase__		,      kind="""mergesort"""      )		,      np.cumsum(UpperCAmelCase__      )[:-1]      )
						for _ in range(UpperCAmelCase__      ):
												UpperCAmelCase__       						=		approximate_mode(UpperCAmelCase__		,      UpperCAmelCase__		,      UpperCAmelCase__      )
												UpperCAmelCase__       						=		class_counts - n_i
												UpperCAmelCase__       						=		approximate_mode(UpperCAmelCase__		,      UpperCAmelCase__		,      UpperCAmelCase__      )

												UpperCAmelCase__       						=		[]
												UpperCAmelCase__       						=		[]

												for i in range(UpperCAmelCase__      ):
																		UpperCAmelCase__       						=		rng.permutation(class_counts[i]      )
																		UpperCAmelCase__       						=		class_indices[i].take(UpperCAmelCase__		,      mode="""clip"""      )
																		train.extend(perm_indices_class_i[: n_i[i]]      )
																		test.extend(perm_indices_class_i[n_i[i] : n_i[i] + t_i[i]]      )
												UpperCAmelCase__       						=		rng.permutation(UpperCAmelCase__      )
												UpperCAmelCase__       						=		rng.permutation(UpperCAmelCase__      )

												yield train, test

