

'''simple docstring'''



from __future__ import annotations

import random

# Maximum size of the population.  Bigger could be faster but is more memory expensive.
__snake_case			:     Union[str, Any]										=      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.
__snake_case			:     Dict										=      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.
__snake_case			:     Optional[int]										=      0.4
# Just a seed to improve randomness required by the algorithm.
random.seed(random.randint(0, 1000))
def 						__lowerCamelCase	(       __snake_case					:		str,		__snake_case					:		str	)	->							tuple[str, float]:






						"""simple docstring"""





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






						"""simple docstring"""





						A__					:       Dict     =random.randint(0,		len(__snake_case	) - 1	)
						A__					:       Tuple     =parent_a[:random_slice] + parent_a[random_slice:]
						A__					:       List[Any]     =parent_a[:random_slice] + parent_a[random_slice:]
						return (child_a, child_a)
def 						__lowerCamelCase	(       __snake_case					:		str,		__snake_case					:		list[str]	)	->							str:






						"""simple docstring"""





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






						"""simple docstring"""





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

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






						"""simple docstring"""





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

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

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

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

												# 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__					:       Optional[int]     =[evaluate(__snake_case,		__snake_case	) for item in population]

												# Check if there is a matching evolution.
												A__					:       Optional[Any]     =sorted(__snake_case,		key=lambda __snake_case	: x[1],		reverse=__snake_case	)
												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__					:       Union[str, Any]     =population[: int(N_POPULATION / 3	)]
												population.clear()
												population.extend(__snake_case	)
												# Normalize population score to be between 0 and 1.
												A__					:       List[str]     =[
												    (item, score / len(__snake_case	)) for item, score in population_score
												]

												# This is selection
												for i in range(__snake_case	):
																		population.extend(select(population_score[int(__snake_case	)],		__snake_case,		__snake_case	)	)
																		# 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(__snake_case	) > N_POPULATION:
																								break


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




'''simple docstring'''



import gc
import unittest

from parameterized import parameterized

from diffusers import FlaxUNetaDConditionModel
from diffusers.utils import is_flax_available
from diffusers.utils.testing_utils import load_hf_numpy, require_flax, slow


if is_flax_available():
		import jax
		import jax.numpy as jnp



