




"""simple docstring"""




from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block






@dataclass
class     __magic_name__							(       _lowerCamelCase				):






				'''simple docstring'''
				__UpperCamelCase			  =						42






class     __magic_name__							(       nn.Module				):






				'''simple docstring'''





				def __init__(       self		, _a=3		, _a=3		, _a=("DownEncoderBlock2D",)		, _a=(64,)		, _a=2		, _a=32		, _a="silu"		, _a=True		, ):





							"""simple docstring"""
							super().__init__()
							lowerCamelCase			      =					layers_per_block

							lowerCamelCase			      =					torch.nn.Convad(
							    lowerCAmelCase_		, block_out_channels[0]		, kernel_size=3		, stride=1		, padding=1		, )

							lowerCamelCase			      =					None
							lowerCamelCase			      =					nn.ModuleList([] )

							# down
							lowerCamelCase			      =					block_out_channels[0]
							for i, down_block_type in enumerate(lowerCAmelCase_ ):
										lowerCamelCase			      =					output_channel
										lowerCamelCase			      =					block_out_channels[i]
										lowerCamelCase			      =					i == len(lowerCAmelCase_ ) - 1

										lowerCamelCase			      =					get_down_block(
										    lowerCAmelCase_		, num_layers=self.layers_per_block		, in_channels=lowerCAmelCase_		, out_channels=lowerCAmelCase_		, add_downsample=not is_final_block		, resnet_eps=1e-6		, downsample_padding=0		, resnet_act_fn=lowerCAmelCase_		, resnet_groups=lowerCAmelCase_		, attention_head_dim=lowerCAmelCase_		, temb_channels=lowerCAmelCase_		, )
										self.down_blocks.append(lowerCAmelCase_ )

							# mid
							lowerCamelCase			      =					UNetMidBlockaD(
							    in_channels=block_out_channels[-1]		, resnet_eps=1e-6		, resnet_act_fn=lowerCAmelCase_		, output_scale_factor=1		, resnet_time_scale_shift="""default"""		, attention_head_dim=block_out_channels[-1]		, resnet_groups=lowerCAmelCase_		, temb_channels=lowerCAmelCase_		, )

							# out
							lowerCamelCase			      =					nn.GroupNorm(num_channels=block_out_channels[-1]		, num_groups=lowerCAmelCase_		, eps=1e-6 )
							lowerCamelCase			      =					nn.SiLU()

							lowerCamelCase			      =					2 * out_channels if double_z else out_channels
							lowerCamelCase			      =					nn.Convad(block_out_channels[-1]		, lowerCAmelCase_		, 3		, padding=1 )

							lowerCamelCase			      =					False





				def    _lowerCAmelCase				(       self		, _a ):





							"""simple docstring"""
							lowerCamelCase			      =					x
							lowerCamelCase			      =					self.conv_in(lowerCAmelCase_ )

							if self.training and self.gradient_checkpointing:

										def create_custom_forward(_a ):
													def custom_forward(*_a ):
																return module(*lowerCAmelCase_ )

													return custom_forward

										# down
										if is_torch_version(""">="""		, """1.11.0""" ):
													for down_block in self.down_blocks:
																lowerCamelCase			      =					torch.utils.checkpoint.checkpoint(
																    create_custom_forward(lowerCAmelCase_ )		, lowerCAmelCase_		, use_reentrant=lowerCAmelCase_ )
													# middle
													lowerCamelCase			      =					torch.utils.checkpoint.checkpoint(
													    create_custom_forward(self.mid_block )		, lowerCAmelCase_		, use_reentrant=lowerCAmelCase_ )
										else:
													for down_block in self.down_blocks:
																lowerCamelCase			      =					torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCAmelCase_ )		, lowerCAmelCase_ )
													# middle
													lowerCamelCase			      =					torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block )		, lowerCAmelCase_ )

							else:
										# down
										for down_block in self.down_blocks:
													lowerCamelCase			      =					down_block(lowerCAmelCase_ )

										# middle
										lowerCamelCase			      =					self.mid_block(lowerCAmelCase_ )

							# post-process
							lowerCamelCase			      =					self.conv_norm_out(lowerCAmelCase_ )
							lowerCamelCase			      =					self.conv_act(lowerCAmelCase_ )
							lowerCamelCase			      =					self.conv_out(lowerCAmelCase_ )

							return sample






