






import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
							from huggingface_hub import snapshot_download
							from pyctcdecode import BeamSearchDecoderCTC

							from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
							from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
							from transformers import WavaVecaForCTC




@require_pyctcdecode
class 		UpperCAmelCase_						(	unittest.TestCase					):


						'''simple docstring'''




						def       _lowerCAmelCase       (	self				) ->				Optional[Any]:
								snake_case_				:							Any						 =       "| <pad> <unk> <s> </s> a b c d e f g h i j k".split()
								snake_case_				:							Any						 =       dict(zip(__a		,				range(len(__a				)				)				)				)

								snake_case_				:							Union[str, Any]						 =       {
								    "unk_token": "<unk>",
								    "bos_token": "<s>",
								    "eos_token": "</s>",
								}
								snake_case_				:							List[str]						 =       {
								    "feature_size": 1,
								    "padding_value": 0.0,
								    "sampling_rate": 1_6000,
								    "return_attention_mask": False,
								    "do_normalize": True,
								}

								snake_case_				:							Optional[Any]						 =       tempfile.mkdtemp()
								snake_case_				:							Tuple						 =       os.path.join(self.tmpdirname		,				VOCAB_FILES_NAMES["vocab_file"]				)
								snake_case_				:							List[Any]						 =       os.path.join(self.tmpdirname		,				__a				)
								with open(self.vocab_file		,				"w"		,				encoding="utf-8"				) as fp:
										fp.write(json.dumps(__a				) + "\n"				)

								with open(self.feature_extraction_file		,				"w"		,				encoding="utf-8"				) as fp:
										fp.write(json.dumps(__a				) + "\n"				)

								# load decoder from hub
								snake_case_				:							List[Any]						 =       "hf-internal-testing/ngram-beam-search-decoder"






						def       _lowerCAmelCase       (	self		,				**_SCREAMING_SNAKE_CASE				) ->				int:
								snake_case_				:							Tuple						 =       self.add_kwargs_tokens_map.copy()
								kwargs.update(__a				)
								return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname		,				**__a				)






						def       _lowerCAmelCase       (	self		,				**_SCREAMING_SNAKE_CASE				) ->				int:
								return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname		,				**__a				)






						def       _lowerCAmelCase       (	self		,				**_SCREAMING_SNAKE_CASE				) ->				Tuple:
								return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name		,				**__a				)






						def       _lowerCAmelCase       (	self				) ->				List[str]:
								shutil.rmtree(self.tmpdirname				)






						def       _lowerCAmelCase       (	self				) ->				Union[str, Any]:
								snake_case_				:							List[Any]						 =       self.get_tokenizer()
								snake_case_				:							Dict						 =       self.get_feature_extractor()
								snake_case_				:							Dict						 =       self.get_decoder()

								snake_case_				:							int						 =       WavaVecaProcessorWithLM(tokenizer=__a		,				feature_extractor=__a		,				decoder=__a				)

								processor.save_pretrained(self.tmpdirname				)
								snake_case_				:							Optional[Any]						 =       WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname				)

								# tokenizer
								self.assertEqual(processor.tokenizer.get_vocab()		,				tokenizer.get_vocab()				)
								self.assertIsInstance(processor.tokenizer		,				__a				)

								# feature extractor
								self.assertEqual(processor.feature_extractor.to_json_string()		,				feature_extractor.to_json_string()				)
								self.assertIsInstance(processor.feature_extractor		,				__a				)

								# decoder
								self.assertEqual(processor.decoder._alphabet.labels		,				decoder._alphabet.labels				)
								self.assertEqual(
								    processor.decoder.model_container[decoder._model_key]._unigram_set		,				decoder.model_container[decoder._model_key]._unigram_set		,				)
								self.assertIsInstance(processor.decoder		,				__a				)






						def       _lowerCAmelCase       (	self				) ->				Optional[int]:
								snake_case_				:							List[Any]						 =       WavaVecaProcessorWithLM(
								    tokenizer=self.get_tokenizer()		,				feature_extractor=self.get_feature_extractor()		,				decoder=self.get_decoder()				)
								processor.save_pretrained(self.tmpdirname				)

								# make sure that error is thrown when decoder alphabet doesn't match
								snake_case_				:							Union[str, Any]						 =       WavaVecaProcessorWithLM.from_pretrained(
								    self.tmpdirname		,				alpha=5.0		,				beta=3.0		,				score_boundary=-7.0		,				unk_score_offset=3				)

								# decoder
								self.assertEqual(processor.language_model.alpha		,				5.0				)
								self.assertEqual(processor.language_model.beta		,				3.0				)
								self.assertEqual(processor.language_model.score_boundary		,				-7.0				)
								self.assertEqual(processor.language_model.unk_score_offset		,				3				)






						def       _lowerCAmelCase       (	self				) ->				Union[str, Any]:
								snake_case_				:							int						 =       self.get_tokenizer()
								# add token to trigger raise
								tokenizer.add_tokens(["xx"]				)
								with self.assertRaisesRegex(__a		,				"include"				):
										WavaVecaProcessorWithLM(
										    tokenizer=__a		,				feature_extractor=self.get_feature_extractor()		,				decoder=self.get_decoder()				)






						def       _lowerCAmelCase       (	self				) ->				List[str]:
								snake_case_				:							Any						 =       self.get_feature_extractor()
								snake_case_				:							Dict						 =       self.get_tokenizer()
								snake_case_				:							str						 =       self.get_decoder()

								snake_case_				:							Tuple						 =       WavaVecaProcessorWithLM(tokenizer=__a		,				feature_extractor=__a		,				decoder=__a				)

								snake_case_				:							Union[str, Any]						 =       floats_list((3, 1000)				)

								snake_case_				:							List[Any]						 =       feature_extractor(__a		,				return_tensors="np"				)
								snake_case_				:							Optional[Any]						 =       processor(__a		,				return_tensors="np"				)

								for key in input_feat_extract.keys():
										self.assertAlmostEqual(input_feat_extract[key].sum()		,				input_processor[key].sum()		,				delta=1e-2				)






						def       _lowerCAmelCase       (	self				) ->				int:
								snake_case_				:							str						 =       self.get_feature_extractor()
								snake_case_				:							List[str]						 =       self.get_tokenizer()
								snake_case_				:							Union[str, Any]						 =       self.get_decoder()

								snake_case_				:							Dict						 =       WavaVecaProcessorWithLM(tokenizer=__a		,				feature_extractor=__a		,				decoder=__a				)

								snake_case_				:							Optional[int]						 =       "This is a test string"

								snake_case_				:							Optional[int]						 =       processor(text=__a				)

								snake_case_				:							Any						 =       tokenizer(__a				)

								for key in encoded_tok.keys():
										self.assertListEqual(encoded_tok[key]		,				encoded_processor[key]				)






						def       _lowerCAmelCase       (	self		,				_SCREAMING_SNAKE_CASE=(2, 10, 16)		,				_SCREAMING_SNAKE_CASE=77				) ->				List[Any]:
								np.random.seed(__a				)
								return np.random.rand(*__a				)






						def       _lowerCAmelCase       (	self				) ->				Union[str, Any]:
								snake_case_				:							Optional[Any]						 =       self.get_feature_extractor()
								snake_case_				:							List[Any]						 =       self.get_tokenizer()
								snake_case_				:							Optional[Any]						 =       self.get_decoder()

								snake_case_				:							Optional[int]						 =       WavaVecaProcessorWithLM(tokenizer=__a		,				feature_extractor=__a		,				decoder=__a				)

								snake_case_				:							Union[str, Any]						 =       self._get_dummy_logits(shape=(10, 16)		,				seed=13				)

								snake_case_				:							List[Any]						 =       processor.decode(__a				)

								snake_case_				:							str						 =       decoder.decode_beams(__a				)[0]

								self.assertEqual(decoded_decoder[0]		,				decoded_processor.text				)
								self.assertEqual("</s> <s> </s>"		,				decoded_processor.text				)
								self.assertEqual(decoded_decoder[-2]		,				decoded_processor.logit_score				)
								self.assertEqual(decoded_decoder[-1]		,				decoded_processor.lm_score				)






						@parameterized.expand([[None], ["fork"], ["spawn"]]				)
						def       _lowerCAmelCase       (	self		,				_SCREAMING_SNAKE_CASE				) ->				Any:
								snake_case_				:							str						 =       self.get_feature_extractor()
								snake_case_				:							int						 =       self.get_tokenizer()
								snake_case_				:							str						 =       self.get_decoder()

								snake_case_				:							Dict						 =       WavaVecaProcessorWithLM(tokenizer=__a		,				feature_extractor=__a		,				decoder=__a				)

								snake_case_				:							Any						 =       self._get_dummy_logits()

								# note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
								#       otherwise, the LM won't be available to the pool's sub-processes.
								# manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
								if pool_context is None:
										snake_case_				:							List[str]						 =       processor.batch_decode(__a				)
								else:
										with get_context(__a				).Pool() as pool:
												snake_case_				:							List[str]						 =       processor.batch_decode(__a		,				__a				)

								snake_case_				:							Any						 =       list(__a				)

								with get_context("fork"				).Pool() as p:
										snake_case_				:							Optional[int]						 =       decoder.decode_beams_batch(__a		,				__a				)

								snake_case_				:							Dict						 =       [], [], []
								for beams in decoded_beams:
										texts_decoder.append(beams[0][0]				)
										logit_scores_decoder.append(beams[0][-2]				)
										lm_scores_decoder.append(beams[0][-1]				)

								self.assertListEqual(__a		,				decoded_processor.text				)
								self.assertListEqual(["<s> <s> </s>", "<s> <s> <s>"]		,				decoded_processor.text				)
								self.assertListEqual(__a		,				decoded_processor.logit_score				)
								self.assertListEqual(__a		,				decoded_processor.lm_score				)






						def       _lowerCAmelCase       (	self				) ->				int:
								snake_case_				:							Dict						 =       self.get_feature_extractor()
								snake_case_				:							Any						 =       self.get_tokenizer()
								snake_case_				:							Optional[int]						 =       self.get_decoder()

								snake_case_				:							Optional[int]						 =       WavaVecaProcessorWithLM(tokenizer=__a		,				feature_extractor=__a		,				decoder=__a				)

								snake_case_				:							int						 =       self._get_dummy_logits()

								snake_case_				:							Tuple						 =       15
								snake_case_				:							List[Any]						 =       -20.0
								snake_case_				:							int						 =       -4.0

								snake_case_				:							Union[str, Any]						 =       processor.batch_decode(
								    __a		,				beam_width=__a		,				beam_prune_logp=__a		,				token_min_logp=__a		,				)
								snake_case_				:							List[str]						 =       decoded_processor_out.text

								snake_case_				:							Optional[int]						 =       list(__a				)

								with get_context("fork"				).Pool() as pool:
										snake_case_				:							List[str]						 =       decoder.decode_beams_batch(
										    __a		,				__a		,				beam_width=__a		,				beam_prune_logp=__a		,				token_min_logp=__a		,				)

								snake_case_				:							int						 =       [d[0][0] for d in decoded_decoder_out]
								snake_case_				:							List[Any]						 =       [d[0][2] for d in decoded_decoder_out]
								snake_case_				:							Optional[Any]						 =       [d[0][3] for d in decoded_decoder_out]

								self.assertListEqual(__a		,				__a				)
								self.assertListEqual(["</s> <s> <s>", "<s> <s> <s>"]		,				__a				)

								self.assertTrue(np.array_equal(__a		,				decoded_processor_out.logit_score				)				)
								self.assertTrue(np.allclose([-20.054, -18.447]		,				__a		,				atol=1e-3				)				)

								self.assertTrue(np.array_equal(__a		,				decoded_processor_out.lm_score				)				)
								self.assertTrue(np.allclose([-15.554, -13.9474]		,				__a		,				atol=1e-3				)				)






						def       _lowerCAmelCase       (	self				) ->				Any:
								snake_case_				:							Optional[int]						 =       self.get_feature_extractor()
								snake_case_				:							Union[str, Any]						 =       self.get_tokenizer()
								snake_case_				:							Any						 =       self.get_decoder()

								snake_case_				:							Dict						 =       WavaVecaProcessorWithLM(tokenizer=__a		,				feature_extractor=__a		,				decoder=__a				)

								snake_case_				:							Union[str, Any]						 =       self._get_dummy_logits()

								snake_case_				:							Optional[Any]						 =       2.0
								snake_case_				:							Union[str, Any]						 =       5.0
								snake_case_				:							Tuple						 =       -20.0
								snake_case_				:							str						 =       True

								snake_case_				:							Any						 =       processor.batch_decode(
								    __a		,				alpha=__a		,				beta=__a		,				unk_score_offset=__a		,				lm_score_boundary=__a		,				)
								snake_case_				:							str						 =       decoded_processor_out.text

								snake_case_				:							Union[str, Any]						 =       list(__a				)
								decoder.reset_params(
								    alpha=__a		,				beta=__a		,				unk_score_offset=__a		,				lm_score_boundary=__a		,				)

								with get_context("fork"				).Pool() as pool:
										snake_case_				:							List[str]						 =       decoder.decode_beams_batch(
										    __a		,				__a		,				)

								snake_case_				:							str						 =       [d[0][0] for d in decoded_decoder_out]

								self.assertListEqual(__a		,				__a				)
								self.assertListEqual(["<s> </s> <s> </s> </s>", "</s> </s> <s> </s> </s>"]		,				__a				)
								snake_case_				:							Tuple						 =       processor.decoder.model_container[processor.decoder._model_key]
								self.assertEqual(lm_model.alpha		,				2.0				)
								self.assertEqual(lm_model.beta		,				5.0				)
								self.assertEqual(lm_model.unk_score_offset		,				-20.0				)
								self.assertEqual(lm_model.score_boundary		,				__a				)






						def       _lowerCAmelCase       (	self				) ->				Dict:
								snake_case_				:							List[Any]						 =       WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"				)

								snake_case_				:							str						 =       processor.decoder.model_container[processor.decoder._model_key]
								snake_case_				:							Optional[Any]						 =       Path(language_model._kenlm_model.path.decode("utf-8"				)				).parent.parent.absolute()

								snake_case_				:							Dict						 =       os.listdir(__a				)
								snake_case_				:							Dict						 =       ["alphabet.json", "language_model"]

								downloaded_decoder_files.sort()
								expected_decoder_files.sort()

								# test that only decoder relevant files from
								# https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
								# are downloaded and none of the rest (e.g. README.md, ...)
								self.assertListEqual(__a		,				__a				)






						def       _lowerCAmelCase       (	self				) ->				Optional[Any]:
								snake_case_				:							List[str]						 =       snapshot_download("hf-internal-testing/processor_with_lm"				)

								snake_case_				:							int						 =       WavaVecaProcessorWithLM.from_pretrained(__a				)

								snake_case_				:							Optional[Any]						 =       processor.decoder.model_container[processor.decoder._model_key]
								snake_case_				:							Dict						 =       Path(language_model._kenlm_model.path.decode("utf-8"				)				).parent.parent.absolute()

								snake_case_				:							Dict						 =       os.listdir(__a				)
								snake_case_				:							Optional[Any]						 =       os.listdir(__a				)

								local_decoder_files.sort()
								expected_decoder_files.sort()

								# test that both decoder form hub and local files in cache are the same
								self.assertListEqual(__a		,				__a				)






						def       _lowerCAmelCase       (	self				) ->				List[str]:
								snake_case_				:							List[Any]						 =       WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"				)
								snake_case_				:							Dict						 =       AutoProcessor.from_pretrained("hf-internal-testing/processor_with_lm"				)

								snake_case_				:							Dict						 =       floats_list((3, 1000)				)

								snake_case_				:							Optional[Any]						 =       processor_wavaveca(__a		,				return_tensors="np"				)
								snake_case_				:							Dict						 =       processor_auto(__a		,				return_tensors="np"				)

								for key in input_wavaveca.keys():
										self.assertAlmostEqual(input_wavaveca[key].sum()		,				input_auto[key].sum()		,				delta=1e-2				)

								snake_case_				:							List[str]						 =       self._get_dummy_logits()

								snake_case_				:							str						 =       processor_wavaveca.batch_decode(__a				)
								snake_case_				:							str						 =       processor_auto.batch_decode(__a				)

								self.assertListEqual(decoded_wavaveca.text		,				decoded_auto.text				)






						def       _lowerCAmelCase       (	self				) ->				Optional[Any]:
								snake_case_				:							Any						 =       self.get_feature_extractor()
								snake_case_				:							Optional[int]						 =       self.get_tokenizer()
								snake_case_				:							Any						 =       self.get_decoder()

								snake_case_				:							Tuple						 =       WavaVecaProcessorWithLM(tokenizer=__a		,				feature_extractor=__a		,				decoder=__a				)

								self.assertListEqual(
								    processor.model_input_names		,				feature_extractor.model_input_names		,				msg="`processor` and `feature_extractor` model input names do not match"		,				)






						@staticmethod
						def       _lowerCAmelCase       (	_SCREAMING_SNAKE_CASE		,				_SCREAMING_SNAKE_CASE				) ->				int:
								snake_case_				:							Dict						 =       [d[key] for d in offsets]
								return retrieved_list






						def       _lowerCAmelCase       (	self				) ->				Optional[Any]:
								snake_case_				:							Any						 =       WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"				)
								snake_case_				:							int						 =       self._get_dummy_logits()[0]

								snake_case_				:							List[str]						 =       processor.decode(__a		,				output_word_offsets=__a				)
								# check Wav2Vec2CTCTokenizerOutput keys for word
								self.assertEqual(len(outputs.keys()				)		,				4				)
								self.assertTrue("text" in outputs				)
								self.assertTrue("word_offsets" in outputs				)
								self.assertTrue(isinstance(__a		,				__a				)				)

								self.assertEqual(" ".join(self.get_from_offsets(outputs["word_offsets"]		,				"word"				)				)		,				outputs.text				)
								self.assertListEqual(self.get_from_offsets(outputs["word_offsets"]		,				"word"				)		,				["<s>", "<s>", "</s>"]				)
								self.assertListEqual(self.get_from_offsets(outputs["word_offsets"]		,				"start_offset"				)		,				[0, 2, 4]				)
								self.assertListEqual(self.get_from_offsets(outputs["word_offsets"]		,				"end_offset"				)		,				[1, 3, 5]				)






						def       _lowerCAmelCase       (	self				) ->				Optional[Any]:
								snake_case_				:							List[str]						 =       WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"				)
								snake_case_				:							List[Any]						 =       self._get_dummy_logits()

								snake_case_				:							str						 =       processor.batch_decode(__a		,				output_word_offsets=__a				)

								# check Wav2Vec2CTCTokenizerOutput keys for word
								self.assertEqual(len(outputs.keys()				)		,				4				)
								self.assertTrue("text" in outputs				)
								self.assertTrue("word_offsets" in outputs				)
								self.assertTrue(isinstance(__a		,				__a				)				)

								self.assertListEqual(
								    [" ".join(self.get_from_offsets(__a		,				"word"				)				) for o in outputs["word_offsets"]]		,				outputs.text				)
								self.assertListEqual(self.get_from_offsets(outputs["word_offsets"][0]		,				"word"				)		,				["<s>", "<s>", "</s>"]				)
								self.assertListEqual(self.get_from_offsets(outputs["word_offsets"][0]		,				"start_offset"				)		,				[0, 2, 4]				)
								self.assertListEqual(self.get_from_offsets(outputs["word_offsets"][0]		,				"end_offset"				)		,				[1, 3, 5]				)






						@slow
						@require_torch
						@require_torchaudio
						def       _lowerCAmelCase       (	self				) ->				Union[str, Any]:
								import torch

								snake_case_				:							Optional[int]						 =       load_dataset("common_voice"		,				"en"		,				split="train"		,				streaming=__a				)
								snake_case_				:							Optional[int]						 =       ds.cast_column("audio"		,				datasets.Audio(sampling_rate=1_6000				)				)
								snake_case_				:							Any						 =       iter(__a				)
								snake_case_				:							Union[str, Any]						 =       next(__a				)

								snake_case_				:							Optional[Any]						 =       AutoProcessor.from_pretrained("patrickvonplaten/wav2vec2-base-100h-with-lm"				)
								snake_case_				:							List[Any]						 =       WavaVecaForCTC.from_pretrained("patrickvonplaten/wav2vec2-base-100h-with-lm"				)

								# compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
								snake_case_				:							int						 =       processor(sample["audio"]["array"]		,				return_tensors="pt"				).input_values

								with torch.no_grad():
										snake_case_				:							Optional[Any]						 =       model(__a				).logits.cpu().numpy()

								snake_case_				:							Dict						 =       processor.decode(logits[0]		,				output_word_offsets=__a				)

								snake_case_				:							Any						 =       model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
								snake_case_				:							str						 =       [
								    {
								        "start_time": d["start_offset"] * time_offset,
								        "end_time": d["end_offset"] * time_offset,
								        "word": d["word"],
								    }
								    for d in output["word_offsets"]
								]

								snake_case_				:							Dict						 =       "WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL"

								# output words
								self.assertEqual(" ".join(self.get_from_offsets(__a		,				"word"				)				)		,				__a				)
								self.assertEqual(" ".join(self.get_from_offsets(__a		,				"word"				)				)		,				output.text				)

								# output times
								snake_case_				:							List[str]						 =       torch.tensor(self.get_from_offsets(__a		,				"start_time"				)				)
								snake_case_				:							List[Any]						 =       torch.tensor(self.get_from_offsets(__a		,				"end_time"				)				)

								# fmt: off
								snake_case_				:							List[Any]						 =       torch.tensor([1.4199, 1.6599, 2.2599, 3.0, 3.24, 3.5999, 3.7999, 4.0999, 4.26, 4.94, 5.28, 5.6599, 5.78, 5.94, 6.32, 6.5399, 6.6599]				)
								snake_case_				:							Tuple						 =       torch.tensor([1.5399, 1.8999, 2.9, 3.16, 3.5399, 3.72, 4.0199, 4.1799, 4.76, 5.1599, 5.5599, 5.6999, 5.86, 6.1999, 6.38, 6.6199, 6.94]				)
								# fmt: on

								self.assertTrue(torch.allclose(__a		,				__a		,				atol=0.01				)				)
								self.assertTrue(torch.allclose(__a		,				__a		,				atol=0.01				)				)








