



import argparse
import os
import shutil
from pathlib import Path

import onnx
import torch
from packaging import version
from torch.onnx import export

from diffusers import OnnxRuntimeModel, OnnxStableDiffusionPipeline, StableDiffusionPipeline


UpperCamelCase_	   =			version.parse(version.parse(torch.__version__).base_version) < version.parse('1.11')




def 		SCREAMING_SNAKE_CASE     (     snake_case__				,   snake_case__				,   snake_case__				,   snake_case__				,   snake_case__				,   snake_case__				,   snake_case__				,   snake_case__=False				,   )   ->	List[str]:
    output_path.parent.mkdir(parents=__lowerCAmelCase				,   exist_ok=__lowerCAmelCase      )
    # PyTorch deprecated the `enable_onnx_checker` and `use_external_data_format` arguments in v1.11,
    # so we check the torch version for backwards compatibility
    if is_torch_less_than_1_11:
        export(
            __lowerCAmelCase				,   __lowerCAmelCase				,   f=output_path.as_posix()				,   input_names=__lowerCAmelCase				,   output_names=__lowerCAmelCase				,   dynamic_axes=__lowerCAmelCase				,   do_constant_folding=__lowerCAmelCase				,   use_external_data_format=__lowerCAmelCase				,   enable_onnx_checker=__lowerCAmelCase				,   opset_version=__lowerCAmelCase				,   )
    else:
        export(
            __lowerCAmelCase				,   __lowerCAmelCase				,   f=output_path.as_posix()				,   input_names=__lowerCAmelCase				,   output_names=__lowerCAmelCase				,   dynamic_axes=__lowerCAmelCase				,   do_constant_folding=__lowerCAmelCase				,   opset_version=__lowerCAmelCase				,   )