@slow
@require_flax
class lowerCamelCase			(  unittest.TestCase						):



			'''simple docstring'''






			def lowercase__						(				self	:						Optional[Any]		,      lowerCAmelCase_	:						Union[str, Any]		,      lowerCAmelCase_	:						List[str]			)      ->		List[str]:





									'''simple docstring'''
									return f"gaussian_noise_s={seed}_shape={'_'.join([str(lowerCAmelCase_			) for s in shape]			)}.npy"


			def lowercase__						(				self	:						Optional[Any]			)      ->		Dict:





									'''simple docstring'''
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()


			def lowercase__						(				self	:						Tuple		,      lowerCAmelCase_	:						str=0		,      lowerCAmelCase_	:						List[str]=(4, 4, 64, 64)		,      lowerCAmelCase_	:						Optional[Any]=False			)      ->		str:





									'''simple docstring'''
									A__					:       Union[str, Any]     =jnp.bfloataa if fpaa else jnp.floataa
									A__					:       Any     =jnp.array(load_hf_numpy(self.get_file_format(lowerCAmelCase_		,      lowerCAmelCase_			)			)		,      dtype=lowerCAmelCase_			)
									return image


			def lowercase__						(				self	:						int		,      lowerCAmelCase_	:						Optional[int]=False		,      lowerCAmelCase_	:						Dict="CompVis/stable-diffusion-v1-4"			)      ->		List[Any]:





									'''simple docstring'''
									A__					:       Any     =jnp.bfloataa if fpaa else jnp.floataa
									A__					:       int     ="""bf16""" if fpaa else None

									A__   ,						A__					:       Any     =FlaxUNetaDConditionModel.from_pretrained(
									    lowerCAmelCase_		,      subfolder="""unet"""		,      dtype=lowerCAmelCase_		,      revision=lowerCAmelCase_			)
									return model, params


			def lowercase__						(				self	:						Optional[Any]		,      lowerCAmelCase_	:						int=0		,      lowerCAmelCase_	:						Optional[Any]=(4, 77, 7_68)		,      lowerCAmelCase_	:						List[str]=False			)      ->		Tuple:





									'''simple docstring'''
									A__					:       Union[str, Any]     =jnp.bfloataa if fpaa else jnp.floataa
									A__					:       Optional[int]     =jnp.array(load_hf_numpy(self.get_file_format(lowerCAmelCase_		,      lowerCAmelCase_			)			)		,      dtype=lowerCAmelCase_			)
									return hidden_states


			@parameterized.expand(
			    [
			        # fmt: off
			        [83, 4, [-0.2323, -0.1304, 0.0813, -0.3093, -0.0919, -0.1571, -0.1125, -0.5806]],
			        [17, 0.55, [-0.0831, -0.2443, 0.0901, -0.0919, 0.3396, 0.0103, -0.3743, 0.0701]],
			        [8, 0.89, [-0.4863, 0.0859, 0.0875, -0.1658, 0.9199, -0.0114, 0.4839, 0.4639]],
			        [3, 10_00, [-0.5649, 0.2402, -0.5518, 0.1248, 1.1328, -0.2443, -0.0325, -1.0078]],
			        # fmt: on
			    ]			)
			def lowercase__						(				self	:						Any		,      lowerCAmelCase_	:						Tuple		,      lowerCAmelCase_	:						Tuple		,      lowerCAmelCase_	:						str			)      ->		int:





									'''simple docstring'''
									A__   ,						A__					:       Tuple     =self.get_unet_model(model_id="""CompVis/stable-diffusion-v1-4"""		,      fpaa=lowerCAmelCase_			)
									A__					:       List[Any]     =self.get_latents(lowerCAmelCase_		,      fpaa=lowerCAmelCase_			)
									A__					:       Optional[int]     =self.get_encoder_hidden_states(lowerCAmelCase_		,      fpaa=lowerCAmelCase_			)

									A__					:       Union[str, Any]     =model.apply(
									    {"""params""": params}		,      lowerCAmelCase_		,      jnp.array(lowerCAmelCase_		,      dtype=jnp.intaa			)		,      encoder_hidden_states=lowerCAmelCase_		,      ).sample

									assert sample.shape == latents.shape

									A__					:       Optional[int]     =jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())			)		,      dtype=jnp.floataa			)
									A__					:       Tuple     =jnp.array(lowerCAmelCase_		,      dtype=jnp.floataa			)

									# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, in the same hardware
									assert jnp.allclose(lowerCAmelCase_		,      lowerCAmelCase_		,      atol=1e-2			)



			@parameterized.expand(
			    [
			        # fmt: off
			        [83, 4, [0.1514, 0.0807, 0.1624, 0.1016, -0.1896, 0.0263, 0.0677, 0.2310]],
			        [17, 0.55, [0.1164, -0.0216, 0.0170, 0.1589, -0.3120, 0.1005, -0.0581, -0.1458]],
			        [8, 0.89, [-0.1758, -0.0169, 0.1004, -0.1411, 0.1312, 0.1103, -0.1996, 0.2139]],
			        [3, 10_00, [0.1214, 0.0352, -0.0731, -0.1562, -0.0994, -0.0906, -0.2340, -0.0539]],
			        # fmt: on
			    ]			)
			def lowercase__						(				self	:						Dict		,      lowerCAmelCase_	:						str		,      lowerCAmelCase_	:						int		,      lowerCAmelCase_	:						Optional[int]			)      ->		List[str]:





									'''simple docstring'''
									A__   ,						A__					:       List[str]     =self.get_unet_model(model_id="""stabilityai/stable-diffusion-2"""		,      fpaa=lowerCAmelCase_			)
									A__					:       Union[str, Any]     =self.get_latents(lowerCAmelCase_		,      shape=(4, 4, 96, 96)		,      fpaa=lowerCAmelCase_			)
									A__					:       Dict     =self.get_encoder_hidden_states(lowerCAmelCase_		,      shape=(4, 77, 10_24)		,      fpaa=lowerCAmelCase_			)

									A__					:       Optional[Any]     =model.apply(
									    {"""params""": params}		,      lowerCAmelCase_		,      jnp.array(lowerCAmelCase_		,      dtype=jnp.intaa			)		,      encoder_hidden_states=lowerCAmelCase_		,      ).sample

									assert sample.shape == latents.shape

									A__					:       Optional[Any]     =jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())			)		,      dtype=jnp.floataa			)
									A__					:       List[Any]     =jnp.array(lowerCAmelCase_		,      dtype=jnp.floataa			)

									# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, on the same hardware
									assert jnp.allclose(lowerCAmelCase_		,      lowerCAmelCase_		,      atol=1e-2			)