from __future__ import annotations

import random

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

def      lowerCAmelCase__	(    _a       :			str  ,      _a       :			str    ):
		snake_case_				:							Union[str, Any]						 =       len([g for position, g in enumerate(_a    ) if g == main_target[position]]    )
		return (item, float(_a    ))

def      lowerCAmelCase__	(    _a       :			str  ,      _a       :			str    ):
		snake_case_				:							Tuple						 =       random.randint(0  ,      len(_a    ) - 1    )
		snake_case_				:							Any						 =       parent_a[:random_slice] + parent_a[random_slice:]
		snake_case_				:							Optional[Any]						 =       parent_a[:random_slice] + parent_a[random_slice:]
		return (child_a, child_a)

def      lowerCAmelCase__	(    _a       :			str  ,      _a       :			list[str]    ):
		snake_case_				:							str						 =       list(_a    )
		if random.uniform(0  ,      1    ) < MUTATION_PROBABILITY:
				snake_case_				:							Optional[Any]						 =       random.choice(_a    )
		return "".join(_a    )

def      lowerCAmelCase__	(    _a       :			tuple[str, float]  ,      _a       :			list[tuple[str, float]]  ,      _a       :			list[str]  ,      ):
		snake_case_				:							Tuple						 =       []
		# Generate more children proportionally to the fitness score.
		snake_case_				:							Optional[int]						 =       int(parent_a[1] * 1_00    ) + 1
		snake_case_				:							Tuple						 =       10 if child_n >= 10 else child_n
		for _ in range(_a    ):
				snake_case_				:							int						 =       population_score[random.randint(0  ,      _a    )][0]

				snake_case_ ,       snake_case_				:							Any						 =       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      lowerCAmelCase__	(    _a       :			str  ,      _a       :			list[str]  ,      _a       :			bool = True    ):

		# Verify if N_POPULATION is bigger than N_SELECTED
		if N_POPULATION < N_SELECTED:
				snake_case_				:							Any						 =       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_				:							Optional[Any]						 =       sorted({c for c in target if c not in genes}    )
		if not_in_genes_list:
				snake_case_				:							str						 =       F'''{not_in_genes_list} is not in genes list, evolution cannot converge'''
				raise ValueError(_a    )

		# Generate random starting population.
		snake_case_				:							Optional[int]						 =       []
		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_				:							Dict						 =       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_				:							Optional[int]						 =       [evaluate(_a  ,      _a    ) for item in population]

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