class     __magic_name__							(       nn.Module				):






				'''simple docstring'''





				def __init__(       self		, _a=3		, _a=3		, _a=("UpDecoderBlock2D",)		, _a=(64,)		, _a=2		, _a=32		, _a="silu"		, _a="group"		, ):





							"""simple docstring"""
							super().__init__()
							lowerCamelCase			      =					layers_per_block

							lowerCamelCase			      =					nn.Convad(
							    lowerCAmelCase_		, block_out_channels[-1]		, kernel_size=3		, stride=1		, padding=1		, )

							lowerCamelCase			      =					None
							lowerCamelCase			      =					nn.ModuleList([] )

							lowerCamelCase			      =					in_channels if norm_type == """spatial""" else None

							# mid
							lowerCamelCase			      =					UNetMidBlockaD(
							    in_channels=block_out_channels[-1]		, resnet_eps=1e-6		, resnet_act_fn=lowerCAmelCase_		, output_scale_factor=1		, resnet_time_scale_shift="""default""" if norm_type == """group""" else norm_type		, attention_head_dim=block_out_channels[-1]		, resnet_groups=lowerCAmelCase_		, temb_channels=lowerCAmelCase_		, )

							# up
							lowerCamelCase			      =					list(reversed(lowerCAmelCase_ ) )
							lowerCamelCase			      =					reversed_block_out_channels[0]
							for i, up_block_type in enumerate(lowerCAmelCase_ ):
										lowerCamelCase			      =					output_channel
										lowerCamelCase			      =					reversed_block_out_channels[i]

										lowerCamelCase			      =					i == len(lowerCAmelCase_ ) - 1

										lowerCamelCase			      =					get_up_block(
										    lowerCAmelCase_		, num_layers=self.layers_per_block + 1		, in_channels=lowerCAmelCase_		, out_channels=lowerCAmelCase_		, prev_output_channel=lowerCAmelCase_		, add_upsample=not is_final_block		, resnet_eps=1e-6		, resnet_act_fn=lowerCAmelCase_		, resnet_groups=lowerCAmelCase_		, attention_head_dim=lowerCAmelCase_		, temb_channels=lowerCAmelCase_		, resnet_time_scale_shift=lowerCAmelCase_		, )
										self.up_blocks.append(lowerCAmelCase_ )
										lowerCamelCase			      =					output_channel

							# out
							if norm_type == "spatial":
										lowerCamelCase			      =					SpatialNorm(block_out_channels[0]		, lowerCAmelCase_ )
							else:
										lowerCamelCase			      =					nn.GroupNorm(num_channels=block_out_channels[0]		, num_groups=lowerCAmelCase_		, eps=1e-6 )
							lowerCamelCase			      =					nn.SiLU()
							lowerCamelCase			      =					nn.Convad(block_out_channels[0]		, lowerCAmelCase_		, 3		, padding=1 )

							lowerCamelCase			      =					False





				def    _lowerCAmelCase				(       self		, _a		, _a=None ):





							"""simple docstring"""
							lowerCamelCase			      =					z
							lowerCamelCase			      =					self.conv_in(lowerCAmelCase_ )

							lowerCamelCase			      =					next(iter(self.up_blocks.parameters() ) ).dtype
							if self.training and self.gradient_checkpointing:

										def create_custom_forward(_a ):
													def custom_forward(*_a ):
																return module(*lowerCAmelCase_ )

													return custom_forward

										if is_torch_version(""">="""		, """1.11.0""" ):
													# middle
													lowerCamelCase			      =					torch.utils.checkpoint.checkpoint(
													    create_custom_forward(self.mid_block )		, lowerCAmelCase_		, lowerCAmelCase_		, use_reentrant=lowerCAmelCase_ )
													lowerCamelCase			      =					sample.to(lowerCAmelCase_ )

													# up
													for up_block in self.up_blocks:
																lowerCamelCase			      =					torch.utils.checkpoint.checkpoint(
																    create_custom_forward(lowerCAmelCase_ )		, lowerCAmelCase_		, lowerCAmelCase_		, use_reentrant=lowerCAmelCase_ )
										else:
													# middle
													lowerCamelCase			      =					torch.utils.checkpoint.checkpoint(
													    create_custom_forward(self.mid_block )		, lowerCAmelCase_		, lowerCAmelCase_ )
													lowerCamelCase			      =					sample.to(lowerCAmelCase_ )

													# up
													for up_block in self.up_blocks:
																lowerCamelCase			      =					torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCAmelCase_ )		, lowerCAmelCase_		, lowerCAmelCase_ )
							else:
										# middle
										lowerCamelCase			      =					self.mid_block(lowerCAmelCase_		, lowerCAmelCase_ )
										lowerCamelCase			      =					sample.to(lowerCAmelCase_ )

										# up
										for up_block in self.up_blocks:
													lowerCamelCase			      =					up_block(lowerCAmelCase_		, lowerCAmelCase_ )

        # post-process
							if latent_embeds is None:
										lowerCamelCase			      =					self.conv_norm_out(lowerCAmelCase_ )
							else:
										lowerCamelCase			      =					self.conv_norm_out(lowerCAmelCase_		, lowerCAmelCase_ )
							lowerCamelCase			      =					self.conv_act(lowerCAmelCase_ )
							lowerCamelCase			      =					self.conv_out(lowerCAmelCase_ )

							return sample






