






"""simple docstring"""



import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
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 			lowerCAmelCase__			(    __magic_name__							,						unittest.TestCase   ):
   SCREAMING_SNAKE_CASE_    =KandinskyImgaImgPipeline
   SCREAMING_SNAKE_CASE_    =['''prompt''', '''image_embeds''', '''negative_image_embeds''', '''image''']
   SCREAMING_SNAKE_CASE_    =[
       '''prompt''',
       '''negative_prompt''',
       '''image_embeds''',
       '''negative_image_embeds''',
       '''image''',
   ]
   SCREAMING_SNAKE_CASE_    =[
       '''generator''',
       '''height''',
       '''width''',
       '''strength''',
       '''guidance_scale''',
       '''negative_prompt''',
       '''num_inference_steps''',
       '''return_dict''',
       '''guidance_scale''',
       '''num_images_per_prompt''',
       '''output_type''',
       '''return_dict''',
   ]
   SCREAMING_SNAKE_CASE_    =False
   @property
   def       __a							(		self      :	Any							):


        '''simple docstring'''
        return 3_2
   @property
   def       __a							(		self      :	List[Any]							):


        '''simple docstring'''
        return 3_2
   @property
   def       __a							(		self      :	Union[str, Any]							):


        '''simple docstring'''
        return self.time_input_dim
   @property
   def       __a							(		self      :	List[Any]							):


        '''simple docstring'''
        return self.time_input_dim * 4
   @property
   def       __a							(		self      :	Union[str, Any]							):


        '''simple docstring'''
        return 1_0_0
   @property
   def       __a							(		self      :	Optional[Any]							):


        '''simple docstring'''
        UpperCAmelCase__						:    List[str]							=							XLMRobertaTokenizerFast.from_pretrained("YiYiXu/tiny-random-mclip-base"							)
        return tokenizer
   @property
   def       __a							(		self      :	int							):


        '''simple docstring'''
        torch.manual_seed(0							)
        UpperCAmelCase__						:    Any							=							MCLIPConfig(
            numDims=self.cross_attention_dim							,			transformerDimensions=self.text_embedder_hidden_size							,			hidden_size=self.text_embedder_hidden_size							,			intermediate_size=3_7							,			num_attention_heads=4							,			num_hidden_layers=5							,			vocab_size=1_0_0_5							,			)

        UpperCAmelCase__						:    Optional[Any]							=							MultilingualCLIP(snake_case__							)
        UpperCAmelCase__						:    Union[str, Any]							=							text_encoder.eval()

        return text_encoder
   @property
   def       __a							(		self      :	str							):


        '''simple docstring'''
        torch.manual_seed(0							)

        UpperCAmelCase__						:    int							=							{
            "in_channels": 4,
            # Out channels is double in channels because predicts mean and variance
            "out_channels": 8,
            "addition_embed_type": "text_image",
            "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": "text_image_proj",
            "cross_attention_dim": self.cross_attention_dim,
            "attention_head_dim": 4,
            "resnet_time_scale_shift": "scale_shift",
            "class_embed_type": None,
        }

        UpperCAmelCase__						:    Optional[int]							=							UNetaDConditionModel(**snake_case__							)
        return model
   @property
   def       __a							(		self      :	Tuple							):


        '''simple docstring'''
        return {
            "block_out_channels": [3_2, 6_4],
            "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
            "in_channels": 3,
            "latent_channels": 4,
            "layers_per_block": 1,
            "norm_num_groups": 8,
            "norm_type": "spatial",
            "num_vq_embeddings": 1_2,
            "out_channels": 3,
            "up_block_types": [
                "AttnUpDecoderBlock2D",
                "UpDecoderBlock2D",
            ],
            "vq_embed_dim": 4,
        }
   @property
   def       __a							(		self      :	Union[str, Any]							):


        '''simple docstring'''
        torch.manual_seed(0							)
        UpperCAmelCase__						:    str							=							VQModel(**self.dummy_movq_kwargs							)
        return model
   def       __a							(		self      :	Dict							):


        '''simple docstring'''
        UpperCAmelCase__						:    List[str]							=							self.dummy_text_encoder
        UpperCAmelCase__						:    Tuple							=							self.dummy_tokenizer
        UpperCAmelCase__						:    int							=							self.dummy_unet
        UpperCAmelCase__						:    Dict							=							self.dummy_movq

        UpperCAmelCase__						:    Dict							=							{
            "num_train_timesteps": 1_0_0_0,
            "beta_schedule": "linear",
            "beta_start": 0.0_0085,
            "beta_end": 0.012,
            "clip_sample": False,
            "set_alpha_to_one": False,
            "steps_offset": 0,
            "prediction_type": "epsilon",
            "thresholding": False,
        }

        UpperCAmelCase__						:    Optional[int]							=							DDIMScheduler(**snake_case__							)

        UpperCAmelCase__						:    Union[str, Any]							=							{
            "text_encoder": text_encoder,
            "tokenizer": tokenizer,
            "unet": unet,
            "scheduler": scheduler,
            "movq": movq,
        }

        return components
   def       __a							(		self      :	List[Any]							,			snake_case__      :	Optional[int]							,			snake_case__      :	int=0							):


        '''simple docstring'''
        UpperCAmelCase__						:    Optional[int]							=							floats_tensor((1, self.cross_attention_dim)							,			rng=random.Random(snake_case__							)							).to(snake_case__							)
        UpperCAmelCase__						:    Any							=							floats_tensor((1, self.cross_attention_dim)							,			rng=random.Random(seed + 1							)							).to(snake_case__							)
        # create init_image
        UpperCAmelCase__						:    Dict							=							floats_tensor((1, 3, 6_4, 6_4)							,			rng=random.Random(snake_case__							)							).to(snake_case__							)
        UpperCAmelCase__						:    Optional[Any]							=							image.cpu().permute(0							,			2							,			3							,			1							)[0]
        UpperCAmelCase__						:    Tuple							=							Image.fromarray(np.uinta(snake_case__							)							).convert("RGB"							).resize((2_5_6, 2_5_6)							)

        if str(snake_case__							).startswith("mps"							):
             UpperCAmelCase__						:    List[str]							=							torch.manual_seed(snake_case__							)
        else:
             UpperCAmelCase__						:    str							=							torch.Generator(device=snake_case__							).manual_seed(snake_case__							)
        UpperCAmelCase__						:    Union[str, Any]							=							{
            "prompt": "horse",
            "image": init_image,
            "image_embeds": image_embeds,
            "negative_image_embeds": negative_image_embeds,
            "generator": generator,
            "height": 6_4,
            "width": 6_4,
            "num_inference_steps": 1_0,
            "guidance_scale": 7.0,
            "strength": 0.2,
            "output_type": "np",
        }
        return inputs




   def       __a							(		self      :	int							):


        '''simple docstring'''
        UpperCAmelCase__						:    List[str]							=							"cpu"

        UpperCAmelCase__						:    Dict							=							self.get_dummy_components()

        UpperCAmelCase__						:    str							=							self.pipeline_class(**snake_case__							)
        UpperCAmelCase__						:    Optional[Any]							=							pipe.to(snake_case__							)

        pipe.set_progress_bar_config(disable=snake_case__							)

        UpperCAmelCase__						:    Dict							=							pipe(**self.get_dummy_inputs(snake_case__							)							)
        UpperCAmelCase__						:    Union[str, Any]							=							output.images

        UpperCAmelCase__						:    Dict							=							pipe(
            **self.get_dummy_inputs(snake_case__							)							,			return_dict=snake_case__							,			)[0]

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

        assert image.shape == (1, 6_4, 6_4, 3)

        UpperCAmelCase__						:    Optional[int]							=							np.array(
            [0.6147_4943, 0.607_3539, 0.4330_8544, 0.592_8269, 0.4749_3595, 0.4675_5973, 0.461_3838, 0.4536_8797, 0.5011_9233]							)
        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 			lowerCAmelCase__			(    unittest.TestCase   ):
   def       __a							(		self      :	str							):


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




   def       __a							(		self      :	Optional[int]							):


        '''simple docstring'''
        UpperCAmelCase__						:    Any							=							load_numpy(
            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
            "/kandinsky/kandinsky_img2img_frog.npy"							)

        UpperCAmelCase__						:    Optional[Any]							=							load_image(
            "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"							)
        UpperCAmelCase__						:    Dict							=							"A red cartoon frog, 4k"

        UpperCAmelCase__						:    Union[str, Any]							=							KandinskyPriorPipeline.from_pretrained(
            "kandinsky-community/kandinsky-2-1-prior"							,			torch_dtype=torch.floataa							)
        pipe_prior.to(snake_case__							)

        UpperCAmelCase__						:    Any							=							KandinskyImgaImgPipeline.from_pretrained(
            "kandinsky-community/kandinsky-2-1"							,			torch_dtype=torch.floataa							)
        UpperCAmelCase__						:    Dict							=							pipeline.to(snake_case__							)

        pipeline.set_progress_bar_config(disable=snake_case__							)

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

        UpperCAmelCase__						:    Dict							=							pipeline(
            snake_case__							,			image=snake_case__							,			image_embeds=snake_case__							,			negative_image_embeds=snake_case__							,			generator=snake_case__							,			num_inference_steps=1_0_0							,			height=7_6_8							,			width=7_6_8							,			strength=0.2							,			output_type="np"							,			)

        UpperCAmelCase__						:    Dict							=							output.images[0]

        assert image.shape == (7_6_8, 7_6_8, 3)

        assert_mean_pixel_difference(snake_case__							,			snake_case__							)