@torch.no_grad()
def 		SCREAMING_SNAKE_CASE     (     snake_case__				,   snake_case__				,   snake_case__				,   snake_case__ = False      )   ->	Dict:
    __UpperCAmelCase			 =torch.floataa if fpaa else torch.floataa
    if fpaa and torch.cuda.is_available():
        __UpperCAmelCase			 ="""cuda"""
    elif fpaa and not torch.cuda.is_available():
        raise ValueError('''`float16` model export is only supported on GPUs with CUDA'''      )
    else:
        __UpperCAmelCase			 ="""cpu"""
    __UpperCAmelCase			 =StableDiffusionPipeline.from_pretrained(__lowerCAmelCase				,   torch_dtype=__lowerCAmelCase      ).to(__lowerCAmelCase      )
    __UpperCAmelCase			 =Path(__lowerCAmelCase      )

    # TEXT ENCODER
    __UpperCAmelCase			 =pipeline.text_encoder.config.max_position_embeddings
    __UpperCAmelCase			 =pipeline.text_encoder.config.hidden_size
    __UpperCAmelCase			 =pipeline.tokenizer(
        '''A sample prompt'''				,   padding='''max_length'''				,   max_length=pipeline.tokenizer.model_max_length				,   truncation=__lowerCAmelCase				,   return_tensors='''pt'''				,   )
    onnx_export(
        pipeline.text_encoder				,   model_args=(text_input.input_ids.to(device=__lowerCAmelCase				,   dtype=torch.intaa      ))				,   output_path=output_path / '''text_encoder''' / '''model.onnx'''				,   ordered_input_names=['''input_ids''']				,   output_names=['''last_hidden_state''', '''pooler_output''']				,   dynamic_axes={
            '''input_ids''': {0: '''batch''', 1: '''sequence'''},
        }				,   opset=__lowerCAmelCase				,   )
    del pipeline.text_encoder

    # UNET
    __UpperCAmelCase			 =pipeline.unet.config.in_channels
    __UpperCAmelCase			 =pipeline.unet.config.sample_size
    __UpperCAmelCase			 =output_path / """unet""" / """model.onnx"""
    onnx_export(
        pipeline.unet				,   model_args=(
            torch.randn(2				,   __lowerCAmelCase				,   __lowerCAmelCase				,   __lowerCAmelCase      ).to(device=__lowerCAmelCase				,   dtype=__lowerCAmelCase      ),
            torch.randn(2      ).to(device=__lowerCAmelCase				,   dtype=__lowerCAmelCase      ),
            torch.randn(2				,   __lowerCAmelCase				,   __lowerCAmelCase      ).to(device=__lowerCAmelCase				,   dtype=__lowerCAmelCase      ),
            False,
        )				,   output_path=__lowerCAmelCase				,   ordered_input_names=['''sample''', '''timestep''', '''encoder_hidden_states''', '''return_dict''']				,   output_names=['''out_sample''']				,   dynamic_axes={
            '''sample''': {0: '''batch''', 1: '''channels''', 2: '''height''', 3: '''width'''},
            '''timestep''': {0: '''batch'''},
            '''encoder_hidden_states''': {0: '''batch''', 1: '''sequence'''},
        }				,   opset=__lowerCAmelCase				,   use_external_data_format=__lowerCAmelCase				,   )
    __UpperCAmelCase			 =str(unet_path.absolute().as_posix()      )
    __UpperCAmelCase			 =os.path.dirname(__lowerCAmelCase      )
    __UpperCAmelCase			 =onnx.load(__lowerCAmelCase      )
    # clean up existing tensor files
    shutil.rmtree(__lowerCAmelCase      )
    os.mkdir(__lowerCAmelCase      )
    # collate external tensor files into one
    onnx.save_model(
        __lowerCAmelCase				,   __lowerCAmelCase				,   save_as_external_data=__lowerCAmelCase				,   all_tensors_to_one_file=__lowerCAmelCase				,   location='''weights.pb'''				,   convert_attribute=__lowerCAmelCase				,   )
    del pipeline.unet

    # VAE ENCODER
    __UpperCAmelCase			 =pipeline.vae
    __UpperCAmelCase			 =vae_encoder.config.in_channels
    __UpperCAmelCase			 =vae_encoder.config.sample_size
    # need to get the raw tensor output (sample) from the encoder
    __UpperCAmelCase			 =lambda snake_case__				,   snake_case__      : vae_encoder.encode(__lowerCAmelCase				,   __lowerCAmelCase      )[0].sample()
    onnx_export(
        __lowerCAmelCase				,   model_args=(
            torch.randn(1				,   __lowerCAmelCase				,   __lowerCAmelCase				,   __lowerCAmelCase      ).to(device=__lowerCAmelCase				,   dtype=__lowerCAmelCase      ),
            False,
        )				,   output_path=output_path / '''vae_encoder''' / '''model.onnx'''				,   ordered_input_names=['''sample''', '''return_dict''']				,   output_names=['''latent_sample''']				,   dynamic_axes={
            '''sample''': {0: '''batch''', 1: '''channels''', 2: '''height''', 3: '''width'''},
        }				,   opset=__lowerCAmelCase				,   )

    # VAE DECODER
    __UpperCAmelCase			 =pipeline.vae
    __UpperCAmelCase			 =vae_decoder.config.latent_channels
    __UpperCAmelCase			 =vae_decoder.config.out_channels
    # forward only through the decoder part
    __UpperCAmelCase			 =vae_encoder.decode
    onnx_export(
        __lowerCAmelCase				,   model_args=(
            torch.randn(1				,   __lowerCAmelCase				,   __lowerCAmelCase				,   __lowerCAmelCase      ).to(device=__lowerCAmelCase				,   dtype=__lowerCAmelCase      ),
            False,
        )				,   output_path=output_path / '''vae_decoder''' / '''model.onnx'''				,   ordered_input_names=['''latent_sample''', '''return_dict''']				,   output_names=['''sample''']				,   dynamic_axes={
            '''latent_sample''': {0: '''batch''', 1: '''channels''', 2: '''height''', 3: '''width'''},
        }				,   opset=__lowerCAmelCase				,   )
    del pipeline.vae

    # SAFETY CHECKER
    if pipeline.safety_checker is not None:
        __UpperCAmelCase			 =pipeline.safety_checker
        __UpperCAmelCase			 =safety_checker.config.vision_config.num_channels
        __UpperCAmelCase			 =safety_checker.config.vision_config.image_size
        __UpperCAmelCase			 =safety_checker.forward_onnx
        onnx_export(
            pipeline.safety_checker				,   model_args=(
                torch.randn(
                    1				,   __lowerCAmelCase				,   __lowerCAmelCase				,   __lowerCAmelCase				,   ).to(device=__lowerCAmelCase				,   dtype=__lowerCAmelCase      ),
                torch.randn(1				,   __lowerCAmelCase				,   __lowerCAmelCase				,   __lowerCAmelCase      ).to(device=__lowerCAmelCase				,   dtype=__lowerCAmelCase      ),
            )				,   output_path=output_path / '''safety_checker''' / '''model.onnx'''				,   ordered_input_names=['''clip_input''', '''images''']				,   output_names=['''out_images''', '''has_nsfw_concepts''']				,   dynamic_axes={
                '''clip_input''': {0: '''batch''', 1: '''channels''', 2: '''height''', 3: '''width'''},
                '''images''': {0: '''batch''', 1: '''height''', 2: '''width''', 3: '''channels'''},
            }				,   opset=__lowerCAmelCase				,   )
        del pipeline.safety_checker
        __UpperCAmelCase			 =OnnxRuntimeModel.from_pretrained(output_path / '''safety_checker'''      )
        __UpperCAmelCase			 =pipeline.feature_extractor
    else:
        __UpperCAmelCase			 =None
        __UpperCAmelCase			 =None

    __UpperCAmelCase			 =OnnxStableDiffusionPipeline(
        vae_encoder=OnnxRuntimeModel.from_pretrained(output_path / '''vae_encoder'''      )				,   vae_decoder=OnnxRuntimeModel.from_pretrained(output_path / '''vae_decoder'''      )				,   text_encoder=OnnxRuntimeModel.from_pretrained(output_path / '''text_encoder'''      )				,   tokenizer=pipeline.tokenizer				,   unet=OnnxRuntimeModel.from_pretrained(output_path / '''unet'''      )				,   scheduler=pipeline.scheduler				,   safety_checker=__lowerCAmelCase				,   feature_extractor=__lowerCAmelCase				,   requires_safety_checker=safety_checker is not None				,   )

    onnx_pipeline.save_pretrained(__lowerCAmelCase      )
    print('''ONNX pipeline saved to'''				,   __lowerCAmelCase      )

    del pipeline
    del onnx_pipeline
    __UpperCAmelCase			 =OnnxStableDiffusionPipeline.from_pretrained(__lowerCAmelCase				,   provider='''CPUExecutionProvider'''      )
    print('''ONNX pipeline is loadable'''      )