class     __magic_name__							(       nn.Module				):






				'''simple docstring'''





				def __init__(       self		, _a		, _a		, _a		, _a=None		, _a="random"		, _a=False		, _a=True ):





							"""simple docstring"""
							super().__init__()
							lowerCamelCase			      =					n_e
							lowerCamelCase			      =					vq_embed_dim
							lowerCamelCase			      =					beta
							lowerCamelCase			      =					legacy

							lowerCamelCase			      =					nn.Embedding(self.n_e		, self.vq_embed_dim )
							self.embedding.weight.data.uniform_(-1.0 / self.n_e		, 1.0 / self.n_e )

							lowerCamelCase			      =					remap
							if self.remap is not None:
										self.register_buffer("""used"""		, torch.tensor(np.load(self.remap ) ) )
										lowerCamelCase			      =					self.used.shape[0]
										lowerCamelCase			      =					unknown_index  # "random" or "extra" or integer
										if self.unknown_index == "extra":
													lowerCamelCase			      =					self.re_embed
													lowerCamelCase			      =					self.re_embed + 1
										print(
										    f'Remapping {self.n_e} indices to {self.re_embed} indices. '
										    f'Using {self.unknown_index} for unknown indices.' )
							else:
										lowerCamelCase			      =					n_e

							lowerCamelCase			      =					sane_index_shape





				def    _lowerCAmelCase				(       self		, _a ):





							"""simple docstring"""
							lowerCamelCase			      =					inds.shape
							assert len(lowerCAmelCase_ ) > 1
							lowerCamelCase			      =					inds.reshape(ishape[0]		, -1 )
							lowerCamelCase			      =					self.used.to(lowerCAmelCase_ )
							lowerCamelCase			      =					(inds[:, :, None] == used[None, None, ...]).long()
							lowerCamelCase			      =					match.argmax(-1 )
							lowerCamelCase			      =					match.sum(2 ) < 1
							if self.unknown_index == "random":
										lowerCamelCase			      =					torch.randint(0		, self.re_embed		, size=new[unknown].shape ).to(device=new.device )
							else:
										lowerCamelCase			      =					self.unknown_index
							return new.reshape(lowerCAmelCase_ )





				def    _lowerCAmelCase				(       self		, _a ):





							"""simple docstring"""
							lowerCamelCase			      =					inds.shape
							assert len(lowerCAmelCase_ ) > 1
							lowerCamelCase			      =					inds.reshape(ishape[0]		, -1 )
							lowerCamelCase			      =					self.used.to(lowerCAmelCase_ )
							if self.re_embed > self.used.shape[0]:  # extra token
										lowerCamelCase			      =					0  # simply set to zero
							lowerCamelCase			      =					torch.gather(used[None, :][inds.shape[0] * [0], :]		, 1		, lowerCAmelCase_ )
							return back.reshape(lowerCAmelCase_ )





				def    _lowerCAmelCase				(       self		, _a ):





							"""simple docstring"""
							lowerCamelCase			      =					z.permute(0		, 2		, 3		, 1 ).contiguous()
							lowerCamelCase			      =					z.view(-1		, self.vq_embed_dim )

							# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
							lowerCamelCase			      =					torch.argmin(torch.cdist(lowerCAmelCase_		, self.embedding.weight )		, dim=1 )

							lowerCamelCase			      =					self.embedding(lowerCAmelCase_ ).view(z.shape )
							lowerCamelCase			      =					None
							lowerCamelCase			      =					None

							# compute loss for embedding
							if not self.legacy:
										lowerCamelCase			      =					self.beta * torch.mean((z_q.detach() - z) ** 2 ) + torch.mean((z_q - z.detach()) ** 2 )
							else:
										lowerCamelCase			      =					torch.mean((z_q.detach() - z) ** 2 ) + self.beta * torch.mean((z_q - z.detach()) ** 2 )

							# preserve gradients
							lowerCamelCase			      =					z + (z_q - z).detach()

							# reshape back to match original input shape
							lowerCamelCase			      =					z_q.permute(0		, 3		, 1		, 2 ).contiguous()

							if self.remap is not None:
										lowerCamelCase			      =					min_encoding_indices.reshape(z.shape[0]		, -1 )  # add batch axis
										lowerCamelCase			      =					self.remap_to_used(lowerCAmelCase_ )
										lowerCamelCase			      =					min_encoding_indices.reshape(-1		, 1 )  # flatten

							if self.sane_index_shape:
										lowerCamelCase			      =					min_encoding_indices.reshape(z_q.shape[0]		, z_q.shape[2]		, z_q.shape[3] )

							return z_q, loss, (perplexity, min_encodings, min_encoding_indices)





				def    _lowerCAmelCase				(       self		, _a		, _a ):





							"""simple docstring"""






							if self.remap is not None:
										lowerCamelCase			      =					indices.reshape(shape[0]		, -1 )  # add batch axis
										lowerCamelCase			      =					self.unmap_to_all(lowerCAmelCase_ )
										lowerCamelCase			      =					indices.reshape(-1 )  # flatten again

							# get quantized latent vectors
							lowerCamelCase			      =					self.embedding(lowerCAmelCase_ )

							if shape is not None:
										lowerCamelCase			      =					z_q.view(lowerCAmelCase_ )
										# reshape back to match original input shape
										lowerCamelCase			      =					z_q.permute(0		, 3		, 1		, 2 ).contiguous()

							return z_q






