
"""simple docstring"""



from typing import List, Optional, Union

import torch

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


a       :						str       		=		logging.get_logger(__name__)  # pylint: disable=invalid-name

a       :						Dict       		=		'''
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained("kandinsky-community/kandinsky-2-2-prior")
        >>> pipe_prior.to("cuda")
        >>> prompt = "red cat, 4k photo"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> zero_image_emb = out.negative_image_embeds
        >>> pipe = KandinskyV22Pipeline.from_pretrained("kandinsky-community/kandinsky-2-2-decoder")
        >>> pipe.to("cuda")
        >>> image = pipe(
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=50,
        ... ).images
        >>> image[0].save("cat.png")
        ```
'''


def 			_UpperCamelCase						(    _A       ,   _A       ,   _A=8   )     ->	Any:





    """simple docstring"""

    _UpperCAmelCase        = height // scale_factor**2
    if height % scale_factor**2 != 0:
        new_height += 1
    _UpperCAmelCase        = width // scale_factor**2
    if width % scale_factor**2 != 0:
        new_width += 1
    return new_height * scale_factor, new_width * scale_factor







class 		a_    (			_UpperCAmelCase						):







    def __init__(      self      :							int  ,       __UpperCamelCase      :							UNetaDConditionModel  ,       __UpperCamelCase      :							DDPMScheduler  ,       __UpperCamelCase      :							VQModel  ,       )		->str:






        '''simple docstring'''






        super().__init__()

        self.register_modules(
            unet=__UpperCamelCase  ,       scheduler=__UpperCamelCase  ,       movq=__UpperCamelCase  ,       )
        _UpperCAmelCase        = 2 ** (len(self.movq.config.block_out_channels  ) - 1)







    def   _snake_case       (      self      :							List[Any]  ,       __UpperCamelCase      :							int  ,       __UpperCamelCase      :							Optional[int]  ,       __UpperCamelCase      :							List[str]  ,       __UpperCamelCase      :							Dict  ,       __UpperCamelCase      :							Optional[Any]  ,       __UpperCamelCase      :							Union[str, Any]  )		->int:






        '''simple docstring'''






        if latents is None:
            _UpperCAmelCase        = randn_tensor(__UpperCamelCase  ,       generator=__UpperCamelCase  ,       device=__UpperCamelCase  ,       dtype=__UpperCamelCase  )
        else:
            if latents.shape != shape:
                raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {shape}"""  )
            _UpperCAmelCase        = latents.to(__UpperCamelCase  )

        _UpperCAmelCase        = latents * scheduler.init_noise_sigma
        return latents







    def   _snake_case       (      self      :							str  ,       __UpperCamelCase      :							Union[str, Any]=0  )		->str:






        '''simple docstring'''






        if is_accelerate_available():
            from accelerate import cpu_offload
        else:
            raise ImportError("""Please install accelerate via `pip install accelerate`"""  )

        _UpperCAmelCase        = torch.device(f"""cuda:{gpu_id}"""  )

        _UpperCAmelCase        = [
            self.unet,
            self.movq,
        ]
        for cpu_offloaded_model in models:
            if cpu_offloaded_model is not None:
                cpu_offload(__UpperCamelCase  ,       __UpperCamelCase  )







    def   _snake_case       (      self      :							Union[str, Any]  ,       __UpperCamelCase      :							Union[str, Any]=0  )		->Any:






        '''simple docstring'''






        if is_accelerate_available() and is_accelerate_version(""">="""  ,       """0.17.0.dev0"""  ):
            from accelerate import cpu_offload_with_hook
        else:
            raise ImportError("""`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher."""  )

        _UpperCAmelCase        = torch.device(f"""cuda:{gpu_id}"""  )

        if self.device.type != "cpu":
            self.to("""cpu"""  ,       silence_dtype_warnings=__UpperCamelCase  )
            torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

        _UpperCAmelCase        = None
        for cpu_offloaded_model in [self.unet, self.movq]:
            _UpperCAmelCase     ,_UpperCAmelCase        = cpu_offload_with_hook(__UpperCamelCase  ,       __UpperCamelCase  ,       prev_module_hook=__UpperCamelCase  )

        # We'll offload the last model manually.
        _UpperCAmelCase        = hook







    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def   _snake_case       (      self      :							str  )		->str:






        '''simple docstring'''






        if not hasattr(self.unet  ,       """_hf_hook"""  ):
            return self.device
        for module in self.unet.modules():
            if (
                hasattr(__UpperCamelCase  ,       """_hf_hook"""  )
                and hasattr(module._hf_hook  ,       """execution_device"""  )
                and module._hf_hook.execution_device is not None
            ):
                return torch.device(module._hf_hook.execution_device  )
        return self.device







    @torch.no_grad()
    @replace_example_docstring(__UpperCamelCase  )
    def __call__(      self      :							int  ,       __UpperCamelCase      :							Union[torch.FloatTensor, List[torch.FloatTensor]]  ,       __UpperCamelCase      :							Union[torch.FloatTensor, List[torch.FloatTensor]]  ,       __UpperCamelCase      :							int = 5_12  ,       __UpperCamelCase      :							int = 5_12  ,       __UpperCamelCase      :							int = 1_00  ,       __UpperCamelCase      :							float = 4.0  ,       __UpperCamelCase      :							int = 1  ,       __UpperCamelCase      :							Optional[Union[torch.Generator, List[torch.Generator]]] = None  ,       __UpperCamelCase      :							Optional[torch.FloatTensor] = None  ,       __UpperCamelCase      :							Optional[str] = "pil"  ,       __UpperCamelCase      :							bool = True  ,       )		->Optional[int]:






        '''simple docstring'''






        _UpperCAmelCase        = self._execution_device

        _UpperCAmelCase        = guidance_scale > 1.0

        if isinstance(__UpperCamelCase  ,       __UpperCamelCase  ):
            _UpperCAmelCase        = torch.cat(__UpperCamelCase  ,       dim=0  )
        _UpperCAmelCase        = image_embeds.shape[0] * num_images_per_prompt
        if isinstance(__UpperCamelCase  ,       __UpperCamelCase  ):
            _UpperCAmelCase        = torch.cat(__UpperCamelCase  ,       dim=0  )

        if do_classifier_free_guidance:
            _UpperCAmelCase        = image_embeds.repeat_interleave(__UpperCamelCase  ,       dim=0  )
            _UpperCAmelCase        = negative_image_embeds.repeat_interleave(__UpperCamelCase  ,       dim=0  )

        _UpperCAmelCase        = torch.cat([negative_image_embeds, image_embeds]  ,       dim=0  ).to(dtype=self.unet.dtype  ,       device=__UpperCamelCase  )

        self.scheduler.set_timesteps(__UpperCamelCase  ,       device=__UpperCamelCase  )
        _UpperCAmelCase        = self.scheduler.timesteps

        _UpperCAmelCase        = self.unet.config.in_channels

        _UpperCAmelCase     ,_UpperCAmelCase        = downscale_height_and_width(__UpperCamelCase  ,       __UpperCamelCase  ,       self.movq_scale_factor  )

        # create initial latent
        _UpperCAmelCase        = self.prepare_latents(
            (batch_size, num_channels_latents, height, width)  ,       image_embeds.dtype  ,       __UpperCamelCase  ,       __UpperCamelCase  ,       __UpperCamelCase  ,       self.scheduler  ,       )

        for i, t in enumerate(self.progress_bar(__UpperCamelCase  )  ):
            # expand the latents if we are doing classifier free guidance
            _UpperCAmelCase        = torch.cat([latents] * 2  ) if do_classifier_free_guidance else latents

            _UpperCAmelCase        = {"""image_embeds""": image_embeds}
            _UpperCAmelCase        = self.unet(
                sample=__UpperCamelCase  ,       timestep=__UpperCamelCase  ,       encoder_hidden_states=__UpperCamelCase  ,       added_cond_kwargs=__UpperCamelCase  ,       return_dict=__UpperCamelCase  ,       )[0]

            if do_classifier_free_guidance:
                _UpperCAmelCase     ,_UpperCAmelCase        = noise_pred.split(latents.shape[1]  ,       dim=1  )
                _UpperCAmelCase     ,_UpperCAmelCase        = noise_pred.chunk(2  )
                _UpperCAmelCase     ,_UpperCAmelCase        = variance_pred.chunk(2  )
                _UpperCAmelCase        = noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
                _UpperCAmelCase        = torch.cat([noise_pred, variance_pred_text]  ,       dim=1  )

            if not (
                hasattr(self.scheduler.config  ,       """variance_type"""  )
                and self.scheduler.config.variance_type in ["learned", "learned_range"]
            ):
                _UpperCAmelCase     ,_UpperCAmelCase        = noise_pred.split(latents.shape[1]  ,       dim=1  )

            # compute the previous noisy sample x_t -> x_t-1
            _UpperCAmelCase        = self.scheduler.step(
                __UpperCamelCase  ,       __UpperCamelCase  ,       __UpperCamelCase  ,       generator=__UpperCamelCase  ,       )[0]
        # post-processing
        _UpperCAmelCase        = self.movq.decode(__UpperCamelCase  ,       force_not_quantize=__UpperCamelCase  )["""sample"""]

        if output_type not in ["pt", "np", "pil"]:
            raise ValueError(f"""Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}"""  )

        if output_type in ["np", "pil"]:
            _UpperCAmelCase        = image * 0.5 + 0.5
            _UpperCAmelCase        = image.clamp(0  ,       1  )
            _UpperCAmelCase        = image.cpu().permute(0  ,       2  ,       3  ,       1  ).float().numpy()

        if output_type == "pil":
            _UpperCAmelCase        = self.numpy_to_pil(__UpperCamelCase  )

        if not return_dict:
            return (image,)

        return ImagePipelineOutput(images=__UpperCamelCase  )

"""simple docstring"""



import math




class 		a_    :







    def   _snake_case       (      self      :							List[Any]  ,       __UpperCamelCase      :							list[list[float]]  ,       __UpperCamelCase      :							list[int]  )		->int:






        '''simple docstring'''






        _UpperCAmelCase        = 0.0
        _UpperCAmelCase        = 0.0
        for i in range(len(__UpperCamelCase  )  ):
            da += math.pow((sample[i] - weights[0][i])  ,       2  )
            da += math.pow((sample[i] - weights[1][i])  ,       2  )
            return 0 if da > da else 1
        return 0







    def   _snake_case       (      self      :							Dict  ,       __UpperCamelCase      :							list[list[int | float]]  ,       __UpperCamelCase      :							list[int]  ,       __UpperCamelCase      :							int  ,       __UpperCamelCase      :							float  )		->list[list[int | float]]:






        '''simple docstring'''






        for i in range(len(__UpperCamelCase  )  ):
            weights[j][i] += alpha * (sample[i] - weights[j][i])
        return weights







def 			_UpperCamelCase						(    )     ->	None:





    """simple docstring"""

    _UpperCAmelCase        = [[1, 1, 0, 0], [0, 0, 0, 1], [1, 0, 0, 0], [0, 0, 1, 1]]

    # weight initialization ( n, C )
    _UpperCAmelCase        = [[0.2, 0.6, 0.5, 0.9], [0.8, 0.4, 0.7, 0.3]]

    # training
    _UpperCAmelCase        = SelfOrganizingMap()
    _UpperCAmelCase        = 3
    _UpperCAmelCase        = 0.5

    for _ in range(_A   ):
        for j in range(len(_A   )   ):
            # training sample
            _UpperCAmelCase        = training_samples[j]

            # Compute the winning vector
            _UpperCAmelCase        = self_organizing_map.get_winner(_A       ,   _A   )

            # Update the winning vector
            _UpperCAmelCase        = self_organizing_map.update(_A       ,   _A       ,   _A       ,   _A   )

    # classify test sample
    _UpperCAmelCase        = [0, 0, 0, 1]
    _UpperCAmelCase        = self_organizing_map.get_winner(_A       ,   _A   )

    # results
    print(F"""Clusters that the test sample belongs to : {winner}"""   )
    print(F"""Weights that have been trained : {weights}"""   )


# running the main() function
if __name__ == "__main__":
 main()