if __name__ == "__main__":
       UpperCamelCase_	   =			argparse.ArgumentParser()

       parser.add_argument(
           '--model_path',
           type=str,
           required=True,
           help='Path to the `diffusers` checkpoint to convert (either a local directory or on the Hub).',
       )

       parser.add_argument('--output_path', type=str, required=True, help='Path to the output model.')

       parser.add_argument(
           '--opset',
           default=1_4,
           type=int,
           help='The version of the ONNX operator set to use.',
       )
       parser.add_argument('--fp16', action='store_true', default=False, help='Export the models in `float16` mode')

       UpperCamelCase_	   =			parser.parse_args()

       convert_models(args.model_path, args.output_path, args.opset, args.fpaa)











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 		_A						( UpperCAmelCase_	,				unittest.TestCase ):
  lowercase_						:			int						=					KandinskyVaaPipeline
  lowercase_						:			Union[str, Any]						=					[
      '''image_embeds''',
      '''negative_image_embeds''',
  ]
  lowercase_						:			int						=					['''image_embeds''', '''negative_image_embeds''']
  lowercase_						:			str						=					[
      '''generator''',
      '''height''',
      '''width''',
      '''latents''',
      '''guidance_scale''',
      '''num_inference_steps''',
      '''return_dict''',
      '''guidance_scale''',
      '''num_images_per_prompt''',
      '''output_type''',
      '''return_dict''',
  ]
  lowercase_						:			str						=					False

  @property
  def    a					(    self						:	int			):
     """simple docstring"""

     return 32

  @property
  def    a					(    self						:	str			):
     """simple docstring"""

     return 32

  @property
  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     return self.time_input_dim

  @property
  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     return self.time_input_dim * 4

  @property
  def    a					(    self						:	Optional[Any]			):
     """simple docstring"""

     return 1_00

  @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""": """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,
     }

     __UpperCamelCase	:			Any					=					UNetaDConditionModel(**lowerCamelCase__			)
     return model

  @property
  def    a					(    self						:	Union[str, Any]			):
     """simple docstring"""

     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						:	int			):
     """simple docstring"""

     torch.manual_seed(0			)
     __UpperCamelCase	:			Tuple					=					VQModel(**self.dummy_movq_kwargs			)
     return model

  def    a					(    self						:	Tuple			):
     """simple docstring"""

     __UpperCamelCase	:			str					=					self.dummy_unet
     __UpperCamelCase	:			Optional[int]					=					self.dummy_movq

     __UpperCamelCase	:			List[str]					=					DDIMScheduler(
         num_train_timesteps=10_00   ,				beta_schedule="""linear"""   ,				beta_start=0.0_0085   ,				beta_end=0.012   ,				clip_sample=lowerCamelCase__   ,				set_alpha_to_one=lowerCamelCase__   ,				steps_offset=1   ,				prediction_type="""epsilon"""   ,				thresholding=lowerCamelCase__   ,				)

     __UpperCamelCase	:			str					=					{
         """unet""": unet,
         """scheduler""": scheduler,
         """movq""": movq,
     }
     return components

  def    a					(    self						:	Optional[int]   ,				lowerCamelCase__						:	Optional[Any]   ,				lowerCamelCase__						:	Any=0			):
     """simple docstring"""

     __UpperCamelCase	:			Tuple					=					floats_tensor((1, self.text_embedder_hidden_size)   ,				rng=random.Random(lowerCamelCase__			)			).to(lowerCamelCase__			)
     __UpperCamelCase	:			Optional[int]					=					floats_tensor((1, self.text_embedder_hidden_size)   ,				rng=random.Random(seed + 1			)			).to(
         lowerCamelCase__			)
     if str(lowerCamelCase__			).startswith("""mps"""			):
        __UpperCamelCase	:			Optional[int]					=					torch.manual_seed(lowerCamelCase__			)
     else:
        __UpperCamelCase	:			int					=					torch.Generator(device=lowerCamelCase__			).manual_seed(lowerCamelCase__			)
     __UpperCamelCase	:			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    a					(    self						:	Dict			):
     """simple docstring"""

     __UpperCamelCase	:			Dict					=					"""cpu"""

     __UpperCamelCase	:			Dict					=					self.get_dummy_components()

     __UpperCamelCase	:			Union[str, Any]					=					self.pipeline_class(**lowerCamelCase__			)
     __UpperCamelCase	:			int					=					pipe.to(lowerCamelCase__			)

     pipe.set_progress_bar_config(disable=lowerCamelCase__			)

     __UpperCamelCase	:			List[Any]					=					pipe(**self.get_dummy_inputs(lowerCamelCase__			)			)
     __UpperCamelCase	:			List[Any]					=					output.images

     __UpperCamelCase	:			int					=					pipe(
         **self.get_dummy_inputs(lowerCamelCase__			)   ,				return_dict=lowerCamelCase__   ,				)[0]

     __UpperCamelCase	:			List[str]					=					image[0, -3:, -3:, -1]
     __UpperCamelCase	:			Tuple					=					image_from_tuple[0, -3:, -3:, -1]

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

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

  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()

  def    a					(    self						:	List[Any]			):
     """simple docstring"""

     __UpperCamelCase	:			int					=					load_numpy(
         """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
         """/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy"""			)

     __UpperCamelCase	:			List[str]					=					KandinskyVaaPriorPipeline.from_pretrained(
         """kandinsky-community/kandinsky-2-2-prior"""   ,				torch_dtype=torch.floataa			)
     pipe_prior.to(lowerCamelCase__			)

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

     __UpperCamelCase	:			Optional[Any]					=					"""red cat, 4k photo"""

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

     __UpperCamelCase	:			Any					=					torch.Generator(device="""cuda"""			).manual_seed(0			)
     __UpperCamelCase	:			Union[str, Any]					=					pipeline(
         image_embeds=lowerCamelCase__   ,				negative_image_embeds=lowerCamelCase__   ,				generator=lowerCamelCase__   ,				num_inference_steps=1_00   ,				output_type="""np"""   ,				)

     __UpperCamelCase	:			Any					=					output.images[0]

     assert image.shape == (5_12, 5_12, 3)

     assert_mean_pixel_difference(lowerCamelCase__   ,				lowerCamelCase__			)