"""simple docstring"""



import functools
import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

import huggingface_hub.utils as hf_hub_utils
from tqdm import auto as tqdm_lib


_lowerCAmelCase	:	Union[str, Any]				=      threading.Lock()
_lowerCAmelCase	:	Optional[logging.Handler]				=      None

_lowerCAmelCase	:	Union[str, Any]				=      {
    """debug""": logging.DEBUG,
    """info""": logging.INFO,
    """warning""": logging.WARNING,
    """error""": logging.ERROR,
    """critical""": logging.CRITICAL,
}

_lowerCAmelCase	:	Dict				=      logging.WARNING

_lowerCAmelCase	:	Optional[Any]				=      True


def 	SCREAMING_SNAKE_CASE__       (  )->		List[str]:






     '''simple docstring'''



     UpperCAmelCase__						:    Union[str, Any]							=							os.getenv("TRANSFORMERS_VERBOSITY"			,					snake_case				)
     if env_level_str:
          if env_level_str in log_levels:
               return log_levels[env_level_str]
          else:
               logging.getLogger().warning(
                   f'Unknown option TRANSFORMERS_VERBOSITY={env_level_str}, '
                   f'has to be one of: { ", ".join(log_levels.keys()				) }'				)
     return _default_log_level


def 	SCREAMING_SNAKE_CASE__       (  )->		str:






     '''simple docstring'''



     return __name__.split("."				)[0]


