





'''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_   :   Union[str, Any]       =					version.parse(version.parse(torch.__version__).base_version) < version.parse("1.11")







def    _lowerCamelCase  (__lowerCamelCase  :     Tuple					,  __lowerCamelCase  :     tuple					,  __lowerCamelCase  :     Path					,  __lowerCamelCase  :     int					,  __lowerCamelCase  :     Any					,  __lowerCamelCase  :     Optional[int]					,  __lowerCamelCase  :     int					,  __lowerCamelCase  :     str=False					,  )   ->    Dict:
			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    _lowerCamelCase  (__lowerCamelCase  :     str					,  __lowerCamelCase  :     str					,  __lowerCamelCase  :     int					,  __lowerCamelCase  :     bool = False     )   ->    Optional[int]:
			a__     = torch.floataa if fpaa else torch.floataa
			if fpaa and torch.cuda.is_available():
						a__     = "cuda"
			elif fpaa and not torch.cuda.is_available():
						raise ValueError("`float16` model export is only supported on GPUs with CUDA"     )
			else:
						a__     = "cpu"
			a__     = StableDiffusionPipeline.from_pretrained(__lowerCamelCase					,  torch_dtype=__lowerCamelCase     ).to(__lowerCamelCase     )
			a__     = Path(__lowerCamelCase     )

			# TEXT ENCODER
			a__     = pipeline.text_encoder.config.max_position_embeddings
			a__     = pipeline.text_encoder.config.hidden_size
			a__     = 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
			a__     = pipeline.unet.config.in_channels
			a__     = pipeline.unet.config.sample_size
			a__     = 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					,  )
			a__     = str(unet_path.absolute().as_posix()     )
			a__     = os.path.dirname(__lowerCamelCase     )
			a__     = 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
			a__     = pipeline.vae
			a__     = vae_encoder.config.in_channels
			a__     = vae_encoder.config.sample_size
			# need to get the raw tensor output (sample) from the encoder
			a__     = lambda __lowerCamelCase					,  __lowerCamelCase     : 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
			a__     = pipeline.vae
			a__     = vae_decoder.config.latent_channels
			a__     = vae_decoder.config.out_channels
			# forward only through the decoder part
			a__     = 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:
						a__     = pipeline.safety_checker
						a__     = safety_checker.config.vision_config.num_channels
						a__     = safety_checker.config.vision_config.image_size
						a__     = 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
						a__     = OnnxRuntimeModel.from_pretrained(output_path / "safety_checker"     )
						a__     = pipeline.feature_extractor
			else:
						a__     = None
						a__     = None

			a__     = 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
			a__     = OnnxStableDiffusionPipeline.from_pretrained(__lowerCamelCase					,  provider="CPUExecutionProvider"     )
			print("ONNX pipeline is loadable"     )


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

		lowerCAmelCase_   :   int       =					parser.parse_args()

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










'''simple docstring'''


import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase_   :   str       =					logging.get_logger(__name__)

lowerCAmelCase_   :   Optional[Any]       =					"▁"

lowerCAmelCase_   :   Union[str, Any]       =					{"vocab_file": "sentencepiece.bpe.model"}

lowerCAmelCase_   :   str       =					{
    "vocab_file": {
        "facebook/xglm-564M": "https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model",
    }
}

lowerCAmelCase_   :   Optional[int]       =					{
    "facebook/xglm-564M": 2048,
}






class       UpperCamelCase__  (		__lowerCAmelCase     ):
	lowerCAmelCase__					:			int								=						VOCAB_FILES_NAMES
	lowerCAmelCase__					:			Optional[int]								=						PRETRAINED_VOCAB_FILES_MAP
	lowerCAmelCase__					:			Tuple								=						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	lowerCAmelCase__					:			Optional[int]								=						["input_ids", "attention_mask"]







	def __init__(				self			:   Optional[Any] , lowerCamelCase			:   Union[str, Any] , lowerCamelCase			:   int="<s>" , lowerCamelCase			:   List[Any]="</s>" , lowerCamelCase			:   Any="</s>" , lowerCamelCase			:   Tuple="<s>" , lowerCamelCase			:   Optional[Any]="<unk>" , lowerCamelCase			:   Tuple="<pad>" , lowerCamelCase			:   Optional[Dict[str, Any]] = None , **lowerCamelCase			:   Union[str, Any] , ):




				'''simple docstring'''
				a__     = {} if sp_model_kwargs is None else sp_model_kwargs

				# Compatibility with the original tokenizer
				a__     = 7
				a__     = [F'''<madeupword{i}>''' for i in range(self.num_madeup_words		)]

				a__     = kwargs.get("additional_special_tokens" , []		)
				kwargs["additional_special_tokens"] += [
				    word for word in madeup_words if word not in kwargs["additional_special_tokens"]
				]

				super().__init__(
				    bos_token=lowerCamelCase , eos_token=lowerCamelCase , unk_token=lowerCamelCase , sep_token=lowerCamelCase , cls_token=lowerCamelCase , pad_token=lowerCamelCase , sp_model_kwargs=self.sp_model_kwargs , **lowerCamelCase , )

				a__     = spm.SentencePieceProcessor(**self.sp_model_kwargs		)
				self.sp_model.Load(str(lowerCamelCase		)		)
				a__     = vocab_file

				# Original fairseq vocab and spm vocab must be "aligned":
				# Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
				# -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
				# fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
				# spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

				# The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
				a__     = 1

				# Mimic fairseq token-to-id alignment for the first 4 token
				a__     = {"<s>": 0, "<pad>": 1, "</s>": 2, "<unk>": 3}

				a__     = len(self.sp_model		)
				a__     = {F'''<madeupword{i}>''': sp_size + i + self.fairseq_offset for i in range(self.num_madeup_words		)}
				self.fairseq_tokens_to_ids.update(lowerCamelCase		)

				a__     = {v: k for k, v in self.fairseq_tokens_to_ids.items()}







	def __getstate__(				self			:   Optional[Any]		):




				'''simple docstring'''
				a__     = self.__dict__.copy()
				a__     = None
				a__     = self.sp_model.serialized_model_proto()
				return state







	def __setstate__(				self			:   Dict , lowerCamelCase			:   List[Any]		):




				'''simple docstring'''
				a__     = d

				# for backward compatibility
				if not hasattr(self , "sp_model_kwargs"		):
							a__     = {}

				a__     = spm.SentencePieceProcessor(**self.sp_model_kwargs		)
				self.sp_model.LoadFromSerializedProto(self.sp_model_proto		)







	def 						__a   (				self			:   Union[str, Any] , lowerCamelCase			:   List[int] , lowerCamelCase			:   Optional[List[int]] = None		):




				'''simple docstring'''

				if token_ids_a is None:
							return [self.sep_token_id] + token_ids_a
				a__     = [self.sep_token_id]
				return sep + token_ids_a + sep + sep + token_ids_a







	def 						__a   (				self			:   int , lowerCamelCase			:   List[int] , lowerCamelCase			:   Optional[List[int]] = None , lowerCamelCase			:   bool = False		):




				'''simple docstring'''

				if already_has_special_tokens:
							return super().get_special_tokens_mask(
							    token_ids_a=lowerCamelCase , token_ids_a=lowerCamelCase , already_has_special_tokens=lowerCamelCase		)

				if token_ids_a is None:
							return [1] + ([0] * len(lowerCamelCase		))
				return [1] + ([0] * len(lowerCamelCase		)) + [1, 1] + ([0] * len(lowerCamelCase		))







	def 						__a   (				self			:   Dict , lowerCamelCase			:   List[int] , lowerCamelCase			:   Optional[List[int]] = None		):




				'''simple docstring'''
				a__     = [self.sep_token_id]

				if token_ids_a is None:
							return len(sep + token_ids_a		) * [0]
				return len(sep + token_ids_a + sep + sep + token_ids_a		) * [0]







	@property
	def 						__a   (				self			:   List[str]		):




				'''simple docstring'''
				return len(self.sp_model		) + self.fairseq_offset + self.num_madeup_words







	def 						__a   (				self			:   Dict		):




				'''simple docstring'''
				a__     = {self.convert_ids_to_tokens(lowerCamelCase		): i for i in range(self.vocab_size		)}
				vocab.update(self.added_tokens_encoder		)
				return vocab







	def 						__a   (				self			:   int , lowerCamelCase			:   str		):




				'''simple docstring'''
				return self.sp_model.encode(lowerCamelCase , out_type=lowerCamelCase		)







	def 						__a   (				self			:   List[str] , lowerCamelCase			:   Union[str, Any]		):




				'''simple docstring'''
				if token in self.fairseq_tokens_to_ids:
							return self.fairseq_tokens_to_ids[token]
				a__     = self.sp_model.PieceToId(lowerCamelCase		)

				# Need to return unknown token if the SP model returned 0
				return spm_id + self.fairseq_offset if spm_id else self.unk_token_id







	def 						__a   (				self			:   Any , lowerCamelCase			:   List[str]		):




				'''simple docstring'''
				if index in self.fairseq_ids_to_tokens:
							return self.fairseq_ids_to_tokens[index]
				return self.sp_model.IdToPiece(index - self.fairseq_offset		)







	def 						__a   (				self			:   int , lowerCamelCase			:   Optional[Any]		):




				'''simple docstring'''
				a__     = "".join(lowerCamelCase		).replace(lowerCamelCase , " "		).strip()
				return out_string







	def 						__a   (				self			:   int , lowerCamelCase			:   str , lowerCamelCase			:   Optional[str] = None		):




				'''simple docstring'''
				if not os.path.isdir(lowerCamelCase		):
							logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''		)
							return
				a__     = os.path.join(
				    lowerCamelCase , (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]		)

				if os.path.abspath(self.vocab_file		) != os.path.abspath(lowerCamelCase		) and os.path.isfile(self.vocab_file		):
							copyfile(self.vocab_file , lowerCamelCase		)
				elif not os.path.isfile(self.vocab_file		):
							with open(lowerCamelCase , "wb"		) as fi:
										a__     = self.sp_model.serialized_model_proto()
										fi.write(lowerCamelCase		)

				return (out_vocab_file,)



