
"""simple docstring"""



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 					lowerCamelCase   (     _snake_case    ,_snake_case    ,_snake_case    ,_snake_case    ,_snake_case    ,_snake_case    ,_snake_case    ,_snake_case=False    ,):
    output_path.parent.mkdir(parents=_snake_case    ,exist_ok=_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(
            _snake_case    ,_snake_case    ,f=output_path.as_posix()    ,input_names=_snake_case    ,output_names=_snake_case    ,dynamic_axes=_snake_case    ,do_constant_folding=_snake_case    ,use_external_data_format=_snake_case    ,enable_onnx_checker=_snake_case    ,opset_version=_snake_case    ,)
    else:
        export(
            _snake_case    ,_snake_case    ,f=output_path.as_posix()    ,input_names=_snake_case    ,output_names=_snake_case    ,dynamic_axes=_snake_case    ,do_constant_folding=_snake_case    ,opset_version=_snake_case    ,)
@torch.no_grad()
def 					lowerCamelCase   (     _snake_case    ,_snake_case    ,_snake_case    ,_snake_case = False ):
    UpperCAmelCase__	:      Optional[int]     							=					torch.floataa if fpaa else torch.floataa
    if fpaa and torch.cuda.is_available():
        UpperCAmelCase__	:      Tuple     							=					"cuda"
    elif fpaa and not torch.cuda.is_available():
        raise ValueError('`float16` model export is only supported on GPUs with CUDA' )
    else:
        UpperCAmelCase__	:      Union[str, Any]     							=					"cpu"
    UpperCAmelCase__	:      Tuple     							=					StableDiffusionPipeline.from_pretrained(_snake_case    ,torch_dtype=_snake_case ).to(_snake_case )
    UpperCAmelCase__	:      str     							=					Path(_snake_case )

    # TEXT ENCODER
    UpperCAmelCase__	:      Any     							=					pipeline.text_encoder.config.max_position_embeddings
    UpperCAmelCase__	:      int     							=					pipeline.text_encoder.config.hidden_size
    UpperCAmelCase__	:      List[str]     							=					pipeline.tokenizer(
        'A sample prompt'    ,padding='max_length'    ,max_length=pipeline.tokenizer.model_max_length    ,truncation=_snake_case    ,return_tensors='pt'    ,)
    onnx_export(
        pipeline.text_encoder    ,model_args=(text_input.input_ids.to(device=_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=_snake_case    ,)
    del pipeline.text_encoder

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

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

    # VAE DECODER
    UpperCAmelCase__	:      int     							=					pipeline.vae
    UpperCAmelCase__	:      Tuple     							=					vae_decoder.config.latent_channels
    UpperCAmelCase__	:      Any     							=					vae_decoder.config.out_channels
    # forward only through the decoder part
    UpperCAmelCase__	:      Any     							=					vae_encoder.decode
    onnx_export(
        _snake_case    ,model_args=(
            torch.randn(1    ,_snake_case    ,_snake_case    ,_snake_case ).to(device=_snake_case    ,dtype=_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=_snake_case    ,)
    del pipeline.vae

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

    UpperCAmelCase__	:      int     							=					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=_snake_case    ,feature_extractor=_snake_case    ,requires_safety_checker=safety_checker is not None    ,)

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

    del pipeline
    del onnx_pipeline
    UpperCAmelCase__	:      List[Any]     							=					OnnxStableDiffusionPipeline.from_pretrained(_snake_case    ,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=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')

    UpperCamelCase__          =	parser.parse_args()

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








from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import convert_to_rgb, normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    OPENAI_CLIP_MEAN,
    OPENAI_CLIP_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
  import PIL


SCREAMING_SNAKE_CASE__							:      List[str]         =    logging.get_logger(__name__)







class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    int  		=					["""pixel_values"""]





   def __init__(			self       ,      snake_case = True       ,      snake_case = None       ,      snake_case = PILImageResampling.BICUBIC       ,      snake_case = True       ,      snake_case = 1 / 255       ,      snake_case = True       ,      snake_case = None       ,      snake_case = None       ,      snake_case = True       ,      **snake_case       ,      )					->	None:
      """simple docstring"""





      super().__init__(**snake_case   )
      a__						:    List[str]   		=	size if size is not None else {"height": 384, "width": 384}
      a__						:    List[str]   		=	get_size_dict(snake_case       ,      default_to_square=snake_case   )

      a__						:    Any   		=	do_resize
      a__						:    Optional[int]   		=	size
      a__						:    int   		=	resample
      a__						:    Optional[Any]   		=	do_rescale
      a__						:    Dict   		=	rescale_factor
      a__						:    Dict   		=	do_normalize
      a__						:    Any   		=	image_mean if image_mean is not None else OPENAI_CLIP_MEAN
      a__						:    List[str]   		=	image_std if image_std is not None else OPENAI_CLIP_STD
      a__						:    str   		=	do_convert_rgb





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case       ,      snake_case = PILImageResampling.BICUBIC       ,      snake_case = None       ,      **snake_case       ,      )					->	np.ndarray:
      """simple docstring"""





      a__						:    Optional[int]   		=	get_size_dict(snake_case       ,      default_to_square=snake_case   )
      if "height" not in size or "width" not in size:
         raise ValueError(F"""The `size` dictionary must contain the keys `height` and `width`. Got {size.keys()}"""   )
      a__						:    Optional[int]   		=	(size["height"], size["width"])
      return resize(snake_case       ,      size=snake_case       ,      resample=snake_case       ,      data_format=snake_case       ,      **snake_case   )





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case       ,      snake_case = None       ,      **snake_case       ,      )					->	List[str]:
      """simple docstring"""





      return rescale(snake_case       ,      scale=snake_case       ,      data_format=snake_case       ,      **snake_case   )





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case = None       ,      **snake_case       ,      )					->	np.ndarray:
      """simple docstring"""





      return normalize(snake_case       ,      mean=snake_case       ,      std=snake_case       ,      data_format=snake_case       ,      **snake_case   )





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = None       ,      snake_case = ChannelDimension.FIRST       ,      **snake_case       ,      )					->	PIL.Image.Image:
      """simple docstring"""





      a__						:    List[str]   		=	do_resize if do_resize is not None else self.do_resize
      a__						:    Optional[Any]   		=	resample if resample is not None else self.resample
      a__						:    int   		=	do_rescale if do_rescale is not None else self.do_rescale
      a__						:    int   		=	rescale_factor if rescale_factor is not None else self.rescale_factor
      a__						:    Union[str, Any]   		=	do_normalize if do_normalize is not None else self.do_normalize
      a__						:    List[Any]   		=	image_mean if image_mean is not None else self.image_mean
      a__						:    int   		=	image_std if image_std is not None else self.image_std
      a__						:    Tuple   		=	do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

      a__						:    Tuple   		=	size if size is not None else self.size
      a__						:    Any   		=	get_size_dict(snake_case       ,      default_to_square=snake_case   )

      a__						:    str   		=	make_list_of_images(snake_case   )

      if not valid_images(snake_case   ):
         raise ValueError(
             "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
             "torch.Tensor, tf.Tensor or jax.ndarray."   )

      if do_resize and size is None or resample is None:
         raise ValueError("Size and resample must be specified if do_resize is True."   )

      if do_rescale and rescale_factor is None:
         raise ValueError("Rescale factor must be specified if do_rescale is True."   )

      if do_normalize and (image_mean is None or image_std is None):
         raise ValueError("Image mean and std must be specified if do_normalize is True."   )

      # PIL RGBA images are converted to RGB
      if do_convert_rgb:
         a__						:    List[str]   		=	[convert_to_rgb(snake_case   ) for image in images]

      # All transformations expect numpy arrays.
      a__						:    Dict   		=	[to_numpy_array(snake_case   ) for image in images]

      if do_resize:
         a__						:    Union[str, Any]   		=	[self.resize(image=snake_case       ,      size=snake_case       ,      resample=snake_case   ) for image in images]

      if do_rescale:
         a__						:    Tuple   		=	[self.rescale(image=snake_case       ,      scale=snake_case   ) for image in images]

      if do_normalize:
         a__						:    List[str]   		=	[self.normalize(image=snake_case       ,      mean=snake_case       ,      std=snake_case   ) for image in images]

      a__						:    Union[str, Any]   		=	[to_channel_dimension_format(snake_case       ,      snake_case   ) for image in images]

      a__						:    List[Any]   		=	BatchFeature(data={"pixel_values": images}       ,      tensor_type=snake_case   )

      return encoded_outputs

