




'''simple docstring'''




from __future__ import annotations

import random

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





def      a_							(      __snake_case							:			str   ,       __snake_case							:			Dict   )							->						tuple[str, float]:



       """simple docstring"""
       lowerCamelCase_							=len([g for position, g in enumerate(_lowerCAmelCase   ) if g == main_target[position]]   )
       return (item, float(_lowerCAmelCase   ))





def      a_							(      __snake_case							:			str   ,       __snake_case							:			Optional[int]   )							->						tuple[str, str]:



       """simple docstring"""
       lowerCamelCase_							=random.randint(0   ,       len(_lowerCAmelCase   ) - 1   )
       lowerCamelCase_							=parent_a[:random_slice] + parent_a[random_slice:]
       lowerCamelCase_							=parent_a[:random_slice] + parent_a[random_slice:]
       return (child_a, child_a)





def      a_							(      __snake_case							:			Optional[Any]   ,       __snake_case							:			Dict   )							->						str:



       """simple docstring"""
       lowerCamelCase_							=list(_lowerCAmelCase   )
       if random.uniform(0   ,       1   ) < MUTATION_PROBABILITY:
              lowerCamelCase_							=random.choice(_lowerCAmelCase   )
       return "".join(_lowerCAmelCase   )





def      a_							(      __snake_case							:			Any   ,       __snake_case							:			Tuple   ,       __snake_case							:			Union[str, Any]   ,       )							->						list[str]:



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

              lowerCamelCase_							=crossover(parent_a[0]   ,       _lowerCAmelCase   )
              # Append new string to the population list.
              pop.append(mutate(_lowerCAmelCase   ,       _lowerCAmelCase   )   )
              pop.append(mutate(_lowerCAmelCase   ,       _lowerCAmelCase   )   )
       return pop





def      a_							(      __snake_case							:			List[Any]   ,       __snake_case							:			Dict   ,       __snake_case							:			str = True   )							->						tuple[int, int, str]:



       """simple docstring"""






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

       # Generate random starting population.
       lowerCamelCase_							=[]
       for _ in range(_lowerCAmelCase   ):
              population.append(''''''.join([random.choice(_lowerCAmelCase   ) for i in range(len(_lowerCAmelCase   )   )]   )   )

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

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

              # 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.
              lowerCamelCase_							=[evaluate(_lowerCAmelCase   ,       _lowerCAmelCase   ) for item in population]

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

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


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









import gc
import random
import unittest

import numpy as np
import torch

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()