class     __magic_name__							(       _lowerCamelCase				):






				'''simple docstring'''





				def __init__(       self		, _a		, _a=False ):





							"""simple docstring"""
							lowerCamelCase			      =					parameters
							lowerCamelCase      ,   lowerCamelCase			      =					torch.chunk(lowerCAmelCase_		, 2		, dim=1 )
							lowerCamelCase			      =					torch.clamp(self.logvar		, -30.0		, 20.0 )
							lowerCamelCase			      =					deterministic
							lowerCamelCase			      =					torch.exp(0.5 * self.logvar )
							lowerCamelCase			      =					torch.exp(self.logvar )
							if self.deterministic:
										lowerCamelCase			      =					lowerCamelCase			      =					torch.zeros_like(
										    self.mean		, device=self.parameters.device		, dtype=self.parameters.dtype )





				def    _lowerCAmelCase				(       self		, _a = None ):





							"""simple docstring"""
							lowerCamelCase			      =					randn_tensor(
							    self.mean.shape		, generator=lowerCAmelCase_		, device=self.parameters.device		, dtype=self.parameters.dtype )
							lowerCamelCase			      =					self.mean + self.std * sample
							return x





				def    _lowerCAmelCase				(       self		, _a=None ):





							"""simple docstring"""






							if self.deterministic:
										return torch.Tensor([0.0] )
							else:
										if other is None:
													return 0.5 * torch.sum(torch.pow(self.mean		, 2 ) + self.var - 1.0 - self.logvar		, dim=[1, 2, 3] )
										else:
													return 0.5 * torch.sum(
													    torch.pow(self.mean - other.mean		, 2 ) / other.var
													    + self.var / other.var
													    - 1.0
													    - self.logvar
													    + other.logvar		, dim=[1, 2, 3]		, )





				def    _lowerCAmelCase				(       self		, _a		, _a=[1, 2, 3] ):





							"""simple docstring"""






							if self.deterministic:
										return torch.Tensor([0.0] )
							lowerCamelCase			      =					np.log(2.0 * np.pi )
							return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean		, 2 ) / self.var		, dim=lowerCAmelCase_ )





				def    _lowerCAmelCase				(       self ):





							"""simple docstring"""
							return self.mean





