





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     __snake_case       (							__lowerCamelCase   ,			unittest.TestCase			):





       '''simple docstring'''







       lowerCAmelCase__      		=     KandinskyVaaPipeline
       lowerCAmelCase__      		=     [
           """image_embeds""",
           """negative_image_embeds""",
       ]
       lowerCAmelCase__      		=     ["""image_embeds""", """negative_image_embeds"""]
       lowerCAmelCase__      		=     [
           """generator""",
           """height""",
           """width""",
           """latents""",
           """guidance_scale""",
           """num_inference_steps""",
           """return_dict""",
           """guidance_scale""",
           """num_images_per_prompt""",
           """output_type""",
           """return_dict""",
       ]
       lowerCAmelCase__      		=     False




       @property
       def 							UpperCAmelCase__ (  self :    List[str]		):
         return 32




       @property
       def 							UpperCAmelCase__ (  self :    List[str]		):
         return 32




       @property
       def 							UpperCAmelCase__ (  self :    List[Any]		):
         return self.time_input_dim




       @property
       def 							UpperCAmelCase__ (  self :    Union[str, Any]		):
         return self.time_input_dim * 4




       @property
       def 							UpperCAmelCase__ (  self :    int		):
         return 100




       @property
       def 							UpperCAmelCase__ (  self :    int		):
         torch.manual_seed(0		)

         __snake_case:						str   =						{
             """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,
         }

         __snake_case:						Optional[int]   =						UNetaDConditionModel(**A		)
         return model




       @property
       def 							UpperCAmelCase__ (  self :    int		):
         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 							UpperCAmelCase__ (  self :    Union[str, Any]		):
         torch.manual_seed(0		)
         __snake_case:						Union[str, Any]   =						VQModel(**self.dummy_movq_kwargs		)
         return model




       def 							UpperCAmelCase__ (  self :    str		):
         __snake_case:						Union[str, Any]   =						self.dummy_unet
         __snake_case:						int   =						self.dummy_movq

         __snake_case:						List[Any]   =						DDIMScheduler(
             num_train_timesteps=1_000							,				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							,				)

         __snake_case:						Tuple   =						{
             """unet""": unet,
             """scheduler""": scheduler,
             """movq""": movq,
         }
         return components




       def 							UpperCAmelCase__ (  self :    List[Any]							,				A :    Optional[int]							,				A :    Tuple=0		):
         __snake_case:						List[str]   =						floats_tensor((1, self.text_embedder_hidden_size)							,				rng=random.Random(A		)		).to(A		)
         __snake_case:						Optional[int]   =						floats_tensor((1, self.text_embedder_hidden_size)							,				rng=random.Random(seed + 1		)		).to(
             A		)
         if str(A		).startswith("""mps"""		):
           __snake_case:						List[str]   =						torch.manual_seed(A		)
         else:
           __snake_case:						str   =						torch.Generator(device=A		).manual_seed(A		)
         __snake_case:						List[Any]   =						{
             """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 							UpperCAmelCase__ (  self :    Any		):
         __snake_case:						str   =						"""cpu"""

         __snake_case:						int   =						self.get_dummy_components()

         __snake_case:						Tuple   =						self.pipeline_class(**A		)
         __snake_case:						Optional[Any]   =						pipe.to(A		)

         pipe.set_progress_bar_config(disable=A		)

         __snake_case:						List[str]   =						pipe(**self.get_dummy_inputs(A		)		)
         __snake_case:						Optional[Any]   =						output.images

         __snake_case:						Optional[int]   =						pipe(
             **self.get_dummy_inputs(A		)							,				return_dict=A							,				)[0]

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

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

         __snake_case:						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     __snake_case       (							unittest.TestCase			):





       '''simple docstring'''







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






       def 							UpperCAmelCase__ (  self :    Optional[int]		):
         __snake_case:						Optional[int]   =						load_numpy(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
             """/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy"""		)

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

         __snake_case:						Tuple   =						KandinskyVaaPipeline.from_pretrained(
             """kandinsky-community/kandinsky-2-2-decoder"""							,				torch_dtype=torch.floataa		)
         __snake_case:						str   =						pipeline.to(A		)
         pipeline.set_progress_bar_config(disable=A		)

         __snake_case:						Tuple   =						"""red cat, 4k photo"""

         __snake_case:						Tuple   =						torch.Generator(device="""cuda"""		).manual_seed(0		)
         __snake_case  ,   __snake_case:						Union[str, Any]   =						pipe_prior(
             A							,				generator=A							,				num_inference_steps=5							,				negative_prompt=""""""							,				).to_tuple()

         __snake_case:						Dict   =						torch.Generator(device="""cuda"""		).manual_seed(0		)
         __snake_case:						Optional[int]   =						pipeline(
             image_embeds=A							,				negative_image_embeds=A							,				generator=A							,				num_inference_steps=100							,				output_type="""np"""							,				)

         __snake_case:						List[Any]   =						output.images[0]

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

         assert_mean_pixel_difference(A							,				A		)













import argparse
import os
import re


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_dummies.py
__UpperCAmelCase       :   Any				       =							"src/diffusers"

# Matches is_xxx_available()
__UpperCAmelCase       :   List[str]				       =							re.compile(R"is\_([a-z_]*)_available\(\)")
# Matches from xxx import bla
__UpperCAmelCase       :   Dict				       =							re.compile(R"\s+from\s+\S*\s+import\s+([^\(\s].*)\n")


__UpperCAmelCase       :   int				       =							"\n{0} = None\n"

__UpperCAmelCase       :   List[str]				       =							"\nclass {0}(metaclass=DummyObject):\n    _backends = {1}\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, {1})\n\n    @classmethod\n    def from_config(cls, *args, **kwargs):\n        requires_backends(cls, {1})\n\n    @classmethod\n    def from_pretrained(cls, *args, **kwargs):\n        requires_backends(cls, {1})\n"


__UpperCAmelCase       :   Tuple				       =							"\ndef {0}(*args, **kwargs):\n    requires_backends({0}, {1})\n"







def 			A__							(    SCREAMING_SNAKE_CASE__)   ->    Any:
  __snake_case:						int   =						_re_backend.findall(SCREAMING_SNAKE_CASE__)
  if len(SCREAMING_SNAKE_CASE__) == 0:
    return None

  return "_and_".join(SCREAMING_SNAKE_CASE__)







def 			A__							(    )   ->    Optional[int]:
  with open(os.path.join(SCREAMING_SNAKE_CASE__					,					"""__init__.py""")					,					"""r"""					,					encoding="""utf-8"""					,					newline="""\n""") as f:
    __snake_case:						Optional[Any]   =						f.readlines()

  # Get to the point we do the actual imports for type checking
  __snake_case:						Tuple   =						0
  __snake_case:						Any   =						{}
  # Go through the end of the file
  while line_index < len(SCREAMING_SNAKE_CASE__):
    # If the line contains is_backend_available, we grab all objects associated with the `else` block
    __snake_case:						List[Any]   =						find_backend(lines[line_index])
    if backend is not None:
      while not lines[line_index].startswith("""else:"""):
        line_index += 1
      line_index += 1
      __snake_case:						Any   =						[]
      # Until we unindent, add backend objects to the list
      while line_index < len(SCREAMING_SNAKE_CASE__) and len(lines[line_index]) > 1:
        __snake_case:						List[Any]   =						lines[line_index]
        __snake_case:						str   =						_re_single_line_import.search(SCREAMING_SNAKE_CASE__)
        if single_line_import_search is not None:
          objects.extend(single_line_import_search.groups()[0].split(""", """))
        elif line.startswith(""" """ * 8):
          objects.append(line[8:-2])
        line_index += 1

      if len(SCREAMING_SNAKE_CASE__) > 0:
        __snake_case:						Optional[Any]   =						objects
    else:
      line_index += 1

  return backend_specific_objects







def 			A__							(    SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)   ->    List[Any]:
  if name.isupper():
    return DUMMY_CONSTANT.format(SCREAMING_SNAKE_CASE__)
  elif name.islower():
    return DUMMY_FUNCTION.format(SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)
  else:
    return DUMMY_CLASS.format(SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)







def 			A__							(    SCREAMING_SNAKE_CASE__=None)   ->    Optional[int]:
  if backend_specific_objects is None:
    __snake_case:						Union[str, Any]   =						read_init()
  # For special correspondence backend to module name as used in the function requires_modulename
  __snake_case:						Union[str, Any]   =						{}

  for backend, objects in backend_specific_objects.items():
    __snake_case:						List[Any]   =						"""[""" + """, """.join(F'''"{b}"''' for b in backend.split("""_and_""")) + """]"""
    __snake_case:						Dict   =						"""# This file is autogenerated by the command `make fix-copies`, do not edit.\n"""
    dummy_file += "from ..utils import DummyObject, requires_backends\n\n"
    dummy_file += "\n".join([create_dummy_object(SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__) for o in objects])
    __snake_case:						List[str]   =						dummy_file

  return dummy_files







def 			A__							(    SCREAMING_SNAKE_CASE__=False)   ->    int:
  __snake_case:						List[str]   =						create_dummy_files()
  # For special correspondence backend to shortcut as used in utils/dummy_xxx_objects.py
  __snake_case:						Dict   =						{"""torch""": """pt"""}

  # Locate actual dummy modules and read their content.
  __snake_case:						Optional[Any]   =						os.path.join(SCREAMING_SNAKE_CASE__					,					"""utils""")
  __snake_case:						List[Any]   =						{
      backend: os.path.join(SCREAMING_SNAKE_CASE__					,					F'''dummy_{short_names.get(SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)}_objects.py''')
      for backend in dummy_files.keys()
  }

  __snake_case:						int   =						{}
  for backend, file_path in dummy_file_paths.items():
    if os.path.isfile(SCREAMING_SNAKE_CASE__):
      with open(SCREAMING_SNAKE_CASE__					,					"""r"""					,					encoding="""utf-8"""					,					newline="""\n""") as f:
        __snake_case:						Any   =						f.read()
    else:
      __snake_case:						Tuple   =						""""""

  for backend in dummy_files.keys():
    if dummy_files[backend] != actual_dummies[backend]:
      if overwrite:
        print(
            F'''Updating diffusers.utils.dummy_{short_names.get(SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)}_objects.py as the main '''
            """__init__ has new objects.""")
        with open(dummy_file_paths[backend]					,					"""w"""					,					encoding="""utf-8"""					,					newline="""\n""") as f:
          f.write(dummy_files[backend])
      else:
        raise ValueError(
            """The main __init__ has objects that are not present in """
            F'''diffusers.utils.dummy_{short_names.get(SCREAMING_SNAKE_CASE__					,					SCREAMING_SNAKE_CASE__)}_objects.py. Run `make fix-copies` '''
            """to fix this.""")


if __name__ == "__main__":
      __UpperCAmelCase       :   Optional[Any]				       =							argparse.ArgumentParser()
      parser.add_argument("--fix_and_overwrite", action="store_true", help="Whether to fix inconsistencies.")
      __UpperCAmelCase       :   List[Any]				       =							parser.parse_args()

      check_dummies(args.fix_and_overwrite)






