

"""simple docstring"""






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetImgaImgPipeline,
    KandinskyVaaPriorEmbaEmbPipeline,
    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        _lowercase					(     __UpperCAmelCase   , unittest.TestCase     ):
     _lowerCamelCase     =       KandinskyVaaControlnetImgaImgPipeline
     _lowerCamelCase     =       ["image_embeds", "negative_image_embeds", "image", "hint"]
     _lowerCamelCase     =       ["image_embeds", "negative_image_embeds", "image", "hint"]
     _lowerCamelCase     =       [
         "generator",
         "height",
         "width",
         "strength",
         "guidance_scale",
         "num_inference_steps",
         "return_dict",
         "guidance_scale",
         "num_images_per_prompt",
         "output_type",
         "return_dict",
     ]
     _lowerCamelCase     =       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       )

      __magic_name__     =      {
          '''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,
      }

      __magic_name__     =      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       )
      __magic_name__     =      VQModel(**self.dummy_movq_kwargs       )
      return model







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      self.dummy_unet
      __magic_name__     =      self.dummy_movq

      __magic_name__     =      {
          '''num_train_timesteps''': 1000,
          '''beta_schedule''': '''linear''',
          '''beta_start''': 0.0_0_0_8_5,
          '''beta_end''': 0.0_1_2,
          '''clip_sample''': False,
          '''set_alpha_to_one''': False,
          '''steps_offset''': 0,
          '''prediction_type''': '''epsilon''',
          '''thresholding''': False,
      }

      __magic_name__     =      DDIMScheduler(**_lowerCamelCase       )

      __magic_name__     =      {
          '''unet''': unet,
          '''scheduler''': scheduler,
          '''movq''': movq,
      }

      return components







     def 	lowerCAmelCase__    (      self ,						UpperCamelCase_ ,						UpperCamelCase_=0       ):
      __magic_name__     =      floats_tensor((1, self.text_embedder_hidden_size) ,						rng=random.Random(_lowerCamelCase       )       ).to(_lowerCamelCase       )
      __magic_name__     =      floats_tensor((1, self.text_embedder_hidden_size) ,						rng=random.Random(seed + 1       )       ).to(
          _lowerCamelCase       )
      # create init_image
      __magic_name__     =      floats_tensor((1, 3, 64, 64) ,						rng=random.Random(_lowerCamelCase       )       ).to(_lowerCamelCase       )
      __magic_name__     =      image.cpu().permute(0 ,						2 ,						3 ,						1       )[0]
      __magic_name__     =      Image.fromarray(np.uinta(_lowerCamelCase       )       ).convert('''RGB'''       ).resize((256, 256)       )
      # create hint
      __magic_name__     =      floats_tensor((1, 3, 64, 64) ,						rng=random.Random(_lowerCamelCase       )       ).to(_lowerCamelCase       )

      if str(_lowerCamelCase       ).startswith('''mps'''       ):
       __magic_name__     =      torch.manual_seed(_lowerCamelCase       )
      else:
       __magic_name__     =      torch.Generator(device=_lowerCamelCase       ).manual_seed(_lowerCamelCase       )
      __magic_name__     =      {
          '''image''': init_image,
          '''image_embeds''': image_embeds,
          '''negative_image_embeds''': negative_image_embeds,
          '''hint''': hint,
          '''generator''': generator,
          '''height''': 64,
          '''width''': 64,
          '''num_inference_steps''': 10,
          '''guidance_scale''': 7.0,
          '''strength''': 0.2,
          '''output_type''': '''np''',
      }
      return inputs







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      '''cpu'''

      __magic_name__     =      self.get_dummy_components()

      __magic_name__     =      self.pipeline_class(**_lowerCamelCase       )
      __magic_name__     =      pipe.to(_lowerCamelCase       )

      pipe.set_progress_bar_config(disable=_lowerCamelCase       )

      __magic_name__     =      pipe(**self.get_dummy_inputs(_lowerCamelCase       )       )
      __magic_name__     =      output.images

      __magic_name__     =      pipe(
          **self.get_dummy_inputs(_lowerCamelCase       ) ,						return_dict=_lowerCamelCase ,						)[0]

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

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

      __magic_name__     =      np.array(
          [0.5_4_9_8_5_0_3_4, 0.5_5_5_0_9_3_6_5, 0.5_2_5_6_1_5_0_4, 0.5_5_7_0_4_9_4, 0.5_5_9_3_8_1_8, 0.5_2_6_3_9_7_9, 0.5_0_2_8_5_6_4_3, 0.5_0_6_9_8_4_6, 0.5_1_1_9_6_7_3_6]       )
      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        _lowercase					(     unittest.TestCase     ):







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







     def 	lowerCAmelCase__    (      self       ):
      __magic_name__     =      load_numpy(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/kandinskyv22/kandinskyv22_controlnet_img2img_robotcat_fp16.npy'''       )

      __magic_name__     =      load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main''' '''/kandinsky/cat.png'''       )
      __magic_name__     =      init_image.resize((512, 512)       )

      __magic_name__     =      load_image(
          '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
          '''/kandinskyv22/hint_image_cat.png'''       )
      __magic_name__     =      torch.from_numpy(np.array(_lowerCamelCase       )       ).float() / 255.0
      __magic_name__     =      hint.permute(2 ,						0 ,						1       ).unsqueeze(0       )

      __magic_name__     =      '''A robot, 4k photo'''

      __magic_name__     =      KandinskyVaaPriorEmbaEmbPipeline.from_pretrained(
          '''kandinsky-community/kandinsky-2-2-prior''' ,						torch_dtype=torch.floataa       )
      pipe_prior.to(_lowerCamelCase       )

      __magic_name__     =      KandinskyVaaControlnetImgaImgPipeline.from_pretrained(
          '''kandinsky-community/kandinsky-2-2-controlnet-depth''' ,						torch_dtype=torch.floataa       )
      __magic_name__     =      pipeline.to(_lowerCamelCase       )

      pipeline.set_progress_bar_config(disable=_lowerCamelCase       )

      __magic_name__     =      torch.Generator(device='''cpu'''       ).manual_seed(0       )

      __magic_name__   ,						__magic_name__     =      pipe_prior(
          _lowerCamelCase ,						image=_lowerCamelCase ,						strength=0.8_5 ,						generator=_lowerCamelCase ,						negative_prompt='''''' ,						).to_tuple()

      __magic_name__     =      pipeline(
          image=_lowerCamelCase ,						image_embeds=_lowerCamelCase ,						negative_image_embeds=_lowerCamelCase ,						hint=_lowerCamelCase ,						generator=_lowerCamelCase ,						num_inference_steps=100 ,						height=512 ,						width=512 ,						strength=0.5 ,						output_type='''np''' ,						)

      __magic_name__     =      output.images[0]

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

      assert_mean_pixel_difference(_lowerCamelCase ,						_lowerCamelCase       )




"""simple docstring"""






import os

import pytest
import yaml

from datasets.features.features import Features, Value
from datasets.info import DatasetInfo, DatasetInfosDict





@pytest.mark.parametrize(
    '''files'''      ,   [
        ['''full:README.md''', '''dataset_infos.json'''],
        ['''empty:README.md''', '''dataset_infos.json'''],
        ['''dataset_infos.json'''],
        ['''full:README.md'''],
    ]      ,   )
def 	lowercase      (   __UpperCamelCase      ,   __UpperCamelCase						) ->      List[str]:
 __magic_name__     =      tmp_path_factory.mktemp('''dset_infos_dir'''						)
 if "full:README.md" in files:
  with open(dataset_infos_dir / '''README.md'''      ,   '''w'''						) as f:
   f.write('''---\ndataset_info:\n  dataset_size: 42\n---'''						)
 if "empty:README.md" in files:
  with open(dataset_infos_dir / '''README.md'''      ,   '''w'''						) as f:
   f.write(''''''						)
    # we want to support dataset_infos.json for backward compatibility
 if "dataset_infos.json" in files:
  with open(dataset_infos_dir / '''dataset_infos.json'''      ,   '''w'''						) as f:
   f.write('''{"default": {"dataset_size": 42}}'''						)
 __magic_name__     =      DatasetInfosDict.from_directory(__UpperCamelCase						)
 assert dataset_infos
 assert dataset_infos["default"].dataset_size == 42





@pytest.mark.parametrize(
    '''dataset_info'''      ,   [
        DatasetInfo(),
        DatasetInfo(
            description='''foo'''      ,   features=Features({'''a''': Value('''int32'''						)}						)      ,   builder_name='''builder'''      ,   config_name='''config'''      ,   version='''1.0.0'''      ,   splits=[{'''name''': '''train'''}]      ,   download_size=42      ,   ),
    ]      ,   )
def 	lowercase      (   __UpperCamelCase      ,   __UpperCamelCase						) ->      Tuple:
 __magic_name__     =      str(__UpperCamelCase						)
 dataset_info.write_to_directory(__UpperCamelCase						)
 __magic_name__     =      DatasetInfo.from_directory(__UpperCamelCase						)
 assert dataset_info == reloaded
 assert os.path.exists(os.path.join(__UpperCamelCase      ,   '''dataset_info.json'''						)						)





def 	lowercase      (   ) ->      Any:
 __magic_name__     =      DatasetInfo(
     description='''foo'''      ,   citation='''bar'''      ,   homepage='''https://foo.bar'''      ,   license='''CC0'''      ,   features=Features({'''a''': Value('''int32'''						)}						)      ,   post_processed={}      ,   supervised_keys=()      ,   task_templates=[]      ,   builder_name='''builder'''      ,   config_name='''config'''      ,   version='''1.0.0'''      ,   splits=[{'''name''': '''train''', '''num_examples''': 42}]      ,   download_checksums={}      ,   download_size=1337      ,   post_processing_size=442      ,   dataset_size=1234      ,   size_in_bytes=1337 + 442 + 1234      ,   )
 __magic_name__     =      dataset_info._to_yaml_dict()
 assert sorted(__UpperCamelCase						) == sorted(DatasetInfo._INCLUDED_INFO_IN_YAML						)
 for key in DatasetInfo._INCLUDED_INFO_IN_YAML:
  assert key in dataset_info_yaml_dict
  assert isinstance(dataset_info_yaml_dict[key]      ,   (list, dict, int, str)						)
 __magic_name__     =      yaml.safe_dump(__UpperCamelCase						)
 __magic_name__     =      yaml.safe_load(__UpperCamelCase						)
 assert dataset_info_yaml_dict == reloaded