'''simple docstring'''







from __future__ import annotations

import random

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







def 				SCREAMING_SNAKE_CASE__	(   __A			,				__A	)       ->	tuple[str, float]:
	_snake_case   						=		len([g for position, g in enumerate(__A	) if g == main_target[position]]	)
	return (item, float(__A	))







def 				SCREAMING_SNAKE_CASE__	(   __A			,				__A	)       ->	tuple[str, str]:
	_snake_case   						=		random.randint(0			,				len(__A	) - 1	)
	_snake_case   						=		parent_a[:random_slice] + parent_a[random_slice:]
	_snake_case   						=		parent_a[:random_slice] + parent_a[random_slice:]
	return (child_a, child_a)







def 				SCREAMING_SNAKE_CASE__	(   __A			,				__A	)       ->	str:
	_snake_case   						=		list(__A	)
	if random.uniform(0			,				1	) < MUTATION_PROBABILITY:
		_snake_case   						=		random.choice(__A	)
	return "".join(__A	)







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

		_snake_case						, _snake_case   						=		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 				SCREAMING_SNAKE_CASE__	(   __A			,				__A			,				__A = True	)       ->	tuple[int, int, str]:

	# Verify if N_POPULATION is bigger than N_SELECTED
	if N_POPULATION < N_SELECTED:
		_snake_case   						=		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.
	_snake_case   						=		sorted({c for c in target if c not in genes}	)
	if not_in_genes_list:
		_snake_case   						=		F'{not_in_genes_list} is not in genes list, evolution cannot converge'
		raise ValueError(__A	)

	# Generate random starting population.
	_snake_case   						=		[]
	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.
	_snake_case						, _snake_case   						=		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.
		_snake_case   						=		[evaluate(__A			,				__A	) for item in population]

		# Check if there is a matching evolution.
		_snake_case   						=		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.
		_snake_case   						=		population[: int(N_POPULATION / 3	)]
		population.clear()
		population.extend(__A	)
		# Normalize population score to be between 0 and 1.
		_snake_case   						=		[
		    (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__":
					lowercase    :							str      =					(
					    "This is a genetic algorithm to evaluate, combine, evolve, and mutate a string!"
					)
					lowercase    :							str      =					list(
					    " ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklm"
					    "nopqrstuvwxyz.,;!?+-*#@^'èéòà€ù=)(&%$£/\\"
					)
					lowercase      ,  lowercase      ,  lowercase    :							Tuple      =					basic(target_str, genes_list)
					print(
					    F'''\nGeneration: {generation}\nTotal Population: {population}\nTarget: {target}'''
					)

