





import gc
import random
import unittest

import numpy as np
import torch

from diffusers import DDIMScheduler, KandinskyVaaPipeline, KandinskyVaaPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.utils import floats_tensor, 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    _UpperCAmelCase							(		_lowerCamelCase			,       unittest.TestCase     ):
      a						       =							KandinskyVaaPipeline
      a						       =							[
          '''image_embeds''',
          '''negative_image_embeds''',
      ]
      a						       =							['''image_embeds''', '''negative_image_embeds''']
      a						       =							[
          '''generator''',
          '''height''',
          '''width''',
          '''latents''',
          '''guidance_scale''',
          '''num_inference_steps''',
          '''return_dict''',
          '''guidance_scale''',
          '''num_images_per_prompt''',
          '''output_type''',
          '''return_dict''',
      ]
      a						       =							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_     :				Union[str, Any]			      =	{
             """in_channels""": 4,
             # Out channels is double in channels because predicts mean and variance
             """out_channels""": 8,
             """addition_embed_type""": """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""": """image_proj""",
             """cross_attention_dim""": self.cross_attention_dim,
             """attention_head_dim""": 4,
             """resnet_time_scale_shift""": """scale_shift""",
             """class_embed_type""": None,
         }

         A_     :				Tuple			      =	UNetaDConditionModel(**a__       )
         return model
      @property
      def 		_lowerCamelCase      (		self       ):
         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 		_lowerCamelCase      (		self       ):
         torch.manual_seed(0       )
         A_     :				Optional[Any]			      =	VQModel(**self.dummy_movq_kwargs       )
         return model
      def 		_lowerCamelCase      (		self       ):
         A_     :				List[Any]			      =	self.dummy_unet
         A_     :				Optional[Any]			      =	self.dummy_movq

         A_     :				Tuple			      =	DDIMScheduler(
             num_train_timesteps=1000      ,       beta_schedule="""linear"""      ,       beta_start=0.0_0085      ,       beta_end=0.012      ,       clip_sample=a__      ,       set_alpha_to_one=a__      ,       steps_offset=1      ,       prediction_type="""epsilon"""      ,       thresholding=a__      ,       )

         A_     :				Any			      =	{
             """unet""": unet,
             """scheduler""": scheduler,
             """movq""": movq,
         }
         return components
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__=0       ):
         A_     :				Optional[Any]			      =	floats_tensor((1, self.text_embedder_hidden_size)      ,       rng=random.Random(a__       )       ).to(a__       )
         A_     :				List[Any]			      =	floats_tensor((1, self.text_embedder_hidden_size)      ,       rng=random.Random(seed + 1       )       ).to(
             a__       )
         if str(a__       ).startswith("""mps"""       ):
            A_     :				Union[str, Any]			      =	torch.manual_seed(a__       )
         else:
            A_     :				Dict			      =	torch.Generator(device=a__       ).manual_seed(a__       )
         A_     :				Dict			      =	{
             """image_embeds""": image_embeds,
             """negative_image_embeds""": negative_image_embeds,
             """generator""": generator,
             """height""": 64,
             """width""": 64,
             """guidance_scale""": 4.0,
             """num_inference_steps""": 2,
             """output_type""": """np""",
         }
         return inputs






      def 		_lowerCamelCase      (		self       ):
         A_     :				Any			      =	"""cpu"""

         A_     :				Optional[Any]			      =	self.get_dummy_components()

         A_     :				Optional[Any]			      =	self.pipeline_class(**a__       )
         A_     :				Tuple			      =	pipe.to(a__       )

         pipe.set_progress_bar_config(disable=a__       )

         A_     :				str			      =	pipe(**self.get_dummy_inputs(a__       )       )
         A_     :				Union[str, Any]			      =	output.images

         A_     :				Any			      =	pipe(
             **self.get_dummy_inputs(a__       )      ,       return_dict=a__      ,       )[0]

         A_     :				List[Any]			      =	image[0, -3:, -3:, -1]
         A_     :				Union[str, Any]			      =	image_from_tuple[0, -3:, -3:, -1]

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

         A_     :				int			      =	np.array(
             [0.623_7976, 1.0, 0.3644_1332, 1.0, 0.7063_9634, 0.2987_7186, 0.8565_2125, 0.521_6843, 0.5445_4046]       )

         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    _UpperCAmelCase							(		unittest.TestCase     ):
      def 		_lowerCamelCase      (		self       ):
         # clean up the VRAM after each test
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()






      def 		_lowerCamelCase      (		self       ):
         A_     :				Union[str, Any]			      =	load_numpy(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
             """/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy"""       )

         A_     :				Dict			      =	KandinskyVaaPriorPipeline.from_pretrained(
             """kandinsky-community/kandinsky-2-2-prior"""      ,       torch_dtype=torch.floataa       )
         pipe_prior.to(a__       )

         A_     :				Optional[int]			      =	KandinskyVaaPipeline.from_pretrained(
             """kandinsky-community/kandinsky-2-2-decoder"""      ,       torch_dtype=torch.floataa       )
         A_     :				int			      =	pipeline.to(a__       )
         pipeline.set_progress_bar_config(disable=a__       )

         A_     :				Optional[int]			      =	"""red cat, 4k photo"""

         A_     :				Any			      =	torch.Generator(device="""cuda"""       ).manual_seed(0       )
         A_      ,				A_     :				Any			      =	pipe_prior(
             a__      ,       generator=a__      ,       num_inference_steps=5      ,       negative_prompt=""""""      ,       ).to_tuple()

         A_     :				int			      =	torch.Generator(device="""cuda"""       ).manual_seed(0       )
         A_     :				str			      =	pipeline(
             image_embeds=a__      ,       negative_image_embeds=a__      ,       generator=a__      ,       num_inference_steps=100      ,       output_type="""np"""      ,       )

         A_     :				Dict			      =	output.images[0]

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

         assert_mean_pixel_difference(a__      ,       a__       )