class SCREAMING_SNAKE_CASE__     (		SCREAMING_SNAKE_CASE__ ,						unittest.TestCase    ):




    '''simple docstring'''






    __lowerCamelCase			: Union[str, Any]					=      KandinskyVaaControlnetPipeline
    __lowerCamelCase			: Optional[int]					=      ["image_embeds", "negative_image_embeds", "hint"]
    __lowerCamelCase			: Dict					=      ["image_embeds", "negative_image_embeds", "hint"]
    __lowerCamelCase			: List[str]					=      [
        "generator",
        "height",
        "width",
        "latents",
        "guidance_scale",
        "num_inference_steps",
        "return_dict",
        "guidance_scale",
        "num_images_per_prompt",
        "output_type",
        "return_dict",
    ]
    __lowerCamelCase			: Dict					=      False






    @property
    def 			_lowerCAmelCase					(       self		):
        return 32






    @property
    def 			_lowerCAmelCase					(       self		):
        return 32






    @property
    def 			_lowerCAmelCase					(       self		):
        return self.time_input_dim






    @property
    def 			_lowerCAmelCase					(       self		):
        return self.time_input_dim * 4






    @property
    def 			_lowerCAmelCase					(       self		):
        return 100






    @property
    def 			_lowerCAmelCase					(       self		):
        torch.manual_seed(0		)

        A :       Any          =   {
            """in_channels""": 8,
            # Out channels is double in channels because predicts mean and variance
            """out_channels""": 8,
            """addition_embed_type""": """image_hint""",
            """down_block_types""": ("""ResnetDownsampleBlock2D""", """SimpleCrossAttnDownBlock2D"""),
            """up_block_types""": ("""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""),
            """mid_block_type""": """UNetMidBlock2DSimpleCrossAttn""",
            """block_out_channels""": (self.block_out_channels_a, self.block_out_channels_a * 2),
            """layers_per_block""": 1,
            """encoder_hid_dim""": self.text_embedder_hidden_size,
            """encoder_hid_dim_type""": """image_proj""",
            """cross_attention_dim""": self.cross_attention_dim,
            """attention_head_dim""": 4,
            """resnet_time_scale_shift""": """scale_shift""",
            """class_embed_type""": None,
        }

        A :       List[str]          =   UNetaDConditionModel(**lowerCamelCase__		)
        return model






    @property
    def 			_lowerCAmelCase					(       self		):
        return {
            "block_out_channels": [32, 32, 64, 64],
            "down_block_types": [
                "DownEncoderBlock2D",
                "DownEncoderBlock2D",
                "DownEncoderBlock2D",
                "AttnDownEncoderBlock2D",
            ],
            "in_channels": 3,
            "latent_channels": 4,
            "layers_per_block": 1,
            "norm_num_groups": 8,
            "norm_type": "spatial",
            "num_vq_embeddings": 12,
            "out_channels": 3,
            "up_block_types": ["AttnUpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D"],
            "vq_embed_dim": 4,
        }






    @property
    def 			_lowerCAmelCase					(       self		):
        torch.manual_seed(0		)
        A :       List[Any]          =   VQModel(**self.dummy_movq_kwargs		)
        return model






    def 			_lowerCAmelCase					(       self		):
        A :       Optional[Any]          =   self.dummy_unet
        A :       Tuple          =   self.dummy_movq

        A :       List[Any]          =   DDIMScheduler(
            num_train_timesteps=1000,		beta_schedule="""linear""",		beta_start=0.0_0085,		beta_end=0.012,		clip_sample=lowerCamelCase__,		set_alpha_to_one=lowerCamelCase__,		steps_offset=1,		prediction_type="""epsilon""",		thresholding=lowerCamelCase__,		)

        A :       int          =   {
            """unet""": unet,
            """scheduler""": scheduler,
            """movq""": movq,
        }
        return components






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__,		lowerCamelCase__=0		):
        A :       Optional[Any]          =   floats_tensor((1, self.text_embedder_hidden_size),		rng=random.Random(lowerCamelCase__		)		).to(lowerCamelCase__		)
        A :       List[Any]          =   floats_tensor((1, self.text_embedder_hidden_size),		rng=random.Random(seed + 1		)		).to(
            lowerCamelCase__		)

        # create hint
        A :       int          =   floats_tensor((1, 3, 64, 64),		rng=random.Random(lowerCamelCase__		)		).to(lowerCamelCase__		)

        if str(lowerCamelCase__		).startswith("""mps"""		):
            A :       Optional[Any]          =   torch.manual_seed(lowerCamelCase__		)
        else:
            A :       str          =   torch.Generator(device=lowerCamelCase__		).manual_seed(lowerCamelCase__		)
        A :       List[str]          =   {
            """image_embeds""": image_embeds,
            """negative_image_embeds""": negative_image_embeds,
            """hint""": hint,
            """generator""": generator,
            """height""": 64,
            """width""": 64,
            """guidance_scale""": 4.0,
            """num_inference_steps""": 2,
            """output_type""": """np""",
        }
        return inputs






    def 			_lowerCAmelCase					(       self		):
        A :       Dict          =   """cpu"""

        A :       List[str]          =   self.get_dummy_components()

        A :       Dict          =   self.pipeline_class(**lowerCamelCase__		)
        A :       Optional[Any]          =   pipe.to(lowerCamelCase__		)

        pipe.set_progress_bar_config(disable=lowerCamelCase__		)

        A :       int          =   pipe(**self.get_dummy_inputs(lowerCamelCase__		)		)
        A :       Union[str, Any]          =   output.images

        A :       str          =   pipe(
            **self.get_dummy_inputs(lowerCamelCase__		),		return_dict=lowerCamelCase__,		)[0]

        A :       Optional[int]          =   image[0, -3:, -3:, -1]
        A :       Any          =   image_from_tuple[0, -3:, -3:, -1]

        assert image.shape == (1, 64, 64, 3)

        A :       Dict          =   np.array(
            [0.695_9826, 0.86_8279, 0.755_8092, 0.6876_9467, 0.8580_5804, 0.6597_7496, 0.4488_5302, 0.595_9111, 0.425_1595]		)

        assert (
            np.abs(image_slice.flatten() - expected_slice		).max() < 1e-2
        ), f''' expected_slice {expected_slice}, but got {image_slice.flatten()}'''

        assert (
            np.abs(image_from_tuple_slice.flatten() - expected_slice		).max() < 1e-2
        ), f''' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'''






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




    '''simple docstring'''






    def 			_lowerCAmelCase					(       self		):
        # clean up the VRAM after each test
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()






    def 			_lowerCAmelCase					(       self		):
        A :       Tuple          =   load_numpy(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
            """/kandinskyv22/kandinskyv22_controlnet_robotcat_fp16.npy"""		)

        A :       Any          =   load_image(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
            """/kandinskyv22/hint_image_cat.png"""		)
        A :       Optional[Any]          =   torch.from_numpy(np.array(lowerCamelCase__		)		).float() / 255.0
        A :       List[str]          =   hint.permute(2,		0,		1		).unsqueeze(0		)

        A :       Union[str, Any]          =   KandinskyVaaPriorPipeline.from_pretrained(
            """kandinsky-community/kandinsky-2-2-prior""",		torch_dtype=torch.floataa		)
        pipe_prior.to(lowerCamelCase__		)

        A :       Tuple          =   KandinskyVaaControlnetPipeline.from_pretrained(
            """kandinsky-community/kandinsky-2-2-controlnet-depth""",		torch_dtype=torch.floataa		)
        A :       Union[str, Any]          =   pipeline.to(lowerCamelCase__		)
        pipeline.set_progress_bar_config(disable=lowerCamelCase__		)

        A :       Optional[Any]          =   """A robot, 4k photo"""

        A :       Union[str, Any]          =   torch.Generator(device="""cuda"""		).manual_seed(0		)
        A     ,							A :       int          =   pipe_prior(
            lowerCamelCase__,		generator=lowerCamelCase__,		num_inference_steps=5,		negative_prompt="""""",		).to_tuple()

        A :       Union[str, Any]          =   torch.Generator(device="""cuda"""		).manual_seed(0		)
        A :       int          =   pipeline(
            image_embeds=lowerCamelCase__,		negative_image_embeds=lowerCamelCase__,		hint=lowerCamelCase__,		generator=lowerCamelCase__,		num_inference_steps=100,		output_type="""np""",		)

        A :       Optional[int]          =   output.images[0]

        assert image.shape == (512, 512, 3)

        assert_mean_pixel_difference(lowerCamelCase__,		lowerCamelCase__		)


