





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


_A :   int  			=  version.parse(version.parse(torch.__version__).base_version) < version.parse('1.11')



def 	_a						(   UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase=False      ,  )    ->    str:







   """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 	_a						(   UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase = False					)    ->    Optional[Any]:







   """simple docstring"""
   lowerCamelCase__       :						Dict        =							torch.floataa if fpaa else torch.floataa
   if fpaa and torch.cuda.is_available():
      lowerCamelCase__       :						Optional[int]        =							'''cuda'''
   elif fpaa and not torch.cuda.is_available():
      raise ValueError('''`float16` model export is only supported on GPUs with CUDA'''					)
   else:
      lowerCamelCase__       :						str        =							'''cpu'''
   lowerCamelCase__       :						List[Any]        =							StableDiffusionPipeline.from_pretrained(UpperCAmelCase      ,  torch_dtype=UpperCAmelCase					).to(UpperCAmelCase					)
   lowerCamelCase__       :						List[str]        =							Path(UpperCAmelCase					)

   # TEXT ENCODER
   lowerCamelCase__       :						int        =							pipeline.text_encoder.config.max_position_embeddings
   lowerCamelCase__       :						List[str]        =							pipeline.text_encoder.config.hidden_size
   lowerCamelCase__       :						List[Any]        =							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
   lowerCamelCase__       :						Optional[int]        =							pipeline.unet.config.in_channels
   lowerCamelCase__       :						int        =							pipeline.unet.config.sample_size
   lowerCamelCase__       :						Tuple        =							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      ,  )
   lowerCamelCase__       :						List[str]        =							str(unet_path.absolute().as_posix()					)
   lowerCamelCase__       :						str        =							os.path.dirname(UpperCAmelCase					)
   lowerCamelCase__       :						Dict        =							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
   lowerCamelCase__       :						Optional[Any]        =							pipeline.vae
   lowerCamelCase__       :						List[str]        =							vae_encoder.config.in_channels
   lowerCamelCase__       :						Optional[Any]        =							vae_encoder.config.sample_size
   # need to get the raw tensor output (sample) from the encoder
   lowerCamelCase__       :						Union[str, Any]        =							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
   lowerCamelCase__       :						Optional[Any]        =							pipeline.vae
   lowerCamelCase__       :						Any        =							vae_decoder.config.latent_channels
   lowerCamelCase__       :						Optional[int]        =							vae_decoder.config.out_channels
   # forward only through the decoder part
   lowerCamelCase__       :						Optional[Any]        =							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:
      lowerCamelCase__       :						Dict        =							pipeline.safety_checker
      lowerCamelCase__       :						Union[str, Any]        =							safety_checker.config.vision_config.num_channels
      lowerCamelCase__       :						int        =							safety_checker.config.vision_config.image_size
      lowerCamelCase__       :						Dict        =							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
      lowerCamelCase__       :						Optional[int]        =							OnnxRuntimeModel.from_pretrained(output_path / '''safety_checker'''					)
      lowerCamelCase__       :						Optional[Any]        =							pipeline.feature_extractor
   else:
      lowerCamelCase__       :						Union[str, Any]        =							None
      lowerCamelCase__       :						Optional[int]        =							None

   lowerCamelCase__       :						List[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=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
   lowerCamelCase__       :						List[Any]        =							OnnxStableDiffusionPipeline.from_pretrained(UpperCAmelCase      ,  provider='''CPUExecutionProvider'''					)
   print('''ONNX pipeline is loadable'''					)


if __name__ == "__main__":
       _A :   str  			=  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')

       _A :   Optional[Any]  			=  parser.parse_args()

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










def 	_a						(   UpperCAmelCase					)    ->    int:







   """simple docstring"""
   if not isinstance(UpperCAmelCase      ,  UpperCAmelCase					) or number < 0:
      raise ValueError('''Input must be a non-negative integer'''					)

   lowerCamelCase__       :						List[str]        =							0
   while number:
      # This way we arrive at next set bit (next 1) instead of looping
      # through each bit and checking for 1s hence the
      # loop won't run 32 times it will only run the number of `1` times
      number &= number - 1
      count += 1
   return count


if __name__ == "__main__":
       import doctest

       doctest.testmod()