from . import __version__

# Backward compatibility imports, to make sure all those objects can be found in file_utils
from .utils import (
    CLOUDFRONT_DISTRIB_PREFIX,
    CONFIG_NAME,
    DISABLE_TELEMETRY,
    DUMMY_INPUTS,
    DUMMY_MASK,
    ENV_VARS_TRUE_AND_AUTO_VALUES,
    ENV_VARS_TRUE_VALUES,
    FEATURE_EXTRACTOR_NAME,
    FLAX_WEIGHTS_NAME,
    HF_MODULES_CACHE,
    HUGGINGFACE_CO_PREFIX,
    HUGGINGFACE_CO_RESOLVE_ENDPOINT,
    MODEL_CARD_NAME,
    MULTIPLE_CHOICE_DUMMY_INPUTS,
    PYTORCH_PRETRAINED_BERT_CACHE,
    PYTORCH_TRANSFORMERS_CACHE,
    S3_BUCKET_PREFIX,
    SENTENCEPIECE_UNDERLINE,
    SPIECE_UNDERLINE,
    TF2_WEIGHTS_NAME,
    TF_WEIGHTS_NAME,
    TORCH_FX_REQUIRED_VERSION,
    TRANSFORMERS_CACHE,
    TRANSFORMERS_DYNAMIC_MODULE_NAME,
    USE_JAX,
    USE_TF,
    USE_TORCH,
    WEIGHTS_INDEX_NAME,
    WEIGHTS_NAME,
    ContextManagers,
    DummyObject,
    EntryNotFoundError,
    ExplicitEnum,
    ModelOutput,
    PaddingStrategy,
    PushToHubMixin,
    RepositoryNotFoundError,
    RevisionNotFoundError,
    TensorType,
    _LazyModule,
    add_code_sample_docstrings,
    add_end_docstrings,
    add_start_docstrings,
    add_start_docstrings_to_model_forward,
    cached_property,
    copy_func,
    default_cache_path,
    define_sagemaker_information,
    get_cached_models,
    get_file_from_repo,
    get_full_repo_name,
    get_torch_version,
    has_file,
    http_user_agent,
    is_apex_available,
    is_bsa_available,
    is_coloredlogs_available,
    is_datasets_available,
    is_detectrona_available,
    is_faiss_available,
    is_flax_available,
    is_ftfy_available,
    is_in_notebook,
    is_ipex_available,
    is_librosa_available,
    is_offline_mode,
    is_onnx_available,
    is_pandas_available,
    is_phonemizer_available,
    is_protobuf_available,
    is_psutil_available,
    is_pyanvml_available,
    is_pyctcdecode_available,
    is_pytesseract_available,
    is_pytorch_quantization_available,
    is_rjieba_available,
    is_sagemaker_dp_enabled,
    is_sagemaker_mp_enabled,
    is_scipy_available,
    is_sentencepiece_available,
    is_seqio_available,
    is_sklearn_available,
    is_soundfile_availble,
    is_spacy_available,
    is_speech_available,
    is_tensor,
    is_tensorflow_probability_available,
    is_tfaonnx_available,
    is_tf_available,
    is_timm_available,
    is_tokenizers_available,
    is_torch_available,
    is_torch_bfaa_available,
    is_torch_cuda_available,
    is_torch_fx_available,
    is_torch_fx_proxy,
    is_torch_mps_available,
    is_torch_tfaa_available,
    is_torch_tpu_available,
    is_torchaudio_available,
    is_training_run_on_sagemaker,
    is_vision_available,
    replace_return_docstrings,
    requires_backends,
    to_numpy,
    to_py_obj,
    torch_only_method,
)