def 	SCREAMING_SNAKE_CASE__       (  )->		logging.Logger:






     '''simple docstring'''



     return logging.getLogger(_get_library_name()				)


def 	SCREAMING_SNAKE_CASE__       (  )->		None:






     '''simple docstring'''



     global _default_handler

     with _lock:
          if _default_handler:
               # This library has already configured the library root logger.
               return
          UpperCAmelCase__						:    Any							=							logging.StreamHandler()  # Set sys.stderr as stream.
          UpperCAmelCase__						:    Union[str, Any]							=							sys.stderr.flush

          # Apply our default configuration to the library root logger.
          UpperCAmelCase__						:    Union[str, Any]							=							_get_library_root_logger()
          library_root_logger.addHandler(_default_handler				)
          library_root_logger.setLevel(_get_default_logging_level()				)
          UpperCAmelCase__						:    int							=							False


def 	SCREAMING_SNAKE_CASE__       (  )->		None:






     '''simple docstring'''



     global _default_handler

     with _lock:
          if not _default_handler:
               return

          UpperCAmelCase__						:    Union[str, Any]							=							_get_library_root_logger()
          library_root_logger.removeHandler(_default_handler				)
          library_root_logger.setLevel(logging.NOTSET				)
          UpperCAmelCase__						:    Optional[int]							=							None


def 	SCREAMING_SNAKE_CASE__       (  )->		Optional[int]:






     '''simple docstring'''



     return log_levels


def 	SCREAMING_SNAKE_CASE__       (  snake_case       : Optional[str] = None				)->		logging.Logger:






     '''simple docstring'''



     if name is None:
          UpperCAmelCase__						:    Tuple							=							_get_library_name()

     _configure_library_root_logger()
     return logging.getLogger(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  )->		int:






     '''simple docstring'''



     _configure_library_root_logger()
     return _get_library_root_logger().getEffectiveLevel()


def 	SCREAMING_SNAKE_CASE__       (  snake_case       : int				)->		None:






     '''simple docstring'''



     _configure_library_root_logger()
     _get_library_root_logger().setLevel(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  )->		Optional[Any]:






     '''simple docstring'''



     return set_verbosity(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  )->		List[str]:






     '''simple docstring'''



     return set_verbosity(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  )->		Tuple:






     '''simple docstring'''



     return set_verbosity(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  )->		str:






     '''simple docstring'''



     return set_verbosity(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  )->		None:






     '''simple docstring'''



     _configure_library_root_logger()

     assert _default_handler is not None
     _get_library_root_logger().removeHandler(_default_handler				)


def 	SCREAMING_SNAKE_CASE__       (  )->		None:






     '''simple docstring'''



     _configure_library_root_logger()

     assert _default_handler is not None
     _get_library_root_logger().addHandler(_default_handler				)