def 	lowercase      (   ) ->      str:
 __magic_name__     =      DatasetInfo()
 __magic_name__     =      dataset_info._to_yaml_dict()
 assert dataset_info_yaml_dict == {}





@pytest.mark.parametrize(
    '''dataset_infos_dict'''      ,   [
        DatasetInfosDict(),
        DatasetInfosDict({'''default''': DatasetInfo()}						),
        DatasetInfosDict({'''my_config_name''': DatasetInfo()}						),
        DatasetInfosDict(
            {
                '''default''': DatasetInfo(
                    description='''foo'''      ,   features=Features({'''a''': Value('''int32'''						)}						)      ,   builder_name='''builder'''      ,   config_name='''config'''      ,   version='''1.0.0'''      ,   splits=[{'''name''': '''train'''}]      ,   download_size=42      ,   )
            }						),
        DatasetInfosDict(
            {
                '''v1''': DatasetInfo(dataset_size=42						),
                '''v2''': DatasetInfo(dataset_size=1337						),
            }						),
    ]      ,   )
def 	lowercase      (   __UpperCamelCase      ,   __UpperCamelCase						) ->      str:
 __magic_name__     =      str(__UpperCamelCase						)
 dataset_infos_dict.write_to_directory(__UpperCamelCase						)
 __magic_name__     =      DatasetInfosDict.from_directory(__UpperCamelCase						)

 # the config_name of the dataset_infos_dict take over the attribute
 for config_name, dataset_info in dataset_infos_dict.items():
  __magic_name__     =      config_name
  # the yaml representation doesn't include fields like description or citation
  # so we just test that we can recover what we can from the yaml
  __magic_name__     =      DatasetInfo._from_yaml_dict(dataset_info._to_yaml_dict()						)
 assert dataset_infos_dict == reloaded

 if dataset_infos_dict:
  assert os.path.exists(os.path.join(__UpperCamelCase      ,   '''README.md'''						)						)

