




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


SCREAMING_SNAKE_CASE__       :      Dict  =   version.parse(version.parse(torch.__version__).base_version) < version.parse('1.11')

def      A      (      _SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE=False ,)	-> List[Any]:
       output_path.parent.mkdir(parents=_SCREAMING_SNAKE_CASE ,exist_ok=_SCREAMING_SNAKE_CASE    )
       # 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(
                  _SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,f=output_path.as_posix() ,input_names=_SCREAMING_SNAKE_CASE ,output_names=_SCREAMING_SNAKE_CASE ,dynamic_axes=_SCREAMING_SNAKE_CASE ,do_constant_folding=_SCREAMING_SNAKE_CASE ,use_external_data_format=_SCREAMING_SNAKE_CASE ,enable_onnx_checker=_SCREAMING_SNAKE_CASE ,opset_version=_SCREAMING_SNAKE_CASE ,)
       else:
              export(
                  _SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,f=output_path.as_posix() ,input_names=_SCREAMING_SNAKE_CASE ,output_names=_SCREAMING_SNAKE_CASE ,dynamic_axes=_SCREAMING_SNAKE_CASE ,do_constant_folding=_SCREAMING_SNAKE_CASE ,opset_version=_SCREAMING_SNAKE_CASE ,)

@torch.no_grad()
def      A      (      _SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE = False    )	-> List[Any]:
       lowerCamelCase   :      Union[str, Any]			 =					torch.floataa if fpaa else torch.floataa
       if fpaa and torch.cuda.is_available():
              lowerCamelCase   :      Union[str, Any]			 =					"cuda"
       elif fpaa and not torch.cuda.is_available():
              raise ValueError("`float16` model export is only supported on GPUs with CUDA"    )
       else:
              lowerCamelCase   :      Dict			 =					"cpu"
       lowerCamelCase   :      int			 =					StableDiffusionPipeline.from_pretrained(_SCREAMING_SNAKE_CASE ,torch_dtype=_SCREAMING_SNAKE_CASE    ).to(_SCREAMING_SNAKE_CASE    )
       lowerCamelCase   :      Tuple			 =					Path(_SCREAMING_SNAKE_CASE    )

       # TEXT ENCODER
       lowerCamelCase   :      Dict			 =					pipeline.text_encoder.config.max_position_embeddings
       lowerCamelCase   :      Dict			 =					pipeline.text_encoder.config.hidden_size
       lowerCamelCase   :      Dict			 =					pipeline.tokenizer(
           "A sample prompt" ,padding="max_length" ,max_length=pipeline.tokenizer.model_max_length ,truncation=_SCREAMING_SNAKE_CASE ,return_tensors="pt" ,)
       onnx_export(
           pipeline.text_encoder ,model_args=(text_input.input_ids.to(device=_SCREAMING_SNAKE_CASE ,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=_SCREAMING_SNAKE_CASE ,)
       del pipeline.text_encoder

       # UNET
       lowerCamelCase   :      int			 =					pipeline.unet.config.in_channels
       lowerCamelCase   :      Any			 =					pipeline.unet.config.sample_size
       lowerCamelCase   :      List[str]			 =					output_path / "unet" / "model.onnx"
       onnx_export(
           pipeline.unet ,model_args=(
               torch.randn(2 ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    ).to(device=_SCREAMING_SNAKE_CASE ,dtype=_SCREAMING_SNAKE_CASE    ),
               torch.randn(2    ).to(device=_SCREAMING_SNAKE_CASE ,dtype=_SCREAMING_SNAKE_CASE    ),
               torch.randn(2 ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    ).to(device=_SCREAMING_SNAKE_CASE ,dtype=_SCREAMING_SNAKE_CASE    ),
               False,
           ) ,output_path=_SCREAMING_SNAKE_CASE ,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=_SCREAMING_SNAKE_CASE ,use_external_data_format=_SCREAMING_SNAKE_CASE ,)
       lowerCamelCase   :      List[str]			 =					str(unet_path.absolute().as_posix()    )
       lowerCamelCase   :      Optional[int]			 =					os.path.dirname(_SCREAMING_SNAKE_CASE    )
       lowerCamelCase   :      str			 =					onnx.load(_SCREAMING_SNAKE_CASE    )
       # clean up existing tensor files
       shutil.rmtree(_SCREAMING_SNAKE_CASE    )
       os.mkdir(_SCREAMING_SNAKE_CASE    )
       # collate external tensor files into one
       onnx.save_model(
           _SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,save_as_external_data=_SCREAMING_SNAKE_CASE ,all_tensors_to_one_file=_SCREAMING_SNAKE_CASE ,location="weights.pb" ,convert_attribute=_SCREAMING_SNAKE_CASE ,)
       del pipeline.unet

       # VAE ENCODER
       lowerCamelCase   :      int			 =					pipeline.vae
       lowerCamelCase   :      Optional[Any]			 =					vae_encoder.config.in_channels
       lowerCamelCase   :      int			 =					vae_encoder.config.sample_size
       # need to get the raw tensor output (sample) from the encoder
       lowerCamelCase   :      str			 =					lambda _SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    : vae_encoder.encode(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    )[0].sample()
       onnx_export(
           _SCREAMING_SNAKE_CASE ,model_args=(
               torch.randn(1 ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    ).to(device=_SCREAMING_SNAKE_CASE ,dtype=_SCREAMING_SNAKE_CASE    ),
               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=_SCREAMING_SNAKE_CASE ,)

       # VAE DECODER
       lowerCamelCase   :      int			 =					pipeline.vae
       lowerCamelCase   :      Optional[int]			 =					vae_decoder.config.latent_channels
       lowerCamelCase   :      str			 =					vae_decoder.config.out_channels
       # forward only through the decoder part
       lowerCamelCase   :      str			 =					vae_encoder.decode
       onnx_export(
           _SCREAMING_SNAKE_CASE ,model_args=(
               torch.randn(1 ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    ).to(device=_SCREAMING_SNAKE_CASE ,dtype=_SCREAMING_SNAKE_CASE    ),
               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=_SCREAMING_SNAKE_CASE ,)
       del pipeline.vae

       # SAFETY CHECKER
       if pipeline.safety_checker is not None:
              lowerCamelCase   :      int			 =					pipeline.safety_checker
              lowerCamelCase   :      str			 =					safety_checker.config.vision_config.num_channels
              lowerCamelCase   :      Tuple			 =					safety_checker.config.vision_config.image_size
              lowerCamelCase   :      int			 =					safety_checker.forward_onnx
              onnx_export(
                  pipeline.safety_checker ,model_args=(
                      torch.randn(
                          1 ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,).to(device=_SCREAMING_SNAKE_CASE ,dtype=_SCREAMING_SNAKE_CASE    ),
                      torch.randn(1 ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE    ).to(device=_SCREAMING_SNAKE_CASE ,dtype=_SCREAMING_SNAKE_CASE    ),
                  ) ,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=_SCREAMING_SNAKE_CASE ,)
              del pipeline.safety_checker
              lowerCamelCase   :      List[Any]			 =					OnnxRuntimeModel.from_pretrained(output_path / "safety_checker"    )
              lowerCamelCase   :      Optional[Any]			 =					pipeline.feature_extractor
       else:
              lowerCamelCase   :      List[Any]			 =					None
              lowerCamelCase   :      Optional[int]			 =					None

       lowerCamelCase   :      str			 =					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=_SCREAMING_SNAKE_CASE ,feature_extractor=_SCREAMING_SNAKE_CASE ,requires_safety_checker=safety_checker is not None ,)

       onnx_pipeline.save_pretrained(_SCREAMING_SNAKE_CASE    )
       print("ONNX pipeline saved to" ,_SCREAMING_SNAKE_CASE    )

       del pipeline
       del onnx_pipeline
       lowerCamelCase   :      List[str]			 =					OnnxStableDiffusionPipeline.from_pretrained(_SCREAMING_SNAKE_CASE ,provider="CPUExecutionProvider"    )
       print("ONNX pipeline is loadable"    )


if __name__ == "__main__":
       SCREAMING_SNAKE_CASE__       :      int  =   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=14,
           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')

       SCREAMING_SNAKE_CASE__       :      Any  =   parser.parse_args()

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












import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
     import torch

if is_vision_available():
     from PIL import Image

     from transformers import GLPNImageProcessor




class       __lowerCAmelCase				(							unittest.TestCase):






       def __init__(      self:    List[Any]				,    _lowerCAmelCase:    str				,    _lowerCAmelCase:    Dict=7				,    _lowerCAmelCase:    Optional[Any]=3				,    _lowerCAmelCase:    List[Any]=18				,    _lowerCAmelCase:    int=30				,    _lowerCAmelCase:    Union[str, Any]=4_00				,    _lowerCAmelCase:    Any=True				,    _lowerCAmelCase:    Union[str, Any]=32				,    _lowerCAmelCase:    int=True				,    ):
         lowercase			:Any							=  parent
         lowercase			:List[str]							=  batch_size
         lowercase			:Any							=  num_channels
         lowercase			:Optional[Any]							=  image_size
         lowercase			:Optional[Any]							=  min_resolution
         lowercase			:str							=  max_resolution
         lowercase			:str							=  do_resize
         lowercase			:Optional[int]							=  size_divisor
         lowercase			:Tuple							=  do_rescale






       def     SCREAMING_SNAKE_CASE			(      self:    Any ):
         return {
             "do_resize": self.do_resize,
             "size_divisor": self.size_divisor,
             "do_rescale": self.do_rescale,
         }







@require_torch
@require_vision
class       __lowerCAmelCase				(							lowerCAmelCase					,       unittest.TestCase):
       _a           =  GLPNImageProcessor if is_vision_available() else None






       def     SCREAMING_SNAKE_CASE			(      self:    List[str] ):
         lowercase			:int							=  GLPNImageProcessingTester(self )






       @property
       def     SCREAMING_SNAKE_CASE			(      self:    str ):
         return self.image_processor_tester.prepare_image_processor_dict()






       def     SCREAMING_SNAKE_CASE			(      self:    Optional[int] ):
         lowercase			:Optional[Any]							=  self.image_processing_class(**self.image_processor_dict )
         self.assertTrue(hasattr(_lowerCAmelCase				,    "do_resize" ) )
         self.assertTrue(hasattr(_lowerCAmelCase				,    "size_divisor" ) )
         self.assertTrue(hasattr(_lowerCAmelCase				,    "resample" ) )
         self.assertTrue(hasattr(_lowerCAmelCase				,    "do_rescale" ) )






       def     SCREAMING_SNAKE_CASE			(      self:    List[Any] ):
         pass






       def     SCREAMING_SNAKE_CASE			(      self:    Any ):
         # Initialize image_processing
         lowercase			:Union[str, Any]							=  self.image_processing_class(**self.image_processor_dict )
         # create random PIL images
         lowercase			:List[str]							=  prepare_image_inputs(self.image_processor_tester				,    equal_resolution=_lowerCAmelCase )
         for image in image_inputs:
           self.assertIsInstance(_lowerCAmelCase				,    Image.Image )

         # Test not batched input (GLPNImageProcessor doesn't support batching)
         lowercase			:str							=  image_processing(image_inputs[0]				,    return_tensors="pt" ).pixel_values
         self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisor == 0 )
         self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisor == 0 )






       def     SCREAMING_SNAKE_CASE			(      self:    Dict ):
         # Initialize image_processing
         lowercase			:Union[str, Any]							=  self.image_processing_class(**self.image_processor_dict )
         # create random numpy tensors
         lowercase			:Union[str, Any]							=  prepare_image_inputs(self.image_processor_tester				,    equal_resolution=_lowerCAmelCase				,    numpify=_lowerCAmelCase )
         for image in image_inputs:
           self.assertIsInstance(_lowerCAmelCase				,    np.ndarray )

         # Test not batched input (GLPNImageProcessor doesn't support batching)
         lowercase			:Optional[int]							=  image_processing(image_inputs[0]				,    return_tensors="pt" ).pixel_values
         self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisor == 0 )
         self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisor == 0 )






       def     SCREAMING_SNAKE_CASE			(      self:    Tuple ):
         # Initialize image_processing
         lowercase			:Optional[Any]							=  self.image_processing_class(**self.image_processor_dict )
         # create random PyTorch tensors
         lowercase			:List[str]							=  prepare_image_inputs(self.image_processor_tester				,    equal_resolution=_lowerCAmelCase				,    torchify=_lowerCAmelCase )
         for image in image_inputs:
           self.assertIsInstance(_lowerCAmelCase				,    torch.Tensor )

         # Test not batched input (GLPNImageProcessor doesn't support batching)
         lowercase			:Dict							=  image_processing(image_inputs[0]				,    return_tensors="pt" ).pixel_values
         self.assertTrue(encoded_images.shape[-1] % self.image_processor_tester.size_divisor == 0 )
         self.assertTrue(encoded_images.shape[-2] % self.image_processor_tester.size_divisor == 0 )