def 	SCREAMING_SNAKE_CASE__       (  snake_case       : logging.Handler				)->		None:






     '''simple docstring'''



     _configure_library_root_logger()

     assert handler is not None
     _get_library_root_logger().addHandler(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  snake_case       : logging.Handler				)->		None:






     '''simple docstring'''



     _configure_library_root_logger()

     assert handler is not None and handler not in _get_library_root_logger().handlers
     _get_library_root_logger().removeHandler(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  )->		None:






     '''simple docstring'''



     _configure_library_root_logger()
     UpperCAmelCase__						:    Dict							=							False


def 	SCREAMING_SNAKE_CASE__       (  )->		None:






     '''simple docstring'''



     _configure_library_root_logger()
     UpperCAmelCase__						:    List[Any]							=							True


def 	SCREAMING_SNAKE_CASE__       (  )->		None:






     '''simple docstring'''



     UpperCAmelCase__						:    Union[str, Any]							=							_get_library_root_logger().handlers

     for handler in handlers:
          UpperCAmelCase__						:    Union[str, Any]							=							logging.Formatter("[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s"				)
          handler.setFormatter(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  )->		None:






     '''simple docstring'''



     UpperCAmelCase__						:    Any							=							_get_library_root_logger().handlers

     for handler in handlers:
          handler.setFormatter(snake_case				)


def 	SCREAMING_SNAKE_CASE__       (  self       : Optional[int]			,					*snake_case       : List[str]			,					**snake_case       : str				)->		Union[str, Any]:






     '''simple docstring'''



     UpperCAmelCase__						:    Union[str, Any]							=							os.getenv("TRANSFORMERS_NO_ADVISORY_WARNINGS"			,					snake_case				)
     if no_advisory_warnings:
          return
     self.warning(*snake_case			,					**snake_case				)


_lowerCAmelCase	:	int				=      warning_advice


@functools.lru_cache(snake_case				)
def 	SCREAMING_SNAKE_CASE__       (  self       : Optional[Any]			,					*snake_case       : Dict			,					**snake_case       : Any				)->		Any:






     '''simple docstring'''



     self.warning(*snake_case			,					**snake_case				)


_lowerCAmelCase	:	Tuple				=      warning_once




class 			lowerCAmelCase__			:
   def __init__(		self      :	List[str]							,			*snake_case__      :	Any							,			**snake_case__      :	List[str]							):  # pylint: disable=unused-argument


        '''simple docstring'''
        UpperCAmelCase__						:    List[Any]							=							args[0] if args else None
   def __iter__(		self      :	Any							):


        '''simple docstring'''
        return iter(self._iterator							)
   def __getattr__(		self      :	List[Any]							,			snake_case__      :	Tuple							):


        '''simple docstring'''

        def empty_fn(*snake_case__      :	Dict							,			**snake_case__      :	str							):  # pylint: disable=unused-argument
             return

        return empty_fn
   def __enter__(		self      :	Optional[Any]							):


        '''simple docstring'''
        return self




   def __exit__(		self      :	Tuple							,			snake_case__      :	Tuple							,			snake_case__      :	Any							,			snake_case__      :	Optional[int]							):


        '''simple docstring'''
        return




class 			lowerCAmelCase__			:
   def __call__(		self      :	Optional[Any]							,			*snake_case__      :	Optional[Any]							,			**snake_case__      :	Tuple							):


        '''simple docstring'''
        if _tqdm_active:
             return tqdm_lib.tqdm(*snake_case__							,			**snake_case__							)
        else:
             return EmptyTqdm(*snake_case__							,			**snake_case__							)
   def       __a							(		self      :	Dict							,			*snake_case__      :	List[str]							,			**snake_case__      :	Union[str, Any]							):


        '''simple docstring'''
        UpperCAmelCase__						:    Dict							=							None
        if _tqdm_active:
             return tqdm_lib.tqdm.set_lock(*snake_case__							,			**snake_case__							)




   def       __a							(		self      :	Tuple							):


        '''simple docstring'''
        if _tqdm_active:
             return tqdm_lib.tqdm.get_lock()




_lowerCAmelCase	:	Optional[int]				=      _tqdm_cls()


def 	SCREAMING_SNAKE_CASE__       (  )->		bool:






     '''simple docstring'''



     global _tqdm_active
     return bool(_tqdm_active				)


def 	SCREAMING_SNAKE_CASE__       (  )->		List[Any]:






     '''simple docstring'''



     global _tqdm_active
     UpperCAmelCase__						:    int							=							True
     hf_hub_utils.enable_progress_bars()



def 	SCREAMING_SNAKE_CASE__       (  )->		List[Any]:






     '''simple docstring'''



     global _tqdm_active
     UpperCAmelCase__						:    Optional[Any]							=							False
     hf_hub_utils.disable_progress_bars()


