




"""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	(   UpperCamelCase__     ,    unittest.TestCase   ):




    """simple docstring"""

    snake_case__ =				KandinskyImgaImgPipeline
    snake_case__ =				["prompt", "image_embeds", "negative_image_embeds", "image"]
    snake_case__ =				[
        "prompt",
        "negative_prompt",
        "image_embeds",
        "negative_image_embeds",
        "image",
    ]
    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",
    ]
    snake_case__ =				False


    @property
    def        a				(	self					:					Union[str, Any]       )				->	Dict:
         return 32


    @property
    def        a				(	self					:					Optional[Any]       )				->	Any:
         return 32


    @property
    def        a				(	self					:					Tuple       )				->	Any:
         return self.time_input_dim


    @property
    def        a				(	self					:					Any       )				->	List[Any]:
         return self.time_input_dim * 4


    @property
    def        a				(	self					:					str       )				->	List[str]:
         return 100


    @property
    def        a				(	self					:					Optional[int]       )				->	List[Any]:
         lowerCAmelCase__          =			XLMRobertaTokenizerFast.from_pretrained("YiYiXu/tiny-random-mclip-base"       )
         return tokenizer


    @property
    def        a				(	self					:					Optional[Any]       )				->	int:
         torch.manual_seed(0       )
         lowerCAmelCase__          =			MCLIPConfig(
             numDims=self.cross_attention_dim       ,     transformerDimensions=self.text_embedder_hidden_size       ,     hidden_size=self.text_embedder_hidden_size       ,     intermediate_size=37       ,     num_attention_heads=4       ,     num_hidden_layers=5       ,     vocab_size=1_005       ,     )

         lowerCAmelCase__          =			MultilingualCLIP(SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			text_encoder.eval()

         return text_encoder


    @property
    def        a				(	self					:					Dict       )				->	Dict:
         torch.manual_seed(0       )

         lowerCAmelCase__          =			{
             "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,
         }

         lowerCAmelCase__          =			UNetaDConditionModel(**SCREAMING_SNAKE_CASE__       )
         return model


    @property
    def        a				(	self					:					int       )				->	Dict:
         return {
             "block_out_channels": [32, 64],
             "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": 12,
             "out_channels": 3,
             "up_block_types": [
                 "AttnUpDecoderBlock2D",
                 "UpDecoderBlock2D",
             ],
             "vq_embed_dim": 4,
         }


    @property
    def        a				(	self					:					Dict       )				->	str:
         torch.manual_seed(0       )
         lowerCAmelCase__          =			VQModel(**self.dummy_movq_kwargs       )
         return model


    def        a				(	self					:					Tuple       )				->	Any:
         lowerCAmelCase__          =			self.dummy_text_encoder
         lowerCAmelCase__          =			self.dummy_tokenizer
         lowerCAmelCase__          =			self.dummy_unet
         lowerCAmelCase__          =			self.dummy_movq

         lowerCAmelCase__          =			{
             "num_train_timesteps": 1_000,
             "beta_schedule": "linear",
             "beta_start": 0.00_085,
             "beta_end": 0.012,
             "clip_sample": False,
             "set_alpha_to_one": False,
             "steps_offset": 0,
             "prediction_type": "epsilon",
             "thresholding": False,
         }

         lowerCAmelCase__          =			DDIMScheduler(**SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			{
             "text_encoder": text_encoder,
             "tokenizer": tokenizer,
             "unet": unet,
             "scheduler": scheduler,
             "movq": movq,
         }

         return components


    def        a				(	self					:					Tuple       ,     SCREAMING_SNAKE_CASE__					:					Dict       ,     SCREAMING_SNAKE_CASE__					:					str=0       )				->	Dict:
         lowerCAmelCase__          =			floats_tensor((1, self.cross_attention_dim)       ,     rng=random.Random(SCREAMING_SNAKE_CASE__       )       ).to(SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			floats_tensor((1, self.cross_attention_dim)       ,     rng=random.Random(seed + 1       )       ).to(SCREAMING_SNAKE_CASE__       )
         # create init_image
         lowerCAmelCase__          =			floats_tensor((1, 3, 64, 64)       ,     rng=random.Random(SCREAMING_SNAKE_CASE__       )       ).to(SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			image.cpu().permute(0       ,     2       ,     3       ,     1       )[0]
         lowerCAmelCase__          =			Image.fromarray(np.uinta(SCREAMING_SNAKE_CASE__       )       ).convert("RGB"       ).resize((256, 256)       )

         if str(SCREAMING_SNAKE_CASE__       ).startswith("mps"       ):
              lowerCAmelCase__          =			torch.manual_seed(SCREAMING_SNAKE_CASE__       )
         else:
              lowerCAmelCase__          =			torch.Generator(device=SCREAMING_SNAKE_CASE__       ).manual_seed(SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			{
             "prompt": "horse",
             "image": init_image,
             "image_embeds": image_embeds,
             "negative_image_embeds": negative_image_embeds,
             "generator": generator,
             "height": 64,
             "width": 64,
             "num_inference_steps": 10,
             "guidance_scale": 7.0,
             "strength": 0.2,
             "output_type": "np",
         }
         return inputs


    def        a				(	self					:					Union[str, Any]       )				->	Optional[Any]:
         lowerCAmelCase__          =			"cpu"

         lowerCAmelCase__          =			self.get_dummy_components()

         lowerCAmelCase__          =			self.pipeline_class(**SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			pipe.to(SCREAMING_SNAKE_CASE__       )

         pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			pipe(**self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )       )
         lowerCAmelCase__          =			output.images

         lowerCAmelCase__          =			pipe(
             **self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )       ,     return_dict=SCREAMING_SNAKE_CASE__       ,     )[0]

         lowerCAmelCase__          =			image[0, -3:, -3:, -1]
         lowerCAmelCase__          =			image_from_tuple[0, -3:, -3:, -1]

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

         lowerCAmelCase__          =			np.array(
             [0.61_474_943, 0.6_073_539, 0.43_308_544, 0.5_928_269, 0.47_493_595, 0.46_755_973, 0.4_613_838, 0.45_368_797, 0.50_119_233]       )
         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   ):




    """simple docstring"""

    def        a				(	self					:					Optional[Any]       )				->	List[str]:
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()


    def        a				(	self					:					str       )				->	List[Any]:
         lowerCAmelCase__          =			load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/kandinsky/kandinsky_img2img_frog.npy"       )

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

         lowerCAmelCase__          =			KandinskyPriorPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-1-prior"       ,     torch_dtype=torch.floataa       )
         pipe_prior.to(SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			KandinskyImgaImgPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-1"       ,     torch_dtype=torch.floataa       )
         lowerCAmelCase__          =			pipeline.to(SCREAMING_SNAKE_CASE__       )

         pipeline.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			torch.Generator(device="cpu"       ).manual_seed(0       )
         lowerCAmelCase__     ,		lowerCAmelCase__          =			pipe_prior(
             SCREAMING_SNAKE_CASE__       ,     generator=SCREAMING_SNAKE_CASE__       ,     num_inference_steps=5       ,     negative_prompt=""       ,     ).to_tuple()

         lowerCAmelCase__          =			pipeline(
             SCREAMING_SNAKE_CASE__       ,     image=SCREAMING_SNAKE_CASE__       ,     image_embeds=SCREAMING_SNAKE_CASE__       ,     negative_image_embeds=SCREAMING_SNAKE_CASE__       ,     generator=SCREAMING_SNAKE_CASE__       ,     num_inference_steps=100       ,     height=768       ,     width=768       ,     strength=0.2       ,     output_type="np"       ,     )

         lowerCAmelCase__          =			output.images[0]

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

         assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       )






import time
from contextlib import contextmanager
from pathlib import Path

import pytest
import requests
from huggingface_hub.hf_api import HfApi, HfFolder


UpperCamelCase								=					'__DUMMY_TRANSFORMERS_USER__'
UpperCamelCase								=					'Dummy User'
UpperCamelCase								=					'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'

UpperCamelCase								=					'https://hub-ci.huggingface.co'
UpperCamelCase								=					CI_HUB_ENDPOINT + '/datasets/{repo_id}/resolve/{revision}/{path}'
UpperCamelCase								=					CI_HUB_ENDPOINT + '/{repo_id}/resolve/{revision}/{filename}'
UpperCamelCase								=					Path('~/.huggingface/hub_ci_token').expanduser()



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: Dict      ):
     """simple docstring"""

     monkeypatch.setattr(
         "huggingface_hub.file_download.HUGGINGFACE_CO_URL_TEMPLATE"       ,    lowerCAmelCase_      )



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: int      ):
     """simple docstring"""

     monkeypatch.setattr("datasets.config.HF_ENDPOINT"       ,    lowerCAmelCase_      )
     monkeypatch.setattr("datasets.config.HUB_DATASETS_URL"       ,    lowerCAmelCase_      )



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: str      ):
     """simple docstring"""

     monkeypatch.setattr("huggingface_hub.hf_api.HfFolder.path_token"       ,    lowerCAmelCase_      )



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: int       ,    lowerCAmelCase_					: List[Any]      ):
     """simple docstring"""

     HfFolder.save_token(lowerCAmelCase_      )
     yield
     HfFolder.delete_token()



@pytest.fixture(scope="session"      )
def 							_A       (						):
     """simple docstring"""

     return HfApi(endpoint=lowerCAmelCase_      )



@pytest.fixture(scope="session"      )
def 							_A       (						lowerCAmelCase_					: HfApi      ):
     """simple docstring"""

     lowerCAmelCase__          =			HfFolder.get_token()
     HfFolder.save_token(lowerCAmelCase_      )
     yield CI_HUB_USER_TOKEN
     if previous_token is not None:
          HfFolder.save_token(lowerCAmelCase_      )



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: Tuple      ):
     """simple docstring"""

     def _cleanup_repo(lowerCAmelCase_					: Optional[Any]      ):
          hf_api.delete_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"      )

     return _cleanup_repo



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: Union[str, Any]      ):
     """simple docstring"""

     @contextmanager
     def _temporary_repo(lowerCAmelCase_					: str      ):
          try:
               yield repo_id
          finally:
               cleanup_repo(lowerCAmelCase_      )

     return _temporary_repo



@pytest.fixture(scope="session"      )
def 							_A       (						lowerCAmelCase_					: HfApi       ,    lowerCAmelCase_					: Union[str, Any]       ,    lowerCAmelCase_					: Any      ):
     """simple docstring"""

     lowerCAmelCase__          =			F'repo_txt_data-{int(time.time() * 1_0E3      )}'
     lowerCAmelCase__          =			F'{CI_HUB_USER}/{repo_name}'
     hf_api.create_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"       ,    private=lowerCAmelCase_      )
     hf_api.upload_file(
         token=lowerCAmelCase_       ,    path_or_fileobj=str(lowerCAmelCase_      )       ,    path_in_repo="data/text_data.txt"       ,    repo_id=lowerCAmelCase_       ,    repo_type="dataset"       ,    )
     yield repo_id
     try:
          hf_api.delete_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"      )
     except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
          pass



@pytest.fixture()
def 							_A       (						lowerCAmelCase_					: Optional[Any]       ,    lowerCAmelCase_					: Optional[Any]       ,    lowerCAmelCase_					: int      ):
     """simple docstring"""

     return hf_private_dataset_repo_txt_data_



@pytest.fixture(scope="session"      )
def 							_A       (						lowerCAmelCase_					: HfApi       ,    lowerCAmelCase_					: Tuple       ,    lowerCAmelCase_					: Tuple      ):
     """simple docstring"""

     lowerCAmelCase__          =			F'repo_zipped_txt_data-{int(time.time() * 1_0E3      )}'
     lowerCAmelCase__          =			F'{CI_HUB_USER}/{repo_name}'
     hf_api.create_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"       ,    private=lowerCAmelCase_      )
     hf_api.upload_file(
         token=lowerCAmelCase_       ,    path_or_fileobj=str(lowerCAmelCase_      )       ,    path_in_repo="data.zip"       ,    repo_id=lowerCAmelCase_       ,    repo_type="dataset"       ,    )
     yield repo_id
     try:
          hf_api.delete_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"      )
     except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
          pass



@pytest.fixture()
def 							_A       (						lowerCAmelCase_					: int       ,    lowerCAmelCase_					: List[str]       ,    lowerCAmelCase_					: Tuple      ):
     """simple docstring"""

     return hf_private_dataset_repo_zipped_txt_data_



@pytest.fixture(scope="session"      )
def 							_A       (						lowerCAmelCase_					: HfApi       ,    lowerCAmelCase_					: List[str]       ,    lowerCAmelCase_					: int      ):
     """simple docstring"""

     lowerCAmelCase__          =			F'repo_zipped_img_data-{int(time.time() * 1_0E3      )}'
     lowerCAmelCase__          =			F'{CI_HUB_USER}/{repo_name}'
     hf_api.create_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"       ,    private=lowerCAmelCase_      )
     hf_api.upload_file(
         token=lowerCAmelCase_       ,    path_or_fileobj=str(lowerCAmelCase_      )       ,    path_in_repo="data.zip"       ,    repo_id=lowerCAmelCase_       ,    repo_type="dataset"       ,    )
     yield repo_id
     try:
          hf_api.delete_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"      )
     except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
          pass






@pytest.fixture()
def 							_A       (						lowerCAmelCase_					: Optional[Any]       ,    lowerCAmelCase_					: Dict       ,    lowerCAmelCase_					: List[str]      ):
     """simple docstring"""

     return hf_private_dataset_repo_zipped_img_data_
