



import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

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


UpperCAmelCase_							:      Dict      =  False






@skip_mps
class      UpperCamelCase    (						_UpperCAmelCase       ,    _UpperCAmelCase       ,    _UpperCAmelCase       ,    unittest.TestCase     ):
	lowerCAmelCase	:   Optional[Any]				=	StableDiffusionAttendAndExcitePipeline
	lowerCAmelCase	:   Optional[Any]				=	False
	lowerCAmelCase	:   Any				=	TEXT_TO_IMAGE_PARAMS
	lowerCAmelCase	:   Dict				=	TEXT_TO_IMAGE_BATCH_PARAMS.union({"""token_indices"""}     )
	lowerCAmelCase	:   List[str]				=	TEXT_TO_IMAGE_IMAGE_PARAMS
	lowerCAmelCase	:   List[Any]				=	TEXT_TO_IMAGE_IMAGE_PARAMS



	@classmethod
	def 			__A							(       cls		):
							super().setUpClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	@classmethod
	def 			__A							(       cls		):
							super().tearDownClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	def 			__A							(       self		):
							torch.manual_seed(0		)
							A__							=       UNetaDConditionModel(
							    block_out_channels=(32, 64)			,	layers_per_block=1			,	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=UpperCAmelCase__			,	)
							A__							=       DDIMScheduler(
							    beta_start=0.00_085			,	beta_end=0.012			,	beta_schedule="scaled_linear"			,	clip_sample=UpperCAmelCase__			,	set_alpha_to_one=UpperCAmelCase__			,	)
							torch.manual_seed(0		)
							A__							=       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		)
							A__							=       CLIPTextConfig(
							    bos_token_id=0			,	eos_token_id=2			,	hidden_size=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			,	hidden_act="gelu"			,	projection_dim=512			,	)
							A__							=       CLIPTextModel(UpperCAmelCase__		)
							A__							=       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		)

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

							return components



	def 			__A							(       self			,	UpperCAmelCase__			,	UpperCAmelCase__=0		):
							if str(UpperCAmelCase__		).startswith("mps"		):
													A__							=       torch.manual_seed(UpperCAmelCase__		)
							else:
													A__							=       torch.Generator(device=UpperCAmelCase__		).manual_seed(UpperCAmelCase__		)
							A__							=       A__							=       {
							    "prompt": "a cat and a frog",
							    "token_indices": [2, 5],
							    "generator": generator,
							    "num_inference_steps": 1,
							    "guidance_scale": 6.0,
							    "output_type": "numpy",
							    "max_iter_to_alter": 2,
							    "thresholds": {0: 0.7},
							}
							return inputs



	def 			__A							(       self		):
							A__							=       "cpu"

							A__							=       self.get_dummy_components()
							A__							=       self.pipeline_class(**UpperCAmelCase__		)
							pipe.to(UpperCAmelCase__		)
							pipe.set_progress_bar_config(disable=UpperCAmelCase__		)

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

							self.assertEqual(image.shape			,	(1, 64, 64, 3)		)
							A__							=       np.array(
							    [0.63_905_364, 0.62_897_307, 0.48_599_017, 0.5_133_624, 0.5_550_048, 0.45_769_516, 0.50_326_973, 0.5_023_139, 0.45_384_496]		)
							A__							=       np.abs(image_slice.flatten() - expected_slice		).max()
							self.assertLessEqual(UpperCAmelCase__			,	1e-3		)



	def 			__A							(       self		):
							super().test_cpu_offload_forward_pass(expected_max_diff=5e-4		)



	def 			__A							(       self		):
							# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
							self._test_inference_batch_consistent(batch_sizes=[1, 2]		)



	def 			__A							(       self		):
							self._test_inference_batch_single_identical(batch_size=2			,	expected_max_diff=7e-4		)



	def 			__A							(       self		):
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3		)



	def 			__A							(       self		):
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4		)



	def 			__A							(       self		):
							super().test_save_load_local(expected_max_difference=5e-4		)



	def 			__A							(       self		):
							super().test_save_load_optional_components(expected_max_difference=4e-4		)






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



	@classmethod
	def 			__A							(       cls		):
							super().setUpClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	@classmethod
	def 			__A							(       cls		):
							super().tearDownClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	def 			__A							(       self		):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()



	def 			__A							(       self		):
							A__							=       torch.manual_seed(51		)

							A__							=       StableDiffusionAttendAndExcitePipeline.from_pretrained(
							    "CompVis/stable-diffusion-v1-4"			,	safety_checker=UpperCAmelCase__			,	torch_dtype=torch.floataa		)
							pipe.to("cuda"		)

							A__							=       "a painting of an elephant with glasses"
							A__							=       [5, 7]

							A__							=       pipe(
							    prompt=UpperCAmelCase__			,	token_indices=UpperCAmelCase__			,	guidance_scale=7.5			,	generator=UpperCAmelCase__			,	num_inference_steps=5			,	max_iter_to_alter=5			,	output_type="numpy"			,	).images[0]

							A__							=       load_numpy(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy"		)
							assert np.abs((expected_image - image).max()		) < 5e-1










from __future__ import annotations

import random

# Maximum size of the population.  Bigger could be faster but is more memory expensive.
UpperCAmelCase_							:      Tuple      =  200
# Number of elements selected in every generation of evolution. The selection takes
# place from best to worst of that generation and must be smaller than N_POPULATION.
UpperCAmelCase_							:      Any      =  50
# Probability that an element of a generation can mutate, changing one of its genes.
# This will guarantee that all genes will be used during evolution.
UpperCAmelCase_							:      Union[str, Any]      =  0.4
# Just a seed to improve randomness required by the algorithm.
random.seed(random.randint(0, 1_000))
def  UpperCamelCase						(       _A			:				str			,							_A			:				str    )->       tuple[str, float]:





						"""simple docstring"""





						A__							=       len([g for position, g in enumerate(_A    ) if g == main_target[position]]    )
						return (item, float(_A    ))
def  UpperCamelCase						(       _A			:				str			,							_A			:				str    )->       tuple[str, str]:





						"""simple docstring"""





						A__							=       random.randint(0			,							len(_A    ) - 1    )
						A__							=       parent_a[:random_slice] + parent_a[random_slice:]
						A__							=       parent_a[:random_slice] + parent_a[random_slice:]
						return (child_a, child_a)
def  UpperCamelCase						(       _A			:				str			,							_A			:				list[str]    )->       str:





						"""simple docstring"""





						A__							=       list(_A    )
						if random.uniform(0			,							1    ) < MUTATION_PROBABILITY:
												A__							=       random.choice(_A    )
						return "".join(_A    )
def  UpperCamelCase						(       _A			:				tuple[str, float]			,							_A			:				list[tuple[str, float]]			,							_A			:				list[str]			,							)->       list[str]:





						"""simple docstring"""





						A__							=       []
						# Generate more children proportionally to the fitness score.
						A__							=       int(parent_a[1] * 100    ) + 1
						A__							=       10 if child_n >= 10 else child_n
						for _ in range(_A    ):
												A__							=       population_score[random.randint(0			,							_A    )][0]

												A__						,       A__							=       crossover(parent_a[0]			,							_A    )
												# Append new string to the population list.
												pop.append(mutate(_A			,							_A    )    )
												pop.append(mutate(_A			,							_A    )    )
						return pop
def  UpperCamelCase						(       _A			:				str			,							_A			:				list[str]			,							_A			:				bool = True    )->       tuple[int, int, str]:





						"""simple docstring"""





						if N_POPULATION < N_SELECTED:
												A__							=       f"""{N_POPULATION} must be bigger than {N_SELECTED}"""
												raise ValueError(_A    )
						# Verify that the target contains no genes besides the ones inside genes variable.
						A__							=       sorted({c for c in target if c not in genes}    )
						if not_in_genes_list:
												A__							=       f"""{not_in_genes_list} is not in genes list, evolution cannot converge"""
												raise ValueError(_A    )

						# Generate random starting population.
						A__							=       []
						for _ in range(_A    ):
												population.append("".join([random.choice(_A    ) for i in range(len(_A    )    )]    )    )

						# Just some logs to know what the algorithms is doing.
						A__						,       A__							=       0, 0

						# This loop will end when we find a perfect match for our target.
						while True:
												generation += 1
												total_population += len(_A    )

												# Random population created. Now it's time to evaluate.

												# Adding a bit of concurrency can make everything faster,
												#
												# import concurrent.futures
												# population_score: list[tuple[str, float]] = []
												# with concurrent.futures.ThreadPoolExecutor(
												#                                   max_workers=NUM_WORKERS) as executor:
												#     futures = {executor.submit(evaluate, item) for item in population}
												#     concurrent.futures.wait(futures)
												#     population_score = [item.result() for item in futures]
												#
												# but with a simple algorithm like this, it will probably be slower.
												# We just need to call evaluate for every item inside the population.
												A__							=       [evaluate(_A			,							_A    ) for item in population]

												# Check if there is a matching evolution.
												A__							=       sorted(_A			,							key=lambda _A    : x[1]			,							reverse=_A    )
												if population_score[0][0] == target:
																		return (generation, total_population, population_score[0][0])

												# Print the best result every 10 generation.
												# Just to know that the algorithm is working.
												if debug and generation % 10 == 0:
																		print(
																		    f"""\nGeneration: {generation}"""
																		    f"""\nTotal Population:{total_population}"""
																		    f"""\nBest score: {population_score[0][1]}"""
																		    f"""\nBest string: {population_score[0][0]}"""    )

												# Flush the old population, keeping some of the best evolutions.
												# Keeping this avoid regression of evolution.
												A__							=       population[: int(N_POPULATION / 3    )]
												population.clear()
												population.extend(_A    )
												# Normalize population score to be between 0 and 1.
												A__							=       [
												    (item, score / len(_A    )) for item, score in population_score
												]

												# This is selection
												for i in range(_A    ):
																		population.extend(select(population_score[int(_A    )]			,							_A			,							_A    )    )
																		# Check if the population has already reached the maximum value and if so,
																		# break the cycle.  If this check is disabled, the algorithm will take
																		# forever to compute large strings, but will also calculate small strings in
																		# a far fewer generations.
																		if len(_A    ) > N_POPULATION:
																								break


if __name__ == "__main__":
						UpperCAmelCase_							:      Any      =  (
						    "This is a genetic algorithm to evaluate, combine, evolve, and mutate a string!"
						)
						UpperCAmelCase_							:      List[str]      =  list(
						    " ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklm"
						    "nopqrstuvwxyz.,;!?+-*#@^'èéòà€ù=)(&%$£/\\"
						)
						UpperCAmelCase_   ,	UpperCAmelCase_   ,	UpperCAmelCase_							:      Optional[int]      =  basic(target_str, genes_list)
						print(
						    F'''\nGeneration: {generation}\nTotal Population: {population}\nTarget: {target}'''
						)





