

'''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


lowerCamelCase			:				int 			=					version.parse(version.parse(torch.__version__).base_version) < version.parse("1.11")


def 	_lowerCAmelCase (						_UpperCamelCase						:					Optional[int]							,			_UpperCamelCase						:					tuple							,			_UpperCamelCase						:					Path							,			_UpperCamelCase						:					Optional[int]							,			_UpperCamelCase						:					Dict							,			_UpperCamelCase						:					List[Any]							,			_UpperCamelCase						:					Tuple							,			_UpperCamelCase						:					Optional[Any]=False							,			):
     """simple docstring"""




     output_path.parent.mkdir(parents=_UpperCamelCase							,			exist_ok=_UpperCamelCase    )
     # 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(
              _UpperCamelCase							,			_UpperCamelCase							,			f=output_path.as_posix()							,			input_names=_UpperCamelCase							,			output_names=_UpperCamelCase							,			dynamic_axes=_UpperCamelCase							,			do_constant_folding=_UpperCamelCase							,			use_external_data_format=_UpperCamelCase							,			enable_onnx_checker=_UpperCamelCase							,			opset_version=_UpperCamelCase							,			)
     else:
          export(
              _UpperCamelCase							,			_UpperCamelCase							,			f=output_path.as_posix()							,			input_names=_UpperCamelCase							,			output_names=_UpperCamelCase							,			dynamic_axes=_UpperCamelCase							,			do_constant_folding=_UpperCamelCase							,			opset_version=_UpperCamelCase							,			)


@torch.no_grad()
def 	_lowerCAmelCase (						_UpperCamelCase						:					str							,			_UpperCamelCase						:					str							,			_UpperCamelCase						:					int							,			_UpperCamelCase						:					bool = False    ):
     """simple docstring"""




     _SCREAMING_SNAKE_CASE							   =torch.floataa if fpaa else torch.floataa
     if fpaa and torch.cuda.is_available():
          _SCREAMING_SNAKE_CASE							   ='cuda'
     elif fpaa and not torch.cuda.is_available():
          raise ValueError('`float16` model export is only supported on GPUs with CUDA'    )
     else:
          _SCREAMING_SNAKE_CASE							   ='cpu'
     _SCREAMING_SNAKE_CASE							   =StableDiffusionPipeline.from_pretrained(_UpperCamelCase							,			torch_dtype=_UpperCamelCase    ).to(_UpperCamelCase    )
     _SCREAMING_SNAKE_CASE							   =Path(_UpperCamelCase    )

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

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

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

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

     # SAFETY CHECKER
     if pipeline.safety_checker is not None:
          _SCREAMING_SNAKE_CASE							   =pipeline.safety_checker
          _SCREAMING_SNAKE_CASE							   =safety_checker.config.vision_config.num_channels
          _SCREAMING_SNAKE_CASE							   =safety_checker.config.vision_config.image_size
          _SCREAMING_SNAKE_CASE							   =safety_checker.forward_onnx
          onnx_export(
              pipeline.safety_checker							,			model_args=(
                  torch.randn(
                      1							,			_UpperCamelCase							,			_UpperCamelCase							,			_UpperCamelCase							,			).to(device=_UpperCamelCase							,			dtype=_UpperCamelCase    ),
                  torch.randn(1							,			_UpperCamelCase							,			_UpperCamelCase							,			_UpperCamelCase    ).to(device=_UpperCamelCase							,			dtype=_UpperCamelCase    ),
              )							,			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=_UpperCamelCase							,			)
          del pipeline.safety_checker
          _SCREAMING_SNAKE_CASE							   =OnnxRuntimeModel.from_pretrained(output_path / 'safety_checker'    )
          _SCREAMING_SNAKE_CASE							   =pipeline.feature_extractor
     else:
          _SCREAMING_SNAKE_CASE							   =None
          _SCREAMING_SNAKE_CASE							   =None

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

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

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


if __name__ == "__main__":
     lowerCamelCase			:				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=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")

     lowerCamelCase			:				Union[str, Any] 			=					parser.parse_args()

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



'''simple docstring'''
def 	_lowerCAmelCase (						_UpperCamelCase						:					str    )				->      bool:
     """simple docstring"""




     _SCREAMING_SNAKE_CASE							   =0
     for ch in input_str:
          _SCREAMING_SNAKE_CASE							   =ord(_UpperCamelCase    )
          _SCREAMING_SNAKE_CASE							   =pow(2							,			_UpperCamelCase    )

          # If we already turned on bit for current character's unicode
          if bitmap >> ch_unicode & 1 == 1:
               return False
          bitmap |= ch_bit_index_on
     return True


if __name__ == "__main__":
     import doctest

     doctest.testmod()
